upstash-workflow

Upstash Workflow 実装ガイド

LobeHub のコードベースで Upstash Workflow と QStash の非同期ワークフローを実装するための標準パターンです。

🎯 3 つの基本パターン

LobeHub のすべてのワークフローは、次の 3 つのパターンを組み合わせています。プラットフォームには 3 つの制約があるため、これらのパターンが必要です。レート制限により無計画なファンアウトは危険であり、ステップ数の上限により 1 つのワークフローの規模が制限され、冪等性を保つには再試行時に二重処理を防ぐ必要があります。

1. 🔍 ドライランモード — 実際の処理を開始せずに統計情報を取得します
2. 🌟 ファンアウトパターン — 大きなバッチを小さなチャンクに分割して並列処理します
3. 🎯 単一タスクの実行 — 各ワークフローの実行で処理する項目は必ず 1 つだけです

---

アーキテクチャの概要

すべてのワークフローは、同じ 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 を参照してください。

---

3 つのパターンを 60 秒で理解する

1. ドライランモード

副作用が発生する前にレイヤー 1 をショートサーキットさせ、呼び出し元が実行内容を事前に確認できるようにします。

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

用途：実行を確定する前に、処理対象となる項目数を確認します。

2. ファンアウトパターン

レイヤー 2 は大きすぎるバッチをチャンクに分割し、各チャンクを使って自身を再帰的に再実行します。これにより、1 ページに含まれる項目が多すぎる場合にワークフローのステップ数の上限に達するのを防ぎます。

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（1 ページあたりの項目数）、CHUNK_SIZE = 20（ファンアウトのチャンクあたりの項目数）。

3. 単一タスクの実行

レイヤー 3 は、呼び出しごとに必ず 1 つの項目だけを処理します。並列処理は、flowControl で制御しながら、レイヤー 2 が多数のレイヤー 3 の呼び出しにファンアウトすることで実現します。

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
2 つの実際のワークフローを最初から最後まで確認する	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

---

新しいワークフローのチェックリスト

計画

• 処理対象となるエンティティ（ユーザー、エージェント、項目など）を特定する
• 項目ごとのビジネスロジックを定義する
• フィルタリングロジック（Redis キャッシュ、データベースの状態など）を決定する

実装

• TypeScript のインターフェースでペイロードの型を定義する
• 静的なトリガーメソッドを備えたワークフロークラスを作成する
• レイヤー 1： ドライランに対応したエントリーポイント
• レイヤー 1： 重複処理を防ぐフィルタリングロジック
• レイヤー 2： ファンアウトを使ったページネーション
• レイヤー 3： 単一タスクの実行（実行ごとに項目は 1 つ）
• 各レイヤーに適切な flowControl を設定する
• ワークフローのプレフィックスを使って一貫性のあるロギングを行う
• 必須のペイロードパラメーターをすべて検証する
• context.run() のステップ名を一意にする

品質とデプロイ

• 一貫性のあるレスポンス形式を返す
• クラウドへのデプロイを設定する（lobehub-cloud を使用する場合は references/cloud.md を参照）
• 統合テストを作成する（dryRun の経路と通常の経路）
• まずドライランでスモークテストを行う
• 全体展開の前に小さなバッチでテストする

---

追加リソース

• Upstash Workflow ドキュメント
• QStash ドキュメント
• コードベース内のワークフローの例
• ワークフロークラス