Flow

Behavior — endpoints, jobs, consumers, and workflows as one species: a typed trigger, declared contracts, and a do that touches the world only through fx.

Flow is how your backend does anything. An HTTP endpoint, a queue consumer, a cron job, a multi-step payment workflow — in other stacks these are four frameworks; here they are one species with one shape: on(trigger, flow). Learn the shape once, and only the trigger ever changes.

The one rule

All world access goes through fx. A direct fetch, Date.now(), or node: import inside a flow is a defect — effects are inferred from what a flow touches through fx, and that inference is what powers the Manifest, the Console, caching, and durability.

Quick start

Write the flow

Four declarations plus a do:

src/flows/orders/create.ts

import { on, flow, http } from "okengine";
import { z } from "zod";

export const createOrder = on(
  http.post("/orders"),
  flow({
    in: z.object({ sku: z.string(), qty: z.number().int().min(1) }),
    out: z.object({ id: z.string() }),
    errors: { OutOfStock: z.object({ left: z.number() }) },
    do: async (input, fx) => {
      const id = fx.id();
      await fx
        .store(db)
        .insert(orders)
        .values({ id, ...input, status: "pending" });
      return { id };
    },
  }),
);

Run it

oke dev picks up every exported flow — no route registration, no controller wiring:

oke dev

Call it

curl -X POST localhost:6530/orders -d '{"sku":"SKU-1","qty":2}' -H 'content-type: application/json'
# { "data": { "id": "…" }, "error": null }

Anatomy of a Flow

Part	Role
trigger	What starts the flow — `http`, a signal, `every`, a row change
`in`	Input contract — validated before `do` runs; bad input is a 422
`out`	Output contract — the return value is checked against it
`errors`	Typed failures — returned with `fx.fail`, never thrown
`do`	The work — every read, write, emit, and call goes through `fx`

Failures are values, not exceptions:

do: async (input, fx) => {
  const [product] = await fx.store(db).select().from(products).where(eq(products.sku, input.sku));
  if (!product || product.stock < input.qty) {
    return fx.fail("OutOfStock", { left: product?.stock ?? 0 }); // declared in errors
  }
  // …
};

Every response follows one envelope — success { data, error: null }, failure { data: null, error: { code, data } } — so clients handle outcomes by error.code, not by parsing status text.

The five triggers

Trigger	Starts when	Replaces
`http.post("/orders")`	a request arrives	endpoint · handler
`orderPlaced` (a signal)	another flow emits	queue consumer
`every("1h")`	time passes	cron job
`db.table(orders).changed()`	a row changes	CDC pipeline
— none	another flow calls `fx.call`	"private" helper

http — a request arrives

export const findOrder = on(
  http.get("/orders/:id").gate(member), // gates evaluate before do runs
  flow({
    in: z.object({ id: z.string() }),
    out: Order,
    errors: { NotFound: z.object({}) },
    do: async ({ id }, fx) => (await fx.store(db).findById(orders, id)) ?? fx.fail("NotFound", {}),
  }),
);

signal — another flow emits

The producer emits through fx (transactionally with its writes); the consumer is the same species — no subscribe(), no listener registration:

await fx.emit(orderPlaced, { orderId: id }); // inside the producing flow

on(
  orderPlaced,
  flow({
    do: async ({ orderId }, fx) => {
      /* … */
    },
  }),
);

clock — time passes

every("1h") is a trigger value, not a registration with a scheduler library:

on(
  every("1h"),
  flow({
    do: async (_, fx) => {
      const cutoff = fx.clock.now() - 30 * 24 * 60 * 60 * 1000;
      await fx.store(db).delete(sessions).where(lt(sessions.createdAt, cutoff));
    },
  }),
);

store change — a row changes

CDC is built in; the flow receives { before, after }:

on(
  db.table(orders).changed("status"),
  flow({
    do: ({ before, after }, fx) => fx.log.info("status", { from: before.status, to: after.status }),
  }),
);

no trigger — a callable flow

Drop on() and it is still a real Flow — contracts, Manifest entry, everything. Other flows call it through fx.call:

export const getOrder = flow({
  in: OrderRef,
  out: Order,
  do: async ({ id }, fx) => {
    /* … */
  },
});

const order = await fx.call(getOrder, { id: orderId }); // from any other flow

fx — the only door

Everything a flow may touch, on one object:

Surface	Effect recorded	What it does
`fx.store(db).select/insert/…`	read / write	SQL, KV, files, index sessions
`fx.emit(signal, payload)`	emit	Publish a signal (transactional with writes)
`fx.send(template, opts)`	send	Reach a human (email · SMS · …)
`fx.ask(prompt, input)`	ask	Call a versioned AI prompt
`fx.run(agent, input)`	ask	Run a bounded agent
`fx.call(flow, input)`	call	Invoke another flow
`fx.vault(contract)`	read	Read a secret (redacted from logs)
`fx.clock.now()` / `.sleep(…)`	—	Injected time / durable sleep
`fx.cache.get/set`	—	Shared cache with effect-aware invalidation
`fx.step(name, fn)`	—	Named durable step — never re-runs on replay
`fx.id()` · `fx.log` · `fx.t`	—	UUIDs, redacting logger, i18n
`fx.auth` · `fx.operator` · `fx.tenant`	—	Who is calling (user / operator / tenant)

Why this strictness pays off

Effects are inferred from fx usage, so the Manifest knows exactly which flows read orders or send PII to a model — without you declaring any of it. That one graph drives the Console panels, cache invalidation, least-privilege tokens, and durable replay.

Durability — flows that survive the process

Set durable: true and every fx call is journaled. Wrap side effects in fx.step and they never re-run on replay:

export const chargeOrder = flow({
  durable: true, // every fx call below is journaled
  in: OrderRef,
  out: z.boolean(),
  do: async ({ orderId }, fx) => {
    const intent = await fx.step("create-intent", () =>
      stripe(fx.vault(stripeKey)).create(orderId),
    );

    await fx.clock.sleep("verify-window", "2m"); // survives restart and deploy

    return fx.step("confirm", () => stripe(fx.vault(stripeKey)).confirm(intent));
  },
});

Consequence: kill the process between the two steps and the run resumes at confirm — completed steps replay from the journal, so the card is not charged twice. This is verified by the engine's own test suite: after resume, create-intent has run exactly once.

Composition is just calls

Wiring is values flowing between flows — declared in code, never configured in a dashboard:

on(http.post("/orders"), createOrder); // ① a request arrives
on(orderPlaced, sendReceipt); // ② its emit starts the consumer
on(every("1h"), sweepExpired); // ③ time passes
on(db.table(orders).changed("status"), reverify); // ④ a row changes
// ⑤ getOrder — nothing starts it; every flow above can fx.call it

Troubleshooting

Learn more

Signal — delivery physics (once · broadcast · live)
Clock — schedules and durable sleep
Console · Flows — the Manifest-derived panel
Runs — how a flow execution is observed