feat: recurring workflow schedules

AGENTS.md

@@ -67,17 +67,57 @@ import { z } from 'zod';
 
 const myWorkflow = workflow(
   'workflow-id',                              // unique string ID
-  async ({ step, input, runId, workflowId, timeline, logger }) => {
+  async ({ step, input, runId, workflowId, timeline, logger, schedule }) => {
     // workflow body with step calls
+    // `schedule` is populated only for runs triggered by a recurring schedule
   },
   {
     inputSchema: z.object({ /* ... */ }),      // optional Zod schema
     timeout: 60000,                            // optional, milliseconds
     retries: 3,                                // optional, max retry count
+    schedule: '*/5 * * * *',                   // optional, recurring schedule (cron or duration)
+    timezone: 'America/New_York',              // optional, only meaningful for cron (default: UTC)
   }
 );
 ```
 
+### Recurring workflows
+
+Workflows can run on a recurring schedule. `schedule` accepts three forms — a
+cron expression, a duration string, or a `DurationObject`:
+
+```typescript
+workflow('cron-style',   handler, { schedule: '0 9 * * 1-5', timezone: 'America/New_York' });
+workflow('every-5-min',  handler, { schedule: '5m' });          // duration string
+workflow('every-hour',   handler, { schedule: '1 hour' });      // natural-language duration
+workflow('every-day',    handler, { schedule: { days: 1 } });   // DurationObject
+```
+
+**Detection rule.** A string that splits into 5 or 6 whitespace tokens of
+cron-charset (`0-9 * / , - ? L W #`) is treated as a cron expression and
+validated by `cron-parser`. Otherwise it's parsed as a duration via
+`parse-duration` and translated into a cron expression — but only if the
+interval divides cleanly (whole minutes that divide 60, whole hours that divide
+24, or 1 day). Non-divisible intervals (`'23m'`, `'7h'`) throw with a clear
+message; use an explicit cron expression for those.
+
+**Schedule context.** Schedule-triggered runs receive `ctx.schedule.timestamp`
+— the time the schedule fired. Manual runs from `engine.startWorkflow()` have
+`ctx.schedule === undefined`. Use that as the "this is a scheduled fire" flag.
+
+```typescript
+async ({ step, schedule, workflowId }) => {
+  // For cursor-style incremental syncs, fetch the previous run separately:
+  const lastRun = await engine.getWorkflowLastRun({ workflowId });
+  const since = lastRun?.completedAt ?? new Date(0);
+  // ... fetch data updated since `since`
+}
+```
+
+**Overlap policy.** Scheduled runs are singletons via pg-boss — if a fire is
+queued while the previous run is still executing, pg-boss handles it
+(configurable overlap policies may be added later).
+
 ### `WorkflowEngine` - Main orchestrator
 
 ```typescript
@@ -190,6 +230,7 @@ await engine.triggerEvent({
 
 // Query runs
 const run = await engine.getRun({ runId, resourceId });
+const lastRun = await engine.getWorkflowLastRun({ workflowId, resourceId }); // null if none
 const progress = await engine.checkProgress({ runId, resourceId });
 const { items, nextCursor, hasMore } = await engine.getRuns({
   resourceId: 'user-123',

examples/cron.ts

@@ -0,0 +1,64 @@
+import { WorkflowEngine, workflow } from '../src';
+
+// A recurring workflow.
+//
+// `schedule` accepts:
+//   - a cron expression: '0 9 * * 1-5' (weekdays at 9am)
+//   - a duration string: '5m', '1 hour', '1 day'
+//   - a DurationObject:  { minutes: 5 }
+//
+// `timezone` is optional and only meaningful for cron expressions (UTC by default).
+// `ctx.schedule.timestamp` is the time this fire was scheduled — present only on
+// schedule-triggered runs. Use `engine.getWorkflowLastRun(...)` to fetch the
+// previous run when you need a cursor for incremental syncs.
+
+const syncOrders = workflow(
+  'sync-orders',
+  async ({ step, schedule, workflowId, logger }) => {
+    logger.log(
+      schedule
+        ? `Cron fire at ${schedule.timestamp.toISOString()}`
+        : 'Manual run (no schedule context)',
+    );
+
+    const lastRun = await engine.getWorkflowLastRun({ workflowId });
+    const since = lastRun?.completedAt ?? new Date(0);
+    logger.log(`Syncing orders changed since ${since.toISOString()}`);
+
+    const orders = await step.run('fetch-new-orders', async () => {
+      return [
+        { id: 'ord_1', total: 99.0 },
+        { id: 'ord_2', total: 149.5 },
+      ];
+    });
+
+    await step.run('write-to-warehouse', async () => ({ written: orders.length }));
+
+    return { synced: orders.length, since: since.toISOString() };
+  },
+  {
+    schedule: '5m',
+    retries: 3,
+  },
+);
+
+const engine = new WorkflowEngine({
+  connectionString: process.env.DATABASE_URL ?? 'postgres://localhost:5432/pg_workflows_example',
+  workflows: [syncOrders],
+});
+
+async function main() {
+  await engine.start();
+  console.warn('Schedule registered. Waiting for triggers (Ctrl+C to stop)...');
+
+  process.on('SIGINT', async () => {
+    console.warn('Shutting down...');
+    await engine.stop();
+    process.exit(0);
+  });
+}
+
+main().catch((err) => {
+  console.error('Example failed:', err);
+  process.exit(1);
+});

examples/package.json

@@ -11,7 +11,8 @@
     "example:timeout": "npm run build:lib && dotenvx run -- tsx timeout.ts",
     "example:polling": "npm run build:lib && dotenvx run -- tsx polling.ts",
     "example:microservices:worker": "npm run build:lib && dotenvx run -- tsx microservices/worker-service.ts",
-    "example:microservices:api": "npm run build:lib && dotenvx run -- tsx microservices/api-service.ts"
+    "example:microservices:api": "npm run build:lib && dotenvx run -- tsx microservices/api-service.ts",
+    "example:cron": "npm run build:lib && dotenvx run -- tsx cron.ts"
   },
   "dependencies": {
     "pg": "^8.13.1",

package.json

@@ -78,6 +78,7 @@
   },
   "dependencies": {
     "@standard-schema/spec": "^1.1.0",
+    "cron-parser": "^5.5.0",
     "es-toolkit": "^1.44.0",
     "ksuid": "^3.0.0",
     "parse-duration": "^2.1.5",

src/constants.ts

@@ -1,6 +1,11 @@
 export const PAUSE_EVENT_NAME = '__internal_pause';
 export const WORKFLOW_RUN_QUEUE_NAME = 'workflow-run';
 export const WORKFLOW_RUN_DLQ_QUEUE_NAME = 'workflow_run_dlq';
+// pg-boss queue names allow only alphanumeric, _, -, ., or / — keep the
+// prefix in that character set so any valid workflow id stays addressable.
+const SCHEDULE_QUEUE_PREFIX = '__pgw_schedule_';
+export const scheduleQueueNameFor = (workflowId: string): string =>
+  `${SCHEDULE_QUEUE_PREFIX}${workflowId}`;
 export const DEFAULT_PGBOSS_SCHEMA = 'pgboss_v12_pgworkflow';
 export const MAX_WORKFLOW_ID_LENGTH = 256;
 export const MAX_RESOURCE_ID_LENGTH = 256;

src/db/migration.ts

@@ -5,7 +5,7 @@ export const MIGRATION_LOCK_ID = 738291645;
 
 // Bump this when adding new migrations. The engine stores the current version
 // in a `workflow_schema_version` table so migrations only run once per version.
-const CURRENT_SCHEMA_VERSION = 4;
+const CURRENT_SCHEMA_VERSION = 5;
 
 export async function runMigrations(db: Db): Promise<void> {
   // Fast path: skip the advisory lock if schema is already current.
@@ -87,6 +87,12 @@ export async function runMigrations(db: Db): Promise<void> {
     );
   }
 
+  if (currentVersion < 5) {
+    commands.push(
+      'ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS scheduled_at timestamp with time zone',
+    );
+  }
+
   // Upsert the schema version
   if (currentVersion === 0) {
     commands.push(

src/db/queries.ts

@@ -29,6 +29,7 @@ type WorkflowRunRow = {
   parent_run_id: string | null;
   parent_step_id: string | null;
   parent_resource_id: string | null;
+  scheduled_at: string | Date | null;
 };
 
 function mapRowToWorkflowRun(row: WorkflowRunRow): WorkflowRun {
@@ -60,6 +61,7 @@ function mapRowToWorkflowRun(row: WorkflowRunRow): WorkflowRun {
     parentRunId: row.parent_run_id,
     parentStepId: row.parent_step_id,
     parentResourceId: row.parent_resource_id,
+    scheduledAt: row.scheduled_at ? new Date(row.scheduled_at) : null,
   };
 }
 
@@ -76,6 +78,7 @@ export async function insertWorkflowRun(
     parentRunId,
     parentStepId,
     parentResourceId,
+    scheduledAt,
   }: {
     resourceId?: string;
     workflowId: string;
@@ -88,6 +91,7 @@ export async function insertWorkflowRun(
     parentRunId?: string;
     parentStepId?: string;
     parentResourceId?: string;
+    scheduledAt?: Date;
   },
   db: Db,
 ): Promise<{ run: WorkflowRun; created: boolean }> {
@@ -111,9 +115,10 @@ export async function insertWorkflowRun(
       idempotency_key,
       parent_run_id,
       parent_step_id,
-      parent_resource_id
+      parent_resource_id,
+      scheduled_at
     )
-    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16)
+    VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15, $16, $17)
     ON CONFLICT (idempotency_key) WHERE idempotency_key IS NOT NULL DO NOTHING
     RETURNING *`,
     [
@@ -133,6 +138,7 @@ export async function insertWorkflowRun(
       parentRunId ?? null,
       parentStepId ?? null,
       parentResourceId ?? null,
+      scheduledAt ?? null,
     ],
   );
 
@@ -187,6 +193,39 @@ export async function getWorkflowRun(
   return mapRowToWorkflowRun(run);
 }
 
+export async function getWorkflowLastRun(
+  {
+    workflowId,
+    resourceId,
+  }: {
+    workflowId: string;
+    resourceId?: string;
+  },
+  db: Db,
+): Promise<WorkflowRun | null> {
+  const result = resourceId
+    ? await db.executeSql(
+        `SELECT * FROM workflow_runs
+         WHERE workflow_id = $1 AND resource_id = $2
+         ORDER BY created_at DESC
+         LIMIT 1`,
+        [workflowId, resourceId],
+      )
+    : await db.executeSql(
+        `SELECT * FROM workflow_runs
+         WHERE workflow_id = $1
+         ORDER BY created_at DESC
+         LIMIT 1`,
+        [workflowId],
+      );
+
+  const run = result.rows[0];
+  if (!run) {
+    return null;
+  }
+  return mapRowToWorkflowRun(run);
+}
+
 export async function updateWorkflowRun(
   {
     runId,

src/db/types.ts

@@ -21,4 +21,6 @@ export type WorkflowRun = {
   parentRunId: string | null;
   parentStepId: string | null;
   parentResourceId: string | null;
+  /** Set when the run was started by a recurring schedule; the timestamp the schedule fired. */
+  scheduledAt: Date | null;
 };

src/definition.ts

@@ -28,6 +28,8 @@ export function createWorkflowRef<
     inputSchema: options?.inputSchema,
     timeout: defineOptions?.timeout,
     retries: defineOptions?.retries,
+    schedule: defineOptions?.schedule,
+    timezone: defineOptions?.timezone,
   })) as WorkflowRef<TInput, TOutput>;
 
   Object.defineProperty(ref, 'id', { value: id, enumerable: true });
@@ -45,7 +47,7 @@ function createWorkflowFactory<TStepExt extends object = object>(
   const factory = (<I extends InputParameters>(
     id: string,
     handler: (context: WorkflowContext<I, StepBaseContext & TStepExt>) => Promise<unknown>,
-    { inputSchema, timeout, retries }: WorkflowOptions<I> = {},
+    { inputSchema, timeout, retries, schedule, timezone }: WorkflowOptions<I> = {},
   ): WorkflowDefinition<I> => ({
     id,
     handler: handler as (
@@ -54,6 +56,8 @@ function createWorkflowFactory<TStepExt extends object = object>(
     inputSchema,
     timeout,
     retries,
+    schedule,
+    timezone,
     plugins: plugins.length > 0 ? (plugins as WorkflowPlugin[]) : undefined,
   })) as WorkflowFactory<TStepExt>;
 

src/duration.ts

@@ -11,11 +11,11 @@ export type DurationObject = {
 
 export type Duration = string | DurationObject;
 
-const MS_PER_SECOND = 1000;
-const MS_PER_MINUTE = 60 * MS_PER_SECOND;
-const MS_PER_HOUR = 60 * MS_PER_MINUTE;
-const MS_PER_DAY = 24 * MS_PER_HOUR;
-const MS_PER_WEEK = 7 * MS_PER_DAY;
+const MS_PER_SECOND: number = 1000;
+export const MS_PER_MINUTE: number = 60 * MS_PER_SECOND;
+export const MS_PER_HOUR: number = 60 * MS_PER_MINUTE;
+export const MS_PER_DAY: number = 24 * MS_PER_HOUR;
+const MS_PER_WEEK: number = 7 * MS_PER_DAY;
 
 export function parseDuration(duration: Duration): number {
   if (typeof duration === 'string') {