# 機能設計書 94-ワークフローオーケストレーション

## 概要

本ドキュメントは、Medusaコマースプラットフォームにおけるワークフローオーケストレーション機能の設計仕様を記述する。ワークフローオーケストレーション機能は、複数ステップで構成されるワークフローの調整・監視・イベント通知を行う高度な制御機能を提供する。

### 本機能の処理概要

ワークフローオーケストレーション機能は、ワークフロー実行の各ステップを調整し、ステップ間のデータフロー、エラーハンドリング、補償処理（compensation）の実行を管理する。Redis実装ではPub/Subを使用して複数インスタンス間でワークフロー状態を同期し、分散環境での一貫性を保証する。

**業務上の目的・背景**：大規模ECプラットフォームでは、ワークフローが複数のサーバーインスタンスにまたがって実行される場合がある。例えば、注文処理ワークフローの途中で支払い確認の非同期ステップを待機する際、別のインスタンスがその完了を処理する可能性がある。オーケストレーション機能により、このような分散環境でのワークフロー状態管理と、ライフサイクルイベントの適切な通知が可能になる。

**機能の利用シーン**：
- 非同期ステップを含むワークフローの状態管理
- ワークフロー完了/失敗時のリアルタイム通知
- 複数インスタンス間でのワークフロー状態同期
- ワークフローのタイムアウト検知と処理
- ステップ単位の進捗監視

**主要な処理内容**：
1. ワークフローライフサイクルイベントの構築と通知
2. 購読者へのイベント配信（Redis Pub/Sub）
3. 親ワークフローへのサブワークフロー完了通知
4. べき等キーの構築と管理
5. 分散環境でのワークフロー状態同期

**関連システム・外部連携**：
- Redis：Pub/Subによるイベント配信、状態管理
- PostgreSQL：ワークフロー実行履歴の永続化
- イベントバス：ワークフロー完了時のドメインイベント発行

**権限による制御**：オーケストレーション機能自体には権限制御は設けられていない。ワークフロー購読は信頼されたサービス間通信として設計されている。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 186 | ワークフロー実行詳細画面 | 補助機能 | ステップ状態の確認 |

## 機能種別

処理制御 / イベント管理 / 分散調整

## 入力仕様

### 入力パラメータ（subscribe）

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| workflowId | string | Yes | 購読するワークフローID | - |
| transactionId | string | No | 特定トランザクションのみ購読する場合 | - |
| subscriber | SubscriberHandler | Yes | イベント受信時のコールバック関数 | - |
| subscriberId | string | No | 購読者の識別子 | - |

### 入力パラメータ（SubscriberHandler引数 - NotifyOptions）

| パラメータ名 | 型 | 必須 | 説明 |
|-------------|-----|-----|------|
| eventType | keyof DistributedTransactionEvents | Yes | イベント種別 |
| isFlowAsync | boolean | Yes | 非同期フローか |
| workflowId | string | Yes | ワークフローID |
| transactionId | string | No | トランザクションID |
| step | TransactionStep | No | ステップ情報 |
| response | unknown | No | ステップレスポンス |
| result | unknown | No | ワークフロー結果 |
| errors | unknown[] | No | エラー情報 |
| state | TransactionState | No | トランザクション状態 |

### イベント種別一覧

| イベント種別 | 発生タイミング |
|------------|--------------|
| onBegin | ワークフロー開始時 |
| onResume | ワークフロー再開時 |
| onTimeout | タイムアウト時 |
| onFinish | ワークフロー完了時 |
| onCompensateBegin | 補償処理開始時 |
| onStepBegin | ステップ開始時 |
| onStepSuccess | ステップ成功時 |
| onStepFailure | ステップ失敗時 |
| onStepAwaiting | ステップ待機時 |
| onCompensateStepSuccess | 補償ステップ成功時 |
| onCompensateStepFailure | 補償ステップ失敗時 |

### 入力データソース

- ワークフロー実行エンジンからの自動呼び出し
- 管理画面からのワークフロー状態監視

## 出力仕様

### 出力データ（notify）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| (void) | void | 非同期で購読者に通知 |

### 出力データ（購読者コールバック）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| NotifyOptions | object | イベント詳細情報 |

### 出力先

- 登録された購読者関数（インプロセス）
- Redis Pub/Subチャネル（他インスタンス）

## 処理フロー

### 処理シーケンス

```
1. イベント発生（ワークフローエンジン内）
   └─ buildWorkflowEvents()で構築されたハンドラーが呼び出し

2. 通知処理（notify）
   ├─ Redis Pub/Subへの発行（非同期フローの場合）
   │      └─ チャネル名: orchestrator:{workflowId}
   │
   └─ ローカル購読者への通知（processSubscriberNotifications）

3. 購読者呼び出し
   ├─ transactionIdに対応する購読者を取得
   ├─ ワイルドカード購読者（AnySubscriber）を取得
   └─ 各購読者を並列実行

4. 完了時の購読者クリーンアップ
   └─ onFinishイベント時にtransactionId購読者を削除
```

### フローチャート

```mermaid
flowchart TD
    A[イベント発生] --> B{非同期フロー?}
    B -->|Yes| C[Redis Pub/Sub発行]
    B -->|No| D[ローカル通知のみ]
    C --> D
    D --> E[購読者リスト取得]
    E --> F{transactionId指定あり?}
    F -->|Yes| G[特定トランザクション購読者取得]
    F -->|No| H[ワイルドカード購読者取得]
    G --> I[購読者を並列呼び出し]
    H --> I
    I --> J{onFinishイベント?}
    J -->|Yes| K[transactionId購読者を削除]
    J -->|No| L[完了]
    K --> L
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-94-1 | Redisチャネル命名 | チャネル名はorchestrator:{workflowId}形式 | Redis Pub/Sub発行時 |
| BR-94-2 | 自インスタンス通知スキップ | Redis経由のメッセージは発行元インスタンスでは処理しない | instanceIdで判定 |
| BR-94-3 | 非同期フロー判定 | hasAsyncSteps がtrueの場合のみRedis発行 | notify()呼び出し時 |
| BR-94-4 | 購読者クリーンアップ | onFinishイベント時にtransactionId購読者を自動削除 | ワークフロー完了時 |
| BR-94-5 | ワイルドカード購読 | transactionIdなしで登録した購読者は全トランザクションを受信 | AnySubscriber定数で管理 |

### 計算ロジック

べき等キーの構築：
```typescript
// 文字列形式のべき等キー
idempotencyKey = `${workflowId}:${transactionId}:${stepId}:${action}`
// 例: "create-order:tx-001:validate-input:invoke"

// オブジェクト形式からの変換
buildIdempotencyKeyAndParts({
  workflowId: "create-order",
  transactionId: "tx-001",
  stepId: "validate-input",
  action: "invoke" // or "compensate"
})
```

## データベース操作仕様

### 操作別データベース影響一覧

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | オーケストレーション自体はDBを使用しない |

### 外部ストレージ操作（Redis）

| 操作 | キー/チャネル形式 | 操作種別 | 概要 |
|-----|---------|---------|------|
| イベント発行 | orchestrator:{workflowId} | PUBLISH | ワークフローイベントを発行 |
| イベント購読 | orchestrator:{workflowId} | SUBSCRIBE | ワークフローイベントを購読 |
| 購読解除 | orchestrator:{workflowId} | UNSUBSCRIBE | イベント購読を解除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | RedisPublishError | Redis発行失敗 | エラーログ出力、処理継続 |
| - | SubscriberError | 購読者内で例外発生 | エラーログ出力、他の購読者の処理継続 |
| - | JsonParseError | Redisメッセージのパース失敗 | エラーログ出力 |

### リトライ仕様

- オーケストレーション通知自体のリトライは行わない
- 購読者内でのリトライは購読者側の責任

## トランザクション仕様

オーケストレーション機能自体はトランザクション管理の対象外。イベント通知は非同期で行われ、通知失敗はワークフロー実行に影響しない。

## パフォーマンス要件

- イベント通知：非ブロッキング（setImmediate使用）
- Redis Pub/Sub：ミリ秒オーダー
- 購読者呼び出し：promiseAll()で並列実行

## セキュリティ考慮事項

- Redis通信はTLS暗号化を推奨（本番環境）
- ワークフロー状態情報に機密データを含めないこと
- 購読者登録はアプリケーション層で制御

## 備考

- instanceIdはulidで生成され、各サービスインスタンスを一意に識別
- WorkflowOrchestratorService.subscribersは静的メンバーとして全インスタンスで共有
- 非同期フローでない場合はRedis発行をスキップしてローカル通知のみ

---

## コードリーディングガイド

本機能を理解するために参照すべきファイルと、推奨する読み解き順序を以下に示す。

### 推奨読解順序

#### Step 1: データ構造を理解する

通知オプションと購読者の構造を理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | workflow-orchestrator.ts | `packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts` | NotifyOptions、SubscribeOptions、Subscribers型の定義 |

**読解のコツ**: Subscribersは`Map<WorkflowId, Map<TransactionId, SubscriberHandler[]>>`の構造で、ワークフローID→トランザクションID→購読者リストの階層になっている。

#### Step 2: イベントハンドラー構築を理解する

ワークフローライフサイクルイベントの構築ロジックを確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | workflow-orchestrator.ts | `packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts` | buildWorkflowEvents()メソッド |

**主要処理フロー**:
- **834-988行目**: buildWorkflowEvents()でライフサイクルイベントハンドラーを構築
- **869-876行目**: onBeginイベントハンドラー
- **899-908行目**: onFinishイベントハンドラー
- **910-931行目**: onStepSuccessイベントハンドラー（resolveValueで結果取得）
- **932-945行目**: onStepFailureイベントハンドラー

#### Step 3: 通知処理を理解する

イベント通知のメカニズムを確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | workflow-orchestrator.ts | `packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts` | notify()とprocessSubscriberNotifications()メソッド |

**主要処理フロー**:
- **758-787行目**: notify()メソッドで通知を処理
- **763-765行目**: 自インスタンスからのメッセージはスキップ
- **770-783行目**: 非同期フローの場合、setImmediateでRedis発行
- **786行目**: setImmediateでローカル購読者通知
- **789-828行目**: processSubscriberNotifications()で購読者を呼び出し

#### Step 4: 購読管理を理解する

購読者の登録と解除のロジックを確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | workflow-orchestrator.ts | `packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts` | subscribe()とunsubscribe()メソッド |

**主要処理フロー**:
- **663-706行目**: subscribe()で購読者を登録
- **674-676行目**: 初回登録時にRedisチャネルを購読
- **684-694行目**: transactionId指定時は特定トランザクションのみ購読
- **697-705行目**: transactionIdなしはワイルドカード購読（AnySubscriber）
- **708-756行目**: unsubscribe()で購読者を解除

### プログラム呼び出し階層図

```
WorkflowOrchestratorService
    │
    ├─ buildWorkflowEvents({ customEventHandlers, workflowId, transactionId })
    │      │
    │      └─ returns DistributedTransactionEvents
    │             ├─ onBegin() ───▶ notify({ eventType: "onBegin" })
    │             ├─ onResume() ───▶ notify({ eventType: "onResume" })
    │             ├─ onTimeout() ───▶ notify({ eventType: "onTimeout" })
    │             ├─ onFinish() ───▶ notify({ eventType: "onFinish" })
    │             ├─ onCompensateBegin() ───▶ notify()
    │             ├─ onStepBegin() ───▶ notify()
    │             ├─ onStepSuccess() ───▶ resolveValue() ───▶ notify()
    │             ├─ onStepFailure() ───▶ notify()
    │             ├─ onStepAwaiting() ───▶ notify()
    │             ├─ onCompensateStepSuccess() ───▶ notify()
    │             └─ onCompensateStepFailure() ───▶ notify()
    │
    ├─ notify(options, publish, instanceId)
    │      ├─ [非同期フロー && publish]
    │      │      └─ setImmediate()
    │      │             └─ redisPublisher.publish(channel, message)
    │      │
    │      └─ setImmediate()
    │             └─ processSubscriberNotifications(options)
    │
    ├─ processSubscriberNotifications(options)
    │      ├─ subscribers.get(workflowId)
    │      ├─ transactionSubscribers.get(transactionId)
    │      ├─ workflowSubscribers.get(AnySubscriber)
    │      ├─ promiseAll(handlers.map(invoke))
    │      └─ [onFinish] subscribers.delete(transactionId)
    │
    ├─ subscribe({ workflowId, transactionId, subscriber, subscriberId })
    │      ├─ subscriber._id = subscriberId
    │      ├─ [初回] redisSubscriber.subscribe(channel)
    │      └─ subscribers.set()
    │
    └─ unsubscribe({ workflowId, transactionId, subscriberOrId })
           ├─ filterSubscribers()
           └─ [購読者なし] redisSubscriber.unsubscribe(channel)
```

### データフロー図

```
[イベント発生]                [オーケストレーター]              [購読者]

ワークフロー実行
    │
    ▼
onStepSuccess() ────▶ buildWorkflowEvents()
                             │
                             ▼
                       notify()
                             │
                    ┌────────┴────────┐
                    │                 │
                    ▼                 ▼
           [非同期フロー]        [同期フロー]
                    │                 │
                    ▼                 │
           Redis PUBLISH              │
           "orchestrator:wf-1"        │
                    │                 │
                    ▼                 │
           他インスタンスの            │
           redisSubscriber.on()       │
                    │                 │
                    ▼                 ▼
           processSubscriberNotifications()
                             │
                    ┌────────┴────────┐
                    │                 │
                    ▼                 ▼
           [transactionId購読者]  [ワイルドカード購読者]
                    │                 │
                    ▼                 ▼
           handler(options)     handler(options)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| workflow-orchestrator.ts | `packages/modules/workflow-engine-redis/src/services/workflow-orchestrator.ts` | ソース | ワークフローオーケストレーター（Redis実装） |
| redis-distributed-transaction-storage.ts | `packages/modules/workflow-engine-redis/src/utils/redis-distributed-transaction-storage.ts` | ソース | 分散トランザクションストレージ |
| workflows-module.ts | `packages/modules/workflow-engine-inmemory/src/services/workflows-module.ts` | ソース | ワークフローモジュールサービス |
| orchestration | `packages/core/framework/src/orchestration/` | ソース | 分散トランザクションフレームワーク |
