#Workflows

$workflow is for a business process that has more than one step and must survive the process dying halfway through. Steps run in order, each step's result is written to the database before the next one starts, and a failure either unwinds the completed steps or stops the execution where it stands.

1import { $workflow } from "alepha/api/workflows";

It builds on the job outbox and a distributed lock, so it is registered as a module:

1import { AlephaApiWorkflows } from "alepha/api/workflows";2 3alepha.with(AlephaApiWorkflows);

#When this rather than $job

A $job is one unit of work: push it, it runs, it retries, it is done. Reach for $workflow when the work has joints - places where you want to know that step three succeeded and step four has not started yet.

That distinction pays for itself in three situations:

• A crash must not redo the whole thing. After a restart, an execution resumes at the step that was pending, not at the beginning. Charging a card twice because the email server was slow is the failure this prevents.
• A late failure must undo earlier success. Reserve stock, charge the card, book the courier: if the courier refuses, the charge has to come back. That is the saga pattern, and it is the default.
• The process waits. "Send a reminder 24 hours later" is not a step that sleeps, it is a step that is parked in the database with a due date.

If none of those apply, a $job is the smaller tool and the right one.

#Declaring one

 1import { z } from "alepha"; 2import { $workflow } from "alepha/api/workflows"; 3  4class Orders { 5  fulfil = $workflow({ 6    schema: z.object({ orderId: z.uuid() }), 7    steps: [ 8      { 9        name: "reserveStock",10        handler: async ({ payload }) => {11          return { reservationId: `res-${payload.orderId}` };12        },13      },14      {15        name: "charge",16        handler: async ({ results }) => {17          const stock = results.reserveStock as { reservationId: string };18          return { chargeId: `ch-${stock.reservationId}` };19        },20      },21    ],22  });23}

schema types the payload. Each step returns a value, and that value is persisted and handed to every later step in results, keyed by step name. A step that returns nothing still records that it ran.

Start it and you get an execution id back:

1const executionId = await this.fulfil.start({ orderId });

start() returns as soon as the execution row exists. The steps run through the job outbox, not inline, so the caller is not holding a request open while a courier API decides.

#Every handler must be idempotent

This is the one rule the engine cannot enforce for you. Crash recovery replays the last unacknowledged unit of work, which means a step that crashed after its side effect but before its result was written will run again. So will a when() guard, and so will a compensation handler.

In practice: pass an idempotency key to whatever you call, or check whether the effect already happened before doing it.

#Failure: compensate or fail

The default is onError: "compensate". When a step throws and its retries are exhausted, every step that already completed runs its compensate handler, in reverse order:

 1import { z } from "alepha"; 2import { $workflow } from "alepha/api/workflows"; 3  4class Orders { 5  fulfil = $workflow({ 6    schema: z.object({ orderId: z.uuid() }), 7    onError: "compensate", 8    steps: [ 9      {10        name: "charge",11        handler: async ({ payload }) => ({ chargeId: payload.orderId }),12        compensate: async ({ result }) => {13          const charge = result as { chargeId: string };14          await Promise.resolve(charge.chargeId);15        },16      },17      {18        name: "book",19        handler: async () => {20          throw new Error("courier refused");21        },22      },23    ],24  });25}

compensate receives that step's own result, so it knows what to undo without looking it up. It does not run for the step that failed: that step did not complete, so there is nothing of its own to reverse.

onError: "fail" skips all of it and marks the execution failed. Use it when the steps are independent enough that a half-finished run is a state a human should look at rather than one the machine should unwind.

#Retries, timeouts and conditions

Per step:

 1import { z } from "alepha"; 2import { $workflow } from "alepha/api/workflows"; 3  4class Orders { 5  fulfil = $workflow({ 6    schema: z.object({ orderId: z.uuid(), express: z.boolean() }), 7    timeout: [1, "hour"], 8    steps: [ 9      {10        name: "book",11        timeout: [30, "seconds"],12        retry: {13          retries: 5,14          backoff: { initial: [2, "seconds"], factor: 2, jitter: true },15          when: (error) => !error.message.includes("invalid address"),16        },17        when: ({ payload }) => payload.express,18        handler: async () => ({ booked: true }),19      },20    ],21  });22}

Retries go through the job outbox, so a retry scheduled two seconds before the process died still fires after it comes back.

#Waiting: delay and repeat

delay parks a step until a moment in the future, counted from when the previous step finished:

1{2  name: "reminder",3  delay: [24, "hours"],4  handler: async ({ payload }) => this.email.send(payload.orderId),5}

The wait is persisted as the step's scheduledAt before any timer is armed. The timer only improves latency; the recovery sweep re-dispatches anything due from the database alone, so a 24 hour wait survives every deploy in between.

repeat makes a step run itself again. The handler resolving with { repeat: true } re-parks the same step after delay; any other resolution is the step's final result and the workflow moves on:

 1import { z } from "alepha"; 2import { $workflow } from "alepha/api/workflows"; 3  4class Shifts { 5  offer = $workflow({ 6    schema: z.object({ shiftId: z.uuid(), candidates: z.array(z.uuid()) }), 7    steps: [ 8      { 9        name: "offerToNextCandidate",10        repeat: { delay: [10, "minutes"], limit: 20 },11        handler: async ({ payload, context }) => {12          const candidate = payload.candidates[context.iteration];13          if (!candidate) {14            return { filled: false };15          }16          const accepted = await Promise.resolve(false);17          return accepted ? { filled: true, candidate } : { repeat: true };18        },19      },20    ],21  });22}

context.iteration is the zero-based round counter, and it is 0 for every step that does not repeat. limit caps the total number of runs; a handler still asking to repeat after the last one fails the step, and normal onError handling takes over.

The retry budget resets on each iteration, which is what you want (a transient failure in round 7 should not be counted against round 8) and is also why iteration handlers have to be idempotent: a crash between the verdict and the re-park replays the same round.

#Deduplication and fan-out

key makes a start idempotent: while an execution is live under that key, a second start() with the same key returns the existing execution instead of a new one.

1await this.fulfil.start({ orderId }, { key: `order:${orderId}` });

The uniqueness only spans live statuses, so once an execution reaches a terminal state its key is free again. That is deliberate (an order can be fulfilled twice over its lifetime) and it is why you should not treat a key as a permanent handle: look executions up by key or payload.

startEach is the same idea over a list:

1await this.fulfil.startEach(orders, (order) => ({2  payload: { orderId: order.id },3  key: `order:${order.id}`,4}));

Each item gets its own execution, its own retry budget, its own logs and its own admin row. The trade against one step looping over the whole list is more rows in exchange for per-item granularity, and re-drivability: run it again after a partial failure and only the missing items start.

#Controlling a running execution

cancelByKey is what a "the customer replied, stop the reminder cascade" listener wants: it does not need to have kept the execution id anywhere.

#Carrying context across steps

Steps run on whatever process the outbox hands them to, which is not necessarily the one that called start(). Anything held in an async-local context is therefore gone by the time a step runs.

context names the atoms to capture at start and restore around every step, when() and compensation handler:

1fulfil = $workflow({2  schema: z.object({ orderId: z.uuid() }),3  context: [currentTenantAtom],4  steps: [/* ... */],5});

Tenancy is the canonical case: each step runs under the tenant that started the execution, with no tenant id hand-carried through the payload and no chance of a recovery-sweep dispatch running against the wrong one. The values are persisted with the execution, so they must survive a JSON round trip.

#Observing

The module declares a hook for every transition: workflow:started, workflow:step:begin, workflow:step:completed, workflow:step:failed, workflow:step:skipped, workflow:step:repeat, workflow:completed, workflow:failed, workflow:compensating, workflow:compensated, workflow:compensation:failed, workflow:cancelled and workflow:timed_out.

1alepha.events.on(2  "workflow:step:failed",3  ({ workflowName, stepName, error }) => {4    // ship it to your error tracker5  },6);

AdminWorkflowController exposes executions, their steps and their captured logs. One caveat: admin action names are global across the app, so a second controller exporting an action called getExecution collides at boot rather than at typecheck.

#Testing

WorkflowTestKit exists because testing a durable engine with a frozen clock has two traps in it, and both produce a hanging test rather than a failing one.

1const kit = alepha.inject(WorkflowTestKit);2 3const executionId = await app.fulfil.start({ orderId });4await kit.awaitParked(executionId, "reminder");5await dateTime.travel([25, "hours"]);6await kit.awaitStatus(executionId, "completed");

• Park before you travel. awaitParked waits for the next step to be pending with its scheduledAt stamp written. Travelling before that stamp exists means the timer was never born and nothing will ever fire it.
• Nudge the sweep afterwards. Once the clock is frozen no cron ticks again on its own, so polling for a status has to drive the recovery sweep. settle and awaitStatus do that for you.

findByPayload finds executions that were started without a key.

#See also

• Background Jobs is the outbox underneath, and the right tool for single-step work
• Unit Tests for travel() and the container lifecycle in tests