# 機能設計書 91-イベント発行

## 概要

本ドキュメントは、Medusaコマースプラットフォームにおけるイベント発行機能の設計仕様を記述する。イベント発行機能は、ドメインイベントをシステム内の購読者に通知するための非同期メッセージング基盤を提供する。

### 本機能の処理概要

イベント発行機能は、システム内で発生したビジネスイベント（注文作成、商品更新、支払い完了など）を、関心を持つコンポーネントやサービスに通知するためのPub/Sub（発行/購読）パターンを実装する。ローカル環境向けのインメモリ実装と、分散環境向けのRedis実装の2つのプロバイダーが用意されている。

**業務上の目的・背景**：ECプラットフォームでは、注文処理、在庫更新、通知送信など、様々なビジネスプロセスが相互に連携する必要がある。イベント駆動アーキテクチャを採用することで、サービス間の疎結合を実現し、システムの拡張性と保守性を向上させる。例えば、注文が作成された際に、在庫の確保、顧客への通知、分析データの記録などを独立して処理できる。

**機能の利用シーン**：
- 注文が作成された際に、在庫システムへの通知、メール送信、Webhook呼び出しを行う
- 商品情報が更新された際に、検索インデックスを更新する
- 分散トランザクション内でイベントをグループ化し、トランザクション完了後に一括発行する
- 外部システムとの連携処理をイベントベースで非同期実行する

**主要な処理内容**：
1. イベントメッセージの作成と発行（単一または複数イベントの一括発行）
2. イベントのグループ化（分散トランザクション対応）
3. グループ化されたイベントの一括リリース
4. イベント発行前のインターセプター処理
5. 遅延発行オプションのサポート

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

**権限による制御**：イベント発行自体には権限制御は設けられていない。イベントを発行するサービス側で適切な認可チェックを行う設計となっている。

## 関連画面

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

## 機能種別

データ連携 / 非同期処理

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| eventsData | Message<T> \| Message<T>[] | Yes | 発行するイベントデータ（単一または配列） | name フィールドは必須 |
| eventsData.name | string | Yes | イベント名（例: "order.created"） | 空文字列不可 |
| eventsData.data | T | No | イベントに付随するデータペイロード | - |
| eventsData.metadata | object | No | イベントメタデータ | - |
| eventsData.metadata.eventGroupId | string | No | イベントグループID（グループ化用） | - |
| options | Record<string, unknown> | No | 発行オプション | - |
| options.internal | boolean | No | 内部イベントフラグ（ログ出力抑制） | - |
| options.delay | number | No | 発行遅延時間（ミリ秒） | 正の整数 |
| options.priority | number | No | 優先度（1〜2097152、低い値が高優先度） | 範囲外の場合デフォルト値に修正 |
| options.attempts | number | No | リトライ回数 | - |
| options.groupedEventsTTL | number | No | グループイベントの有効期限（秒） | デフォルト600秒 |

### 入力データソース

- アプリケーション内のサービス層から直接呼び出し
- ワークフローステップからの呼び出し
- APIルートハンドラーからの呼び出し

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| (void) | void | イベント発行は非同期で行われ、戻り値は返さない |

### 出力先

- ローカル実装：Node.js EventEmitterを通じてインプロセスで購読者に通知
- Redis実装：BullMQキューを通じてワーカープロセスに通知

## 処理フロー

### 処理シーケンス

```
1. イベントデータの正規化
   └─ 単一イベントを配列形式に変換

2. イベントグループ判定
   └─ metadata.eventGroupIdの有無を確認

3. グループ化イベントの場合
   ├─ ステージングキュー（Map/Redis List）に追加
   └─ TTL設定（Redis実装のみ）

4. 非グループ化イベントの場合
   ├─ インターセプターの呼び出し
   ├─ 遅延オプションの適用
   ├─ 購読者への通知
   └─ ログ出力（internalフラグがfalseの場合）

5. 完了
```

### フローチャート

```mermaid
flowchart TD
    A[emit呼び出し] --> B[イベントデータ正規化]
    B --> C{eventGroupIdあり?}
    C -->|Yes| D[ステージングキューに追加]
    D --> E[TTL設定]
    E --> F[完了]
    C -->|No| G[インターセプター呼び出し]
    G --> H{delayオプションあり?}
    H -->|Yes| I[遅延待機]
    I --> J[イベント発行]
    H -->|No| J
    J --> K{購読者存在?}
    K -->|Yes| L[購読者に通知]
    L --> M[ログ出力]
    K -->|No| M
    M --> F
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-91-1 | グループ化イベント遅延発行 | eventGroupIdが設定されたイベントは即座に発行されず、releaseGroupedEvents呼び出しまで保持される | 分散トランザクション内でのイベント発行時 |
| BR-91-2 | ワイルドカード購読者 | "*"で購読登録された購読者は全てのイベントを受信する | 監査ログ、デバッグ等の目的で全イベントを監視する場合 |
| BR-91-3 | 購読者エラー隔離 | 購読者でエラーが発生しても他の購読者や発行処理には影響しない | 購読者処理中の例外発生時 |
| BR-91-4 | 優先度範囲制御 | 優先度は1〜2097152の範囲。範囲外の場合はデフォルト値（100）に修正 | Redis実装でのイベント処理順序制御時 |

### 計算ロジック

イベント優先度の決定ロジック：
```typescript
// 優先度階層（低い値が高優先度）
EventPriority.CRITICAL = 10      // 重要ビジネスイベント
EventPriority.DEFAULT = 100      // 通常イベント
EventPriority.LOWEST = 2097152   // 内部イベント

// 優先度決定の優先順位
1. メッセージレベルオプション（eventData.options.priority）
2. 発行レベルオプション（options.priority）
3. モジュールレベルジョブオプション（this.jobOptions_.priority）
4. 内部フラグデフォルト（options.internal ? LOWEST : DEFAULT）
```

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

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

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| - | - | - | データベース操作なし（インメモリまたはRedis使用） |

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

| 操作 | キー形式 | 操作種別 | 概要 |
|-----|---------|---------|------|
| グループイベント追加 | staging:{eventGroupId} | RPUSH | イベントデータをリストに追加 |
| グループイベント取得 | staging:{eventGroupId} | LRANGE | グループ内全イベントを取得 |
| グループイベント削除 | staging:{eventGroupId} | UNLINK | グループを削除 |
| TTL設定 | staging:{eventGroupId} | EXPIRE | 有効期限を設定 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | SubscriberError | 購読者内で例外発生 | エラーログ出力後、処理継続 |
| - | ConnectionError | Redis接続失敗 | 再接続試行（BullMQ管理） |
| INVALID_DATA | ValidationError | 優先度が範囲外 | 警告ログ出力、デフォルト値に修正 |

### リトライ仕様

- ローカル実装：リトライなし（同期的に処理）
- Redis実装：
  - attempts オプションで指定されたリトライ回数を使用
  - 失敗した購読者のみをリトライ対象として追跡
  - completedSubscriberIds で成功済み購読者を管理

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

イベント発行自体はトランザクション管理の対象外。ただし、eventGroupIdを使用することで、外部トランザクション（分散トランザクション）と連携したイベント発行が可能：

1. トランザクション開始時にeventGroupIdを生成
2. トランザクション内のイベントはeventGroupIdを付与して発行（ステージング）
3. トランザクションコミット後にreleaseGroupedEventsを呼び出し
4. トランザクションロールバック時はclearGroupedEventsで破棄

## パフォーマンス要件

- ローカル実装：インプロセス処理のため、数ミリ秒以内
- Redis実装：
  - イベント発行：10ms以内（ネットワーク遅延を除く）
  - 一括発行（addBulk）によりラウンドトリップを最小化

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

- イベントペイロードに機密情報を含める場合は、適切なマスキングを推奨
- Redis通信はTLS暗号化を推奨（本番環境）
- イベント購読者の登録はアプリケーション起動時に制限（動的登録の制限）

## 備考

- EventEmitterのmaxListenersはInfinityに設定されており、購読者数に制限なし
- ワイルドカード購読者（"*"）は監査・デバッグ目的での利用を想定
- インターセプター機能により、イベント発行前に共通処理（ログ記録、変換等）が可能

---

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

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

### 推奨読解順序

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

まず、イベントメッセージの型定義を理解することが重要。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | types | `packages/core/types/src/event-bus/common.ts` | Messageインターフェースの構造（name, data, metadata）を確認 |

**読解のコツ**: Message<T>はジェネリック型で、dataフィールドの型がパラメータ化されている。metadataにはeventGroupIdなどのシステム情報が含まれる。

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

共通機能を提供する抽象クラスの実装を確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | index.ts | `packages/core/utils/src/event-bus/index.ts` | AbstractEventBusModuleServiceの実装を確認 |

**主要処理フロー**:
- **12-16行目**: eventToSubscribersMap_で購読者をイベント名ごとに管理
- **18行目**: interceptorSubscribers_でインターセプターを管理
- **85-110行目**: subscribe()メソッドで購読者を登録
- **169-181行目**: callInterceptors()でイベント発行前にインターセプターを呼び出し

#### Step 3: ローカル実装を理解する

インメモリでのイベント発行実装を確認。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | event-bus-local.ts | `packages/modules/event-bus-local/src/services/event-bus-local.ts` | LocalEventBusServiceの実装 |

**主要処理フロー**:
- **20-21行目**: EventEmitterのインスタンス作成、maxListenersをInfinityに設定
- **65-79行目**: emit()メソッドでイベントを発行
- **86-118行目**: groupOrEmitEvent()でグループ化判定と実際の発行処理
- **121-130行目**: groupEvent()でステージングキューに追加
- **132-165行目**: releaseGroupedEvents()でグループイベントを一括発行
- **182-210行目**: subscribe()で購読者を登録、エラーハンドリングをラップ

#### Step 4: Redis実装を理解する

分散環境向けのRedis実装を確認。

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

**主要処理フロー**:
- **92-106行目**: BullMQのQueueとWorkerを初期化
- **137-190行目**: buildEvents()でキュー用のイベントデータを構築、優先度を設定
- **197-261行目**: emit()でイベントを発行（グループ化判定含む）
- **271-279行目**: groupEvents()でRedisリストにイベントを追加
- **291-320行目**: releaseGroupedEvents()でグループイベントを発行
- **373-475行目**: worker_()でキューからイベントを取り出し購読者に配信

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

```
LocalEventBusService / RedisEventBusService
    │
    ├─ emit(eventsData, options)
    │      ├─ groupOrEmitEvent() / buildEvents()
    │      │      ├─ groupEvent() / groupEvents() [グループ化の場合]
    │      │      │      └─ Map.set() / Redis.rpush()
    │      │      │
    │      │      └─ [即時発行の場合]
    │      │             ├─ callInterceptors()
    │      │             └─ eventEmitter_.emit() / queue_.addBulk()
    │      │
    │      └─ logProcessingEvent()
    │
    ├─ releaseGroupedEvents(eventGroupId)
    │      ├─ getGroupedEvents() [Redis実装のみ]
    │      ├─ callInterceptors()
    │      ├─ eventEmitter_.emit() / queue_.addBulk()
    │      └─ clearGroupedEvents()
    │
    └─ subscribe(event, subscriber, context)
           └─ storeSubscribers() [抽象クラス]
```

### データフロー図

```
[入力]                    [処理]                         [出力]

emit()呼び出し
    │
    ▼
Message<T> ────────▶ groupOrEmitEvent() ────────▶ [グループ化]
  - name                    │                      ステージングキュー
  - data                    │                      (Map/Redis List)
  - metadata                │
  - options                 │
                            │
                            ▼
                     [即時発行]
                            │
                            ├───▶ EventEmitter.emit() ───▶ 購読者関数
                            │     (ローカル実装)
                            │
                            └───▶ BullMQ Queue ───▶ Worker ───▶ 購読者関数
                                  (Redis実装)
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| event-bus-local.ts | `packages/modules/event-bus-local/src/services/event-bus-local.ts` | ソース | ローカル（インメモリ）イベントバス実装 |
| event-bus-redis.ts | `packages/modules/event-bus-redis/src/services/event-bus-redis.ts` | ソース | Redis（分散）イベントバス実装 |
| index.ts | `packages/core/utils/src/event-bus/index.ts` | ソース | 抽象基底クラス（AbstractEventBusModuleService） |
| common.ts | `packages/core/types/src/event-bus/common.ts` | 型定義 | Message、Subscriber等の型定義 |
| types.ts | `packages/modules/event-bus-redis/src/types.ts` | 型定義 | Redis実装固有の型定義 |
