# 機能設計書 92-イベント購読

## 概要

本ドキュメントは、Medusaコマースプラットフォームにおけるイベント購読機能の設計仕様を記述する。イベント購読機能は、システム内で発行されたドメインイベントを受信し、対応する処理を実行するためのサブスクライバー管理機能を提供する。

### 本機能の処理概要

イベント購読機能は、イベントバスに対して購読者（サブスクライバー）を登録し、特定のイベントが発行された際に自動的に呼び出されるコールバック関数を管理する。Redis実装では、BullMQのWorkerを使用してキューからイベントを取得し、登録された購読者に配信する。

**業務上の目的・背景**：ECプラットフォームでは、様々なビジネスイベントに対してリアクティブな処理を行う必要がある。例えば、注文作成イベントに対して在庫の確保、メール通知、外部システムへの連携など、複数の独立した処理を実行する。イベント購読機能により、これらの処理を疎結合に実装でき、新しい処理の追加や既存処理の修正が容易になる。

**機能の利用シーン**：
- 注文イベント（order.created, order.updated）に対して通知処理を実行
- 商品更新イベントに対して検索インデックスを更新
- 支払い完了イベントに対してフルフィルメント処理を開始
- 全イベントをワイルドカード購読してログを記録

**主要な処理内容**：
1. 購読者の登録（subscribe）
2. 購読者の解除（unsubscribe）
3. イベント受信時の購読者呼び出し
4. 購読者エラーのハンドリングとリトライ
5. ワイルドカード購読のサポート

**関連システム・外部連携**：
- Redis：分散環境でのイベントキュー管理（BullMQライブラリ使用）
- Node.js EventEmitter：ローカル環境でのイベント受信

**権限による制御**：イベント購読自体には権限制御は設けられていない。購読者内で必要に応じて権限チェックを実装する。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| - | - | - | 本機能は直接UIを持たないバックエンド機能 |

## 機能種別

データ連携 / 非同期処理

## 入力仕様

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

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| event | string \| symbol | Yes | 購読するイベント名（"*"でワイルドカード） | - |
| subscriber | Subscriber | Yes | イベント受信時に呼び出されるコールバック関数 | typeof === 'function' |
| context | SubscriberContext | No | 購読者のコンテキスト情報 | - |
| context.subscriberId | string | No | 購読者の一意識別子 | 重複不可 |

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

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| event | string \| symbol | Yes | 購読解除するイベント名 | - |
| subscriber | Subscriber | Yes | 解除する購読者関数 | - |
| context | SubscriberContext | No | 購読者のコンテキスト情報 | - |

### 購読者関数の引数（Event）

| パラメータ名 | 型 | 必須 | 説明 |
|-------------|-----|-----|------|
| name | string | Yes | イベント名 |
| data | T | No | イベントペイロード |
| metadata | object | No | イベントメタデータ |

### 入力データソース

- アプリケーション起動時のモジュール初期化処理
- サービスクラスのコンストラクタ
- ワークフロー定義内での動的登録

## 出力仕様

### 出力データ（subscribe/unsubscribe）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| this | EventBusModuleService | メソッドチェーン用にサービスインスタンスを返却 |

### 出力データ（購読者処理結果）

| 項目名 | 型 | 説明 |
|--------|-----|------|
| result | unknown[] | 各購読者の処理結果（Redis実装のworker）|

### 出力先

- 購読者関数の戻り値はイベントバスによって収集される（Redis実装）
- ローカル実装では戻り値は破棄される

## 処理フロー

### 処理シーケンス（購読登録）

```
1. 購読者関数の検証
   └─ typeof === 'function' であることを確認

2. subscriberIdの決定
   └─ context.subscriberId または 自動生成（event-ulid形式）

3. 重複チェック
   └─ 同一subscriberIdが既に登録されていないことを確認

4. 購読者の登録
   ├─ eventToSubscribersMap_に追加
   └─ EventEmitterにリスナー登録（ローカル実装）

5. 完了（thisを返却）
```

### 処理シーケンス（イベント受信・配信）

```
1. イベント受信
   └─ EventEmitter.on() または BullMQ Worker

2. 購読者の取得
   ├─ イベント名に対応する購読者リストを取得
   └─ ワイルドカード購読者（"*"）を追加

3. 各購読者への配信
   ├─ イベントデータを引数として購読者関数を呼び出し
   └─ エラー発生時はログ出力して継続

4. リトライ処理（Redis実装）
   ├─ 失敗した購読者を特定
   ├─ completedSubscriberIdsを更新
   └─ 再実行

5. 完了
```

### フローチャート

```mermaid
flowchart TD
    A[subscribe呼び出し] --> B{関数か?}
    B -->|No| C[Error: Subscriber must be a function]
    B -->|Yes| D[subscriberId決定]
    D --> E{重複チェック}
    E -->|重複| F[Error: Subscriber already exists]
    E -->|OK| G[eventToSubscribersMapに登録]
    G --> H[EventEmitterに登録]
    H --> I[thisを返却]

    J[イベント受信] --> K[購読者リスト取得]
    K --> L[ワイルドカード購読者追加]
    L --> M[各購読者を呼び出し]
    M --> N{エラー発生?}
    N -->|Yes| O[エラーログ出力]
    O --> P{リトライ設定あり?}
    P -->|Yes| Q[リトライキューに追加]
    P -->|No| R[処理継続]
    N -->|No| R
    Q --> R
    R --> S[完了]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-92-1 | 購読者ID一意性 | 同一イベントに対して同一subscriberIdの購読者は登録不可 | subscribe()呼び出し時 |
| BR-92-2 | ワイルドカード購読 | "*"で登録した購読者は全イベントを受信する | イベント配信時 |
| BR-92-3 | エラー隔離 | 購読者でエラーが発生しても他の購読者の処理は継続 | 購読者処理中の例外発生時 |
| BR-92-4 | サーバーモード制限 | worker_mode === "server" の場合、unsubscribeは無効 | 非ワーカーモード時 |
| BR-92-5 | 成功済み購読者スキップ | リトライ時は既に成功した購読者をスキップ | リトライ処理時 |

### 計算ロジック

購読者ID生成ロジック：
```typescript
// context.subscriberIdが指定されていない場合
const subscriberId = `${event}-${ulid()}`
// 例: "order.created-01ARZ3NDEKTSV4RRFFQ69G5FAV"
```

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし（インメモリ管理） |

### インメモリデータ構造

| 構造名 | 型 | 用途 |
|-------|-----|------|
| eventToSubscribersMap_ | Map<string \| symbol, SubscriberDescriptor[]> | イベント名から購読者リストへのマッピング |
| SubscriberDescriptor | { subscriber: Subscriber, id: string } | 購読者関数とIDのペア |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | TypeError | subscriberが関数でない | Error("Subscriber must be a function") をスロー |
| - | Error | 同一subscriberIdが既に登録済み | Error(`Subscriber with id ${subscriberId} already exists`) をスロー |
| - | SubscriberError | 購読者内で例外発生 | エラーログ出力後、他の購読者の処理を継続 |

### リトライ仕様（Redis実装）

```typescript
// リトライ判定
const shouldRetry =
  didSubscribersFail &&      // 失敗した購読者がいる
  isRetriesConfigured &&     // attempts > 1 が設定されている
  !isFinalAttempt            // 最終試行ではない

// 成功した購読者の追跡
job.data.completedSubscriberIds = [
  ...completedSubscribers,
  ...completedSubscribersInCurrentAttempt
]
```

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

イベント購読処理自体はトランザクション管理の対象外。購読者内で必要に応じてトランザクションを管理する。

## パフォーマンス要件

- 購読者登録：O(1)（Map操作）
- 購読者検索：O(1)（Map.get）
- イベント配信：O(n)（nは購読者数）
- Redis実装では並列処理によりスループットを確保

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

- 購読者関数は信頼されたコードのみを登録すること
- 購読者内での外部リソースアクセスは適切な認証・認可を実装すること
- エラーログに機密情報が含まれないよう注意

## 備考

- workerモード以外（serverモード）ではunsubscribeは無効化される
- BullMQ Workerはautorun: falseで初期化され、onApplicationStartで開始
- 購読者関数にはsubscriberIdプロパティが付与される

---

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

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

### 推奨読解順序

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

購読者の管理構造を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | index.ts | `packages/core/utils/src/event-bus/index.ts` | SubscriberDescriptor型とeventToSubscribersMap_の構造 |

**読解のコツ**: eventToSubscribersMap_はMap<string | symbol, SubscriberDescriptor[]>型で、イベント名をキーとして購読者の配列を保持する。

#### Step 2: 抽象基底クラスの購読処理を理解する

共通の購読ロジックを確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | index.ts | `packages/core/utils/src/event-bus/index.ts` | subscribe()とunsubscribe()メソッドの実装 |

**主要処理フロー**:
- **58-83行目**: storeSubscribers()で購読者をMapに追加、重複チェックを実施
- **85-110行目**: subscribe()で購読者を登録、subscriberIdを決定
- **112-138行目**: unsubscribe()で購読者を解除、workerモードチェックを実施

#### Step 3: ローカル実装の購読処理を理解する

Node.js EventEmitterを使用した購読実装。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | event-bus-local.ts | `packages/modules/event-bus-local/src/services/event-bus-local.ts` | subscribe()とunsubscribe()のオーバーライド |

**主要処理フロー**:
- **182-210行目**: subscribe()でEventEmitterにリスナーを登録、エラーハンドリングをラップ
- **192-200行目**: wrappedSubscriberでtry-catchによるエラー処理
- **212-236行目**: unsubscribe()でEventEmitterからリスナーを削除

#### Step 4: Redis実装のワーカー処理を理解する

BullMQ Workerによるイベント配信処理。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | event-bus-redis.ts | `packages/modules/event-bus-redis/src/services/event-bus-redis.ts` | worker_メソッドの実装 |

**主要処理フロー**:
- **98-106行目**: Worker初期化（autorun: false）
- **373-475行目**: worker_()メソッドでキューからイベントを処理
- **375-378行目**: イベント購読者とワイルドカード購読者を取得
- **384-387行目**: 成功済み購読者をフィルタリング
- **415-437行目**: 各購読者を並列実行、成功した購読者を追跡
- **447-465行目**: リトライ判定とジョブデータ更新

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

```
AbstractEventBusModuleService
    │
    ├─ subscribe(event, subscriber, context)
    │      ├─ 関数チェック
    │      ├─ subscriberId決定（context.subscriberId または ulid生成）
    │      └─ storeSubscribers()
    │             └─ eventToSubscribersMap_.set()
    │
    └─ unsubscribe(event, subscriber, context)
           └─ eventToSubscribersMap_.get().splice()

LocalEventBusService extends AbstractEventBusModuleService
    │
    ├─ subscribe() オーバーライド
    │      ├─ super.subscribe()
    │      ├─ wrappedSubscriber作成（エラーハンドリング）
    │      └─ eventEmitter_.on(event, wrappedSubscriber)
    │
    └─ unsubscribe() オーバーライド
           ├─ super.unsubscribe()
           └─ eventEmitter_.off(event, subscriber)

RedisEventBusService extends AbstractEventBusModuleService
    │
    └─ worker_(job)
           ├─ eventToSubscribersMap.get(name)
           ├─ eventToSubscribersMap.get("*")
           ├─ completedSubscriberIds除外
           ├─ Promise.all(subscribers.map(invoke))
           └─ リトライ判定・ジョブデータ更新
```

### データフロー図

```
[購読登録]

subscribe(event, subscriber)
         │
         ▼
    subscriberId決定
         │
         ▼
    eventToSubscribersMap_
    {
      "order.created": [
        { id: "sub-1", subscriber: fn1 },
        { id: "sub-2", subscriber: fn2 }
      ],
      "*": [
        { id: "wildcard", subscriber: logFn }
      ]
    }

[イベント配信]

イベント受信 ────▶ eventToSubscribersMap_.get(eventName)
                          │
                          ├───▶ [購読者1] ───▶ 処理実行
                          │
                          ├───▶ [購読者2] ───▶ 処理実行
                          │
                          └───▶ [ワイルドカード購読者] ───▶ 処理実行
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| index.ts | `packages/core/utils/src/event-bus/index.ts` | ソース | 抽象基底クラス（購読管理の共通ロジック） |
| event-bus-local.ts | `packages/modules/event-bus-local/src/services/event-bus-local.ts` | ソース | ローカル実装（EventEmitterへのリスナー登録） |
| event-bus-redis.ts | `packages/modules/event-bus-redis/src/services/event-bus-redis.ts` | ソース | Redis実装（BullMQ Workerによるイベント配信） |
| common.ts | `packages/core/types/src/event-bus/common.ts` | 型定義 | Subscriber、SubscriberContext等の型定義 |
