workflow-management

工作流管理技能

此技能可帮助你在 apps/web/src/workflows/ 中使用 Vercel WDK（Workflow Development Kit）工作流。

适用场景

• 添加用于数据抓取的新定时工作流
• 调试工作流执行错误
• 修改现有工作流的计划或逻辑
• 将新的数据源集成到更新流水线中
• 添加新的社交媒体发布工作流

工作流架构

项目使用 Vercel WDK 工作流，结构如下：

apps/web/src/workflows/
├── cars.ts              # Car registration data workflow
├── coe.ts               # COE bidding data workflow
├── deregistrations.ts   # Deregistration data workflow
├── regenerate-post.ts   # Blog post regeneration workflow
└── shared.ts            # Shared workflow steps (social media, cache)

关键模式

1. 工作流定义

使用 Vercel WDK 指令定义工作流：

import { fetch } from "workflow";

export async function carsWorkflow(payload: CarsWorkflowPayload) {
  "use workflow";

  // Enable WDK's durable fetch for AI SDK
  globalThis.fetch = fetch;

  // Step 1: Process data
  const result = await processCarsData();

  // Step 2: Revalidate cache
  await revalidateCarsCache(month, year);

  // Step 3: Generate blog post
  const post = await generateCarsPost(data, month);

  return { postId: post.postId };
}

async function processCarsData(): Promise<UpdaterResult> {
  "use step";
  // Processing logic - each step is durable and can retry
  return await updateCars();
}

2. 工作流调度

通过在 apps/web/vercel.json 中配置的 Vercel Cron 计划触发工作流：

{
  "crons": [
    { "path": "/api/workflows/cars", "schedule": "0 10 * * *" },
    { "path": "/api/workflows/coe", "schedule": "0 10 * * *" },
    { "path": "/api/workflows/deregistrations", "schedule": "0 10 * * *" }
  ]
}

3. API 路由集成

API 路由使用 WDK 的 start() 函数触发工作流：

import { start } from "workflow/api";
import { carsWorkflow } from "@web/workflows/cars";

export async function POST(request: Request) {
  const payload = await request.json();
  const run = await start(carsWorkflow, [payload]);
  return Response.json({ message: "Workflow started", runId: run.runId });
}

4. 共享步骤

将常用操作提取到 shared.ts 中：

export async function publishToSocialMedia(title: string, link: string) {
  "use step";
  await socialMediaManager.publishToAll({ message: `📰 ${title}`, link });
}

export async function revalidatePostsCache() {
  "use step";
  revalidateTag("posts:list", "max");
}

5. 错误处理

步骤失败时会自动重试。自定义错误处理方式如下：

async function processData() {
  "use step";
  try {
    const result = await updateData();
    return result;
  } catch (error) {
    console.error("Step failed:", error);
    throw error; // Re-throw for automatic retry
  }
}

常见任务

添加新工作流

1. 在 apps/web/src/workflows/ 中创建工作流文件
2. 使用 "use workflow" 指令定义工作流函数
3. 使用 "use step" 指令将逻辑拆分为多个步骤
4. 在 apps/web/src/app/api/workflows/ 中创建 API 路由
5. 如有需要，在 vercel.json 中添加 Vercel Cron 计划
6. 为工作流逻辑添加测试

调试工作流故障

1. 在 Vercel 控制面板中查看工作流执行日志
2. 查看 Vercel 中的函数日志
3. 确认环境变量设置正确
4. 使用开发服务器在本地测试工作流
5. 检查数据库连接和 Redis 是否可用

修改现有工作流

1. 阅读现有的工作流实现
2. 确定需要修改的步骤
3. 在保留错误处理的同时更新步骤逻辑
4. 在本地测试更改
5. 部署并监控执行情况

环境变量

工作流通常需要：

• DATABASE_URL - PostgreSQL 连接
• UPSTASH_REDIS_REST_URL / UPSTASH_REDIS_REST_TOKEN - Redis
• 服务专用令牌（Discord webhook、Twitter API 等）

测试工作流

运行工作流测试：

pnpm -F @sgcarstrends/web test -- src/workflows

在本地测试单个工作流：

# Start dev server
pnpm dev

# Trigger workflow via HTTP
curl -X POST http://localhost:3000/api/workflows/cars

参考资料

• Vercel WDK 文档：https://vercel.com/docs/workflow-kit
• 相关文件：
  • apps/web/src/app/api/workflows/ - 工作流路由处理程序
  • apps/web/src/config/platforms.ts - 社交媒体配置
  • apps/web/vercel.json - Vercel Cron 计划
  • apps/web/AGENTS.md - Web 应用文档

最佳实践

1. 幂等性：确保工作流可安全重试，且不会重复生成数据
2. 步骤粒度：使用 "use step" 将工作流拆分为目标明确的小步骤
3. 持久化抓取：为 AI SDK 调用设置 globalThis.fetch = fetch
4. 日志记录：添加全面的日志，便于调试
5. 测试：为工作流逻辑编写单元测试
6. 监控：在 Vercel 控制面板中跟踪工作流执行情况