@dmytromykhailiuk/flow-engine

Typed flow engine with JSON-serializable flows, safe string expressions, pause/cancel/suspend, checkpoints and restore. Works the same in the browser and on the backend.

Full documentation: Docs — every node type, every option, with examples.

The one rule: steps are code, flows are data. Anything that talks to the world is registered as a typed step handler in the builder; everything that decides what happens when — branching, loops, retries, conditions — lives in a JSON config that can be stored, sent from a backend, versioned and validated.

You have multi-step processes: a checkout, an onboarding, a document pipeline, a sync job. The steps themselves are easy — the hard part is the choreography around them: branching on intermediate results, retrying the flaky ones, cancelling when the user leaves, surviving a page refresh or a worker crash, and not littering the codebase with one-off state machines. This package is that choreography, factored out once.

Each run carries an explicit context — { input, steps, vars } — where every step's output is addressable by name. Conditions and data wiring are written as string expressions ("steps.loadUser.user.age >= 18") evaluated by a built-in safe interpreter: no eval, no new Function, no reaching globals or prototypes, so configs from a backend are data, not code. After every node the engine emits a JSON checkpoint; restore() continues an unfinished run after a refresh — or on a different machine.

Install

npm i @dmytromykhailiuk/flow-engine

Quick start

import { createFlowRunner } from "@dmytromykhailiuk/flow-engine";

const runner = createFlowRunner((b) =>
  b
    .registerStep("loadUser", async (input: { id: string }, _cfg: void, ctx) => ({
      user: await api.getUser(input.id, { signal: ctx.signal }),
    }))
    .registerStep("greet", (input: { name: string }, cfg: { greeting: string }) => ({
      message: `${cfg.greeting}, ${input.name}!`,
    }))
    .registerCondition("isPremium", (ctx) => (ctx.steps.loadUser as any)?.user.plan === "premium")
    .registerFlow("welcome", {
      nodes: [
        { step: "loadUser", input: { id: "input.userId" } },
        {
          if: [
            {
              when: "$isPremium",
              flow: [{ step: "greet", input: { name: "steps.loadUser.user.name" }, config: { greeting: "Welcome back" } }],
            },
          ],
          else: [{ step: "greet", input: { name: "steps.loadUser.user.name" }, config: { greeting: "Hello" } }],
        },
      ],
      output: "steps.greet.message",
    }),
);

const handle = runner.run("welcome", { userId: "42" });

handle.pause();            // parks at the next step boundary
handle.resume();
handle.cancel("navigated away"); // aborts ctx.signal in the running step

const result = await handle.getResult(); // never rejects
if (result.status === "completed") console.log(result.output); // "Welcome back, Ada!"

What's in a flow

Ten node types cover the control flow; each is JSON:

| Node | Purpose | | --- | --- | | step | run a registered handler — with input mapping, config, timeout, retry, onError | | if | first branch whose when holds wins; optional else | | loop | { times }, { while, max } or { forEach, as } over data | | parallel | branches via Promise.all; settle: "all" to collect every error | | subflow | call another flow like a function — own context, output stored under as | | assign | compute vars with expressions — the safe replacement for an eval step | | finish | early exit; with a tag it unwinds nested flows like a labeled break | | break / continue | loop control, optionally by loop label | | try | try/catch/finally over nodes; catch sees the error root |

Expressions may read input, steps, vars — plus loop inside loops, error inside catch/retry.when — and call a whitelist of pure methods. "$name" references a predicate registered in code.

Never lose a run

const runner = createFlowRunner(setup, {
  onEvent: (e) => {
    if (e.type === "checkpoint") localStorage.setItem(`flow:${e.runId}`, JSON.stringify(e.snapshot));
    if (e.type === "flowEnd") localStorage.removeItem(`flow:${e.runId}`);
  },
});

// after a refresh — continue everything unfinished:
for (const key of Object.keys(localStorage)) {
  if (key.startsWith("flow:")) runner.restore(JSON.parse(localStorage.getItem(key)!));
}

Snapshots are self-contained JSON (position, context, and the flow config itself). A step interrupted mid-flight re-runs on restore — at-least-once — and ctx.executionKey gives you a stable idempotency key to dedupe side effects.

For long external waits there is ctx.suspend(): the run ends in this process with a final snapshot, freeing memory, queue slots and locks; when the reply arrives, restore() re-runs the waiting step, which picks the result up by its executionKey.

Sequential when you say so

Runs are parallel by default. Give flows a queue and they serialize FIFO:

runner.run("syncCart", input, { queue: "cart" }); // or  { queue: "cart" }  in the flow config

Queues ride on @dmytromykhailiuk/execution-blocker — the package's only dependency, itself zero-dep. Pass your own instance via queueAdapter to share queues with non-flow code, or a Redis-backed { run(id, fn) } adapter to make them cluster-wide.

And the rest

• Guards — declarative preconditions with onSuccess/onFailure side flows; a failing guard rejects the run (status: "rejected"), which is neither success nor error.
• Hooks — named entry points: runner.runHook("onLogin", input) runs the first entry whose condition matches.
• Background flows — "when this condition over external sources becomes true, run that flow"; sources are anything with { getSnapshot, subscribe } (signals, Redux, useSyncExternalStore).
• applyConfig — load or update the whole declarative layer (flows/hooks/guards/background) from JSON at runtime, atomically; active runs keep the version they started with.
• Events — a typed stream (flowStart, stepEnd, checkpoint, heartbeat, …) for logging, devtools or distributed-lock lease renewal.
• validateFlow — every expression parsed, every reference checked, with exact paths; run() refuses a broken config synchronously.

TypeScript

The builder accumulates a registry in its generics: step ids autocomplete inside configs, each step's config is type-checked by id, registered flow and predicate names are suggested. What the type system cannot see — the contents of expression strings, JSON from a backend — is covered by runtime validation with precise error paths. FlowResult is a discriminated union (completed | rejected | cancelled | failed | suspended), so handling every outcome is a switch, not guesswork.

License

MIT