pg-workflows

Durable workflows for TypeScript, backed by PostgreSQL. Each step's result is saved, a retried run skips the steps that already finished, and a run can pause for an event, a timer, or a polled condition. There's no Redis, broker, or scheduler to run.

Install with your coding agent

Paste this into Claude Code, Cursor, Codex, or any agent that can fetch a URL:

Add pg-workflows to this project. Fetch https://pgworkflows.dev/skill.md and follow it step by step: pick the right layout for this codebase (monolith, web app plus worker, or microservices), install and verify the engine, add the @pg-workflows/ui dashboard for our stack, then ask me whether to add OpenTelemetry tracing.

Or install the skill: npx skills add SokratisVidros/pg-workflows --skill pg-workflows-install. See Install with an agent.

Quickstart

1. Start Postgres (skip if you already have one):

docker run -d --name pg-workflows-db -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres:17

2. Create a project:

mkdir workflows-demo && cd workflows-demo
npm init -y
npm install pg-workflows pg zod
npm install -D tsx

3. Save this as index.ts:

import { WorkflowEngine, WorkflowStatus, workflow } from 'pg-workflows'
import { z } from 'zod'

const greetUser = workflow(
  'greet-user',
  async ({ step, input }) => {
    const user = await step.run('load-user', async () => {
      return { name: input.name, signedUpAt: new Date().toISOString() }
    })

    const message = await step.run('build-message', async () => {
      return `Welcome, ${user.name}!`
    })

    return { message }
  },
  { inputSchema: z.object({ name: z.string() }) },
)

async function main() {
  const engine = new WorkflowEngine({
    connectionString: process.env.DATABASE_URL ?? 'postgres://postgres:postgres@localhost:5432/postgres',
    workflows: [greetUser],
  })
  await engine.start()

  const run = await engine.startWorkflow({
    workflowId: 'greet-user',
    input: { name: 'Ada' },
  })

  let result = await engine.getRun({ runId: run.id })
  while (result.status === WorkflowStatus.PENDING || result.status === WorkflowStatus.RUNNING) {
    await new Promise((resolve) => setTimeout(resolve, 200))
    result = await engine.getRun({ runId: run.id })
  }

  console.log(result.status, result.output)
  await engine.stop()
}

main()

4. Run it:

npx tsx index.ts

After the engine's startup logs, you should see:

completed { message: 'Welcome, Ada!' }

engine.start() creates its tables on first run. Each step.run result is saved on the run's row in workflow_runs. When a run is retried (set retries on the workflow), completed steps return their saved result instead of running again.

Wait for an event

step.waitFor pauses the run until your code calls triggerEvent. A paused run holds no worker and no connection.

import { WorkflowEngine, WorkflowStatus, workflow } from 'pg-workflows'
import { z } from 'zod'

const approveExpense = workflow(
  'approve-expense',
  async ({ step, input }) => {
    await step.run('notify-manager', async () => {
      return { notified: `manager of ${input.employee}` }
    })

    const review = await step.waitFor('wait-for-review', {
      eventName: 'expense-reviewed',
      schema: z.object({ approved: z.boolean() }),
    })

    return { amount: input.amount, approved: review.approved }
  },
  { inputSchema: z.object({ employee: z.string(), amount: z.number() }) },
)

const sleep = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))

async function main() {
  const engine = new WorkflowEngine({
    connectionString: process.env.DATABASE_URL ?? 'postgres://postgres:postgres@localhost:5432/postgres',
    workflows: [approveExpense],
  })
  await engine.start()

  const run = await engine.startWorkflow({
    workflowId: 'approve-expense',
    input: { employee: 'ada', amount: 120 },
  })

  // In a real app, this is a separate request: an approval button, a webhook, and so on.
  while ((await engine.getRun({ runId: run.id })).status !== WorkflowStatus.PAUSED) await sleep(200)

  await engine.triggerEvent({
    runId: run.id,
    eventName: 'expense-reviewed',
    data: { approved: true },
  })

  let result = await engine.getRun({ runId: run.id })
  while (result.status !== WorkflowStatus.COMPLETED && result.status !== WorkflowStatus.FAILED) {
    await sleep(200)
    result = await engine.getRun({ runId: run.id })
  }

  console.log(result.status, result.output) // completed { amount: 120, approved: true }
  await engine.stop()
}

main()

Features

| Feature | API | |---------|-----| | Durable steps | step.run(id, fn) | | Wait for external events | step.waitFor(id, { eventName, timeout?, schema? }) + engine.triggerEvent() | | Timers | step.delay(id, '3 days'), step.waitUntil(id, date) | | Polling | step.poll(id, fn, { interval, timeout }) | | Manual pause and resume | step.pause(id), engine.resumeWorkflow() | | Child workflows | step.invokeChildWorkflow(id, ref, input) | | Recurring schedules | workflow(id, fn, { schedule: '0 9 * * 1-5' }) | | Retries | workflow(id, fn, { retries: 3 }) | | Priorities | workflow(id, fn, { priority: 'high' }) | | One run at a time | workflow(id, fn, { singleton: true }) | | Deduplicated starts | startWorkflow({ idempotencyKey }) | | Tenant scoping | resourceId on every run and every API call | | Typed input | Any Standard Schema library (Zod, Valibot, ArkType) | | API/worker split | WorkflowClient from pg-workflows/client |

Documentation

Full docs are at pgworkflows.dev.

| Guide | Covers | |-------|--------| | Core concepts | Steps, events, timers, polling, child workflows, schedules, retries, priorities, singletons, idempotency, input validation | | Architectures | Single service, or microservices with web and worker services | | Examples | Conditional steps, fan-out loops, reminders, polling, retries, progress | | AI and agent workflows | Durable LLM pipelines, human review, retrieval | | API reference | WorkflowEngine, WorkflowClient, WorkflowRef, workflow(), types | | Configuration | Environment variables, database objects, requirements |

Packages

| Package | Purpose | |---------|---------| | pg-workflows | The engine and client | | @pg-workflows/otel | OpenTelemetry spans for workflow runs and steps | | @pg-workflows/ui | React dashboard, components, and hooks. Try it with npx @pg-workflows/ui |

Runnable scripts live in examples/.

Requirements

• Node.js >= 18
• PostgreSQL >= 10
• pg >= 8 (peer dependency). pg-boss ships with the engine and needs no setup.

Acknowledgments

Temporal, Inngest, Trigger.dev, and DBOS pioneered the durable execution patterns this project builds on.

License

MIT