Workflows

const order = app.workflow<
  { orderId: string },          // input type
  { shipped: boolean }          // output type
>("order", async ({ step, input, runId }) => {
  const payment = await step.run("charge", () => stripe.charge(input));
  await step.sleep("cooloff", "10m");
  const approval = await step.waitForEvent<{ by: string }>(
    "approval",
    "order.approved",
    { timeout: "1h" },
  );
  if (!approval) return { shipped: false };       // timed out
  await step.run("ship", () => shipping.create(payment));
  return { shipped: true };
});

// The closure runs ONCE. The result is journaled (JSON) and every
// later attempt returns it without re-executing.
const user = await step.run("fetch-user", () => db.users.find(id));

// Throw inside the closure = step failure → retry with backoff.
// Completed steps never re-run during those retries.
const charge = await step.run("charge", async () => {
  const res = await stripe.charge(amount, {
    idempotencyKey: ctx.idempotencyKey("charge"),  // guard the side effect
  });
  if (!res.ok) throw new Error(res.reason);
  return res.id;
});

// ctx.idempotencyKey(label) is deterministic across retries + replays —
// derived from the run id — so an effect that succeeds but crashes before
// its result is journaled won't duplicate. One distinct label per effect.

await step.sleep("cooloff", "10m");   // minutes
await step.sleep("digest-gap", "7d");  // …or days. Zero resources held.

// Suspend until app.emit("invoice.paid", payload) — or 24h pass.
const paid = await step.waitForEvent<{ amount: number }>(
  "payment",
  "invoice.paid",
  { timeout: "24h" },
);

if (paid === null) {
  await step.run("remind", () => emails.push({ to, subject: "Reminder" }));
} else {
  await step.run("receipt", () => emails.push({ to, subject: "Receipt" }));
}

// Somewhere else — a webhook handler, another job, anywhere:
app.emit("invoice.paid", { amount: 4200 });

const notify = app.workflow<{ userId: string }, void>("notify", async ({ step, input }) => {
  await step.run("push", () => pushNotification(input.userId));
});

const signup = app.workflow("signup", async ({ step, input }) => {
  await step.run("create", () => accounts.create(input));
  // Child runs durably; parent suspends holding nothing.
  // A failed child makes this THROW the child's error (catchable).
  await step.invoke("welcome", notify, { userId: input.id });
});

// Parallel steps with independent memoization: if upsert-3 fails,
// the retry fast-forwards 0..2 and 4 from the journal and re-runs only 3.
const results = await step.all(
  rows.map((row, i) => () => step.run(`upsert-${i}`, () => db.upsert(row))),
);

// Fire and forget — durable from this moment:
const { runId } = await order.trigger(
  { orderId: "o_42" },
  { idempotencyKey: "o_42", delay: "10s" },
);

// Trigger and wait for the output (throws on failure/cancel/timeout):
const result = await order.triggerAndWait({ orderId: "o_43" }, { timeout: "2m" });

// Inspect any run:
const run = await order.getRun(runId);
// → { runId, workflow, status, output?, error? }
// status: running | sleeping | waitingEvent | waitingChild
//       | completed | failed | cancelled

// Cancel a run and all its child runs:
await order.cancel(runId);

// Realtime: stream status + step events until the run finishes.
// Store-backed, so it works across processes — pipe it to SSE/WebSocket.
for await (const u of app.subscribe(runId, { interval: 250 })) {
  console.log(u.status, u.steps.length); // ends when u.terminal
}