workflow

Files (2)

• SKILL.md
• options.md · reference/

SKILL.md (first lines)

# @falcondev-oss/workflow

A long-running async process — send an email, sleep three days, send a follow-up — dies with the process that runs it. This library makes it survivable: the workflow body is an ordinary async function, but every `step` it takes is persisted to Redis, so a crash, deploy, or retry **replays** the function from the top and skips straight past the steps that already finished.

Three layers, each minted by the one above:

| Layer | Owns |
|---|---|
| `WorkflowNamespace` | The redis connection, the pub/sub connection, a cross-workflow concurrency cap |
| `WorkflowDefinition` | One queue + one job type: input schema and options. Enqueues, schedules, inspects |
| `Workflow` | A definition plus its `run` handler, so it can also `work()` |
| `WorkflowJob` | One enqueued run: `id`, `groupId`, `wait()` |

Everything is exported from the package root: `WorkflowNamespace`, `createRedis`, `expBackoff`, `TimeoutError`, `ResultExpiredError`, `JobAlreadyExistsError`, `NonRecoverableError`. Always mint workflows through the namespace (`createWorkflow` / `defineWorkflow`), never `new Workflow()`.

## Setting up

```ts
import { createRedis, WorkflowNamespace } from '@falcondev-oss/workflow'

const namespace = new WorkflowNamespace({
  id: 'app',
  redis: await createRedis({ url: process.env.REDIS_URL }),
  logger: console, // opt-in; no logging by default
})
```

The namespace owns all configuration: `redis`, `logger` (`info`/`success`/`error`/`debug`/`warn`, all optional), and the option bags its workflows inherit. Without `redis` it falls back to a process-wide connection to `localhost:6379`. `createRedis` applies the connection options the library needs (`maxRetriesPerRequest: null`, offline queue off, a capped retry strategy), so build connections with it rather than `new IORedis()`.

The namespace registers its own exit hook, so `SIGTERM` drains workers and disconnects. Pass `autoClose: false` and call `namespace.close()` yourself in tests.

## Defining a workflow

```ts
const onboarding = namespace.createWorkflow({
  id: 'onboarding',
  schema: z.object({ userId: z.string(), name: z.string() }),
  getGroupId: (input) => input.userId,
  async run({ input, step }) {
    await step.do('welcome', () => sendWelcome(input.userId))
    await step.wait('settle', 3 * 24 * 60 * 60 * 1000)
    const engaged = await step.do('check engagement', () => isEngaged(input.userId))
    return { engaged }
  },
})
```

- **`id`** scopes every redis key for this workflow. Changing it orphans in-flight jobs.
- **`schema`** is any [Standard Schema](https://standardschema.dev) (zod, arktype, …), and validates twice: at enqueue and again in the worker. A failure throws `Invalid workflow input`. Omit it and `input` is typed `undefined`.
- **`getGroupId`** derives the group from validated input; an explicit `groupId` on the run call wins over it.
- **`run`** receives `{ input, step, span }` and its resolved value becomes the job output. Payload, step results, and output all round-trip through superjson, so `Date`, `Map`, `Set`, and `BigInt` survive intact.
- `queueOptions` / `workerOptions` / `jobOptions` set per-workflow defaults; the same keys on the namespace are shallow-merged underneath them. See [`reference/options.md`](reference/options.md).

## Definitions and handlers

`createWorkflow` puts the handler in the workflow's module, so every module that enqueues it also loads everything the handler imports. When enqueueing code should stay cheap (client modules, import cycles, slow test startup), split it:

```ts
// jobs/onboarding.ts: imported wherever jobs are enqueued
export const onboarding = namespace
  .defineWorkflow({ id: 'onboarding', schema: z.object({ userId: z.string() }) })
  .returns<{ engaged: boolean }>()

// handlers/onboarding.ts: imported only by the worker process
export default onboarding.handle(async ({ input, step }) => ({ engaged: await check(input.userId) }))

// worker.ts
import onboarding from './handlers/onboarding'
await onboarding.work()
```

- `defineWorkflow` takes every `createWorkflow` option except `run`. The definition has every workflow method except `work()`.
- `returns<T>()` declares the output type, so `job.wait()` is typed for enqueueing code. Without it the definition's output is `unknown`.
- `handle(run)` returns a full `Workflow`. Its handler must return the declared type; without `returns<T>()` the output is inferred from the handler.
- `handle()` and `returns()` are type errors on a workflow that already has a handler. The library never `import()`s handlers, so the worker entry must import every handler module it serves.


```ts
await onboarding.work() // start a worker in this process

const job = await onboarding.run({ userId: 'u1', name: 'John' })
const { engaged } = await job.wait() // typed as the run() return
```

`run(input, opts?)` enqueues now, `runIn(input, delayMs, opts?)` after a delay, `runAt(input, date, opts?)` at a timestamp. All three resolve to a `WorkflowJob` as soon as it is enqueued — they do not wait for it to execute. `opts` accepts `groupId`, `priority`, `maxAttempts`, and `jobId`.

`job.wait(timeoutMs?)` blocks on a pub/sub notification and returns the deserialized output. It works from a pure producer with no worker in the process. It rejects with:

| Error | When |
|---|---|
| the workflow's own `Error` | `run` threw and exhausted its attempts — the original message is preserved |
| `TimeoutError` | `timeoutMs` elapsed first (no default; waits forever without it) |
| `ResultExpiredError` | the job finished but its result record aged out of `resultTtl` (default 300 s) |

Passing an explicit `jobId` that is already in use rejects with `JobAlreadyExistsError` — the deduplication handle for "enqueue this at most once".

## Steps and replay

`step.do(name, run)` is the unit of durability. It memoizes **on completion**: once it returns, its result is written to the job's step hash, and any later attempt returns the cached value without re-running the body. So a retry replays completed steps instantly and re-executes only the step that failed.

```ts
const user = await step.do('load user', () => db.user.find(input.userId))
await step.do('charge', async ({ step, span }) => {
  span.setAttribute('user.plan', user.plan)
  await step.do('authorize', () => gateway.authorize(user)) // nested → 'charge|authorize'
  await step.do('capture', () => gateway.capture(user))
})
```

The callback receives `{ step, span }`: a nested `step` whose names are prefixed with the parent's (`parent|child`), and the OTel span for the step. Step names are the memoization key — **keep them stable and unique within a job**, and treat renaming a step as invalidating its cached result.

`step.wait(name, durationMs)` is a durable sleep: it persists `startedAt` on first entry, so a replay sleeps only the *remaining* time rather than the full duration again. `step.waitUntil(name, date)` is the same thing against an absolute time.

Three properties worth holding onto:

- **No step-level retry.** `step.do` takes no options bag; a throw fails the whole job, and retries are job-level (`maxAttempts`).
- **Cancellation is cooperative.** A worker that loses its claim (lock stolen, heartbeat lost) aborts the job's signal: `step.wait` rejects immediately, and `step.do` refuses to *start* new work — but a step already running is not interrupted. Long CPU-bound steps hold their slot until they return.
- **`Promise.all` over steps does not cancel siblings.** When one parallel step throws, the others keep running to completion; the job failure waits for them (and logs a warning). Sequence steps that must not outlive a failure.

From falcondev-oss/skills, MIT.

More from falcondev-oss/skills