upstash-workflow

Upstash Workflow 实现指南

在 LobeHub 代码库中实现 Upstash Workflow + QStash 异步工作流的标准模式。

🎯 三种核心模式

LobeHub 中的每个工作流都结合了这三种模式。它们之所以存在，是因为平台有三方面的限制：速率限制意味着盲目扇出存在风险，步骤数量限制了单个工作流的规模，而幂等性要求重试时不能重复处理。

1. 🔍 Dry-Run 模式 — 获取统计信息，但不触发实际执行
2. 🌟 扇出模式 — 将大型批次拆分为较小的块，以便并行处理
3. 🎯 单任务执行 — 每次工作流执行恰好处理一项

---

架构概览

所有工作流都遵循相同的三层架构：

Layer 1: Entry Point (process-*)
  ├─ Validates prerequisites
  ├─ Calculates total items to process
  ├─ Filters existing items
  ├─ Supports dry-run mode (statistics only)
  └─ Triggers Layer 2 if work is needed

Layer 2: Pagination (paginate-*)
  ├─ Handles cursor-based pagination
  ├─ Implements fan-out for large batches
  ├─ Recursively processes all pages
  └─ Triggers Layer 3 for each item

Layer 3: Single Task Execution (execute-* / generate-*)
  └─ Performs actual business logic for ONE item

此代码库中的实际示例： topicAutoSummary、agentEvalRun — 请参阅 references/examples.md。

---

60 秒了解三种模式

1. Dry-Run 模式

在产生任何副作用之前，于第 1 层提前返回，让调用方预览将要发生的操作：

if (dryRun) {
  return {
    ...result,
    dryRun: true,
    message: `[DryRun] Would process ${itemsNeedingProcessing.length} items`,
  };
}

适用场景：在确认执行前检查将处理多少项。

2. 扇出模式

第 2 层将超大批次拆分为多个块，并使用每个块递归地重新触发自身。当单页包含过多项目时，这可以避免触及工作流步骤数量限制：

const CHUNK_SIZE = 20;

if (itemIds.length > CHUNK_SIZE) {
  const chunks = chunk(itemIds, CHUNK_SIZE);
  await Promise.all(
    chunks.map((ids, idx) =>
      context.run(`workflow:fanout:${idx + 1}/${chunks.length}`, () =>
        WorkflowClass.triggerPaginateItems({ itemIds: ids }),
      ),
    ),
  );
}

默认值：PAGE_SIZE = 50（每页的项目数），CHUNK_SIZE = 20（每个扇出块的项目数）。

3. 单任务执行

第 3 层每次调用始终恰好处理一项。并行处理由第 2 层扇出并调用多个第 3 层实例实现，并通过 flowControl 控制：

app.post(
  '/execute-item',
  serve<ExecutePayload>(
    async (context) => {
      const { itemId } = context.requestPayload ?? {};
      if (!itemId) return { success: false, error: 'Missing itemId' };

      const item = await context.run('workflow:get-item', () => getItem(itemId));
      const result = await context.run('workflow:execute', () => processItem(item));
      await context.run('workflow:save', () => saveResult(itemId, result));

      return { success: true, itemId, result };
    },
    {
      flowControl: { key: 'workflow.execute', parallelism: 10, ratePerSecond: 5 },
    },
  ),
);

---

文件结构

src/app/(backend)/api/workflows/
└── [[...route]]/route.ts             # Single catch-all — forwards every request to the Hono app below

apps/server/src/router-hono/workflows/
├── index.ts                          # Mounts each workflow's Hono app at /api/workflows/{workflow-name}
└── {workflow-name}/
    ├── index.ts                       # Hono app — one `app.post('/{layer}', serve(handler, options))` per layer
    ├── dispatch.ts / paginate-*.ts    # Layer 1 + 2 handler(s) — entry point, dry-run, pagination, fan-out
    └── execute.ts / execute-*.ts      # Layer 3 handler — single-task execution

apps/server/src/workflows/
└── {workflowName}/
    └── index.ts                       # Workflow class — static trigger*() methods that POST to the routes above

每一层都是挂载在 apps/server/src/router-hono/workflows/ 下、按工作流划分的 Hono 应用上的处理器；src/app 下的 Next.js 路由只负责向其分发请求。

---

后续阅读

选择与你当前任务对应的参考文档：

你想要……	阅读
从头编写 Workflow 类和 3 个路由	references/implementation.md
调整 flowControl、错误处理、日志和测试	references/best-practices.md
查看两个完整的实际工作流	references/examples.md
在 lobehub-cloud 上部署（重新导出、仅限云端的操作）	references/cloud.md

---

环境变量

# Required for all workflows
APP_URL=https://your-app.com # Base URL for workflow endpoints
QSTASH_TOKEN=qstash_xxx      # QStash authentication token

# Optional (for custom QStash URL)
QSTASH_URL=https://custom-qstash.com

---

新建工作流检查清单

规划

• 确定要处理的实体（用户、Agent、项目等）
• 定义每项的业务逻辑
• 确定筛选逻辑（Redis 缓存、数据库状态等）

实现

• 使用 TypeScript 接口定义负载类型
• 创建包含静态触发方法的工作流类
• 第 1 层： 支持 dry-run 的入口
• 第 1 层： 避免重复处理的筛选逻辑
• 第 2 层： 使用 扇出 进行分页
• 第 3 层： 单任务执行（每次运行处理一项）
• 为每一层配置适当的 flowControl
• 使用工作流前缀保持日志一致
• 验证所有必需的负载参数
• 确保 context.run() 步骤名称唯一

质量与部署

• 返回结构一致的响应
• 配置云端部署（如果使用 lobehub-cloud，请参阅 references/cloud.md）
• 编写集成测试（dryRun 路径和完整路径）
• 先使用 dry-run 进行冒烟测试
• 全面上线前先使用小批次进行测试

其他资源

• Upstash Workflow 文档
• QStash 文档
• 代码库中的工作流示例
• 工作流类