Event-driven workflows

Not every workflow starts from a direct start call, and not every wait is for a signal aimed at one specific run. Events are name-based pub/sub layered on top of the same suspend machinery: engine.publishEvent(name, payload) can start a fresh run of any workflow subscribed to that name, and can resume any in-flight run parked on ctx.waitForEvent(name) — in one call, fanning out to as many runs as care about it.

Publishing an event

Call engine.publishEvent(name, payload, opts?) — or WorkflowService.publishEvent, the same method exposed for DI — from anywhere: a controller, another workflow's step, a queue consumer.

@Controller('webhooks')
export class BillingWebhookController {
  constructor(private readonly workflows: WorkflowService) {}

  @Post('payment-settled')
  async onPaymentSettled(@Body() body: { orderId: string; amount: number }) {
    const touched = await this.workflows.publishEvent('payment.settled', body);
    return { touched };
  }
}

publishEvent does two things, and returns the sum of how many runs it touched:

1. Resumes every in-flight run currently parked on ctx.waitForEvent(name, { match }) whose match the payload satisfies — fan-out, unlike a signal's point-to-point token.
2. Starts a fresh run of every workflow registered with onEvent: [name] (via @Workflow({ onEvent }) or @OnEvent), passing payload as that run's input.

Publishing is idempotent per subscriber: pass opts.id to dedupe redeliveries of the same logical event. Each started run gets the id evt:<id>:<workflow>, so redelivering the same id finds the existing run instead of starting a second one. Omit id and the engine generates a fresh uuid per publish, so every call starts its subscribers exactly once.

A subscriber whose validateInput rejects the payload simply never starts — it does not block the other subscribers or the waiters being resumed.

Starting a workflow from an event

Subscribe a @Workflow to one or more event names with onEvent, or the equivalent @OnDurableEvent(...) class decorator — the two are merged, so you can use whichever reads better, or both:

@Workflow({ name: 'send-welcome', version: '1', onEvent: ['user.registered'] })
export class SendWelcomeWorkflow {
  constructor(private readonly mailer: MailerService) {}

  async run(ctx: WorkflowCtx, input: { email: string }) {
    await ctx.step(this.mailer.sendWelcome, input.email);
    return { sent: true };
  }
}

// Equivalent, class-decorator form — merges with any `onEvent` array on the same class.
@OnDurableEvent('user.registered', 'user.invited')
@Workflow({ name: 'audit-signup', version: '1' })
export class AuditSignupWorkflow {
  async run(_ctx: WorkflowCtx, input: { email: string }) {
    /* ... */
  }
}

The next time anyone calls publishEvent('user.registered', { email }), both workflows above start a fresh run with { email } as input. A publish can resume waiters and start subscribers in the same call — the returned count covers both.

Coalescing bursts: debounce and batch

A source that fires many events in quick succession — a stream of file uploads, a chatty webhook — usually shouldn't start one run per event. debounce and batch coalesce an onEvent subscription so the workflow starts once per burst instead:

export interface WorkflowOptions {
  // ...
  /** Start one run only once events have been quiet for this long (resets on each event), with the
   *  LAST payload. */
  debounce?: string | number;
  /** Start one run with ALL payloads (`{ events: [...] }`) once maxSize is reached or `within`
   *  elapses from the first event. */
  batch?: { maxSize: number; within: string | number };
}

Both are built the same way under the hood: instead of starting the target workflow directly, the engine routes each published event into a per-workflow accumulator — a long-lived built-in workflow (__evt_debounce / __evt_batch, registered once at engine construction) that collects payloads via signalWithStart, then startChilds the real target once, and finally continueAsNews to re-arm itself for the next burst.

debounce — "reindex at most once a minute"

@Workflow({
  name: 'reindex-search',
  version: '1',
  onEvent: ['catalog.item-changed'],
  debounce: '1m',
})
export class ReindexSearchWorkflow {
  constructor(private readonly search: SearchService) {}

  async run(ctx: WorkflowCtx, lastChange: { itemId: string }) {
    // Runs once the catalog has been quiet for a minute, with only the LAST change payload.
    await ctx.step(this.search.reindex, lastChange);
  }
}

Every catalog.item-changed publish resets the quiet-window clock; the accumulator only fires ReindexSearchWorkflow once nothing new has arrived for the debounce window ('1m'), and the run it starts receives the last payload published during the burst — earlier ones in the same burst are discarded. Three publishes within the window produce exactly one run, carrying the last payload.

batch — "process uploads in batches"

@Workflow({
  name: 'process-uploads',
  version: '1',
  onEvent: ['upload.completed'],
  batch: { maxSize: 100, within: '10s' },
})
export class ProcessUploadsWorkflow {
  constructor(private readonly ingest: IngestService) {}

  async run(ctx: WorkflowCtx, batch: { events: Array<{ fileId: string }> }) {
    await ctx.step(this.ingest.processBatch, batch.events);
  }
}

The accumulator collects payloads as upload.completed is published and fires ProcessUploadsWorkflow as soon as either condition is met: maxSize events have arrived, or within has elapsed since the first one in the batch — whichever comes first. The target's input is always { events: [...] } — the full array of payloads collected in that batch, in publish order: three hit events with maxSize: 3 produce one run with { events: [{ n: 1 }, { n: 2 }, { n: 3 }] }.

debounce and batch are mutually exclusive per workflow, and both are declared once on the @Workflow (or via the plain WorkflowOptions). Their windows (debounce: '30s', or the within of a batch) accept the same duration strings as ctx.sleep.

Waiting for an event mid-run

Where onEvent starts a new run, ctx.waitForEvent<TPayload>(name, opts?) lets an already running workflow suspend until a matching event arrives, then resume with its payload:

@Workflow({ name: 'order', version: '1' })
export class OrderWorkflow {
  async run(ctx: WorkflowCtx, input: { orderId: string }) {
    await ctx.step(this.orders.markAwaitingPayment, input.orderId);

    // Suspends with zero compute until a `payment.settled` publish whose payload has this orderId.
    const payment = await ctx.waitForEvent<{ amount: number }>('payment.settled', {
      match: { orderId: input.orderId },
      timeoutMs: 24 * 60 * 60 * 1000, // 1 day
    });

    return { orderId: input.orderId, paid: payment.amount };
  }
}

match is an optional subset of the payload that must deep-equal for this waiter to be resumed — engine.publishEvent lists every waiter registered for name, and only signals the ones whose match keys equal the corresponding payload values. Omit match and the waiter is a broadcast: it resumes on any payload published under that name. With match, a publish only reaches the runs it concerns — two OrderWorkflow runs waiting on different orderIds each resume only from the payment.settled publish carrying their own id.

Pass { timeoutMs } to bound the wait — same shape as ctx.waitForSignal: if the deadline passes first, the call throws SignalTimeoutError instead of waiting forever. Omit it to wait indefinitely, consuming zero compute either way. The same determinism cost as any bounded wait applies: an unbounded waitForEvent consumes one logical position, a bounded one consumes two (deadline + wait) — see the note under Timeouts.

Under the hood, waitForEvent is a waitForSignal whose token encodes the event name and the match criteria (base64-encoded so they never collide with the token's : delimiter) — engine.publishEvent finds candidate waiters by listing every token with that name's prefix, then checks each one's decoded match against the payload before signalling it.

Events vs signals vs updates

All three suspend a run with zero compute until something external resumes it, but they differ in addressing and delivery:

An event is a signal with the addressing inverted: a signal's token is chosen by the caller to hit one run, while an event's name is a stable, shared channel that any number of runs can subscribe to (by parking on waitForEvent) or be created by (via onEvent) — match narrows a broadcast down to the runs a given publish actually concerns. Reach for an update instead when the caller needs synchronous, validated rejection of a bad delivery — events and signals both always land.