工作流（@weifuwujs/workflow）

状态：已发布 1.2.0（v1 内核 + L2 交付面）：故障恢复、等待、信号、取消、超时、重试、并行/ map、版本化、契约套件全部有测试证据；HTTP + OpenAPI、weifuwu-workflow CLI 已交付。 第一契约：至少一次——崩溃后从最后已提交状态续跑，未提交步骤会重放；步骤必须幂等（用 ctx.run.id + ctx.step 作幂等键）。

在既有 Postgres 上获得多步流程的持久编排：流程是数据（状态 + 转移），步骤是函数（副作用只在转移时提交）。 不需要新基础设施、不需要 Temporal 式的确定性重放：进程崩了、机器重启、跨天等待，都从最后已提交状态接着走。

import { defineWorkflow } from '@weifuwujs/workflow'
import postgres from 'postgres'

const flow = defineWorkflow({ sql: postgres(process.env.DATABASE_URL!), workerId: 'worker-1' })
await flow.migrate() // 幂等：建 `_workflow_*` 表（advisory 锁串行化）

何时不该用：需要毫秒级调度、需要跨库分布式事务、需要人工拖拽 UI、或者单进程内存队列就够——别上工作流引擎。

安装

npm i @weifuwujs/workflow postgres   # 零运行时依赖；postgres 是宿主自己的驱动

无数据库想先试试？用 memory 驱动（零依赖、单进程，语义同族、parallel/map 明确不支持）：

import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

5 分钟上手

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const flow = createMemoryWorkflow({ workerId: 'w1' })
flow.step('send-mail', async (ctx, { input }) => ({ to: input['to'] }))
flow.define('welcome', {
  initial: 'send',
  states: {
    send: { step: 'send-mail', on: { ok: 'done' } },
    done: { end: true },
  },
})

const run = await flow.start('welcome', { to: '[email protected]' })
await flow.drain() // 生产中由 startWorker() 持续轮询；测试里 drain 手动推完
const finished = await flow.get(run.id)
assert.equal(finished?.status, 'done')
assert.deepEqual(finished?.output, { to: '[email protected]' })

换成 Postgres 只需把 createMemoryWorkflow({...}) 换成 defineWorkflow({ sql: postgres(URL), ... }) 并先 await flow.migrate()。

核心概念

• 状态是数据：define(name, { initial, states })；每个状态恰好是 step | map | parallel | wait | waitForSignal | end 之一，转移由 on.ok/fail/signal/timeout/cancel 声明。
• 步骤是函数：flow.step(name, fn)；fn(ctx, { input, prev }) 返回的对象成为下一步的 prev。
• 转移即提交：每次转移是单条 CAS（state_version）+ 一行 history，同事务；崩溃只可能丢「未提交的那一步」。

配方

幂等键：至少一次下的安全重放

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const charged = new Map() // 假装是支付网关
const flow = createMemoryWorkflow()
flow.step('charge', async (ctx) => {
  const key = `${ctx.run.id}:${ctx.step}` // 崩溃重放时同一个 key：网关侧去重
  charged.set(key, (charged.get(key) ?? 0) + 1)
  return { charged: charged.get(key) }
})
flow.define('pay', { initial: 's', states: { s: { step: 'charge', on: { ok: 'done' } }, done: { end: true } } })

const first = await flow.start('pay', {}, { idempotencyKey: 'order-42' })
const retry = await flow.start('pay', {}, { idempotencyKey: 'order-42' }) // 双保险之一：入口去重
assert.equal(retry.id, first.id)
await flow.drain()
assert.equal(charged.size, 1) // 双保险之二：步骤内幂等键

Saga：失败走补偿，终态可查

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const calls: string[] = []
const flow = createMemoryWorkflow()
flow.step('charge', async () => (calls.push('charge'), {}))
flow.step('ship', async () => {
  calls.push('ship')
  throw new Error('仓库罢工')
})
flow.step('refund', async () => (calls.push('refund'), {}))
flow.define('order', {
  initial: 'pay',
  states: {
    pay: { step: 'charge', on: { ok: 'deliver', fail: 'refund' } },
    deliver: { step: 'ship', on: { ok: 'done', fail: 'refund' } },
    refund: { step: 'refund', on: { ok: 'aborted' } },
    done: { end: true },
    aborted: { end: 'failed' },
  },
})

const run = await flow.start('order', {})
await flow.drain()
assert.deepEqual(calls, ['charge', 'ship', 'refund']) // 补偿恰好一次
assert.equal((await flow.get(run.id))?.status, 'failed')
assert.deepEqual(
  (await flow.history(run.id)).map((h) => h.action),
  ['ok', 'fail', 'ok', 'end'], // 顺序在 history 可查
)

人工审批：跨天等待不占进程

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const flow = createMemoryWorkflow()
flow.step('record', async (_ctx, { prev }) => ({ approvedBy: (prev as { by: string }).by }))
flow.define('expense', {
  initial: 'wait',
  states: {
    wait: { waitForSignal: { name: 'approve', timeoutMs: 7 * 24 * 3600 * 1000 }, on: { signal: 'save', timeout: 'expire' } },
    save: { step: 'record', on: { ok: 'done' } },
    expire: { end: 'failed' },
    done: { end: true },
  },
})

const run = await flow.start('expense', {})
await flow.signal(run.id, 'approve', { by: 'boss' })
await flow.drain()
assert.deepEqual((await flow.get(run.id))?.output, { approvedBy: 'boss' })

取消与超时：协作式、可审计

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const flow = createMemoryWorkflow()
flow.step('slow', async () => new Promise((resolve) => setTimeout(resolve, 60)))
flow.define('long', {
  initial: 'wait',
  states: { wait: { wait: { ms: 86_400_000 }, on: { ok: 'work' } }, work: { step: 'slow', on: { ok: 'done' } }, done: { end: true } },
})

const waiting = await flow.start('long', {})
assert.deepEqual(await flow.cancel(waiting.id), { cancelled: true, status: 'cancelled' }) // 等待中：立即终态
assert.deepEqual((await flow.history(waiting.id)).map((h) => h.action), ['cancel'])

const timed = await flow.start('long', {}, { timeoutMs: 30 })
await new Promise((resolve) => setTimeout(resolve, 50)) // 等待中也受 deadline 管：过期即被领取终结
await flow.drain()
assert.match((await flow.get(timed.id))?.error ?? '', /TIMEOUT/) // 运行级超时：不落 ok

并行与 map：有界并发、确定性 join

import assert from 'node:assert/strict'
import { createMemoryWorkflow } from '@weifuwujs/workflow/memory'

const flow = createMemoryWorkflow()
flow.step('double', async (_ctx, { prev }) => ({ value: (prev as { value: number }).value * 2 }))
flow.step('ok', async () => ({}))
flow.step('boom', async () => {
  throw new Error('分支炸了')
})

flow.define('map-fan', {
  initial: 'work',
  states: {
    work: { map: { step: 'double', over: 'items', concurrency: 2 }, on: { ok: 'done' } },
    done: { end: true },
  },
})
const mapRun = await flow.start('map-fan', { items: [{ value: 1 }, { value: 2 }, { value: 3 }] })
await flow.drain()
const mapOutput = (await flow.get(mapRun.id))?.output
assert.deepEqual((mapOutput!.results as Array<{ output: { value: number } }>).map((item) => item.output.value), [2, 4, 6]) // join 按输入序

flow.define('par', {
  initial: 'work',
  states: {
    work: {
      parallel: {
        branches: [
          { name: 'a', step: 'ok' },
          { name: 'b', step: 'boom' },
          { name: 'c', step: 'ok' },
        ],
        onFailure: 'collect',
      },
      on: { ok: 'done' },
    },
    done: { end: true },
  },
})
const parRun = await flow.start('par', {})
await flow.drain()
const parOutput = (await flow.get(parRun.id))?.output
assert.equal((parOutput as { succeeded: number }).succeeded, 2) // collect：部分失败不中断
assert.equal((parOutput as { failed: number }).failed, 1)
assert.deepEqual((parOutput!.results as Array<{ name: string }>).map((item) => item.name), ['a', 'b', 'c']) // 分支按声明序

运维

await flow.runs({ status: 'running', limit: 50 }) // 找卡住/在飞的 run
await flow.history(runId)                        // 每个 attempt 一行；取消/超时/补偿动作都在
await flow.retry(runId)                          // 死 run **绝不自动复活**，只能显式重排
await flow.cancel(runId)                         // 协作式：执行中的步骤经 ctx.signal 收到 abort

flow.runs({ status: 'running' }) 里 deadlineAt 已过的 run 会在下次领取时被按超时处置。

运行数据生命周期（1.3）：清理与导出

_runs/_history/_signals 只增不减。清理不自动跑（没有后台定时器），必须由 cron/运维显式触发； 且只删终态 run（done|failed|cancelled）——活跃 run（running，含等待信号/未推进）永远不删：

await flow.purgeRuns({ olderThan, now, dryRun: true })   // 预览：{ matched, deleted: 0, activeProtected }
await flow.purgeRuns({ olderThan, now, dryRun: false })  // 幂等：二次执行为 { matched: 0, deleted: 0 }
await flow.purgeRuns({ keepLatestPerName: 100, now, dryRun: false }) // 每个流程保留最近 100 个终态 run

weifuwu-workflow purge --older-than-ms 2592000000 --dry-run   # stdout 纯 JSON，可 | jq
weifuwu-workflow purge --keep-latest-per-name 100             # 按流程名配额（不跨流程）
weifuwu-workflow export <id>                                   # 快照：run + history + signals（纯读）

语义（三条都要读）：

1. 活跃免疫写在删除语句里（status in ('done','failed','cancelled') + RETURNING 真删数）： 即使某 failed run 在清理瞬间被 retry 排回 running，也不会被删；
2. olderThan 必须早于 now（--older-than-ms 0 会被显式拒绝），基于 updated_at 严格早于；
3. exportRun(id) 是纯读快照：不迁移、不 touch updated_at，未知 id 返回 null（CLI exit 1）； 删 run 会级联删它的 history/signals——所以「先导出再清理」是排障取证的正确顺序。

删除用到 {prefix}_runs_purge_idx (status, updated_at)；迁移是差集语义（_migrations 里已应用语句的并集 vs 目标 DDL）， 所以存量库下次 migrate() 会自动补齐新增索引（有回归测试在案）。

保证、不承诺与非目标

有保证（测试证据见 test/，每个文件对应一组语义）：

• 至少一次：未提交步骤会重放（spike.e2e、engine.e2e）；用幂等键消化重放。
• 转移恰好一次：CAS + history 同事务，双 worker 竞争只有一个赢（spike.e2e S2、retry.e2e）。
• 取消可审计：取消是普通转移，history 有 cancel 行；已提交步骤不回滚（cancel.e2e）。
• 超时不半提交：检查点式超时，绝不落 ok；on.timeout 优先（timeout.e2e）。
• 版本固定：run 钉住启动时的 flow_version，中途注册新版本不漂移（version.e2e）。

不承诺（诚实边界）：

• 不硬杀：悬挂中的步骤不会被强制中断；取消与超时只在领取点/转移点生效（协作式 ctx.signal 供步骤自行退出）。
• 跨进程取消不即时：别的进程发起的取消，本进程步骤只能在转移点感知（cancel_requested_at 由运行方分流）。
• 不做跨版本迁移：run 不会从旧 flow_version 迁到新版本；要改行为就发新流程即可。
• 版本化只固定定义数据（状态图/等待/重试/并行定义），不固定 step 函数实现：同名 step 跨版本共享，函数必须向后兼容。
• 无 UI、无 Temporal 式确定性重放、无跨库事务。

运维：HTTP 与 CLI

HTTP 交付面默认拒绝一切（缺 authorize → 全部数据端点 403）：

四包默认拒绝语义的对照表见 docs/authorization.md（可执行契约）。

import { createApp } from '@weifuwujs/weifuwu'
import { workflowApi } from '@weifuwujs/workflow'

const app = createApp().mount(
  workflowApi(flow, {
    basePath: '/wf',
    authorize: (ctx, { action }) => ctx['user']?.admin === true, // read/start/signal/retry/cancel
  }),
  { openapi: '/openapi.json' }, // 内核注册；文档不含业务数据
)

错误映射（稳定契约）：400 INVALID_ARGUMENT、403 FORBIDDEN、404 RUN_NOT_FOUND/UNKNOWN_FLOW、409 SIGNAL_DUPLICATE/NOT_EXPECTED/EXPIRED。

CLI（配置默认 weifuwu-workflow.config.ts，默认导出 Workflow 或其工厂；stdout 纯 JSON，诊断 CODE: 于 stderr）：

weifuwu-workflow migrate
weifuwu-workflow runs --status running        # 卡死自查：输出含 deadlineAt / leasedUntil / updatedAt
weifuwu-workflow start daily-report --input '{"date":"2026-10-05"}' --timeout-ms 3600000
weifuwu-workflow worker --drain               # 推进队列（--once 只领一个）
weifuwu-workflow get <id> | history <id>
weifuwu-workflow signal <id> approve --payload '{"by":"boss"}'
weifuwu-workflow retry <id> | cancel <id>
weifuwu-workflow export <id>                  # 运行数据快照（run+history+signals）
weifuwu-workflow purge [--older-than-ms <n>] [--keep-latest-per-name <n>] [--dry-run]

CLI 退出码（1.3 起全仓约定）：0 成功、1 运行时错误（如 RUN_NOT_FOUND/CONFIG_NOT_FOUND）、 2 用法错误（未知命令/缺参数/坏 flag，全部在加载配置之前判定）。

配置为 ESM：weifuwu-workflow.config.ts 用 Node 原生类型剥离（项目需 "type": "module"），或直接写 weifuwu-workflow.config.mjs。

同一份 config 驱动 HTTP 与 CLI——两条入口读到的 runs/get/history 逐字一致（真库双入口测试在案）。

开发

npm test                          # 包内全部（含真库 e2e；无库时明示 skip，绝不假绿）
WORKFLOW_TEST_NO_DB=1 npm test    # 无数据库：契约套件（memory 驱动）+ 单元照跑，e2e 显式跳过
docker compose up -d              # 真库（仓库根目录）

memory 驱动同样支持 parallel / map（与 Postgres 共享同一份纯逻辑核心，契约套件双驱动对照）；差异只在持久性：memory 是单进程、进程退即丢。