ElementsFlow

Consumers

Asynchronous Flows for Signal emissions, named Clock ticks, and SQL row changes.

Consumers are Flows that run when something else happens — a Signal emit, a named Clock tick, or a SQL row change — instead of waiting for an HTTP request.

For developers wiring background work on okengine — bind the trigger, keep do on fx.

The one rule

Bind with on(signal), on(clockDecl), or on(db.table(…).changed()). Delivery physics live on the Signal; cron / every live on the Clock; CDC input is { before, after } plus table / action / id. World access still goes through fx.

Smallest Example

Bind and emit are independent

The consumer and the producer share one Signal handle. They can live in different files, written in any order — emit is not "step 2" after bind.

Bind a Signal consumer

src/flows/notifications/welcome.ts
import { on, flow } from "okengine";
import { userSignedUp } from "@/signals";
import { welcomeEmail } from "@/channels/welcome";

export const sendWelcome = on(
  userSignedUp,
  flow("notifications.welcome", {
    do: async ({ userId, email }, fx) => {
      await fx.send(welcomeEmail, {
        to: email,
        data: { userId },
      });
    },
  }),
);

Emit from any Flow

await fx.emit(userSignedUp, { userId: "usr_123", email: "alice@example.com" });

The compiler records emits: ["users.signed-up"] on the producer. The consumer runs after the emit commits — the HTTP request does not wait for the welcome mail.

Jobs are consumers

A named Clock bound with on(clockDecl, flow) is the same species — an asynchronous Flow. There is no separate job runner. See Clock jobs.

Progressive Patterns

Explore consumers from a typed queue worker to a cron job and a table-change handler:

Declare delivery physics on the Signal, then bind the worker with on(handle, flow):

src/signals/email.ts
import { signal } from "okengine";
import { z } from "zod";

export const emailTask = signal.once("tasks.email", {
  schema: z.object({ to: z.string().email(), body: z.string() }),
  retries: 3,
  deadLetter: true,
});
src/flows/workers/email.ts
import { on, flow } from "okengine";
import { emailTask } from "@/signals/email";
import { rawEmail } from "@/channels/email";

export const processEmail = on(
  emailTask,
  flow("workers.email", {
    do: async ({ to, body }, fx) => {
      await fx.send(rawEmail, { to, body });
    },
  }),
);

Trigger Reference

TriggerSignaturePurposedo input
Signalon(handle, flow("name", { do }))Queue (once) or fan-out (broadcast)Payload (schema)
Clockon(clockDecl, flow("name", { do })) or inline clock.every(…)Interval or cron ticknone (_)
CDC anyon(db.table(t).changed(), flow)Every insert / update / delete{ before, after, table, action, id }
CDC columnon(db.table(t).changed("col"), flow)Same writes; column stamped on the Manifest{ before, after, table, action, id }

signal.live is an HTTP SSE tape — bind it with http.live, not as a worker. A Flow with no trigger is call-only.

Signal is always an exported const (fx.emit needs the handle). Clock may write clock.every(…) inside on() — Clock · Inline or named export. OKE1072 if nameless.

Signal Consumers

Detailed section

If you only need a worker, jump to Once below. Physics live on signal.once / broadcast / live — same idea as http.get / http.post. Defaults: retries: 3, deadLetter: true, optional: false.

Each emit is handled according to the Signal helper you declared. The Flow is the subscriber.

Competing workers — exactly one consumer claims each message. Failed attempts retry, then dead-letter when deadLetter is true (default).

Two different Flows on one once signal fail OKE1071 — see Once · Competing consumers. For every bound Flow to run, use signal.broadcast.

src/signals/orders.ts
import { signal } from "okengine";
import { z } from "zod";

export const orderPlaced = signal.once("orders.placed", {
  schema: z.object({
    orderId: z.string(),
    amount: z.number(),
    userId: z.string(),
  }),
  retries: 5,
  deadLetter: true,
});
src/flows/orders/fulfill.ts
import { on, flow } from "okengine";
import { orderPlaced } from "@/signals/orders";

export const fulfill = on(
  orderPlaced,
  flow("orders.fulfill", {
    do: async ({ orderId }, fx) => {
      await fx.call(chargeAndShip, { orderId });
    },
  }),
);

Visibility lease defaults to 30s. An inflight worker that dies is reclaimed on the next claim.

Clock Jobs

Detailed section

Prefer named helpers (clock.every / daily / cron). Bind with on(clockDecl, flow("name", { do })), or write clock.every inside on() — Clock · Inline or named export.

Named clocks reconcile into the Store at boot. The scheduler leader-elects so N instances do not double-fire. do receives no payload — read time through fx.clock.

Human durations: "200ms" · "30s" · "5m" · "1h" · "7d" (integer + unit, no weeks):

src/clocks/health.ts
import { clock } from "okengine";

export const pingClock = clock.every("health.pingExternal", "30s");
src/flows/health/ping.ts
import { on, flow } from "okengine";
import { pingClock } from "@/clocks/health";

export const pingExternal = on(
  pingClock,
  flow("health.pingExternal", {
    plane: "operator",
    do: async (_, fx) => {
      await fx.call(pingUpstream);
    },
  }),
);

CDC

Detailed section

If you only need any-write, jump to Bare or enriched below. The handle is db.table(table).changed(column?) — db is a store.sql declaration, table is a schema handle. changed("insert") is not an op filter; it stamps a column named insert.

SQL writes through fx.store notify CDC after commit. The Flow input is always { before, after, table, action, id } (CdcPayload).

Bare or enriched

StyleWhen
({ before, after })Bound to one table — images are enough (search reindex, listing cache)
({ table, action, id })Log, route, or branch — kind of change and which record (audit log)

Both styles receive the same object. There is no second dispatch path, no performance difference, and no correctness difference — the choice is which fields this handler destructures.

{ table, action, id } are always populated; omitting them from do does not drop them from the payload.

Bound to notes — the table is already in the trigger. Images decide upsert vs drop; the row's id is on the surviving image:

src/flows/search/reindex.ts
import { on, flow } from "okengine";
import { db } from "@/core";
import { notes } from "@/schema";

export const reindexNotes = on(
  db.table(notes).changed(),
  flow("search.reindexNotes", {
    plane: "operator",
    do: async ({ before, after }, fx) => {
      if (!after) {
        await fx.call(dropNoteIndex, { id: String(before?.id ?? "") });
        return;
      }
      await fx.call(upsertNoteIndex, { id: String(after.id) });
    },
  }),
);

changed("status") stamps trigger.cdc.column on the Manifest. Still the same payload — filter in do when you only care about that field:

src/flows/tasks/on-status.ts
import { on, flow } from "okengine";
import { db } from "@/core";
import { tasks } from "@/schema";

export const onStatus = on(
  db.table(tasks).changed("status"),
  flow("tasks.onStatus", {
    plane: "operator",
    do: async ({ before, after, id }, fx) => {
      if (before?.status === after?.status) return;
      await fx.emit(taskStatusChanged, {
        id,
        from: before?.status ?? null,
        to: after?.status ?? null,
      });
    },
  }),
);

Execution

Consumers share the Flow species with HTTP. The differences are the trigger and how failure is retried.

KindStartFailureTime
Signal oncefx.emitSignal retries then DLQfx.clock.now()
Signal broadcastfx.emitPer-subscriber; no once DLQfx.clock.now()
Clockscheduler tickFlow retry if set; no catch-up burstfx.clock.*
CDCcommitted SQL writeFlow retry if setfx.clock.now()

Mark long work durable: true and wrap side effects in fx.step — see Workflows.

Clock drivers: postgres in dev/prod, frozen in test. Signal drivers: redis in dev/prod, memory in test.

Troubleshooting

Learn more

Next

Durable Workflows

Step journaling and multi-step distributed execution.

Signal Element

Delivery physics — once, broadcast, and live tapes.

Clock Element

Named schedules, intervals, and durable sleep.

HTTP

Synchronous REST, QUERY, resources, and live SSE.

Routing

File-tree stamps for HTTP paths, Flow names, and client units from on(http.*, flow) under src/flows.

Workflows

Journaled multi-step Flows — fx.step replay, LIFO undo, durable sleep, and crash resume.

On this page

Smallest ExampleBind a Signal consumerEmit from any FlowProgressive PatternsTrigger ReferenceSignal ConsumersClock JobsCDCBare or enrichedExecutionTroubleshootingLearn moreNext