# 機能設計書 37-Messenger

## 概要

本ドキュメントは、Symfony Messengerコンポーネントの機能設計を記述する。Messengerコンポーネントは、メッセージキュー経由でのメッセージ送受信（ディスパッチ・消費）機能を提供する。

### 本機能の処理概要

Messengerコンポーネントは、メッセージバスを通じたメッセージのディスパッチ（送信）とハンドリング（処理）の基盤を提供する。同期処理と非同期処理の両方に対応し、AMQP、Doctrine、Redis等の様々なトランスポートを通じたメッセージキューイングを実現する。ミドルウェアパターンにより処理パイプラインを柔軟にカスタマイズできる。

**業務上の目的・背景**：Webアプリケーションでは、メール送信、画像処理、レポート生成等の重い処理をリクエスト処理から切り離して非同期で実行することが重要である。Messengerコンポーネントは、メッセージ駆動のアーキテクチャを提供することで、処理のスケーラビリティと応答性を向上させる。CQRSやイベントソーシングパターンの実装基盤としても利用できる。

**機能の利用シーン**：メール送信の非同期化、画像リサイズ処理の非同期化、外部API連携の非同期化、定期バッチ処理のキューイング、イベント駆動アーキテクチャの実装など、非同期処理が必要なあらゆる場面で使用される。Mailerコンポーネント、Schedulerコンポーネントとも密接に連携する。

**主要な処理内容**：
1. MessageBusによるメッセージのディスパッチ（Envelope + Stampパターン）
2. ミドルウェアパイプラインによるメッセージ処理の連鎖
3. ハンドラーによるメッセージの実際の処理（MessageHandlerInterface）
4. トランスポートによるメッセージのキューイング（送信・受信）
5. Workerによるキューからのメッセージ消費
6. リトライ機構（RetryStrategy）
7. 失敗メッセージの管理（FailedMessagesCommand群）
8. Stamp機構によるメッセージメタデータの付加
9. TraceableMessageBusによるデバッグ・プロファイリング
10. RoutableMessageBusによるメッセージバスの動的ルーティング

**関連システム・外部連携**：AMQP（RabbitMQ等）、Doctrine DBAL、Redis、Amazon SQS、Beanstalkd等のメッセージキューシステムと連携する。Mailerコンポーネント（メール送信のキューイング）、Schedulerコンポーネント（定期メッセージ生成）とも連携する。

**権限による制御**：メッセージ自体に権限制御はないが、消費ワーカーのプロセス権限で実行レベルを制御する。

## 関連画面

| 画面No | 画面名 | 関連種別 | 関連する操作・処理 |
|--------|--------|----------|------------------|
| 32 | Messengerパネル | 主機能 | Messengerコンポーネントによるメッセージバスのディスパッチ情報の表示 |

## 機能種別

メッセージング処理（非同期メッセージディスパッチ / キューイング）

## 入力仕様

### 入力パラメータ

| パラメータ名 | 型 | 必須 | 説明 | バリデーション |
|-------------|-----|-----|------|---------------|
| message | object | Yes | ディスパッチするメッセージオブジェクト | オブジェクト型であること |
| stamps | StampInterface[] | No | メッセージに付加するスタンプ | StampInterfaceを実装していること |

### 入力データソース

- アプリケーションコードからのメッセージオブジェクト生成
- トランスポート（キュー）からのメッセージ受信

## 出力仕様

### 出力データ

| 項目名 | 型 | 説明 |
|--------|-----|------|
| envelope | Envelope | スタンプ付きのメッセージエンベロープ |

### 出力先

- 同期: メッセージハンドラー
- 非同期: トランスポート（メッセージキュー）

## 処理フロー

### 処理シーケンス

```
■ ディスパッチ処理
1. MessageBus::dispatch() の呼び出し
   ├─ Envelope::wrap() でメッセージをエンベロープにラップ
   ├─ ミドルウェアチェーンの構築（StackMiddleware）
   └─ 最初のミドルウェアのhandle()を呼び出し
       ├─ SendMessageMiddleware: トランスポートへの送信判定
       │   ├─ 送信対象: トランスポートに送信してreturn
       │   └─ 送信対象外: 次のミドルウェアに委譲
       ├─ HandleMessageMiddleware: ハンドラーの実行
       └─ 他のカスタムミドルウェア

■ 消費処理（Worker）
1. Worker::run() の呼び出し
   ├─ 各レシーバーからメッセージを取得
   ├─ ReceivedStampを付加
   ├─ MessageBus::dispatch() で再ディスパッチ
   ├─ ハンドラーの実行
   ├─ ACK/REJECT の処理
   └─ イベントのディスパッチ（WorkerMessageReceivedEvent等）
```

### フローチャート

```mermaid
flowchart TD
    A[MessageBus::dispatch] --> B[Envelope::wrap]
    B --> C[StackMiddleware構築]
    C --> D[SendMessageMiddleware]
    D --> E{トランスポート送信対象?}
    E -->|Yes| F[トランスポートに送信]
    E -->|No| G[HandleMessageMiddleware]
    G --> H[MessageHandler実行]
    H --> I[結果のEnvelope返却]
    F --> J[Worker::run]
    J --> K[レシーバーからメッセージ取得]
    K --> L[ReceivedStamp付加]
    L --> M[MessageBus::dispatch 再ディスパッチ]
    M --> N[HandleMessageMiddleware]
    N --> O[MessageHandler実行]
    O --> P{成功?}
    P -->|Yes| Q[ACK]
    P -->|No| R{リトライ?}
    R -->|Yes| S[リトライキューに送信]
    R -->|No| T[REJECT / 失敗キュー]
```

## ビジネスルール

### 業務ルール

| ルールNo | ルール名 | 内容 | 適用条件 |
|---------|---------|------|---------|
| BR-37-01 | ミドルウェアチェーン | メッセージは登録されたミドルウェアを順次通過して処理される | 常時 |
| BR-37-02 | 非同期送信 | SenderStampが付加されたメッセージはトランスポートに送信される | SendMessageMiddleware通過時 |
| BR-37-03 | リトライ戦略 | 失敗したメッセージはRetryStrategyに基づいてリトライされる | メッセージ処理失敗時 |
| BR-37-04 | スタンプ不変性 | Envelopeのwith/withoutメソッドはイミュータブルで、常に新しいEnvelopeを返す | 常時 |
| BR-37-05 | ワーカーのスリープ | メッセージがない場合、Workerは設定された時間（デフォルト1秒）スリープする | キューが空の場合 |
| BR-37-06 | レートリミッター | RateLimiterとの連携によりメッセージ処理速度を制限できる | rateLimiters設定時 |

### 計算ロジック

特になし。

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

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

DoctrineトランスポートはDBALテーブルでメッセージキューを管理する。

| 操作 | 対象テーブル | 操作種別 | 概要 |
|-----|-------------|---------|------|
| 送信 | messenger_messages | INSERT | メッセージをキューテーブルに挿入 |
| 受信 | messenger_messages | SELECT + UPDATE | メッセージを取得し、処理中としてマーク |
| ACK | messenger_messages | DELETE | 処理完了したメッセージを削除 |
| リジェクト | messenger_messages | DELETE/INSERT | 失敗メッセージの処理 |

### テーブル別操作詳細

#### messenger_messages（Doctrineトランスポート使用時）

| 操作 | 項目（カラム名） | 更新値・取得条件 | 備考 |
|-----|-----------------|-----------------|------|
| INSERT | body, headers, queue_name, created_at, available_at | メッセージ本体とメタデータ | 送信時 |
| SELECT | * | available_at <= NOW() AND delivered_at IS NULL | 受信可能なメッセージの取得 |
| DELETE | - | id = ? | 処理完了時に削除 |

## エラー処理

### エラーケース一覧

| エラーコード | エラー種別 | 発生条件 | 対処方法 |
|------------|----------|---------|---------|
| - | HandlerFailedException | メッセージハンドラーで例外発生 | ハンドラーの実装を確認する |
| - | NoHandlerForMessageException | メッセージに対応するハンドラーがない | ハンドラーを登録する |
| - | RejectRedeliveredMessageException | 再配信メッセージの拒否 | リトライ設定を確認する |
| - | RuntimeException | トランスポート接続エラー | トランスポート設定を確認する |

### リトライ仕様

- RetryStrategyInterface により、最大リトライ回数、遅延時間、倍率を設定可能
- MultiplierRetryStrategy: 初期遅延 * 倍率^(リトライ回数) の指数バックオフ
- 最大リトライ回数超過時は失敗キュー（failure transport）に移動

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

- DoctrineTransactionMiddlewareにより、メッセージ処理をDoctrineトランザクション内で実行可能
- Doctrineトランスポートはデータベースレベルのトランザクションでメッセージの原子的な取得を保証

## パフォーマンス要件

- Workerのスリープ設定により、CPUリソースの消費を制御
- レートリミッターによるスループット制御
- 複数ワーカーの並列実行によるスケーラビリティ

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

- メッセージは serialize/unserialize されるため、信頼できないソースからのメッセージの処理に注意
- トランスポートの接続情報（DSN）は環境変数で管理し、ソースコードにハードコードしない

## 備考

- Messengerは Symfony のコアコンポーネントの一つであり、Mailer、Scheduler等の他コンポーネントの非同期処理基盤としても機能する
- HandleTraitを使ってコントローラーから同期的にメッセージをディスパッチし、結果を取得できる
- TraceableMessageBusでディスパッチされたメッセージとスタンプをWebProfilerで確認できる

---

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

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

### 推奨読解順序

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

EnvelopeとStampパターンを理解する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 1-1 | Envelope.php | `src/Symfony/Component/Messenger/Envelope.php` | メッセージのラッパー。stamps配列によるメタデータ管理 |
| 1-2 | StampInterface.php | `src/Symfony/Component/Messenger/Stamp/StampInterface.php` | スタンプの基底インターフェース |

**読解のコツ**: EnvelopeはイミュータブルなValue Objectパターンで、with()メソッドは常に新しいインスタンスを返す。stamps配列はクラス名をキーとして管理される。

#### Step 2: エントリーポイントを理解する

MessageBusのディスパッチ処理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 2-1 | MessageBusInterface.php | `src/Symfony/Component/Messenger/MessageBusInterface.php` | メッセージバスのインターフェース |
| 2-2 | MessageBus.php | `src/Symfony/Component/Messenger/MessageBus.php` | メッセージバスの実装。ミドルウェアチェーンの構築と実行 |

**主要処理フロー**:
1. **54-70行目** (MessageBus.php): `dispatch()` - Envelope::wrap()、ミドルウェアイテレータ取得、StackMiddleware構築、最初のミドルウェアへの委譲

#### Step 3: ミドルウェアを理解する

ミドルウェアパイプラインの構造を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 3-1 | MiddlewareInterface.php | `src/Symfony/Component/Messenger/Middleware/MiddlewareInterface.php` | ミドルウェアインターフェース |
| 3-2 | StackMiddleware.php | `src/Symfony/Component/Messenger/Middleware/StackMiddleware.php` | ミドルウェアスタック |

#### Step 4: ワーカーを理解する

メッセージ消費処理を確認する。

| 順序 | ファイル | パス | 読解ポイント |
|-----|---------|------|-------------|
| 4-1 | Worker.php | `src/Symfony/Component/Messenger/Worker.php` | メッセージ消費ワーカー |

**主要処理フロー**:
- **59-72行目** (Worker.php): コンストラクタ。receivers, bus, eventDispatcher, logger, rateLimiters, clockの注入

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

```
MessageBus::dispatch()
    |
    +-- Envelope::wrap()
    +-- StackMiddleware (ミドルウェアチェーン)
            |
            +-- SendMessageMiddleware::handle()
            |       +-- SenderInterface::send() [非同期送信]
            |
            +-- HandleMessageMiddleware::handle()
            |       +-- HandlersLocator::getHandlers()
            |       +-- MessageHandlerInterface::__invoke()
            |
            +-- 他のミドルウェア

Worker::run()
    |
    +-- ReceiverInterface::get()
    +-- MessageBus::dispatch() [ReceivedStamp付き]
    +-- ReceiverInterface::ack() / reject()
    +-- EventDispatcher::dispatch() [各種イベント]
```

### データフロー図

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

メッセージ ──────────> MessageBus::dispatch()
オブジェクト               |
                      Envelope::wrap()
                           |
                      ミドルウェアパイプライン
                           |
                      ┌────┴────┐
                      |         |
                  同期処理    非同期送信
                      |         |
                  Handler    Transport
                  実行        (キュー)
                      |         |
                      |     Worker::run()
                      |         |
                      |     Handler実行
                      |         |
                      v         v
                  処理結果    ACK/REJECT
```

### 関連ファイル一覧

| ファイル | パス | 種別 | 役割 |
|---------|------|------|------|
| MessageBus.php | `src/Symfony/Component/Messenger/MessageBus.php` | ソース | メッセージバス実装 |
| MessageBusInterface.php | `src/Symfony/Component/Messenger/MessageBusInterface.php` | ソース | メッセージバスインターフェース |
| Envelope.php | `src/Symfony/Component/Messenger/Envelope.php` | ソース | メッセージエンベロープ |
| Worker.php | `src/Symfony/Component/Messenger/Worker.php` | ソース | メッセージ消費ワーカー |
| HandleTrait.php | `src/Symfony/Component/Messenger/HandleTrait.php` | ソース | 同期ハンドル用トレイト |
| TraceableMessageBus.php | `src/Symfony/Component/Messenger/TraceableMessageBus.php` | ソース | デバッグ用メッセージバス |
| RoutableMessageBus.php | `src/Symfony/Component/Messenger/RoutableMessageBus.php` | ソース | 動的ルーティング用バス |
| Middleware/ | `src/Symfony/Component/Messenger/Middleware/` | ソース | ミドルウェア群 |
| Transport/ | `src/Symfony/Component/Messenger/Transport/` | ソース | トランスポート群 |
| Handler/ | `src/Symfony/Component/Messenger/Handler/` | ソース | ハンドラー関連 |
| Stamp/ | `src/Symfony/Component/Messenger/Stamp/` | ソース | スタンプ群 |
| Retry/ | `src/Symfony/Component/Messenger/Retry/` | ソース | リトライ戦略 |
| DataCollector/ | `src/Symfony/Component/Messenger/DataCollector/` | ソース | WebProfiler用データコレクター |
