イベントフロー:

[Order Service]
  │ publish: order.created
  │
  ├──→ [Payment Service]
  │     subscribe: order.created
  │     publish: payment.completed / payment.failed
  │
  ├──→ [Inventory Service]
  │     subscribe: payment.completed
  │     publish: stock.reserved / stock.insufficient
  │
  ├──→ [Shipping Service]
  │     subscribe: stock.reserved
  │     publish: shipment.created
  │
  └──→ [Notification Service]
        subscribe: order.created, payment.completed,
                   stock.reserved, shipment.created

イベント一覧:
  1. order.created       → 注文が作成された
  2. payment.completed   → 決済が完了した
  3. payment.failed      → 決済が失敗した
  4. stock.reserved      → 在庫が確保された
  5. stock.insufficient  → 在庫が不足
  6. shipment.created    → 配送が手配された

判断: 注文確定後のフロー全体を非同期イベントで連携
      （ユーザーには「注文受付済み」を即座に返す）

選定: Apache Kafka

理由:
1. 順序保証: orderId をパーティションキーにすれば、
   同一注文のイベントの順序が保証される

2. 再処理: Kafka はメッセージを保持するため、
   offset を巻き戻して再処理が可能

3. リアルタイム分析: Kafka Streams や ksqlDB で
   ストリーム処理を後から追加できる

4. 複数Consumer Group: 決済、在庫、通知の各サービスが
   独立して同じイベントを読める

Topic設計:
  - order-events (partitions: 12, key: orderId)
  - payment-events (partitions: 6, key: orderId)
  - inventory-events (partitions: 6, key: productId)
  - shipment-events (partitions: 6, key: orderId)

// 1. order.created
interface OrderCreatedEvent {
  specversion: "1.0";
  id: string;
  source: "/order-service";
  type: "com.example.order.created";
  time: string;
  datacontenttype: "application/json";
  correlationid: string;
  data: {
    orderId: string;
    userId: string;
    items: Array<{
      productId: string;
      quantity: number;
      unitPrice: number;
    }>;
    totalAmount: number;
    currency: string;
    shippingAddress: {
      prefecture: string;
      city: string;
      line1: string;
      postalCode: string;
    };
  };
}

// 2. payment.completed
interface PaymentCompletedEvent {
  specversion: "1.0";
  id: string;
  source: "/payment-service";
  type: "com.example.payment.completed";
  time: string;
  correlationid: string;
  causationid: string;  // order.created のID
  data: {
    paymentId: string;
    orderId: string;
    amount: number;
    currency: string;
    method: "credit_card" | "bank_transfer" | "convenience_store";
    transactionId: string;
  };
}

// 3. stock.reserved
interface StockReservedEvent {
  specversion: "1.0";
  id: string;
  source: "/inventory-service";
  type: "com.example.stock.reserved";
  time: string;
  correlationid: string;
  causationid: string;  // payment.completed のID
  data: {
    reservationId: string;
    orderId: string;
    items: Array<{
      productId: string;
      quantity: number;
      warehouseId: string;
    }>;
    expiresAt: string;  // 予約の有効期限
  };
}

// 1. Payment Service: orderIdをキーにした冪等性
class PaymentEventHandler {
  async handleOrderCreated(event: OrderCreatedEvent): Promise<void> {
    const orderId = event.data.orderId;

    // orderIdで既に決済済みか確認
    const existingPayment = await this.paymentRepo.findByOrderId(orderId);
    if (existingPayment) {
      console.log(`Payment already processed for order: ${orderId}`);
      return; // 二重決済を防止
    }

    // 決済処理を実行
    const payment = await this.paymentGateway.charge({
      amount: event.data.totalAmount,
      idempotencyKey: `pay-${orderId}`, // 決済ゲートウェイにも冪等キー
    });

    await this.paymentRepo.save({ orderId, ...payment });
  }
}

// 2. Inventory Service: eventIdによるデデュプリケーション
class InventoryEventHandler {
  async handlePaymentCompleted(event: PaymentCompletedEvent): Promise<void> {
    // イベントIDで重複チェック
    if (await this.processedEvents.has(event.id)) return;

    await this.db.transaction(async (tx) => {
      for (const item of event.data.items) {
        await tx.query(
          `UPDATE inventory
           SET reserved = reserved + $1
           WHERE product_id = $2 AND available >= $1`,
          [item.quantity, item.productId]
        );
      }
      // 処理済みイベントを記録（同一トランザクション内）
      await tx.query(
        `INSERT INTO processed_events (event_id) VALUES ($1) ON CONFLICT DO NOTHING`,
        [event.id]
      );
    });
  }
}

// 3. Notification Service: 通知IDによる重複チェック
class NotificationEventHandler {
  async handleEvent(event: DomainEvent): Promise<void> {
    const notificationKey = `${event.type}:${event.data.orderId}`;

    // 送信済みか確認
    if (await this.notificationLog.exists(notificationKey)) return;

    await this.sendNotification(event);
    await this.notificationLog.record(notificationKey);
  }
}