# @cascade-flow/backend-postgres 中文文档

规范地址：https://npmcn.com/package/%40cascade-flow/backend-postgres/

版本：0\.2\.31
npm 元数据更新时间：2026-08-11T22:08:37.672Z
中文译文更新时间：2026-11-09T12:20:20.772Z
软件包许可：MIT

本文由上游 README 翻译，可能存在偏差，请结合原文核对 API、命令与配置。版权与许可归原作者所有。

原始 npm 页面：https://www.npmjs.com/package/%40cascade-flow/backend-postgres
原文对照：https://npmcn.com/package/%40cascade-flow/backend-postgres/?lang=original
项目主页：https://github.com/cascadeflow/cascadeflow#readme
源码仓库：https://github.com/cascadeflow/cascadeflow

## 关键信息（来自 npm 元数据）

- 包体积（解压后）：873 kB，10 个文件
- 模块格式：ES Module
- TypeScript 类型：包内自带
- 运行依赖：2 个
- peer 依赖：@cascade-flow/backend-interface ^0\.2\.17，typescript ^5
- 首次发布：2025-11-20T10:08:30.511Z
- 最新版本发布：2026-08-11T22:08:37.521Z
- 近 30 天下载量：2787（2026-10-09 至 2026-11-07，npm 官方下载统计）
- 已发布版本数：31（近 12 个月 31 个）

## 中文简介

用于 CascadeFlow 工作流程调谐器的 PostgreSQL 后端执行

## 安装

```sh
npm install @cascade-flow/backend-postgres
```

也可以使用：`pnpm add @cascade-flow/backend-postgres`、`yarn add @cascade-flow/backend-postgres`、`bun add @cascade-flow/backend-postgres`

## README

# 后端邮递

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

## 特征

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

## 安装

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

## 快速启动

### 基本用途

```typescript
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` 环境变量 :

```bash
# 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 '{...}'
```

### 程序用途

```typescript
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 作为第四个构造参数 :

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

## 连接字符串格式

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

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

实例:

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

# With SSL
postgres://user:pass@db.example.com:5432/workflows?sslmode=require

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

# Cloud providers (example: Neon, Supabase, etc.)
postgres://user:password@ep-xyz.region.provider.com/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`数字 :

```typescript
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()` 函数 :

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

### 最佳做法

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

✅ **不错** (专有):

```sql
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;
```

❌ **坏东西** (非专有):

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

## 结构细节

### 事件判定

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

```typescript
// 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` 活动与工作流程活动:

```typescript
// 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` 种族安全货币:

```typescript
// 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
```

执行(简化):

```sql
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` 字段),用于同时执行中正确事件排序:

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

## 测试

### 整合测试

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

```bash
# 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

```bash
# 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领域增加指数:

```sql
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` 不需要改变.

## 与文件系统备份的比较

| 特性 | 文件系统备份 | 邮递端 |
| --- | --- | --- |
| 存储 | 本地文件系统 | PostgreSQL 数据库 |
| 货币 | 文件锁定 | 数据库交易 |
| 可缩放性 | 单机 | 多机组群 |
| 持久性 | 当地 `.runs/` 目录 | 远程数据库 |
| 设置 | 零配置 | 需要 PostgreSQL 服务器 |
| 最好还是 | 开发,单机 | 生产、分配工人 |

## 解决问题

### 连接问题

```typescript
// 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
```

### 移徙失败

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

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

### 业绩问题

```bash
# 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 (英语)](https://npmcn.com/package/CLAUDE.md) - 后端架构概览
-   [后端接口](https://npmcn.com/package/%40cascade-flow/backend-interface/README.md) - 抽象后端合同
-   [后端文件系统](https://npmcn.com/package/%40cascade-flow/backend-filesystem/README.md) - 文件系统实施
-   [工人读取](https://npmcn.com/package/%40cascade-flow/worker/README.md) - 分布式工人建筑
-   [客户端 README](https://npmcn.com/package/%40cascade-flow/client/README.md) - 程序化API的使用

## 常见问题

### @cascade-flow/backend-postgres 是什么？

用于 CascadeFlow 工作流程调谐器的 PostgreSQL 后端执行

### 如何安装 @cascade-flow/backend-postgres？

在项目目录运行 npm install @cascade-flow/backend-postgres，或使用 pnpm add / yarn add / bun add。它还需要宿主项目提供 peer 依赖：@cascade-flow/backend-interface（^0\.2\.17）、typescript（^5）。npm 7 及以上与 pnpm 8 及以上默认会自动安装必需的 peer 依赖，Yarn 需要手动安装。

### @cascade-flow/backend-postgres 支持 TypeScript 吗？

支持。包内自带 TypeScript 类型声明（package\.json 声明了 types 入口），安装后即可获得类型提示，无需再安装 @types 包。

### @cascade-flow/backend-postgres 支持 ESM 还是 CommonJS？

以 ES Module 形式发布，推荐使用 import 引入。在 CommonJS 项目中，Node\.js 20\.19\+ / 22\.12\+ 可以直接 require\(\) 不含顶层 await 的 ES Module，更早的版本需要改用动态 import\(\)。

### @cascade-flow/backend-postgres 可以商用吗？

@cascade-flow/backend-postgres 使用 MIT 许可。宽松许可：允许商业使用、修改和再分发，通常只需在分发时保留原版权与许可声明。以上为概括，不构成法律意见，请以许可原文为准。

### @cascade-flow/backend-postgres 的下载量有多少？

根据 npm 官方下载统计，@cascade-flow/backend-postgres 在 2026/10/09 – 2026/11/07 这 30 天内被下载 2,787 次，日均约 93 次。下载量会把 CI 构建、镜像同步和作为其他包依赖的安装一并计入，适合横向比较热度，不能直接等同于使用人数。

### @cascade-flow/backend-postgres 还在维护吗？

最新版本 0\.2\.31 发布于 2026/08/12（2 个月前）；至今共发布 31 个版本，其中过去 12 个月发布了 31 个；首次发布于 2025/11/20。发布频率只是参考信号：功能稳定的工具可能很少需要更新，建议结合源码仓库的 issue 和提交记录判断。