后端邮递

基于 PostgreSQL 对工作流程管弦乐器的后端接口实施.

特征

• 事件判定: 在 PostgreSQL 中只保存不可移动附件的事件
• 自动 Schema 初始化: 第一次连接时自动运行的异能迁移
• 种族安全步骤要求: 声称使用原子步骤 SELECT FOR UPDATE SKIP LOCKED
• 事件- 队列模式: 事件产生的队列状态( 没有单独的队列表)
• JSONB 存储器: 有 JSONB 列的弹性计划演变
• 完整类型脚本支持: 在所有业务中完全类型安全
• 易用数据库迁移页:1 pg 隔离在 db.ts 方便的图书馆交换

安装

bun add backend-postgres pg
bun add -d @types/pg

快速启动

基本用途

import { PostgresBackend } from "backend-postgres";

// Create backend with connection string
const backend = new PostgresBackend("postgres://user:password@localhost:5432/workflow_db");

// The database schema is automatically initialized on first use
// No manual migration commands needed!

// Use with the runner
import { runAll } from "runner";
const result = await runAll(workflow, input, { backend });

// Don't forget to close the connection pool when done
await backend.close();

带CLI

您可以通过设置 CLI 使用 Postgres 后端 POSTGRES_URL 环境变量 :

# Set the Postgres connection string
export POSTGRES_URL="postgres://user:password@localhost:5432/workflow_db"

# If POSTGRES_URL uses transaction-pooled PgBouncer, use a direct connection
# for schema migrations and concurrent index creation.
export POSTGRES_MIGRATIONS_URL="postgres://user:password@direct-host:5432/workflow_db"

# Run workflows
wfo run my-workflow --input '{"key": "value"}'

# Queue-based execution
wfo worker start  # Uses Postgres backend
wfo submit my-workflow --input '{...}'

程序用途

import { PostgresBackend } from "backend-postgres";
import { Client } from "client";

const backend = new PostgresBackend(process.env.POSTGRES_URL!);
const client = new Client({ backend });

// Submit a workflow run
const runId = await client.submit({
  workflowSlug: "my-workflow",
  input: { message: "Hello" },
  tags: ["production"],
  idempotencyKey: "unique-key-123", // Optional deduplication
});

// Wait for completion
const result = await client.waitForCompletion(runId);
console.log("Output:", result.output);

// Cleanup
await backend.close();

当直接用交易集合的运行时间 URL构建后端时, 将直接迁移 URL 作为第四个构造参数 :

const backend = new PostgresBackend(
  process.env.POSTGRES_URL!,
  "cascadeflow",
  undefined,
  process.env.POSTGRES_MIGRATIONS_URL
);

连接字符串格式

连接字符串遵循标准 PostgreSQL 格式 :

postgres://[user[:password]@][host][:port][/database][?option=value]

实例:

# Local development
postgres://postgres:postgres@localhost:5432/workflow_dev

# With SSL
postgres://user:[email protected]:5432/workflows?sslmode=require

# Unix socket
postgres://user@/dbname?host=/var/run/postgresql

# Cloud providers (example: Neon, Supabase, etc.)
postgres://user:[email protected]/dbname?sslmode=require

数据库Schema

后端创建和管理以下表格:

事件表

workflow_events - 工作流程一级的活动

• id (序列) - 主密钥
• event_id (文本) - 微秒印字符串
• workflow_slug (文本) - 工作流程标识符
• run_id (文本) - 运行标识符
• timestamp_us (bigint) - 事件时间戳以微秒计
• category (文本) - 总是“ 工作流程 ”
• type (text) - 事件类型(WorkflowStarted,Workflow完成等)
• data (jsonb) - 全部事件有效载荷

step_events - 分级活动

• id (序列) - 主密钥
• event_id (文本) - 微秒印字符串
• workflow_slug (文本) - 工作流程标识符
• run_id (文本) - 运行标识符
• step_id (正文) - 步骤标识符
• timestamp_us (bigint) - 事件时间戳以微秒计
• category (文本) - 总是"步骤"
• type (正文) - 事件类型(已定下时间表,StepStarted等)
• data (jsonb) - 全部事件有效载荷

登记表

workflow_metadata - 工作流程登记册

• slug (正文) - 主密钥
• name (文本) - 显示名称
• description (文本) - 可选说明
• input_schema_json (jsonb) - JSON Schema 用于输入验证
• tags (text[]) - 标记数组

step_definitions - 步骤定义

• workflow_slug (文本) - 参考资料工作流程 元数据(slug)
• id (正文) - 步骤标识符
• dependencies (jsonb) - 步骤依赖
• export_output (boolean) - 输出是否导出
• input_schema_json (jsonb) - JSON Schema 用于步骤输入
• timeout_ms (整数) - 步骤超时
• max_retries (整数) - 最大重试尝试
• retry_delay_ms (整数) - 复试之间的延迟

辅助表格

step_outputs - 顺序步骤产出

• workflow_slug, (中文(简体) ), run_id, (中文(简体) ), step_id, (中文(简体) ), attempt_number - 复合主要钥匙
• output (jsonb) - 步骤输出数据

idempotency_keys - 重复

• hash (正文) - 主键 (一能键的SHA256)
• run_id (文本) - 关联运行的ID

移徙战略

后端使用 一元移民 安全地重新运行。 在第一个数据库连接中自动执行迁移.

当前移徙情况

1. 移徙 001: 创建表格 - 用途 CREATE TABLE IF NOT EXISTS
2. 移徙 002: 创建索引 - 使用 CREATE INDEX IF NOT EXISTS

添加新迁移

增加一个新的迁移:

1. 添加一个新的迁移函数到 src/migrations.ts数字 :

async function migration003_addNewFeature(pool: Pool): Promise<void> {
  const client = await pool.connect();
  try {
    // Your migration code here (must be idempotent!)
    await client.query(`
      ALTER TABLE workflow_metadata
      ADD COLUMN IF NOT EXISTS new_field TEXT
    `);

    console.log("[Migration 003] Added new feature");
  } catch (error) {
    console.error("[Migration 003] Error:", error);
    throw error;
  } finally {
    client.release();
  }
}

2. 把它加进 runMigrations() 函数 :

export async function runMigrations(pool: Pool): Promise<void> {
  await migration001_createTables(pool);
  await migration002_createIndexes(pool);
  await migration003_addNewFeature(pool);  // Add here
}

最佳做法

所有迁移必须具有一能(安全运行多次):

✅ 不错 (专有):

CREATE TABLE IF NOT EXISTS my_table (...);
CREATE INDEX IF NOT EXISTS idx_name ON table (column);
ALTER TABLE table ADD COLUMN IF NOT EXISTS new_col TEXT;

❌ 坏东西 (非专有):

CREATE TABLE my_table (...);  -- Fails if table exists
ALTER TABLE table ADD COLUMN new_col TEXT;  -- Fails if column exists

结构细节

事件判定

所有状态变化被存储为不可改变的事件 :

// Events are never updated or deleted
await backend.appendEvent(workflowSlug, runId, {
  category: "step",
  type: "StepStarted",
  eventId: "1234567890123456",
  timestampUs: 1234567890123456,
  stepId: "my-step",
  workerId: "worker-1",
  // ... other event-specific fields
});

// Current state is computed by projecting events
const events = await backend.loadEvents(workflowSlug, runId);
const currentState = projectRunStateFromEvents(events, workflowSlug);

事件- 队列模式

没有单独的"排行"表. 队列状态通过投影生成 RunSubmitted 活动与工作流程活动:

// Submitting creates a RunSubmitted event
const runId = await backend.submitRun({
  workflowSlug: "my-workflow",
  input: { data: "..." },
});

// Listing runs projects events to compute status
const runs = await backend.listRuns({
  statuses: ["pending", "running"],
});

原子步骤要求

步称使用 PostgreSQL SELECT FOR UPDATE SKIP LOCKED 种族安全货币:

// Multiple workers can safely claim different steps
const claimed = await backend.claimScheduledStep(
  workflowSlug,
  runId,
  stepId,
  "worker-1"
);

// Only one worker succeeds per step
console.log(claimed); // true or false

执行(简化):

BEGIN;

-- Acquire lock (SKIP LOCKED prevents blocking)
SELECT * FROM step_events
WHERE workflow_slug = $1 AND run_id = $2 AND step_id = $3
ORDER BY timestamp_us DESC
LIMIT 1
FOR UPDATE SKIP LOCKED;

-- Verify step is in scheduled state
-- Write StepStarted event

COMMIT;

时间精确度

使用所有时间戳 微秒精度 (单位:千美元)timestampUs 字段),用于同时执行中正确事件排序:

// Example: 1762916097401523 (microseconds since epoch)
const timestamp = getMicrosecondTimestamp();

测试

整合测试

集成测试需要一个运行中的 PostgreSQL 数据库 :

# Set test database URL
export POSTGRES_TEST_URL="postgres://postgres:postgres@localhost:5432/workflow_test"

# Run tests
bun test

# Run specific test file
bun test tests/integration/postgres-backend.integration.test.ts

用于测试的嵌入器 PostgreSQL

# Start PostgreSQL in Docker
docker run -d \
  --name postgres-test \
  -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=workflow_test \
  -p 5432:5432 \
  postgres:16

# Run tests
export POSTGRES_TEST_URL="postgres://postgres:postgres@localhost:5432/workflow_test"
bun test

# Cleanup
docker stop postgres-test
docker rm postgres-test

业绩考量

索引

后端为常见查询模式创建优化的索引 :

• 事件查询 : (workflow_slug, run_id, timestamp_us)
• 台阶 : (workflow_slug, run_id, step_id, type, timestamp_us)
• 书记官处查询: (workflow_slug)

连接集合

后端使用 pg.Pool 为了高效的连接管理。 通过连接字符串配置池大小 :

postgres://user:pass@host/db?max=20&min=5&idle_timeout=10000

JSONB 存储器

JSONB列允许灵活的计划进化,但有存储/平滑取舍:

• * 没有迁移的灵活计划变化
• · 如果需要,可编制特定 JSONB 字段索引
• * 储存量略大于普通表格
• QQ 嵌入式 JSONB 的复杂查询可能较慢

对于高产量生产用途,考虑在经常点火的JSONB领域增加指数:

CREATE INDEX idx_event_type ON workflow_events ((data->>'type'));

切换数据库库

全体 pg 用户在 src/db.ts,使得可以轻松地交换到另一个图书馆:

1. 安装新库( 例如 ), postgres, (中文(简体) ), kysely, (中文(简体) ), drizzle(中文(简体) )
2. 更新 src/db.ts 以使用新库的 API
3. 保持函数签名不变
4. 更新 src/migrations.ts 必要时

这个 PostgresBackend 类 src/index.ts 不需要改变.

与文件系统备份的比较

解决问题

连接问题

// Error: connection refused
// Fix: Check PostgreSQL is running and connection string is correct

// Error: authentication failed
// Fix: Verify username/password in connection string

// Error: database does not exist
// Fix: Create database first: createdb workflow_db

移徙失败

// Error: relation already exists
// Fix: Ensure migrations use IF NOT EXISTS

// Error: permission denied
// Fix: Grant required permissions to database user

业绩问题

# Check for missing indexes
EXPLAIN ANALYZE SELECT ...;

# Monitor connection pool
SELECT * FROM pg_stat_activity WHERE datname = 'workflow_db';

# Check table sizes
SELECT pg_size_pretty(pg_total_relation_size('step_events'));

另见

• 主 CLAUDE.md (英语) - 后端架构概览
• 后端接口 - 抽象后端合同
• 后端文件系统 - 文件系统实施
• 工人读取 - 分布式工人建筑
• 客户端 README - 程序化API的使用