Quickstart

Overview

The Workflows fragment provides durable, replayable workflow execution with steps, timers, retries, and external events. This quickstart walks through defining a workflow, wiring the runner, and exposing routes.

Installation

Install the fragment, dispatcher, and database packages:

npm install @fragno-dev/fragment-workflows @fragno-dev/workflows-dispatcher-node @fragno-dev/db

Define a Workflow

Create a workflow class and a registry of workflows:

lib/workflows.ts

import {
  WorkflowEntrypoint,
  type WorkflowEvent,
  type WorkflowStep,
} from "@fragno-dev/fragment-workflows";

type ApprovalParams = {
  requestId: string;
  amount: number;
};

type ApprovalEvent = { approved: boolean };

type FulfillmentEvent = { confirmationId: string };

export class ApprovalWorkflow extends WorkflowEntrypoint<unknown, ApprovalParams> {
  async run(event: WorkflowEvent<ApprovalParams>, step: WorkflowStep) {
    const approval = await step.waitForEvent<ApprovalEvent>("approval", {
      type: "approval",
      timeout: "15 min",
    });

    await step.sleep("cooldown", "2 s");

    const fulfillment = await step.waitForEvent<FulfillmentEvent>("fulfillment", {
      type: "fulfillment",
      timeout: "15 min",
    });

    return { request: event.payload, approval, fulfillment };
  }
}

export const workflows = {
  approval: { name: "approval-workflow", workflow: ApprovalWorkflow },
} as const;

Create the Fragment Server

Wire the workflow runner, dispatcher, and fragment definition:

lib/workflows-fragment.ts

import { defaultFragnoRuntime, instantiate } from "@fragno-dev/core";
import type { DatabaseAdapter } from "@fragno-dev/db";
import {
  createWorkflowsRunner,
  workflowsFragmentDefinition,
  workflowsRoutesFactory,
} from "@fragno-dev/fragment-workflows";
import { createInProcessDispatcher } from "@fragno-dev/workflows-dispatcher-node";
import { workflows } from "./workflows";

export function createWorkflowsFragmentServer(adapter: DatabaseAdapter<any>) {
  const runtime = defaultFragnoRuntime;
  let runner: ReturnType<typeof createWorkflowsRunner> | null = null;
  const dispatcher = createInProcessDispatcher({
    wake: () => {
      if (!runner) {
        return;
      }
      void runner.tick({ maxInstances: 5, maxSteps: 50 });
    },
    pollIntervalMs: 2000,
  });

  const config = { workflows, dispatcher, runtime, enableRunnerTick: true };
  const fragment = instantiate(workflowsFragmentDefinition)
    .withConfig(config)
    .withRoutes([workflowsRoutesFactory])
    .withOptions({ databaseAdapter: adapter })
    .build();

  runner = createWorkflowsRunner({ fragment, workflows, runtime });
  config.runner = runner;

  return { fragment, dispatcher };
}

The in-process dispatcher is ideal for local dev. For Cloudflare deployments use @fragno-dev/workflows-dispatcher-cloudflare-do.

Mount the Routes

Mount the fragment using your framework adapter. See Integrating a Fragment for framework-specific examples.

Run Database Migrations

Generate a schema and run migrations with the Fragno CLI:

npx fragno-cli db generate lib/workflows-fragment.ts -o db/workflows.schema.ts
npx fragno-cli db migrate lib/workflows-fragment.ts

Test Workflows

Use the test harness to drive the runner with a controllable clock and deterministic runtime:

lib/workflows.test.ts

import {
  createWorkflowsTestHarness,
  createWorkflowsTestRuntime,
} from "@fragno-dev/fragment-workflows/test";
import { workflows } from "./workflows";

const runtime = createWorkflowsTestRuntime({ startAt: 0, seed: 123 });
const harness = await createWorkflowsTestHarness({
  workflows,
  adapter: { type: "drizzle-pglite" },
  runtime,
});

const instanceId = await harness.createInstance("approval", {
  params: { requestId: "req_1", amount: 125 },
});

await harness.runUntilIdle();
await harness.sendEvent("approval", instanceId, {
  type: "approval",
  payload: { approved: true },
});

harness.clock.advanceBy("2 s");
await harness.runUntilIdle();

Drive the Runner

If you want an external scheduler to drive execution, keep enableRunnerTick: true and send POST /_runner/tick to your fragment:

await fetch("https://your-app.example.com/api/workflows/_runner/tick", {
  method: "POST",
  headers: { "Content-Type": "application/json" },
  body: JSON.stringify({ maxInstances: 10, maxSteps: 100 }),
});

Use the CLI

The fragno-wf CLI uses the HTTP API to list workflows, inspect instances, and send events:

fragno-wf workflows list -b https://your-app.example.com/api/workflows

fragno-wf instances get -b https://your-app.example.com/api/workflows -w approval -i inst_123

Next Steps

Build a client or CLI on top of the HTTP API in the fragment.
Review workflow history and logs using the /history routes.

Quickstart

On this page