event-driven-architecture

イベント駆動アーキテクチャ

概要

このスキルでは、非同期メッセージングによってサービスを疎結合にする、イベント駆動システムの設計と実装を扱います。メッセージブローカーの選定と統合（Kafka、RabbitMQ、SQS/SNS、NATS）、イベントソーシングとCQRSのパターン、分散トランザクションにおけるサーガオーケストレーション、障害処理のためのデッドレターキュー、冪等性パターン、イベントスキーマの進化、トランザクショナルアウトボックスパターン、コンシューマーグループの管理について説明します。

マイクロサービスアーキテクチャの構築、分散ワークフローの実装、独立したデプロイを可能にするサービスの疎結合、結果整合性への対応、または同期的なサービス間呼び出しの非同期イベントへの置き換えを行う際に、このスキルを使用してください。

---

基本原則

1. イベントは事実、コマンドは要求です - イベントは発生した事実（OrderPlaced、PaymentReceived）を表します。コマンドはアクション（ProcessPayment、ShipOrder）を要求します。この違いが適切なシステム設計の指針となります。イベントはブロードキャストされ、コマンドはポイントツーポイントで送信されます。
2. 冪等性は必須です - メッセージは少なくとも1回は配信されます（場合によってはそれ以上です）。すべてのコンシューマーは、重複メッセージを適切に処理する必要があります。重複排除には、イベントID、バージョンチェック、またはデータベース制約を使用してください。
3. コンシューマーを壊さないスキーマの進化 - イベントスキーマは変化します。後方互換性のある進化（フィールドの追加は行い、削除や名前変更は行わない）を使用してください。スキーマに明示的なバージョンを付け、移行中は複数のバージョンをサポートしてください。
4. デッドレターはゴミではありません - 処理できないメッセージはデッドレターキューに送られます。これらはバグ、データの問題、またはエッジケースを示しています。DLQを監視し、増加時にアラートを発し、メッセージを再生するためのツールを構築してください。
5. ローカルトランザクションと結果整合性 - 各サービスは自身のデータを管理し、ローカルトランザクション内でイベントを処理します。サービスをまたぐ整合性は即時ではなく、結果整合性です。UIとワークフローはこれに対応できるよう設計してください。

---

主要なパターン

パターン1：トランザクショナルアウトボックスによるイベント発行

使用する場面： データベースの更新とイベントの発行をアトミックに行う必要がある場合です。これにより、データベースへの書き込みは成功したもののイベントの発行が失敗する（またはその逆）「二重書き込み」問題を防げます。

実装：

// The Transactional Outbox Pattern
// 1. Write the event to an outbox table in the SAME database transaction as the business data
// 2. A separate poller/CDC process reads the outbox and publishes to the message broker

interface OutboxEvent {
  id: string;
  aggregateType: string;
  aggregateId: string;
  eventType: string;
  payload: Record<string, unknown>;
  createdAt: Date;
  publishedAt: Date | null;
}

// Step 1: Business operation + outbox write in one transaction
async function placeOrder(order: CreateOrderInput): Promise<Order> {
  return await prisma.$transaction(async (tx) => {
    // Business logic
    const createdOrder = await tx.order.create({
      data: {
        customerId: order.customerId,
        items: { create: order.items },
        total: calculateTotal(order.items),
        status: "PLACED",
      },
      include: { items: true },
    });

    // Write event to outbox (same transaction!)
    await tx.outboxEvent.create({
      data: {
        id: crypto.randomUUID(),
        aggregateType: "Order",
        aggregateId: createdOrder.id,
        eventType: "OrderPlaced",
        payload: {
          orderId: createdOrder.id,
          customerId: createdOrder.customerId,
          items: createdOrder.items,
          total: createdOrder.total,
          placedAt: new Date().toISOString(),
        },
      },
    });

    return createdOrder;
  });
}

// Step 2: Outbox poller publishes events to Kafka/RabbitMQ
async function processOutbox(): Promise<void> {
  const unpublished = await prisma.outboxEvent.findMany({
    where: { publishedAt: null },
    orderBy: { createdAt: "asc" },
    take: 100,
  });

  for (const event of unpublished) {
    try {
      await kafka.producer.send({
        topic: `${event.aggregateType}.${event.eventType}`,
        messages: [
          {
            key: event.aggregateId,
            value: JSON.stringify({
              eventId: event.id,
              eventType: event.eventType,
              aggregateId: event.aggregateId,
              payload: event.payload,
              timestamp: event.createdAt.toISOString(),
            }),
            headers: {
              "event-type": event.eventType,
              "event-id": event.id,
            },
          },
        ],
      });

      await prisma.outboxEvent.update({
        where: { id: event.id },
        data: { publishedAt: new Date() },
      });
    } catch (error) {
      console.error(`Failed to publish outbox event ${event.id}:`, error);
      // Will be retried on next poll cycle
    }
  }
}

// Run poller on interval
setInterval(processOutbox, 5000); // Every 5 seconds

理由： 二重書き込み問題は、イベント駆動システムにおけるデータ不整合の最も一般的な原因です。データベースに書き込んでからイベントを発行する場合、データベースへの書き込みが成功した後で発行が失敗する可能性があります。アウトボックスパターンにより、業務データが書き込まれれば、イベントも最終的に必ず発行されます。

---

パターン2：冪等なイベントコンシューマー

使用する場面： すべてのイベントコンシューマーで使用します。主要なブローカーでは、メッセージは少なくとも1回配信されます。

実装：

// Kafka consumer with idempotency
import { Kafka, EachMessagePayload } from "kafkajs";

const kafka = new Kafka({
  clientId: "order-service",
  brokers: [process.env.KAFKA_BROKERS!],
});

const consumer = kafka.consumer({ groupId: "payment-processor" });

interface DomainEvent {
  eventId: string;
  eventType: string;
  aggregateId: string;
  payload: Record<string, unknown>;
  timestamp: string;
}

// Event handler registry
const handlers: Record<string, (event: DomainEvent) => Promise<void>> = {
  OrderPlaced: async (event) => {
    const { orderId, total, customerId } = event.payload as {
      orderId: string;
      total: number;
      customerId: string;
    };

    // Create payment intent (idempotent via orderId)
    await processPayment(orderId, total, customerId);
  },

  PaymentFailed: async (event) => {
    const { orderId } = event.payload as { orderId: string };
    await cancelOrder(orderId);
  },
};

async function startConsumer() {
  await consumer.connect();
  await consumer.subscribe({
    topics: ["Order.OrderPlaced", "Payment.PaymentFailed"],
    fromBeginning: false,
  });

  await consumer.run({
    eachMessage: async ({ topic, partition, message }: EachMessagePayload) => {
      const event: DomainEvent = JSON.parse(message.value!.toString());

      // 1. Idempotency check - have we already processed this event?
      const alreadyProcessed = await prisma.processedEvent.findUnique({
        where: { eventId: event.eventId },
      });

      if (alreadyProcessed) {
        console.log(`Skipping duplicate event: ${event.eventId}`);
        return;
      }

      // 2. Process the event
      const handler = handlers[event.eventType];
      if (!handler) {
        console.warn(`No handler for event type: ${event.eventType}`);
        return;
      }

      try {
        await prisma.$transaction(async (tx) => {
          // Record event as processed (within the same transaction as business logic)
          await tx.processedEvent.create({
            data: {
              eventId: event.eventId,
              eventType: event.eventType,
              processedAt: new Date(),
            },
          });

          // Execute business logic
          await handler(event);
        });
      } catch (error) {
        console.error(`Error processing event ${event.eventId}:`, error);
        // Don't commit offset - message will be redelivered
        throw error;
      }
    },
  });
}

理由： Kafka、RabbitMQ、SQSはいずれも少なくとも1回の配信を保証しているため、リバランス、再試行、ネットワークの問題が発生すると、コンシューマーは重複メッセージを受け取ります。業務ロジックと同じトランザクション内で処理済みイベントIDを記録することで、厳密に1回だけ処理するセマンティクスを保証できます。

---

パターン3：サーガオーケストレーション

使用する場面： 業務プロセスが複数のサービスにまたがり、いずれかのステップが失敗した場合に、連携してロールバックする必要があるときです。

実装：

// Saga orchestrator for order fulfillment
// Order -> Payment -> Inventory -> Shipping

interface SagaStep {
  name: string;
  execute: (context: SagaContext) => Promise<void>;
  compensate: (context: SagaContext) => Promise<void>;
}

interface SagaContext {
  orderId: string;
  customerId: string;
  items: OrderItem[];
  total: number;
  // Accumulated state from previous steps
  paymentId?: string;
  reservationId?: string;
  shipmentId?: string;
}

const orderFulfillmentSaga: SagaStep[] = [
  {
    name: "processPayment",
    execute: async (ctx) => {
      const payment = await paymentService.charge({
        orderId: ctx.orderId,
        amount: ctx.total,
        customerId: ctx.customerId,
      });
      ctx.paymentId = payment.id;
    },
    compensate: async (ctx) => {
      if (ctx.paymentId) {
        await paymentService.refund(ctx.paymentId);
      }
    },
  },
  {
    name: "reserveInventory",
    execute: async (ctx) => {
      const reservation = await inventoryService.reserve({
        orderId: ctx.orderId,
        items: ctx.items,
      });
      ctx.reservationId = reservation.id;
    },
    compensate: async (ctx) => {
      if (ctx.reservationId) {
        await inventoryService.release(ctx.reservationId);
      }
    },
  },
  {
    name: "createShipment",
    execute: async (ctx) => {
      const shipment = await shippingService.create({
        orderId: ctx.orderId,
        items: ctx.items,
        customerId: ctx.customerId,
      });
      ctx.shipmentId = shipment.id;
    },
    compensate: async (ctx) => {
      if (ctx.shipmentId) {
        await shippingService.cancel(ctx.shipmentId);
      }
    },
  },
];

async function executeSaga(
  steps: SagaStep[],
  context: SagaContext
): Promise<void> {
  const completedSteps: SagaStep[] = [];

  for (const step of steps) {
    try {
      console.log(`Executing saga step: ${step.name}`);
      await step.execute(context);
      completedSteps.push(step);
    } catch (error) {
      console.error(`Saga step "${step.name}" failed:`, error);

      // Compensate in reverse order
      for (const completedStep of completedSteps.reverse()) {
        try {
          console.log(`Compensating: ${completedStep.name}`);
          await completedStep.compensate(context);
        } catch (compensateError) {
          // Log compensation failure for manual intervention
          console.error(
            `CRITICAL: Compensation for "${completedStep.name}" failed:`,
            compensateError
          );
          await alertOps({
            type: "saga_compensation_failure",
            saga: "orderFulfillment",
            step: completedStep.name,
            orderId: context.orderId,
            error: compensateError,
          });
        }
      }

      throw new SagaFailedError(step.name, error as Error);
    }
  }
}

理由： サービス間の分散トランザクションは実用的ではありません（2フェーズコミットはスケールしません）。サーガは代替手段として、ローカルトランザクションを順に実行し、ロールバック用の補償アクションを行います。支払いは成功したものの在庫の確保に失敗した場合、サーガが自動的に支払いを返金します。

---

パターン 4：デッドレターキューの処理

使用するタイミング： イベント駆動システムにはすべて、処理できないメッセージに対する方針が必要です。

実装：

// SQS dead letter queue handling pattern
import { SQSClient, SendMessageCommand, ReceiveMessageCommand } from "@aws-sdk/client-sqs";

// Configure main queue with DLQ
const queueConfig = {
  RedrivePolicy: JSON.stringify({
    deadLetterTargetArn: process.env.DLQ_ARN,
    maxReceiveCount: 3, // Move to DLQ after 3 failed processing attempts
  }),
};

// DLQ monitoring and replay
async function monitorDLQ(): Promise<void> {
  const messages = await sqs.send(new ReceiveMessageCommand({
    QueueUrl: process.env.DLQ_URL,
    MaxNumberOfMessages: 10,
    WaitTimeSeconds: 20,
    MessageAttributeNames: ["All"],
  }));

  if (messages.Messages && messages.Messages.length > 0) {
    // Alert operations team
    await alertOps({
      type: "dlq_messages",
      count: messages.Messages.length,
      queue: "order-events-dlq",
      sample: messages.Messages[0].Body,
    });
  }
}

// Replay DLQ messages back to the main queue
async function replayDLQMessages(
  count: number,
  filter?: (msg: Record<string, unknown>) => boolean
): Promise<{ replayed: number; skipped: number }> {
  let replayed = 0;
  let skipped = 0;

  for (let i = 0; i < count; i++) {
    const response = await sqs.send(new ReceiveMessageCommand({
      QueueUrl: process.env.DLQ_URL,
      MaxNumberOfMessages: 1,
      WaitTimeSeconds: 0,
    }));

    const message = response.Messages?.[0];
    if (!message) break;

    const body = JSON.parse(message.Body!);

    if (filter && !filter(body)) {
      skipped++;
      continue;
    }

    // Replay to main queue
    await sqs.send(new SendMessageCommand({
      QueueUrl: process.env.MAIN_QUEUE_URL,
      MessageBody: message.Body!,
      MessageAttributes: message.MessageAttributes,
    }));

    // Delete from DLQ
    await sqs.send(new DeleteMessageCommand({
      QueueUrl: process.env.DLQ_URL,
      ReceiptHandle: message.ReceiptHandle!,
    }));

    replayed++;
  }

  return { replayed, skipped };
}

理由： デッドレターキューは、すべての再試行後も処理に失敗したメッセージを保持します。DLQ がないと、失敗したメッセージがキューを塞いだり、黙って破棄されたりします。DLQ の監視により、バグ（不正な形式のイベント、ハンドラーの不足）を検出できます。再生ツールを使えば、データを失わずに一時的な問題（下流サービスの停止）から復旧できます。

---

メッセージブローカー選定ガイド

ブローカー	最適な用途	順序保証	スループット	永続化
Kafka	高スループットのイベントストリーミング、ログコンパクション	パーティション単位	非常に高い（毎秒数百万件）	保持期間を設定可能
RabbitMQ	タスクキュー、ルーティング、リクエスト・リプライ	キュー単位	高い（毎秒 10 万件）	任意
SQS/SNS	サーバーレス、AWS ネイティブ、運用負荷の低減	FIFO キュー単位	高い（FIFO は毎秒 3,000 件、標準キューは無制限）	14 日間保持
NATS	低レイテンシ、クラウドネイティブ、軽量	JetStream	非常に高い	永続化には JetStream を使用

---

アンチパターン

アンチパターン	問題点	より良いアプローチ
アウトボックスを使わない二重書き込み（DB + イベント）	片方の書き込みが失敗するとデータの不整合が発生する	トランザクショナルアウトボックスパターン
コンシューマーの冪等性がない	再配信時に重複処理が発生する	同一トランザクション内でイベント ID による重複排除を行う
イベントペイロードが大きすぎる（> 1MB）	処理が遅くなり、ブローカーの制限に抵触する	データをサービスに保存し、参照を送信する（クレームチェック）
同期的なイベント処理	非同期アーキテクチャの目的を損なう	キューを使ってイベントを非同期に処理する
DLQ が未設定	失敗したメッセージがキューを塞ぐ、または消失する	アラート機能を含め、必ず DLQ を設定する
イベントスキーマの密結合	スキーマ変更でコンシューマーが破損する	後方互換性を保った進化と明示的なバージョン管理
複数箇所からドメインイベントを発行する	イベント形式に一貫性がなくなり、イベントが欠落する	発行箇所を一元化する（集約ルート／サービスレイヤー）

---

チェックリスト

• DB とイベントの書き込みをアトミックに行うトランザクショナルアウトボックスパターン
• すべてのコンシューマーが冪等である（イベント ID による重複排除）
• アラート機能と再生ツールを備えたデッドレターキューを設定する
• 後方互換性を保って進化できるよう、イベントスキーマをバージョン管理する
• サービスごとにコンシューマーグループを設定する
• DLQ に送る前に指数バックオフを伴う再試行ポリシーを適用する
• サービス間ワークフローに Saga パターンまたはコレオグラフィーパターンを適用する
• 監視項目：コンシューマーのラグ、DLQ の深さ、処理エラー
• 必要な箇所でイベントの順序を維持する（パーティションキー）
• メッセージブローカーのクラスターが耐障害性を備えている（レプリケーション）

---

関連リソース

• スキル： monitoring-observability（コンシューマーのラグ監視）、serverless-development（SQS/Lambda）
• スキル： payment-integration（決済イベントのフロー）、email-systems（非同期イベントとしてのメール）
• ルール： docs/reference/stacks/fullstack-nextjs-nestjs.md（NestJS の CQRS モジュール）