Do slow work outside the database.

Pick a pattern

A worker is your own process that does slow or external work. A live query tells it what to do; a mutation decides whether to accept the result. Nothing new runs inside the database.

Both loops come from @flower-js/sdk/worker and run in Node or a browser. Results land in ordinary collections. To run and scale queue workers, see Worker pools.

Keep a result current

An external value stores each result with the exact input that produced it. When the input changes, the old result stops counting immediately.

This app keeps a SHA-256 digest of every document. Download reactive-worker.ts.

import { collection, define, external, mutation, query, v } from "@flower-js/sdk";

const documentId = v.string({ min: 1, max: 256 });
const documents = collection("documents", v.object({ text: v.string({ max: 100_000 }) }));

// Everything that affects the digest is in its input. Workers publish results
// tagged with that input; a changed document makes its old digest stale at once.
const digest = external("digest", {
  input: (ctx, id: string) => {
    const document = ctx.get(documents, id);
    return document && { recipe: "sha256-v1", text: document.text };
  },
  result: v.string({ pattern: /^[0-9a-f]{64}$/ }),
  each: documents,
});

const put = mutation("document.put", { args: v.object({ id: documentId, text: v.string({ max: 100_000 }) }) }, (ctx, input) => {
  ctx.set(documents, input.id, { text: input.text });
  return null;
});

const remove = mutation("document.delete", { args: documentId }, (ctx, id) => {
  ctx.delete(documents, id);
  return null;
});

const get = query("document.get", { args: documentId }, (ctx, id) => {
  const document = ctx.get(documents, id);
  return document && { text: document.text, digest: ctx.get(digest, id) };
});

const app = define({
  uses: [digest],
  http: { "document.put": put, "document.delete": remove, "document.get": get, ...digest.http("digest") },
});
export default app;

• input(ctx, args) returns everything that affects the result: source values, recipe, model and prompt versions, locale, configuration. null means there is nothing to compute. Don't use a global database revision.
• ctx.get(digest, id) is { status: "pending" }, { status: "ready", value }, or null when there is no input. Read it like any derived value; it stays current.
• digest.http("digest") exposes digest.pending (the work for one key, or null) and digest.publish. With each, it adds digest.next, which lists stale keys oldest first for worker pools.
• publish stores a result only if its key still names the current input, and returns { accepted }. Among racing workers, the first result wins. A result schema checks every value.
• The input must not depend on its own result. If it goes A → B → A, the stored A result counts again.

Run the worker

import { FlowerClient } from "@flower-js/sdk";
import { reconcile } from "@flower-js/sdk/worker";
import { webcrypto } from "node:crypto";
import { fileURLToPath } from "node:url";
import type app from "./reactive-worker.ts";

type Input = { recipe: string; text: string };

// Keep every document's digest current, or only one when id is given. Several
// processes may run at once: publication keeps the first result for an input.
export async function runWorker(client: FlowerClient<typeof app>, signal: AbortSignal, id?: string) {
  await reconcile<string, Input, string>(client, {
    external: "digest",
    ...(id === undefined ? {} : { args: id }),
    signal,
    async compute(input) {
      if (input.recipe !== "sha256-v1") throw new Error("Unsupported worker recipe");
      const digest = await webcrypto.subtle.digest("SHA-256", new TextEncoder().encode(input.text));
      return Buffer.from(digest).toString("hex");
    },
    onEvent: (event) => console.log(event.type, "key" in event ? event.key : event.error),
  });
}

if (process.argv[1] === fileURLToPath(import.meta.url)) {
  const shutdown = new AbortController();
  process.once("SIGINT", () => shutdown.abort());
  process.once("SIGTERM", () => shutdown.abort());
  await runWorker(new FlowerClient(process.env.FLOWER_URL ?? "http://127.0.0.1:7101"), shutdown.signal, process.argv[2]);
}

• With args, reconcile keeps one key current by watching digest.pending. Without, it drains every stale key through digest.next. shard: [index, count] splits keys between processes by hash; concurrency runs several computations at once.
• compute may run more than once for the same input. Failures back off per key, and publications retry with one request ID.
• next notices a key only when its row in the each collection is written. If the input also reads other records, a change there makes the value pending, and digest.pending reports it, but next won't list it until that row changes. Keep pool inputs to the row itself, or run a per-key worker.

Try it: keep a digest current

Run the document reconciler against a fresh local app. It needs Node.js and no API keys.

Start a local node, then run from the checkout:

npm run build
node sdk/cli.ts deploy docs/reactive-worker.ts
node sdk/cli.ts call document.put '{"id":"notes","text":"Hello, garden."}' --request-id notes-v1
node sdk/cli.ts watch document.get '"notes"'

node docs/reactive-worker-client.ts
# FLOWER_URL defaults to http://127.0.0.1:7101. Pass a document ID to follow just one.
# Start the same command in another terminal to exercise competing workers.

node sdk/cli.ts call document.put '{"id":"notes","text":"The garden has changed."}' --request-id notes-v2

The view shows a digest only for the current text. Stop all workers, edit, and restart one: it catches up on its own. Download the application and client.

To run a pool of queue workers instead, see Worker pools.

Complete every action

When every action matters, create a job in the same mutation as the business change. Workers lease jobs, do the work, and record the outcome.

For example, the mutation that marks an invoice paid also calls receipts.enqueue(ctx, id, payload), so a paid invoice always has its receipt job. Put everything the worker needs in the payload. Queues & expiry has the full example.

1. Wait and claim. runQueueWorker watches the queue's ready query and claims only when a lane is free.
2. Do the work. Your work(job, signal) runs while the lease is renewed; the signal aborts shortly before the lease would end.
3. Report through Flower. The worker completes or fails the job with the lease exactly as claimed, retrying lost replies under one request ID.
4. Drain, then wait again. A lane claims until the queue is empty, then goes back to watching.

• Idempotency: use the business operation's ID across retries and leases; the job ID is a good one. A lease token identifies one attempt, not the operation. Lease expiry can't stop a paused process, so the receiving service must deduplicate or fence. If it fences, keep enforcing after a disaster restore.
• Retries: the queue's retry policy requeues failed jobs with backoff. Look up an unknown outcome by business ID before calling it failed.
• The same mutation can update business records when you complete jobs yourself with jobs.complete(ctx, lease, result).