ハイブリッドなトランザクショナル・アウトボックス方式のイベント配信パターン

1
┌─────────────────────────────────────────────────────────────┐
2
│                     AIチャットサービス                      │
3
└───┬─────────────────┬─────────────────┬─────────────────────┘
4
    │                 │                 │
5
    │ 1. 即時イベント │ 2. 将来の       │ 3. 大量データ
6
    │                 │ イベントを予約  │ を蓄積
7
    ▼                 ▼                 ▼
8
┌─────────────────────────────────────────────────────────────┐
9
│                      3つの専用テーブル                      │
10
├─────────────────────────────────────────────────────────────┤
11
│  ┌───────────────────────────────────────────────────────┐  │
12
│  │ outbox_events: ハイブリッド (AI応答, 決済)            │  │
13
│  │   → 即時発行 + 30秒ごとのスキャナfallback             │  │
14
│  │   → インデックス: (status, created_at) で高速スキャン │  │
15
│  │   → クリーンアップ: 30日後にアーカイブ                │  │
16
│  └───────────────────────────────────────────────────────┘  │
17
│  ┌───────────────────────────────────────────────────────┐  │
18
│  │ scheduled_events: 将来のトリガー (サブスク)           │  │
19
│  │   → scheduled_at <= NOW() をポーリング                │  │
20
│  │   → インデックス: (scheduled_at, status) で時刻検索   │  │
21
│  │   → クリーンアップ: 実行後に削除                      │  │
22
│  └───────────────────────────────────────────────────────┘  │
23
│  ┌───────────────────────────────────────────────────────┐  │
24
│  │ accumulation_buffer: バッチ処理 (トークン使用量)      │  │
25
│  │   → MQ不要、5分ごとに直接集約                         │  │
26
│  │   → インデックス: (user_id, created_at) でグループ化  │  │
27
│  │   → クリーンアップ: 集約後に即削除                    │  │
28
│  └───────────────────────────────────────────────────────┘  │
29
└─────────────────────────────────────────────────────────────┘
30
                          │
31
                          │ { event_id: UUID または series }
32
                          │
33
                          ▼
34
                    ┌──────────┐
35
                    │    MQ    │
36
                    └─────┬────┘
37
                          │
38
                          ▼
39
                    ┌──────────┐
40
                    │ Consumer │
41
                    └──────────┘

1
async fn save_ai_response(
2
    pool: &PgPool,
3
    mq: &MessageQueue,
4
    response: AiResponse,
5
) -> Result<()> {
6
    let mut tx = pool.begin().await?;
7
    // ここで何か処理をしてからイベントを発行する
8
    let event = sqlx::query_as!(
9
        OutboxEvent,
10
        r#"
11
        INSERT INTO outbox_events (event_type, payload, status)
12
        VALUES ($1, $2, $3)
13
        RETURNING id, event_type, payload, status, created_at, sent_at, processed_at
14
        "#,
15
        "ai_response_ready",
16
        sqlx::types::Json(EventPayload::AiResponseReady { /* ... */ }) as _,
17
        EventStatus::Pending as EventStatus,
18
    )
19
    .fetch_one(&mut *tx)
20
    .await?;
21
    tx.commit().await?;
22

23
    // コミット後に即時発行 (ノンブロッキング)
24
    let event_id = event.id;
25
    let mq = mq.clone();
26
    tokio::spawn(async move {
27
        if let Err(e) = eager_publish(&mq, event_id).await {
28
            tracing::warn!("AI応答の即時発行に失敗: {:?}", e);
29
            // スキャナが30秒以内に拾う
30
        }
31
    });
32

33
    Ok(())
34
}

1
async fn create_subscription(
2
    pool: &PgPool,
3
    user_id: Uuid,
4
    plan: SubscriptionPlan,
5
    duration_days: i32,
6
) -> Result<Subscription> {
7
    let mut tx = pool.begin().await?;
8

9
    let expires_at = Utc::now() + Duration::days(duration_days as i64);
10

11
    // サブスクリプションを作成
12
    let subscription = sqlx::query_as!(
13
        Subscription,
14
        "INSERT INTO subscriptions (user_id, plan, expires_at, status)
15
         VALUES ($1, $2, $3, $4)
16
         RETURNING *",
17
        user_id,
18
        plan as SubscriptionPlan,
19
        expires_at,
20
        SubscriptionStatus::Active as SubscriptionStatus,
21
    )
22
    .fetch_one(&mut *tx)
23
    .await?;
24

25
    // 専用テーブルに失効イベントを予約
26
    sqlx::query!(
27
        r#"
28
        INSERT INTO scheduled_events (event_type, payload, scheduled_at, status)
29
        VALUES ($1, $2, $3, $4)
30
        "#,
31
        "subscription_expiry",
32
        ScheduledEventPayload::SubscriptionExpiry{ /* ... */ },
33
        expires_at,
34
        EventStatus::Pending as EventStatus,
35
    )
36
    .execute(&mut *tx)
37
    .await?;
38

39
    tx.commit().await?;
40

41
    Ok(subscription)
42
}
43

44
// スキャナは毎分実行され、期限切れのサブスクをチェックする
45
async fn scan_scheduled_events(pool: &PgPool, mq: &MessageQueue) -> Result<()> {
46
    let now = Utc::now();
47

48
    let due_events = sqlx::query_as!(
49
        ScheduledEvent,
50
        r#"
51
        SELECT id, event_type, payload, scheduled_at,
52
               status as "status: EventStatus", sent_at, processed_at
53
        FROM scheduled_events
54
        WHERE status = $1 AND scheduled_at <= $2
55
        LIMIT 100
56
        "#,
57
        EventStatus::Pending as EventStatus,
58
        now,
59
    )
60
    .fetch_all(pool)
61
    .await?;
62

63
    for event in due_events {
64
        publish_scheduled_event(pool, mq, event.id)?;
65
    }
66

67
    Ok(())
68
}
69

70
// コンシューマ: サブスク失効を処理
71
async fn handle_subscription_expiry(
72
    pool: &PgPool,
73
    event: OutboxEvent,
74
) -> Result<()> {
75
    let payload: EventPayload = serde_json::from_value(event.payload)?;
76

77
    if let EventPayload::SubscriptionExpiry { subscription_id, user_id } = payload {
78
        // サブスクリプションのステータスを更新
79
        sqlx::query!(
80
            "UPDATE subscriptions SET status = $1 WHERE id = $2",
81
            SubscriptionStatus::Expired as SubscriptionStatus,
82
            subscription_id,
83
        )
84
        .execute(pool)
85
        .await?;
86

87
        // API アクセスを取り消す
88
        revoke_api_keys(user_id).await?;
89

90
        // 通知メールを送信
91
        send_expiry_notification(user_id).await?;
92
    }
93

94
    Ok(())
95
}

1
// バックグラウンドジョブは5分ごとに実行される
2
async fn accumulate_token_usage(pool: &PgPool) -> Result<()> {
3
    // accumulation buffer から未処理のトークン使用量をすべて取得
4
    let usage_records = sqlx::query!(
5
        r#"
6
        SELECT id, user_id, tokens, model, created_at
7
        FROM accumulation_buffer
8
        LIMIT 10000
9
        "#,
10
    )
11
    .fetch_all(pool)
12
    .await?;
13

14
    if usage_records.is_empty() {
15
        return Ok(());
16
    }
17

18
    // user_id でグループ化しトークンを合計
19
    let mut usage_by_user: HashMap<Uuid, i32> = HashMap::new();
20
    for record in &usage_records {
21
        *usage_by_user.entry(record.user_id).or_insert(0) += record.tokens;
22
    }
23

24
    // ユーザーのクォータを一括更新
25
    let mut tx = pool.begin().await?;
26
    // ここで何か処理をする
27
    tx.commit().await?;
28

29
    tracing::info!(
30
        "{} 件のトークン使用量レコードを {} 人のユーザー分集約した",
31
        usage_records.len(),
32
        usage_by_user.len()
33
    );
34

35
    Ok(())
36
}
37

38
// アキュムレータを定期的に実行
39
async fn run_token_accumulator(pool: PgPool) {
40
    let mut interval = tokio::time::interval(Duration::from_secs(300)); // 5分
41

42
    loop {
43
        interval.tick().await;
44

45
        if let Err(e) = accumulate_token_usage(&pool).await {
46
            tracing::error!("トークンアキュムレータのエラー: {:?}", e);
47
        }
48
    }
49
}

1
このイベントは outbox_events、scheduled_events、accumulation_buffer のどれを使うべきか？
2

3
├─ 実行時刻が既知の将来イベントか？
4
│  └─ YES → scheduled_events
5
│
6
├─ 大量 (>1000/秒) かつ集約可能か？
7
│  └─ YES → accumulation_buffer
8
│
9
└─ 低レイテンシ配信が必要か？
10
   ├─ YES → outbox_events (ハイブリッド)
11
   └─ NO → classic_outbox (ポーリングのみ)

1
#[derive(Debug, Clone, Copy, PartialEq, Eq, sqlx::Type)]
2
#[sqlx(type_name = "event_status", rename_all = "lowercase")]
3
pub enum EventStatus {
4
    Pending,    // 配信待ち
5
    Processed,  // コンシューマ完了
6
    Failed,     // 恒久的な失敗
7
}

1
#[derive(Debug, Serialize, Deserialize)]
2
#[serde(tag = "event_name")]
3
pub enum EventPayload {
4
    AiResponseReady {
5
        conversation_id: Uuid,
6
        message_id: Uuid,
7
        redis_key: String,
8
        provider: AiProvider,
9
        model: String,
10
        tokens_used: i32,
11
    },
12
    PaymentCompleted {
13
        user_id: Uuid,
14
        payment_id: Uuid,
15
        plan: SubscriptionPlan,
16
        amount: Decimal,
17
    },
18
    // ... その他のハイブリッドイベント
19
}

1
#[derive(Debug, sqlx::FromRow)]
2
pub struct OutboxEvent {
3
    pub id: Uuid,
4
    pub event_type: String,
5
    pub payload: sqlx::types::Json<EventPayload>,
6
    pub status: EventStatus,
7
    pub created_at: DateTime<Utc>,
8
    pub sent_at: Option<DateTime<Utc>>,
9
    pub processed_at: Option<DateTime<Utc>>,
10
}
11

12
#[derive(Debug, sqlx::FromRow)]
13
pub struct ScheduledEvent {
14
    // ...
15
}
16

17
#[derive(Debug, sqlx::FromRow)]
18
pub struct AccumulationRecord {
19
    // ...
20
}

1
$ cargo sqlx prepare
2
Connecting to database...
3
Building query metadata for 94 queries...
4
Successfully saved query metadata to .sqlx/
5

6
$ cargo build
7
   Compiling outbox-service v0.1.0
8
    Finished dev [unoptimized + debuginfo] target(s) in 1.14s

1
$ cargo build
2
error: error returned from database: column "scheduled_for" does not exist
3
  --> src/scanner.rs:23:5