キュー抽象化設計 (Queue Abstraction)¶

対象サービス: 全マイクロサービス横断 作成日: 2026-04-19 ステータス: 承認待ち 関連レビュー: Notion Design Docs（2026-04-18）「SQSを採用するようにしているが、Oracleや自前のバッジ処理などで利用する場合などを複数検討した上で、システムを提案するように」

1. 背景と目的¶

Recerdo の 基本的方針（ポリシー） により、AWS サービスは Cognito のみ利用し、SQS / SNS / SES などは使用しない。メッセージングは以下のフェーズで 異なる基盤 を使う：

• Closed Beta（セルフホスト）: XServer VPS 上の Redis + BullMQ（Node ワーカー） / asynq（Go ワーカー） を第一選択
• 本番（OCI ファースト）: OCI Queue Service を第一選択
• 将来: 地域拡張時も OCI を軸に、必要に応じて RabbitMQ / NATS へ拡張（AWS SQS / GCP Pub/Sub は採用しない）

これらを コード変更なし で切り替え可能にするため、キュー層をアダプタパターンで抽象化する。

2. 評価対象の OSS / マネージド キュー¶

2.1 候補一覧と特性¶

2.2 詳細比較¶

2.3 Beta フェーズ推奨：Redis + BullMQ¶

理由: - 既存 VPS 内の Redis を共用できる（追加リソース不要） - ダッシュボード（Bull Board / BullMQ UI）が標準提供され、管理者コンソールへの組み込みが容易 - 遅延実行・レート制限・リトライ戦略を柔軟に設定可能（リマインダー通知・Timeline 非同期処理に最適） - Sidekiq との選択は Admin Console を Rails で作るか Next.js で作るか に連動（Admin Console 設計 参照）

2.4 本番フェーズ推奨：OCI Queue Service¶

理由: - OCI ファースト戦略と整合（クロスクラウドコスト最小化） - AMQP 1.0 互換のため、RabbitMQ ライブラリ資産を流用可能 - マネージドのためシステム運用・スケーリングをオフロード - 月 100万リクエストまで無料枠あり

フォールバック先として RabbitMQ（OCI Container Instances 上で自己ホスト）も同じ Port 実装で選択可能。AWS SQS は 基本的方針 により採用しない。

3. アーキテクチャ設計¶

3.1 全体構成¶

flowchart LR
  subgraph Producers["プロデューサー (全マイクロサービス)"]
    Core["Core Svc"]
    Auth["Auth Svc"]
    Album["Album Svc"]
    Events["Events Svc"]
    Admin["Admin Console Svc"]
  end

  subgraph QueuePort["Queue Port (ドメイン層インタフェース)"]
    P["QueuePort.Enqueue / Dequeue / ACK / NACK"]
  end

  subgraph Adapters["Queue Adapters (差替え可能)"]
    BullMQ["RedisBullMQAdapter (Beta / Node)"]
    Asynq["AsynqAdapter (Beta / Go)"]
    Sidekiq["SidekiqAdapter (Rails Admin 時のみ)"]
    OCIQ["OCIQueueAdapter (本番)"]
    Rabbit["RabbitMQAdapter (将来)"]
  end

  subgraph Consumers["コンシューマー (Worker)"]
    NotifW["Notification Worker"]
    AuditW["Audit Worker"]
    TimelineW["Timeline Fanout Worker"]
    MediaW["Media Processing Worker"]
  end

  subgraph FF["Feature Flag Svc"]
    Flag["infra.queue.provider"]
  end

  Producers --> P
  P -.DI で選択.-> BullMQ
  P -.DI で選択.-> Asynq
  P -.DI で選択.-> Sidekiq
  P -.DI で選択.-> OCIQ
  P -.DI で選択.-> Rabbit
  Flag -.実行時評価.-> P

  BullMQ --> Consumers
  Asynq --> Consumers
  Sidekiq --> Consumers
  OCIQ --> Consumers
  Rabbit --> Consumers

3.2 キュー命名規約（Topic / Stream）¶

すべてのアダプタで共通に以下の命名規則を使う。アダプタ内でそれぞれのプラットフォームにマッピング。

3.3 Port 定義¶

// domain/port/queue.go
package port

import "context"

type Job struct {
    ID          string                 // UUID v7
    Topic       string                 // 論理トピック名（例: "audit.events"）
    Payload     []byte                 // JSON エンコード済みペイロード
    Priority    int                    // 0(高) 〜 9(低)、サポートしないアダプタは無視
    DelaySec    int                    // 実行遅延（秒）
    Attempt     int                    // 試行回数
    MaxAttempts int                    // 最大試行回数
    Headers     map[string]string      // トレーシング等のメタデータ
    EnqueuedAt  time.Time
}

type QueuePort interface {
    Enqueue(ctx context.Context, job Job) error
    // Dequeue は1件ブロッキング取得。キャンセルは ctx で。
    Dequeue(ctx context.Context, topic string) (*Job, AckFn, error)
    HealthCheck(ctx context.Context) error
}

type AckFn func(result AckResult) error

type AckResult struct {
    Status AckStatus // SUCCESS | RETRY | DEAD
    Error  error
}

3.4 実装上のガイドライン¶

• 冪等性: Job.ID を利用者側で採番し、Worker は Idempotency Key として扱う
• 再試行: アダプタ内で指数バックオフ（1s → 2s → 4s → ...、最大 5 分）
• DLQ: MaxAttempts 超過時は *.dead サフィックス付きトピックへ移送
• トレーシング: Headers["traceparent"] に W3C Trace Context を埋め込み（OpenTelemetry 準拠）

4. アダプタ別実装ノート¶

4.1 BullMQ Adapter（Beta 第一推奨）¶

• Redis 6.2+ を使用
• Queue 名は recerdo:queue:<topic>
• ダッシュボード: Bull Board を Admin Console Svc に組み込み（Admin Console 設計）
• リトライ: backoff: { type: 'exponential', delay: 1000 }
• DLQ: failed キューを audit.events.dead として扱う

// adapter/queue/bullmq_adapter.ts
import { Queue } from 'bullmq';
const queue = new Queue('recerdo:queue:audit.events', { connection: redis });
await queue.add('audit', payload, { attempts: 5, backoff: { type: 'exponential', delay: 1000 } });

4.2 Sidekiq Adapter（Rails Admin 採用時の代替）¶

• Redis 共用
• ActiveJob 経由でドメインから呼び出し
• 優先度キュー: critical / default / low
• メリット: Rails 側の Sidekiq-Web ダッシュボードが標準提供される

4.3 OCI Queue Adapter（本番第一候補）¶

• AMQP 1.0 または OCI SDK を使用
• Queue ID は OCID（Oracle Cloud Identifier）で指定、論理名は環境変数 QUEUE_OCI_MAP_* で解決
• Visibility Timeout: 30 秒〜
• Batch 受信最大 32 件

4.4 asynq Adapter（Beta Go ワーカー第一推奨）¶

• Redis 共用（XServer VPS、DB=1 を asynq 用に割当）
• Queue 名: recerdo:asynq:<topic>
• ダッシュボード: asynqmon を admin-console-svc に iframe 埋め込み
• リトライ: asynq.MaxRetry(5) + 指数バックオフ
• DLQ: archived キューを <topic>.dead として扱う

// adapter/queue/asynq_adapter.go
import "github.com/hibiken/asynq"
client := asynq.NewClient(asynq.RedisClientOpt{Addr: redisAddr})
task := asynq.NewTask("audit.events", payload)
_, err := client.Enqueue(task, asynq.MaxRetry(5), asynq.Timeout(300*time.Second))

4.5 RabbitMQ Adapter（オンプレ移行先候補）¶

• Exchange: recerdo.topic（topic exchange）
• Routing Key: 論理トピック名
• DLQ: x-dead-letter-exchange で recerdo.dlx にルーティング
• 管理UI 内蔵（15672）、Admin Console から iframe 埋め込み可能

5. マイグレーションパス¶

5.1 BullMQ → OCI Queue（Beta → 本番）¶

Phase 1（Dual-Write）: - infra.queue.dualWrite=true で両方に書き込む - Workers は infra.queue.readFrom=old で BullMQ からのみ消費 - 両者でジョブ数が一致するか監視

Phase 2（Read Switch）: - infra.queue.readFrom=new に切替 → OCI Queue の Worker が活性化 - BullMQ 側に残留ジョブがないことを確認

Phase 3（Single-Write）: - infra.queue.dualWrite=false に戻し、OCI Queue のみ稼働 - BullMQ は冗長として 2 週間維持、問題なければ撤去

5.2 切替時のデータ喪失ゼロ戦略¶

• In-flight ジョブ: 切替直前に BullMQ の一時停止（pause）→ 完了待ち → 新系統解放
• 重複実行: Worker 側が冪等性キーで排除
• データ差分監視: Prometheus + Grafana / OCI Monitoring で処理レート・失敗率を 1 分粒度で比較（AWS CloudWatch は不使用）

6. 観測可能性（Observability）¶

6.1 メトリクス（Prometheus Exposure）¶

6.2 アラート¶

• P1: queue_dlq_total が 5分で 10 件超過 → PagerDuty
• P2: queue_depth が 15 分で 1,000 超過 → Slack #alerts
• P2: queue_duration_seconds{quantile=0.95} が 1 分超過 → Slack

6.3 ログフォーマット¶

{
  "ts": "2026-04-19T12:34:56Z",
  "level": "info",
  "svc": "notification-worker",
  "queue.provider": "bullmq",
  "queue.topic": "notification.push",
  "job.id": "01HMFG2...",
  "job.attempt": 2,
  "trace_id": "0af7651916cd43dd8448eb211c80319c",
  "event": "job.completed",
  "duration_ms": 342
}

7. セキュリティ¶

• 認可: プロデューサ/コンシューマは サービスアカウント で認証、Topic 単位の ACL（書き/読み）を設定
• 暗号化: 全アダプタで TLS 必須、ペイロードに PII が含まれる場合はアプリ層で追加暗号化
• 監査: Queue 操作は Audit Svc へサンプリング 10% で送信、Kill Switch 操作は 100%

8. 参考文献¶

• BullMQ Documentation
• Sidekiq Documentation
• OCI Queue Service
• hibiken/asynq — Simple, reliable, and efficient distributed task queue in Go
• CloudEvents Specification
• omniqueue-rs — Abstraction Layer Reference Implementation
• When Services Need to Talk: Kafka vs RabbitMQ vs BullMQ

9. 関連ドキュメント¶

• デプロイメント戦略
• 環境抽象化 & Feature Flag
• Admin Console Svc（MS）
• Feature Flag 管理システム（MS）

最終更新: 2026-04-19 ポリシー適用