Event-Driven Architecture 実践ガイド：CQRS、Event Sourcing、Sagaパターン

// TypeScript - Domain Event basic structure
interface DomainEvent {
  eventId: string
  eventType: string
  aggregateId: string
  aggregateType: string
  timestamp: Date
  version: number
  payload: Record<string, unknown>
  metadata: {
    correlationId: string
    causationId: string
    userId?: string
  }
}

// Order creation event example
const orderPlacedEvent: DomainEvent = {
  eventId: 'evt-550e8400-e29b-41d4-a716-446655440000',
  eventType: 'OrderPlaced',
  aggregateId: 'order-12345',
  aggregateType: 'Order',
  timestamp: new Date('2026-03-14T09:00:00Z'),
  version: 1,
  payload: {
    customerId: 'cust-67890',
    items: [
      { productId: 'prod-001', quantity: 2, price: 29900 },
      { productId: 'prod-002', quantity: 1, price: 15000 },
    ],
    totalAmount: 74800,
    shippingAddress: {
      city: 'ソウル',
      district: '江南区',
      detail: 'テヘラン路123',
    },
  },
  metadata: {
    correlationId: 'corr-abc123',
    causationId: 'cmd-place-order-001',
    userId: 'user-admin-01',
  },
}

// TypeScript - CQRS Command Side

// Command definition
interface PlaceOrderCommand {
  type: 'PlaceOrder'
  customerId: string
  items: Array<{
    productId: string
    quantity: number
    price: number
  }>
  shippingAddress: string
}

// Command Handler
class PlaceOrderHandler {
  constructor(
    private orderRepository: OrderWriteRepository,
    private eventBus: EventBus,
    private inventoryService: InventoryService
  ) {}

  async handle(command: PlaceOrderCommand): Promise<string> {
    // 1. ビジネスルール検証
    await this.inventoryService.validateStock(command.items)

    // 2. Aggregate作成
    const order = Order.create({
      customerId: command.customerId,
      items: command.items,
      shippingAddress: command.shippingAddress,
    })

    // 3. 書き込みストアに保存
    await this.orderRepository.save(order)

    // 4. ドメインイベント発行（読み取りモデル同期用）
    for (const event of order.getDomainEvents()) {
      await this.eventBus.publish(event)
    }

    return order.id
  }
}

// Query Side - 読み取り専用モデル
interface OrderReadModel {
  orderId: string
  customerName: string
  orderDate: string
  status: string
  totalAmount: number
  itemCount: number
  lastUpdated: string
}

// Query Handler
class GetOrdersQueryHandler {
  constructor(private readDb: ReadDatabase) {}

  async handle(query: {
    customerId: string
    status?: string
    page: number
    limit: number
  }): Promise<OrderReadModel[]> {
    // 読み取り専用の非正規化テーブルから直接照会
    return this.readDb.query(
      `SELECT order_id, customer_name, order_date, status,
              total_amount, item_count, last_updated
       FROM order_read_model
       WHERE customer_id = ?
       ${query.status ? 'AND status = ?' : ''}
       ORDER BY order_date DESC
       LIMIT ? OFFSET ?`,
      [query.customerId, query.status, query.limit, query.page * query.limit]
    )
  }
}

// TypeScript - Event-based Projection

class OrderProjection {
  constructor(private readDb: ReadDatabase) {}

  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderPlaced':
        await this.onOrderPlaced(event)
        break
      case 'OrderShipped':
        await this.onOrderShipped(event)
        break
      case 'OrderCancelled':
        await this.onOrderCancelled(event)
        break
    }
  }

  private async onOrderPlaced(event: DomainEvent): Promise<void> {
    const payload = event.payload as {
      customerId: string
      items: Array<{ quantity: number; price: number }>
      totalAmount: number
    }

    await this.readDb.upsert('order_read_model', {
      order_id: event.aggregateId,
      customer_id: payload.customerId,
      status: 'PLACED',
      total_amount: payload.totalAmount,
      item_count: payload.items.reduce((sum, i) => sum + i.quantity, 0),
      order_date: event.timestamp,
      last_updated: event.timestamp,
      version: event.version,
    })
  }

  private async onOrderShipped(event: DomainEvent): Promise<void> {
    const payload = event.payload as { trackingNumber: string }

    // 冪等性保証：バージョンチェック
    await this.readDb.updateWhere(
      'order_read_model',
      {
        status: 'SHIPPED',
        tracking_number: payload.trackingNumber,
        last_updated: event.timestamp,
        version: event.version,
      },
      { order_id: event.aggregateId, version: event.version - 1 }
    )
  }

  private async onOrderCancelled(event: DomainEvent): Promise<void> {
    const payload = event.payload as { reason: string }

    await this.readDb.updateWhere(
      'order_read_model',
      {
        status: 'CANCELLED',
        cancellation_reason: payload.reason,
        last_updated: event.timestamp,
        version: event.version,
      },
      { order_id: event.aggregateId }
    )
  }
}

// TypeScript - Event Sourcing Aggregate

abstract class EventSourcedAggregate {
  private uncommittedEvents: DomainEvent[] = []
  protected version: number = 0

  abstract get id(): string

  protected apply(event: DomainEvent): void {
    this.when(event)
    this.version++
    this.uncommittedEvents.push(event)
  }

  protected abstract when(event: DomainEvent): void

  loadFromHistory(events: DomainEvent[]): void {
    for (const event of events) {
      this.when(event)
      this.version++
    }
  }

  getUncommittedEvents(): DomainEvent[] {
    return [...this.uncommittedEvents]
  }

  clearUncommittedEvents(): void {
    this.uncommittedEvents = []
  }
}

class Order extends EventSourcedAggregate {
  private _id: string = ''
  private _customerId: string = ''
  private _items: OrderItem[] = []
  private _status: OrderStatus = OrderStatus.DRAFT
  private _totalAmount: number = 0

  get id(): string {
    return this._id
  }

  static create(params: { orderId: string; customerId: string; items: OrderItem[] }): Order {
    const order = new Order()
    const totalAmount = params.items.reduce((sum, item) => sum + item.price * item.quantity, 0)

    order.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderPlaced',
      aggregateId: params.orderId,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: 1,
      payload: {
        customerId: params.customerId,
        items: params.items,
        totalAmount,
      },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'create',
      },
    })

    return order
  }

  confirm(): void {
    if (this._status !== OrderStatus.PLACED) {
      throw new Error(`Cannot confirm order in status: ${this._status}`)
    }

    this.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderConfirmed',
      aggregateId: this._id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version + 1,
      payload: { confirmedAt: new Date().toISOString() },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'confirm',
      },
    })
  }

  cancel(reason: string): void {
    if (this._status === OrderStatus.CANCELLED) {
      return // 冪等性保証
    }

    this.apply({
      eventId: crypto.randomUUID(),
      eventType: 'OrderCancelled',
      aggregateId: this._id,
      aggregateType: 'Order',
      timestamp: new Date(),
      version: this.version + 1,
      payload: { reason, cancelledAt: new Date().toISOString() },
      metadata: {
        correlationId: crypto.randomUUID(),
        causationId: 'cancel',
      },
    })
  }

  protected when(event: DomainEvent): void {
    switch (event.eventType) {
      case 'OrderPlaced': {
        const p = event.payload as {
          customerId: string
          items: OrderItem[]
          totalAmount: number
        }
        this._id = event.aggregateId
        this._customerId = p.customerId
        this._items = p.items
        this._totalAmount = p.totalAmount
        this._status = OrderStatus.PLACED
        break
      }
      case 'OrderConfirmed':
        this._status = OrderStatus.CONFIRMED
        break
      case 'OrderCancelled':
        this._status = OrderStatus.CANCELLED
        break
    }
  }
}

enum OrderStatus {
  DRAFT = 'DRAFT',
  PLACED = 'PLACED',
  CONFIRMED = 'CONFIRMED',
  SHIPPED = 'SHIPPED',
  CANCELLED = 'CANCELLED',
}

interface OrderItem {
  productId: string
  quantity: number
  price: number
}

# Python - Snapshot-based Event Store

from dataclasses import dataclass, field
from datetime import datetime
from typing import Any
import json


@dataclass
class Snapshot:
    aggregate_id: str
    aggregate_type: str
    version: int
    state: dict
    created_at: datetime = field(default_factory=datetime.utcnow)


class EventStore:
    SNAPSHOT_INTERVAL = 100  # 100イベントごとにスナップショット作成

    def __init__(self, db_connection):
        self.db = db_connection

    async def save_events(
        self, aggregate_id: str, events: list[dict], expected_version: int
    ) -> None:
        """楽観的同時実行制御とともにイベントを保存"""
        async with self.db.transaction():
            current_version = await self._get_current_version(aggregate_id)
            if current_version != expected_version:
                raise ConcurrencyError(
                    f"Expected version {expected_version}, "
                    f"but current version is {current_version}"
                )

            for i, event in enumerate(events):
                version = expected_version + i + 1
                await self.db.execute(
                    """INSERT INTO event_store
                       (event_id, aggregate_id, aggregate_type,
                        event_type, version, payload, metadata, created_at)
                       VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
                    (
                        event["event_id"],
                        aggregate_id,
                        event["aggregate_type"],
                        event["event_type"],
                        version,
                        json.dumps(event["payload"]),
                        json.dumps(event["metadata"]),
                        datetime.utcnow(),
                    ),
                )

            new_version = expected_version + len(events)
            if new_version % self.SNAPSHOT_INTERVAL == 0:
                await self._create_snapshot(aggregate_id, new_version)

    async def load_aggregate(self, aggregate_id: str) -> tuple[list[dict], int]:
        """スナップショットからイベントをロードして状態を復元"""
        snapshot = await self._get_latest_snapshot(aggregate_id)

        if snapshot:
            events = await self.db.fetch_all(
                """SELECT * FROM event_store
                   WHERE aggregate_id = ? AND version > ?
                   ORDER BY version ASC""",
                (aggregate_id, snapshot.version),
            )
            return events, snapshot
        else:
            events = await self.db.fetch_all(
                """SELECT * FROM event_store
                   WHERE aggregate_id = ?
                   ORDER BY version ASC""",
                (aggregate_id,),
            )
            return events, None

    async def _create_snapshot(
        self, aggregate_id: str, version: int
    ) -> None:
        """現在の状態をスナップショットとして保存"""
        events, _ = await self.load_aggregate(aggregate_id)
        aggregate = self._rebuild_aggregate(events)
        await self.db.execute(
            """INSERT INTO snapshots
               (aggregate_id, aggregate_type, version, state, created_at)
               VALUES (?, ?, ?, ?, ?)""",
            (
                aggregate_id,
                aggregate.aggregate_type,
                version,
                json.dumps(aggregate.to_dict()),
                datetime.utcnow(),
            ),
        )

    async def _get_current_version(self, aggregate_id: str) -> int:
        result = await self.db.fetch_one(
            "SELECT MAX(version) FROM event_store WHERE aggregate_id = ?",
            (aggregate_id,),
        )
        return result[0] if result[0] else 0


class ConcurrencyError(Exception):
    pass

// TypeScript - Saga Orchestrator (Order Processing)

interface SagaStep {
  name: string
  action: () => Promise<void>
  compensation: () => Promise<void>
}

class OrderSagaOrchestrator {
  private completedSteps: SagaStep[] = []
  private sagaLog: SagaLogEntry[] = []

  constructor(
    private paymentService: PaymentService,
    private inventoryService: InventoryService,
    private shippingService: ShippingService,
    private sagaStore: SagaStore
  ) {}

  async execute(orderId: string, orderData: OrderData): Promise<SagaResult> {
    const sagaId = crypto.randomUUID()

    const steps: SagaStep[] = [
      {
        name: 'ProcessPayment',
        action: async () => {
          await this.paymentService.processPayment({
            orderId,
            amount: orderData.totalAmount,
            customerId: orderData.customerId,
          })
        },
        compensation: async () => {
          await this.paymentService.refundPayment({
            orderId,
            amount: orderData.totalAmount,
          })
        },
      },
      {
        name: 'ReserveInventory',
        action: async () => {
          await this.inventoryService.reserve({
            orderId,
            items: orderData.items,
          })
        },
        compensation: async () => {
          await this.inventoryService.releaseReservation({
            orderId,
            items: orderData.items,
          })
        },
      },
      {
        name: 'CreateShipment',
        action: async () => {
          await this.shippingService.createShipment({
            orderId,
            address: orderData.shippingAddress,
            items: orderData.items,
          })
        },
        compensation: async () => {
          await this.shippingService.cancelShipment({ orderId })
        },
      },
    ]

    try {
      for (const step of steps) {
        await this.logStep(sagaId, step.name, 'STARTED')

        try {
          await step.action()
          this.completedSteps.push(step)
          await this.logStep(sagaId, step.name, 'COMPLETED')
        } catch (error) {
          await this.logStep(sagaId, step.name, 'FAILED', error)
          await this.compensate(sagaId)
          return {
            success: false,
            sagaId,
            failedStep: step.name,
            error: (error as Error).message,
          }
        }
      }

      await this.sagaStore.markCompleted(sagaId)
      return { success: true, sagaId }
    } catch (compensationError) {
      await this.sagaStore.markRequiresIntervention(sagaId)
      throw new SagaCompensationFailedError(sagaId, compensationError as Error)
    }
  }

  private async compensate(sagaId: string): Promise<void> {
    const stepsToCompensate = [...this.completedSteps].reverse()

    for (const step of stepsToCompensate) {
      try {
        await this.logStep(sagaId, step.name, 'COMPENSATING')
        await step.compensation()
        await this.logStep(sagaId, step.name, 'COMPENSATED')
      } catch (error) {
        await this.logStep(sagaId, step.name, 'COMPENSATION_FAILED', error)
        await this.sagaStore.enqueueRetry(sagaId, step.name)
        throw error
      }
    }
  }

  private async logStep(
    sagaId: string,
    stepName: string,
    status: string,
    error?: unknown
  ): Promise<void> {
    const entry: SagaLogEntry = {
      sagaId,
      stepName,
      status,
      timestamp: new Date(),
      error: error ? (error as Error).message : undefined,
    }
    this.sagaLog.push(entry)
    await this.sagaStore.appendLog(entry)
  }
}

interface SagaResult {
  success: boolean
  sagaId: string
  failedStep?: string
  error?: string
}

interface SagaLogEntry {
  sagaId: string
  stepName: string
  status: string
  timestamp: Date
  error?: string
}

# Python - Choreography-based Saga (event subscription approach)

from abc import ABC, abstractmethod
from dataclasses import dataclass
from enum import Enum
import asyncio
import json


class EventBus:
    """Simple in-memory event bus (use Kafka/RabbitMQ in production)"""

    def __init__(self):
        self._handlers: dict[str, list] = {}

    def subscribe(self, event_type: str, handler):
        if event_type not in self._handlers:
            self._handlers[event_type] = []
        self._handlers[event_type].append(handler)

    async def publish(self, event_type: str, payload: dict):
        handlers = self._handlers.get(event_type, [])
        for handler in handlers:
            await handler(payload)


class PaymentService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        event_bus.subscribe("OrderCreated", self.on_order_created)
        event_bus.subscribe("InventoryReservationFailed", self.on_inventory_failed)

    async def on_order_created(self, payload: dict):
        """注文作成時に決済処理"""
        try:
            order_id = payload["order_id"]
            amount = payload["total_amount"]

            payment_result = await self._process_payment(
                order_id, amount
            )

            await self.event_bus.publish("PaymentProcessed", {
                "order_id": order_id,
                "payment_id": payment_result["payment_id"],
                "amount": amount,
            })
        except Exception as e:
            await self.event_bus.publish("PaymentFailed", {
                "order_id": payload["order_id"],
                "reason": str(e),
            })

    async def on_inventory_failed(self, payload: dict):
        """在庫不足時に決済返金（補償トランザクション）"""
        order_id = payload["order_id"]
        await self._refund_payment(order_id)
        await self.event_bus.publish("PaymentRefunded", {
            "order_id": order_id,
        })

    async def _process_payment(self, order_id: str, amount: int) -> dict:
        return {"payment_id": f"pay-{order_id}"}

    async def _refund_payment(self, order_id: str) -> None:
        pass


class InventoryService:
    def __init__(self, event_bus: EventBus):
        self.event_bus = event_bus
        event_bus.subscribe("PaymentProcessed", self.on_payment_processed)

    async def on_payment_processed(self, payload: dict):
        """決済完了時に在庫確保"""
        try:
            order_id = payload["order_id"]
            await self._reserve_inventory(order_id)

            await self.event_bus.publish("InventoryReserved", {
                "order_id": order_id,
            })
        except InsufficientStockError:
            await self.event_bus.publish("InventoryReservationFailed", {
                "order_id": payload["order_id"],
                "reason": "insufficient_stock",
            })


class InsufficientStockError(Exception):
    pass

// TypeScript - Idempotency guarantee pattern

class IdempotentEventHandler {
  constructor(
    private processedEvents: ProcessedEventStore,
    private handler: EventHandler
  ) {}

  async handle(event: DomainEvent): Promise<void> {
    const isProcessed = await this.processedEvents.exists(event.eventId)
    if (isProcessed) {
      console.log(`Event ${event.eventId} already processed, skipping`)
      return
    }

    try {
      await this.handler.handle(event)

      await this.processedEvents.markAsProcessed(event.eventId, {
        processedAt: new Date(),
        ttl: 60 * 60 * 24 * 7,
      })
    } catch (error) {
      throw error
    }
  }
}

class RedisProcessedEventStore implements ProcessedEventStore {
  constructor(private redis: Redis) {}

  async exists(eventId: string): Promise<boolean> {
    const result = await this.redis.exists(`processed:${eventId}`)
    return result === 1
  }

  async markAsProcessed(
    eventId: string,
    options: { processedAt: Date; ttl: number }
  ): Promise<void> {
    await this.redis.setex(`processed:${eventId}`, options.ttl, options.processedAt.toISOString())
  }
}

// TypeScript - Event Upcaster (Schema Evolution)

class EventUpcaster {
  private upcasters: Map<string, Map<number, (event: DomainEvent) => DomainEvent>> = new Map()

  register(
    eventType: string,
    fromVersion: number,
    upcaster: (event: DomainEvent) => DomainEvent
  ): void {
    if (!this.upcasters.has(eventType)) {
      this.upcasters.set(eventType, new Map())
    }
    this.upcasters.get(eventType)!.set(fromVersion, upcaster)
  }

  upcast(event: DomainEvent): DomainEvent {
    const typeUpcasters = this.upcasters.get(event.eventType)
    if (!typeUpcasters) return event

    let currentEvent = event
    let schemaVersion = (event.metadata as any).schemaVersion || 1

    while (typeUpcasters.has(schemaVersion)) {
      const upcasterFn = typeUpcasters.get(schemaVersion)!
      currentEvent = upcasterFn(currentEvent)
      schemaVersion++
    }

    return currentEvent
  }
}

const upcaster = new EventUpcaster()

upcaster.register('OrderPlaced', 1, (event) => {
  const payload = event.payload as any
  return {
    ...event,
    payload: {
      ...payload,
      shippingAddress: {
        full: payload.shippingAddress,
        city: '',
        zipCode: '',
      },
    },
    metadata: {
      ...event.metadata,
      schemaVersion: 2,
    },
  }
})