feat: recurring workflow schedules

[marcelom97]

Summary

Adds a schedule option to workflow() for recurring runs. Accepts:

• a cron expression: '0 9 * * 1-5'
• a parse-duration string: '5m', '1 hour', '1 day'
• a DurationObject: { minutes: 5 }, { hours: 6 }

Detection is automatic — 5–6 cron-charset tokens go through cron-parser; anything else is parsed as a duration and translated into a cron expression (only when the interval divides cleanly into 60 minutes, 24 hours, or 1 day; non-divisible intervals throw with a clear message).

Usage

import { workflow, WorkflowEngine } from 'pg-workflows';

const sync = workflow(
  'sync-orders',
  async ({ step, schedule, workflowId }) => {
    // ctx.schedule is only defined for schedule-triggered runs
    if (schedule) {
      console.log(`Fire scheduled for: ${schedule.timestamp.toISOString()}`);
    }

    // For incremental syncs, fetch the previous run as a cursor
    const lastRun = await engine.getWorkflowLastRun({ workflowId });
    const since = lastRun?.completedAt ?? new Date(0);

    await step.run('fetch', async () => fetchSince(since));
  },
  {
    schedule: '5m',                             // duration → '*/5 * * * *'
    // schedule: '0 9 * * 1-5',                 // explicit cron
    // timezone: 'America/New_York',            // optional, cron-only (default: UTC)
  }
);

const engine = new WorkflowEngine({ connectionString: '...', workflows: [sync] });
await engine.start();

Design

• Single option, not two. schedule does the work of both a cron and an every option — one thing to learn.
• Flat timezone sibling, not nested. Timezone is only meaningful for cron, but its position next to schedule is unambiguous.
• Minimal ctx.schedule. Only timestamp (the dynamic per-fire value). Timezone is static config the author already wrote; previous-run state is derived, not echoed back into context.
• No trigger_source column. A single scheduled_for timestamp with time zone column on workflow_runs is set when the run came from a schedule; existence acts as the flag.
• History via a generic helper. engine.getWorkflowLastRun({ workflowId, resourceId? }) returns the latest run for a workflow — useful beyond cron (debugging, manual cursoring).
• Singleton overlap policy. Scheduled runs don't overlap (pg-boss schedule() semantics). Configurable overlap: 'skip' | 'queue' | 'allow' is deferred to a follow-up.

Migration

Schema v5 adds:

ALTER TABLE workflow_runs ADD COLUMN IF NOT EXISTS scheduled_for timestamp with time zone;

Runs automatically on engine.start().

Test plan

• Unit (14 tests, src/schedule.test.ts) — schedule resolver: cron passthrough, timezone propagation, duration translation for clean divisors, errors for non-divisible intervals ('23m', '7h', '2d'), errors for sub-minute durations, empty string, invalid cron expressions, DurationObject form.
• Unit (5 tests, src/engine.test.ts) — registerWorkflow schedule validation: cron string + timezone, duration string, DurationObject, rejects invalid cron, rejects non-divisible duration.
• Unit (3 tests, src/engine.test.ts) — getWorkflowLastRun: null when no runs, returns latest, scopes by resourceId.
• All 161 unit tests pass; lint clean; build green.

[SokratisVidros]

@marcelom97 thanks for submitting this. Can you please elaborate on the incremental sync?

Other than that, I need to think about the DX a bit more. I will start by adding inline comments.

[SokratisVidros]

How about using a cron: string column and a timezone, both being optional to avoid a nesting object?

[SokratisVidros]

Instead of a specific getLastCronCompletedAt method, I'd suggest introducing a getWorkflowLastRun that works for all workflows. This feels more generic and aligns with the naming of the rest of the methods.

[SokratisVidros]

Suggested change

	private async setupCronSchedule(wf: InternalWorkflowDefinition): Promise<void> {
	private async scheduleCronWorkflow(wf: InternalWorkflowDefinition): Promise<void> {

[SokratisVidros]

Please refer to my previous comment about modeling scheduleContext.

[SokratisVidros]

Following on the modeling, lastTimestamp comes from the workflow_runs table. We can replace it with the timestamp of the latest record.

[SokratisVidros]

Do we need triggerSource? If not let's remove it.

[SokratisVidros]

Let's create the following DX:

1. The cron expression can be a valid cron string or a human friendly cron string such as https://github.com/rainder/human-to-cron. Note, this library is quite old. Let's see if there is a modern one.

2. Cron can be either a string or an object of the expression and the timezone. If the timezone is not specified, we assume UTC or the current timezone of the running container.

[marcelom97]

I looked into this — human-to-cron has critical bugs (e.g. "every hour" produces * */1 * * * which fires every minute, and unrecognized input silently becomes * * * * *). I couldn't find a modern,
well-maintained alternative either.

I'd suggest we defer this and stick with standard cron expressions for now. We can revisit if a reliable library comes along, or build a small parser ourselves for a limited set of human-friendly strings (e.g.
"every 5 minutes", "daily at 9am").

[SokratisVidros]

@marcelom97 Any updates on this?

[marcelom97]

Hey @SokratisVidros, thanks for the review and for your suggestions.

I will take care of this over the weekend.

[marcelom97]

@SokratisVidros incremental sync is about only processing data that changed since the last cron run, instead of re-processing everything each time.

The schedule context gives cron workflows a lastTimestamp (derived from the previous completed run's completedAt), so the workflow can use it as a cursor:

const sync = workflow('sync-data', async ({ step, schedule }) => {
const since = schedule?.lastTimestamp ?? new Date(0);
const data = await step.run('fetch', async () => fetchSince(since));
await step.run('write', async () => writeToDB(data));
}, { cron: '*/15 * * * *' });

Without this, each cron run would need to either re-process all data or manually track its own high-water mark somewhere. lastTimestamp makes that built-in.

If you think it's not useful we can remove it, but I find it quite useful to have it.

[SokratisVidros]

@marcelom97 Thanks for the updates. I will review them by the end of this week and get back to you.

[marcelom97]

@SokratisVidros any news for this PR?

[SokratisVidros]

@marcelom97 I revised the work on this PR and updated the DX a little bit. Please take a look and let me know if it addresses your usecases.