Event Sourcing + CQRS アーキテクチャ実践実装ガイド：イベントストアからプロジェクション・Sagaパターンまで

現在の状態 = fold(初期状態, [イベント1, イベント2, ..., イベントN])

初期残高: 0円
  -> AccountOpened(金額: 0円)         => 残高: 0円
  -> MoneyDeposited(金額: 100万円)    => 残高: 100万円
  -> MoneyWithdrawn(金額: 30万円)     => 残高: 70万円
  -> MoneyDeposited(金額: 50万円)     => 残高: 120万円

// 誤った出金が発生した場合 - イベントを削除しない
// 代わりに補償イベントを発行する
interface MoneyWithdrawnCompensated {
  type: 'MoneyWithdrawnCompensated'
  originalEventId: string
  amount: number
  reason: string
  timestamp: Date
}

[クライアント]
    |
    |-- Command(書込) --> [Command Handler] --> [Write Model / Event Store]
    |                                                     |
    |                                              イベント発行
    |                                                     |
    |                                                     v
    |                                           [Projection Engine]
    |                                                     |
    |                                                     v
    +-- Query(読取) ----> [Query Handler] ----> [Read Model / View DB]

// Command定義
interface CreateOrderCommand {
  orderId: string
  customerId: string
  items: Array<{ productId: string; quantity: number; price: number }>
}

// Command Handler
class OrderCommandHandler {
  constructor(
    private readonly eventStore: EventStore,
    private readonly orderRepository: EventSourcedRepository<Order>
  ) {}

  async handle(command: CreateOrderCommand): Promise<void> {
    // 1. Aggregateロード（イベント再生）
    const order = await this.orderRepository.load(command.orderId)

    // 2. ビジネスロジック実行（イベント生成）
    order.create(command.customerId, command.items)

    // 3. イベント保存
    await this.orderRepository.save(order)
  }
}

CREATE TABLE events (
    -- グローバル一意識別子
    event_id        UUID PRIMARY KEY DEFAULT gen_random_uuid(),
    -- イベントが属するストリーム（Aggregate）識別子
    stream_id       VARCHAR(255) NOT NULL,
    -- ストリーム内のイベント順序（楽観的同時実行制御に使用）
    stream_version  BIGINT NOT NULL,
    -- イベントタイプ（デシリアライズに使用）
    event_type      VARCHAR(255) NOT NULL,
    -- イベントペイロード（JSON）
    event_data      JSONB NOT NULL,
    -- メタデータ（相関ID、ユーザー情報など）
    metadata        JSONB DEFAULT '{}',
    -- イベント発生時刻
    created_at      TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    -- グローバル順序（プロジェクションで全体順序追跡に使用）
    global_position BIGSERIAL NOT NULL,

    -- 同一ストリーム内でのバージョン重複防止（楽観的同時実行制御）
    UNIQUE(stream_id, stream_version)
);

-- ストリーム別イベント照会インデックス
CREATE INDEX idx_events_stream ON events(stream_id, stream_version);
-- 全体イベント順序インデックス（プロジェクション用）
CREATE INDEX idx_events_global ON events(global_position);
-- イベントタイプ別照会インデックス
CREATE INDEX idx_events_type ON events(event_type);

import { Pool } from 'pg'

interface DomainEvent {
  eventType: string
  data: Record<string, unknown>
  metadata?: Record<string, unknown>
}

interface StoredEvent extends DomainEvent {
  eventId: string
  streamId: string
  streamVersion: number
  globalPosition: number
  createdAt: Date
}

class PostgresEventStore {
  constructor(private readonly pool: Pool) {}

  /**
   * イベントをストリームに追加する。
   * expectedVersionで楽観的同時実行制御を行う。
   */
  async appendToStream(
    streamId: string,
    expectedVersion: number,
    events: DomainEvent[]
  ): Promise<StoredEvent[]> {
    const client = await this.pool.connect()
    try {
      await client.query('BEGIN')

      // 楽観的同時実行検査
      const versionCheck = await client.query(
        'SELECT MAX(stream_version) as current_version FROM events WHERE stream_id = $1',
        [streamId]
      )
      const currentVersion = versionCheck.rows[0].current_version ?? -1

      if (currentVersion !== expectedVersion) {
        throw new ConcurrencyError(
          `Expected version ${expectedVersion}, but current version is ${currentVersion}`
        )
      }

      const storedEvents: StoredEvent[] = []
      for (let i = 0; i < events.length; i++) {
        const event = events[i]
        const version = expectedVersion + i + 1

        const result = await client.query(
          `INSERT INTO events (stream_id, stream_version, event_type, event_data, metadata)
           VALUES ($1, $2, $3, $4, $5)
           RETURNING event_id, global_position, created_at`,
          [
            streamId,
            version,
            event.eventType,
            JSON.stringify(event.data),
            JSON.stringify(event.metadata ?? {}),
          ]
        )

        storedEvents.push({
          eventId: result.rows[0].event_id,
          streamId,
          streamVersion: version,
          globalPosition: result.rows[0].global_position,
          createdAt: result.rows[0].created_at,
          ...event,
        })
      }

      await client.query('COMMIT')
      return storedEvents
    } catch (error) {
      await client.query('ROLLBACK')
      throw error
    } finally {
      client.release()
    }
  }

  /**
   * ストリームのすべてのイベントを順序通りに読み取る。
   */
  async readStream(streamId: string, fromVersion = 0): Promise<StoredEvent[]> {
    const result = await this.pool.query(
      `SELECT * FROM events
       WHERE stream_id = $1 AND stream_version >= $2
       ORDER BY stream_version ASC`,
      [streamId, fromVersion]
    )
    return result.rows.map(this.mapToStoredEvent)
  }

  /**
   * 全イベントをグローバル順序で読み取る（プロジェクション用）。
   */
  async readAll(fromPosition = 0, limit = 1000): Promise<StoredEvent[]> {
    const result = await this.pool.query(
      `SELECT * FROM events
       WHERE global_position > $1
       ORDER BY global_position ASC
       LIMIT $2`,
      [fromPosition, limit]
    )
    return result.rows.map(this.mapToStoredEvent)
  }

  private mapToStoredEvent(row: any): StoredEvent {
    return {
      eventId: row.event_id,
      streamId: row.stream_id,
      streamVersion: row.stream_version,
      globalPosition: row.global_position,
      eventType: row.event_type,
      data: row.event_data,
      metadata: row.metadata,
      createdAt: row.created_at,
    }
  }
}

abstract class EventSourcedAggregate {
  private uncommittedEvents: DomainEvent[] = []
  private _version: number = -1

  get version(): number {
    return this._version
  }

  /**
   * 外部から呼び出して状態を変更する。
   * 内部的にイベントを生成し適用する。
   */
  protected apply(event: DomainEvent): void {
    this.when(event)
    this.uncommittedEvents.push(event)
    this._version++
  }

  /**
   * イベントに基づいて内部状態を変更するハンドラ。
   * サブクラスで実装する。
   */
  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 = []
  }
}

// Domain Events
interface OrderCreated {
  eventType: 'OrderCreated'
  data: {
    orderId: string
    customerId: string
    items: Array<{ productId: string; quantity: number; unitPrice: number }>
    totalAmount: number
  }
}

interface OrderConfirmed {
  eventType: 'OrderConfirmed'
  data: {
    orderId: string
    confirmedAt: string
  }
}

interface OrderCancelled {
  eventType: 'OrderCancelled'
  data: {
    orderId: string
    reason: string
    cancelledAt: string
  }
}

type OrderEvent = OrderCreated | OrderConfirmed | OrderCancelled

// Order Aggregate
class Order extends EventSourcedAggregate {
  private orderId!: string
  private customerId!: string
  private items: Array<{ productId: string; quantity: number; unitPrice: number }> = []
  private totalAmount: number = 0
  private status: 'CREATED' | 'CONFIRMED' | 'CANCELLED' = 'CREATED'

  static create(
    orderId: string,
    customerId: string,
    items: Array<{ productId: string; quantity: number; unitPrice: number }>
  ): Order {
    const order = new Order()
    const totalAmount = items.reduce((sum, item) => sum + item.quantity * item.unitPrice, 0)

    // ビジネスルール検証
    if (items.length === 0) {
      throw new Error('注文には最低1つ以上の商品が必要です')
    }
    if (totalAmount <= 0) {
      throw new Error('注文合計は0より大きくなければなりません')
    }

    order.apply({
      eventType: 'OrderCreated',
      data: { orderId, customerId, items, totalAmount },
    })

    return order
  }

  confirm(): void {
    if (this.status !== 'CREATED') {
      throw new Error('CREATED状態の注文のみ確認できます')
    }
    this.apply({
      eventType: 'OrderConfirmed',
      data: { orderId: this.orderId, confirmedAt: new Date().toISOString() },
    })
  }

  cancel(reason: string): void {
    if (this.status === 'CANCELLED') {
      throw new Error('既にキャンセルされた注文です')
    }
    this.apply({
      eventType: 'OrderCancelled',
      data: { orderId: this.orderId, reason, cancelledAt: new Date().toISOString() },
    })
  }

  protected when(event: DomainEvent): void {
    const orderEvent = event as OrderEvent
    switch (orderEvent.eventType) {
      case 'OrderCreated':
        this.orderId = orderEvent.data.orderId
        this.customerId = orderEvent.data.customerId
        this.items = orderEvent.data.items
        this.totalAmount = orderEvent.data.totalAmount
        this.status = 'CREATED'
        break
      case 'OrderConfirmed':
        this.status = 'CONFIRMED'
        break
      case 'OrderCancelled':
        this.status = 'CANCELLED'
        break
    }
  }
}

interface ProjectionCheckpoint {
  projectionName: string
  lastProcessedPosition: number
}

abstract class Projection {
  abstract readonly name: string

  abstract handle(event: StoredEvent): Promise<void>

  canHandle(eventType: string): boolean {
    return this.handledEventTypes().includes(eventType)
  }

  abstract handledEventTypes(): string[]
}

class ProjectionEngine {
  private projections: Projection[] = []

  constructor(
    private readonly eventStore: PostgresEventStore,
    private readonly checkpointStore: CheckpointStore
  ) {}

  register(projection: Projection): void {
    this.projections.push(projection)
  }

  /**
   * 登録された全プロジェクションを実行する。
   * 各プロジェクションは最後に処理した位置から再開する。
   */
  async run(): Promise<void> {
    while (true) {
      for (const projection of this.projections) {
        const checkpoint = await this.checkpointStore.get(projection.name)
        const lastPosition = checkpoint?.lastProcessedPosition ?? 0

        const events = await this.eventStore.readAll(lastPosition, 100)

        for (const event of events) {
          if (projection.canHandle(event.eventType)) {
            try {
              await projection.handle(event)
            } catch (error) {
              console.error(
                `Projection ${projection.name} failed at position ${event.globalPosition}:`,
                error
              )
              break
            }
          }
          await this.checkpointStore.save({
            projectionName: projection.name,
            lastProcessedPosition: event.globalPosition,
          })
        }
      }
      // ポーリング間隔
      await new Promise((resolve) => setTimeout(resolve, 500))
    }
  }
}

class OrderDashboardProjection extends Projection {
  readonly name = 'order-dashboard'

  constructor(private readonly db: Pool) {
    super()
  }

  handledEventTypes(): string[] {
    return ['OrderCreated', 'OrderConfirmed', 'OrderCancelled']
  }

  async handle(event: StoredEvent): Promise<void> {
    switch (event.eventType) {
      case 'OrderCreated':
        await this.db.query(
          `INSERT INTO order_dashboard (order_id, customer_id, total_amount, status, created_at)
           VALUES ($1, $2, $3, 'CREATED', $4)
           ON CONFLICT (order_id) DO NOTHING`,
          [event.data.orderId, event.data.customerId, event.data.totalAmount, event.createdAt]
        )
        break

      case 'OrderConfirmed':
        await this.db.query(
          `UPDATE order_dashboard SET status = 'CONFIRMED', confirmed_at = $2 WHERE order_id = $1`,
          [event.data.orderId, event.data.confirmedAt]
        )
        break

      case 'OrderCancelled':
        await this.db.query(
          `UPDATE order_dashboard SET status = 'CANCELLED', cancel_reason = $2 WHERE order_id = $1`,
          [event.data.orderId, event.data.reason]
        )
        break
    }
  }
}

// Saga状態定義
type OrderSagaStatus =
  | 'STARTED'
  | 'PAYMENT_PENDING'
  | 'PAYMENT_COMPLETED'
  | 'INVENTORY_RESERVED'
  | 'COMPLETED'
  | 'COMPENSATING'
  | 'FAILED'

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

class OrderSaga {
  private status: OrderSagaStatus = 'STARTED'
  private completedSteps: SagaStep[] = []

  constructor(
    private readonly orderId: string,
    private readonly paymentService: PaymentService,
    private readonly inventoryService: InventoryService,
    private readonly sagaLog: SagaLogStore
  ) {}

  async execute(orderData: {
    customerId: string
    items: any[]
    totalAmount: number
  }): Promise<void> {
    const steps: SagaStep[] = [
      {
        name: 'ProcessPayment',
        execute: async () => {
          await this.paymentService.processPayment(
            this.orderId,
            orderData.customerId,
            orderData.totalAmount
          )
          this.status = 'PAYMENT_COMPLETED'
        },
        compensate: async () => {
          await this.paymentService.refundPayment(this.orderId)
        },
      },
      {
        name: 'ReserveInventory',
        execute: async () => {
          await this.inventoryService.reserve(this.orderId, orderData.items)
          this.status = 'INVENTORY_RESERVED'
        },
        compensate: async () => {
          await this.inventoryService.releaseReservation(this.orderId)
        },
      },
    ]

    for (const step of steps) {
      try {
        await this.sagaLog.record(this.orderId, step.name, 'EXECUTING')
        await step.execute()
        await this.sagaLog.record(this.orderId, step.name, 'COMPLETED')
        this.completedSteps.push(step)
      } catch (error) {
        await this.sagaLog.record(this.orderId, step.name, 'FAILED')
        console.error(`Saga step ${step.name} failed:`, error)

        // 補償トランザクション実行（逆順）
        this.status = 'COMPENSATING'
        await this.compensate()
        this.status = 'FAILED'
        return
      }
    }

    this.status = 'COMPLETED'
    await this.sagaLog.record(this.orderId, 'Saga', 'COMPLETED')
  }

  private async compensate(): Promise<void> {
    // 完了した段階を逆順で補償
    const stepsToCompensate = [...this.completedSteps].reverse()
    for (const step of stepsToCompensate) {
      try {
        await this.sagaLog.record(this.orderId, step.name, 'COMPENSATING')
        await step.compensate()
        await this.sagaLog.record(this.orderId, step.name, 'COMPENSATED')
      } catch (compensateError) {
        // 補償失敗時は手動介入が必要
        await this.sagaLog.record(this.orderId, step.name, 'COMPENSATION_FAILED')
        console.error(`Compensation failed for step ${step.name}:`, compensateError)
      }
    }
  }
}

interface Snapshot {
  streamId: string
  version: number
  state: Record<string, unknown>
  createdAt: Date
}

class SnapshotRepository {
  constructor(private readonly pool: Pool) {}

  async save(snapshot: Snapshot): Promise<void> {
    await this.pool.query(
      `INSERT INTO snapshots (stream_id, version, state, created_at)
       VALUES ($1, $2, $3, $4)
       ON CONFLICT (stream_id) DO UPDATE SET version = $2, state = $3, created_at = $4`,
      [snapshot.streamId, snapshot.version, JSON.stringify(snapshot.state), snapshot.createdAt]
    )
  }

  async load(streamId: string): Promise<Snapshot | null> {
    const result = await this.pool.query('SELECT * FROM snapshots WHERE stream_id = $1', [streamId])
    return result.rows.length > 0 ? result.rows[0] : null
  }
}

class EventSourcedRepository<T extends EventSourcedAggregate> {
  private readonly SNAPSHOT_INTERVAL = 100 // 100イベントごとにスナップショット

  constructor(
    private readonly eventStore: PostgresEventStore,
    private readonly snapshotRepo: SnapshotRepository,
    private readonly factory: () => T
  ) {}

  async load(streamId: string): Promise<T> {
    const aggregate = this.factory()

    // 1. スナップショットロード試行
    const snapshot = await this.snapshotRepo.load(streamId)
    let fromVersion = 0

    if (snapshot) {
      ;(aggregate as any).restoreFromSnapshot(snapshot.state)
      ;(aggregate as any)._version = snapshot.version
      fromVersion = snapshot.version + 1
    }

    // 2. スナップショット以降のイベントのみ再生
    const events = await this.eventStore.readStream(streamId, fromVersion)
    aggregate.loadFromHistory(events)

    return aggregate
  }

  async save(aggregate: T, streamId: string): Promise<void> {
    const uncommittedEvents = aggregate.getUncommittedEvents()
    if (uncommittedEvents.length === 0) return

    await this.eventStore.appendToStream(
      streamId,
      aggregate.version - uncommittedEvents.length,
      uncommittedEvents
    )

    // スナップショット作成要否確認
    if (aggregate.version > 0 && aggregate.version % this.SNAPSHOT_INTERVAL === 0) {
      await this.snapshotRepo.save({
        streamId,
        version: aggregate.version,
        state: (aggregate as any).toSnapshot(),
        createdAt: new Date(),
      })
    }

    aggregate.clearUncommittedEvents()
  }
}