@usu-accessibility/workflows

Type-safe durable workflows for TypeScript with:

• Zod runtime validation for workflow input and each step input/output
• Type-safe step chaining (output of step N becomes the typed input shape for step N+1)
• Prisma-backed durability for run state and resumability

Install

npm install @usu-accessibility/workflow zod
npm install -D prisma typescript

Prisma setup

Add the following to your Prisma schema (copy/paste):

enum WorkflowRunStatus {
  RUNNING
  COMPLETED
  FAILED
}

model WorkflowRun {
  id           String            @id @default(cuid())
  workflowName String
  status       WorkflowRunStatus @default(RUNNING)
  input        Json
  currentData  Json
  outputData   Json?
  stepResults  Json
  currentStep  Int               @default(0)
  error        String?
  createdAt    DateTime          @default(now())
  updatedAt    DateTime          @updatedAt
  startedAt    DateTime          @default(now())
  finishedAt   DateTime?

  @@index([workflowName, status])
}

Then run:

npx prisma generate
npx prisma migrate dev --name add_workflow_runs

Usage

import { z } from "zod";
import { PrismaClient } from "@prisma/client";
import {
  createWorkflow,
  PrismaWorkflowStore,
} from "@usu-accessibility/workflows";

const prisma = new PrismaClient();
const store = new PrismaWorkflowStore(prisma);

const workflow = createWorkflow({
  name: "welcome-email",
  input: z.object({ userId: z.string().uuid() }),
  store,
})
  .step({
    name: "load-user",
    input: z.object({ userId: z.string().uuid() }),
    output: z.object({ email: z.string().email() }),
    run: async (input) => {
      // Look up user in your own DB
      return { email: `user-${input.userId}@example.com` };
    },
  })
  .step({
    name: "send-email",
    input: z.object({ email: z.string().email() }),
    output: z.object({ sent: z.boolean() }),
    run: async (input, context) => {
      console.log("Running", context.stepName, "for run", context.runId);
      return { sent: true };
    },
  });

const result = await workflow.run({
  userId: "0f6f5e15-a321-4e67-8225-f76d9f0c4bff",
});
// result: { sent: boolean }

Workflows without initial input

const workflow = createWorkflow({
  name: "nightly-sync",
  store,
}).step({
  name: "seed",
  input: z.void(),
  output: z.object({ value: z.number() }),
  run: () => ({ value: 42 }),
});

const result = await workflow.run();

Resume failed workflows

try {
  await workflow.run({ userId: "..." });
} catch (error) {
  if (error instanceof WorkflowExecutionError) {
    await workflow.resume(error.runId);
  }
}

Fire-and-forget runs

const { runId } = await workflow.start({
  userId: "0f6f5e15-a321-4e67-8225-f76d9f0c4bff",
});

// Continue doing other work, then check run state later.
const run = await workflow.getRun(runId);
console.log(run?.status);

Queue workflow runs automatically (built-in)

const workflow = createWorkflow({
  name: "welcome-email",
  input: z.object({ userId: z.string().uuid() }),
  store,
  maxConcurrentRuns: 5,
});
// ...steps;

// Runs are queued internally; only 5 execute at a time.
const results = await Promise.all([
  workflow.run({ userId: "11111111-1111-1111-1111-111111111111" }),
  workflow.run({ userId: "22222222-2222-2222-2222-222222222222" }),
  workflow.run({ userId: "33333333-3333-3333-3333-333333333333" }),
]);

Development

npm install
npm run typecheck
npm test