Signal
Data in motion — queues, pub/sub, and streams as one declaration whose delivery physics you choose explicitly.
Signal is how your app moves data between flows and to clients: background jobs, events, fan-out notifications, live feeds. What other stacks split into a queue library, a pub/sub client, and a streaming platform is one declaration here — you pick the delivery physics, and the driver underneath is swappable per environment.
The one rule
delivery is mandatory with no default. Delivery physics is a semantic decision about your
data — guessing it produces silent, expensive bugs, so omitting it is a type error.
Quick start
Declare the signal
A signal is a named, typed channel. The schema describes the payload; delivery describes its physics:
import { signal } from "okengine";
import { z } from "zod";
export const orderPlaced = signal("order-placed", {
schema: z.object({ orderId: z.string(), total: z.number() }),
delivery: "once", // queue physics: one consumer, retries, DLQ
retries: 3,
deadLetter: true,
});Emit it from a Flow
Producers never touch a broker client — they call fx.emit:
do: async (input, fx) => {
await fx.store(db).insert(orders).values({ id: input.id /* … */ });
await fx.emit(orderPlaced, { orderId: input.id, total: input.total });
// With the postgres driver this enrols in the same transaction as the insert —
// the message is only ever sent if the row is committed.
};Consume it with a Flow
A signal is a trigger like any other — the consumer is an ordinary Flow:
export const sendConfirmation = on(
orderPlaced,
flow({
in: z.object({ orderId: z.string(), total: z.number() }),
do: async (input, fx) => {
await fx.channel(email).send(/* … */);
},
}),
);That's the loop: declare → fx.emit → on(signal, flow). The runtime handles routing, retries, and dead-lettering.
The three delivery physics
delivery | Semantics | Use for |
|---|---|---|
"once" | Queue: competing consumers, retries, dead-letter queue | Jobs that must happen exactly once — emails, payments sync |
"broadcast" | Pub/sub: every subscriber receives a copy | Cache invalidation, cross-service events |
"live" | Stream: client-subscribable and replayable | Live feeds, dashboards, progress updates |
The declaration is identical in shape for all three — switching physics later is a one-word change, not a migration to another library.
Options
| Option | Type | Default | Meaning |
|---|---|---|---|
delivery | "once" | "broadcast" | "live" | — (required) | Delivery physics |
schema | zod / Standard Schema | — | Payload contract; typed emits and Manifest docs |
retries | number | 3 | Max delivery attempts before dead-letter (once) |
deadLetter | boolean | true | Preserve exhausted messages in the DLQ (once) |
optional | boolean | false | Allow emitting while nobody subscribes (skip the orphan check) |
When delivery fails
once signals retry automatically. Every attempt keeps a typed failure reason, and the full attempt history survives into the DLQ — so when a message lands there you see why each attempt failed, not just that it did.
The Console (:6533 → Signals) shows the topology (which flows emit and consume each signal), per-subscriber delivery stats, and the DLQ contents with replay / discard controls — no separate broker UI to run.
Orphan emits fail loudly
Emitting a signal with zero subscribers is normally an error — it almost always means a typo or a
forgotten consumer. Set optional: true on the declaration only when zero-subscriber emits are
genuinely expected (e.g. an integration hook).
Per-environment drivers
drivers: {
signal: { local: "memory", docker: "postgres", test: "memory", prod: "postgres" },
},| Driver | Runs as | Best for |
|---|---|---|
memory | in-process | Local loop + tests — zero infrastructure |
postgres | your existing Postgres container | Default for docker/prod — emits join your DB transaction (outbox pattern) |
redis | Redis container | Explicit alternative when throughput outgrows Postgres |
nats | NATS container | High-throughput fan-out |
postgres is the default outside local dev for a reason: fx.emit enrols in the caller's transaction, so "write row + emit event" commits or rolls back atomically — the classic dual-write bug is designed out.
Troubleshooting
You emitted a signal that no flow consumes. Either wire a consumer with on(signal, flow), or set optional: true on the declaration if zero-subscriber emits are intentional.
After retries attempts the message moves to the DLQ — it is not lost. Open Console → Signals, inspect the typed failure reasons on each attempt, fix the consumer, then replay.
Ask: how many consumers should process each message? One → once. All of them → broadcast. Clients over time, with replay → live. If you need retries and a DLQ, you want once.
That is the dual-write problem, and it means the signal driver is not postgres. With the postgres driver the emit joins your SQL transaction; with redis / nats an outbox relay keeps the guarantee, but postgres is the only mode where atomicity is literal.
Learn more
- Flow —
on(trigger, flow)andfx.emit - Console · Signals — topology, delivery stats, DLQ replay
- Clock — scheduled and delayed work