Event SourcingとCQRSの基礎｜Node.js + PostgreSQL 実装ガイド

Event Sourcing と CQRS をなぜ学ぶのか

マイクロサービス、決済、在庫、サブスクリプションなど 「いつ・誰が・何をしたか」 がビジネス上重要なドメインでは、従来の CRUD（UPDATE orders SET status = 'paid'）だけでは次の要求に弱くなります。

• 監査証跡: コンプライアンスやサポート調査で「支払い前の状態に戻したい」
• 時点復元: 「先月末時点の在庫」を再計算したい
• 複数の読み取りビュー: 管理画面用の集計と、モバイル用の軽量一覧を別最適化したい
• 非同期連携: 注文確定イベントを在庫・配送・メール各サービスが購読したい

Event Sourcing（イベントソーシング） は、こうした要求に対して「状態そのもの」ではなく 「状態が変わった事実」 を正（Source of Truth）として保存します。CQRS（Command Query Responsibility Segregation） は、状態を変える経路と読む経路を分け、書き込みの複雑さと読み取りの性能要件を独立してスケールさせます。

本記事では、EC の 注文（Order） ドメインを題材に、TypeScript + pg + PostgreSQL で最小かつ実務に耐える骨格を組み立てます。

従来 CRUD と Event Sourcing の比較

CRUD が「今の値」に最適化されているのに対し、Event Sourcing は 「どう変わったか」 に最適化されています。すべてのドメインをイベントソーシング化する必要はありません。境界づけられたコンテキスト（Bounded Context） ごとに、注文・台帳・在庫など「イベントが自然な言語」になる部分だけ適用するのが現実的です。

アーキテクチャの全体像

典型的な構成は次の 4 層に分かれます。

[ HTTP API ]
     │
     ├─ Query  ──► order_summaries（読み取りモデル / 投影テーブル）
     │
     └─ Command ──► CommandHandler ──► Aggregate ──► EventStore（events 追記）
                              │
                              ▼
                    Projection Worker（非同期）
                              │
                              ▼
                    order_summaries を UPDATE

書き込み経路と読み取り経路が 物理的にも論理的にも分離 されているのが CQRS の核心です。読み取りは RDB の denormalized テーブルでも Elasticsearch でも Redis でも構いません。本記事では PostgreSQL 内に投影テーブルを置く シンプルな構成 を採用します。

PostgreSQL スキーマ設計

イベントストアは 追記専用（append-only） です。UPDATE や DELETE で過去イベントを改変しない運用ルールを徹底します（法的削除要求は別ストリームで打ち消しイベントを追記）。

マイグレーション SQL

-- migrations/001_event_store.sql

CREATE TABLE events (
  id            BIGSERIAL PRIMARY KEY,
  stream_id     TEXT        NOT NULL,
  version       INTEGER     NOT NULL,
  event_type    TEXT        NOT NULL,
  payload       JSONB       NOT NULL,
  metadata      JSONB       NOT NULL DEFAULT '{}',
  occurred_at   TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  CONSTRAINT events_stream_version_unique UNIQUE (stream_id, version)
);

CREATE INDEX events_stream_id_version_idx ON events (stream_id, version);
CREATE INDEX events_occurred_at_idx ON events (occurred_at);
CREATE INDEX events_event_type_idx ON events (event_type);

-- 読み取りモデル（投影）
CREATE TABLE order_summaries (
  order_id      UUID PRIMARY KEY,
  customer_id   UUID        NOT NULL,
  status        TEXT        NOT NULL,
  total_amount  INTEGER     NOT NULL,
  item_count    INTEGER     NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL,
  updated_at    TIMESTAMPTZ NOT NULL
);

CREATE INDEX order_summaries_customer_id_idx ON order_summaries (customer_id);
CREATE INDEX order_summaries_status_idx ON order_summaries (status);

-- コマンドべき等性（再送対策）
CREATE TABLE processed_commands (
  idempotency_key TEXT PRIMARY KEY,
  command_type    TEXT        NOT NULL,
  result_body     JSONB       NOT NULL,
  http_status     INTEGER     NOT NULL,
  processed_at    TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

-- 投影の進捗（グローバルシーケンス位置）
CREATE TABLE projection_checkpoints (
  projection_name TEXT PRIMARY KEY,
  last_event_id   BIGINT      NOT NULL DEFAULT 0,
  updated_at      TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

stream_id は集約インスタンスごとの識別子です。注文なら order-<uuid> のように命名します。version はストリーム内で 1 から単調増加し、楽観的並行制御 に使います。

接続プールとの関係

投影ワーカーが大量イベントをリプレイするとき、1 接続を長時間占有すると API 用プールが枯渇します。pg.Pool の max は インスタンス数 × max が PostgreSQL の max_connections を超えない よう設計し、リプレイはバッチ単位で接続を返却します。詳細な調査手順と対策は PostgreSQL 接続プール枯渇の対処 を参照してください。

ドメインイベントと集約の型定義

イベントは 過去形の事実 として命名します（CreateOrder ではなく OrderCreated）。payload はその時点で確定したデータだけを含め、集約外の参照は ID に留めます。

// src/domain/order/events.ts

export type OrderId = string;
export type CustomerId = string;

export interface OrderLineItem {
  sku: string;
  quantity: number;
  unitPrice: number;
}

export interface OrderCreatedPayload {
  orderId: OrderId;
  customerId: CustomerId;
  items: OrderLineItem[];
  currency: 'JPY';
}

export interface PaymentCapturedPayload {
  orderId: OrderId;
  paymentId: string;
  amount: number;
  capturedAt: string;
}

export interface OrderCancelledPayload {
  orderId: OrderId;
  reason: string;
  cancelledAt: string;
}

export type OrderEvent =
  | { type: 'OrderCreated'; payload: OrderCreatedPayload; occurredAt: string }
  | { type: 'PaymentCaptured'; payload: PaymentCapturedPayload; occurredAt: string }
  | { type: 'OrderCancelled'; payload: OrderCancelledPayload; occurredAt: string };

export type OrderStatus = 'pending' | 'paid' | 'cancelled';

// src/domain/order/order-aggregate.ts

import { randomUUID } from 'node:crypto';
import type {
  OrderEvent,
  OrderCreatedPayload,
  PaymentCapturedPayload,
  OrderCancelledPayload,
  OrderStatus,
  OrderLineItem,
  OrderId,
  CustomerId,
} from './events.js';

export class ConcurrencyError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'ConcurrencyError';
  }
}

export class DomainError extends Error {
  constructor(message: string) {
    super(message);
    this.name = 'DomainError';
  }
}

interface OrderState {
  orderId: OrderId | null;
  customerId: CustomerId | null;
  status: OrderStatus | null;
  items: OrderLineItem[];
  totalAmount: number;
  version: number;
}

export class OrderAggregate {
  private state: OrderState = {
    orderId: null,
    customerId: null,
    status: null,
    items: [],
    totalAmount: 0,
    version: 0,
  };

  private uncommittedEvents: OrderEvent[] = [];

  static rehydrate(history: OrderEvent[]): OrderAggregate {
    const agg = new OrderAggregate();
    for (const event of history) {
      agg.apply(event);
      agg.state.version += 1;
    }
    return agg;
  }

  static createNew(
    customerId: CustomerId,
    items: OrderLineItem[],
  ): OrderAggregate {
    const agg = new OrderAggregate();
    const orderId = randomUUID();
    const totalAmount = items.reduce(
      (sum, item) => sum + item.unitPrice * item.quantity,
      0,
    );
    if (items.length === 0) {
      throw new DomainError('注文には1件以上の明細が必要です');
    }
    if (totalAmount <= 0) {
      throw new DomainError('合計金額は正の整数である必要があります');
    }
    const payload: OrderCreatedPayload = {
      orderId,
      customerId,
      items,
      currency: 'JPY',
    };
    const event: OrderEvent = {
      type: 'OrderCreated',
      payload,
      occurredAt: new Date().toISOString(),
    };
    agg.raise(event);
    return agg;
  }

  capturePayment(paymentId: string, amount: number): void {
    if (this.state.status !== 'pending') {
      throw new DomainError('支払い済みまたはキャンセル済みの注文には課金できません');
    }
    if (amount !== this.state.totalAmount) {
      throw new DomainError('支払い金額が注文合計と一致しません');
    }
    const payload: PaymentCapturedPayload = {
      orderId: this.state.orderId!,
      paymentId,
      amount,
      capturedAt: new Date().toISOString(),
    };
    this.raise({
      type: 'PaymentCaptured',
      payload,
      occurredAt: new Date().toISOString(),
    });
  }

  cancel(reason: string): void {
    if (this.state.status === 'cancelled') {
      throw new DomainError('すでにキャンセル済みです');
    }
    if (this.state.status === 'paid') {
      throw new DomainError('支払い済み注文はキャンセルできません（返金フローを利用してください）');
    }
    const payload: OrderCancelledPayload = {
      orderId: this.state.orderId!,
      reason,
      cancelledAt: new Date().toISOString(),
    };
    this.raise({
      type: 'OrderCancelled',
      payload,
      occurredAt: new Date().toISOString(),
    });
  }

  private raise(event: OrderEvent): void {
    this.apply(event);
    this.uncommittedEvents.push(event);
  }

  private apply(event: OrderEvent): void {
    switch (event.type) {
      case 'OrderCreated': {
        this.state.orderId = event.payload.orderId;
        this.state.customerId = event.payload.customerId;
        this.state.items = event.payload.items;
        this.state.totalAmount = event.payload.items.reduce(
          (sum, item) => sum + item.unitPrice * item.quantity,
          0,
        );
        this.state.status = 'pending';
        break;
      }
      case 'PaymentCaptured': {
        this.state.status = 'paid';
        break;
      }
      case 'OrderCancelled': {
        this.state.status = 'cancelled';
        break;
      }
      default: {
        const _exhaustive: never = event;
        return _exhaustive;
      }
    }
  }

  getOrderId(): OrderId | null {
    return this.state.orderId;
  }

  getVersion(): number {
    return this.state.version;
  }

  getUncommittedEvents(): readonly OrderEvent[] {
    return this.uncommittedEvents;
  }

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

集約は 外部から直接状態を書き換えず、コマンドメソッド経由でイベントを発行します。rehydrate は永続化されたイベント列を先頭から apply し、メモリ上の状態を再構築します。これが Event Sourcing の「再生」です。

EventStore リポジトリ（PostgreSQL）

// src/infrastructure/event-store.ts

import type { Pool, PoolClient } from 'pg';
import type { OrderEvent } from '../domain/order/events.js';
import { ConcurrencyError } from '../domain/order/order-aggregate.js';

interface StoredEventRow {
  version: number;
  event_type: string;
  payload: OrderEvent['payload'];
  occurred_at: Date;
}

export class EventStore {
  constructor(private readonly pool: Pool) {}

  streamIdForOrder(orderId: string): string {
    return `order-${orderId}`;
  }

  async loadOrderEvents(orderId: string): Promise<OrderEvent[]> {
    const streamId = this.streamIdForOrder(orderId);
    const result = await this.pool.query<StoredEventRow>(
      `
        SELECT version, event_type, payload, occurred_at
        FROM events
        WHERE stream_id = $1
        ORDER BY version ASC
      `,
      [streamId],
    );

    return result.rows.map((row) => {
      const occurredAt = row.occurred_at.toISOString();
      switch (row.event_type) {
        case 'OrderCreated':
          return {
            type: 'OrderCreated',
            payload: row.payload as OrderEvent extends { type: 'OrderCreated' } ? OrderEvent['payload'] : never,
            occurredAt,
          };
        case 'PaymentCaptured':
          return {
            type: 'PaymentCaptured',
            payload: row.payload as OrderEvent extends { type: 'PaymentCaptured' } ? OrderEvent['payload'] : never,
            occurredAt,
          };
        case 'OrderCancelled':
          return {
            type: 'OrderCancelled',
            payload: row.payload as OrderEvent extends { type: 'OrderCancelled' } ? OrderEvent['payload'] : never,
            occurredAt,
          };
        default:
          throw new Error(`未知のイベント型: ${row.event_type}`);
      }
    });
  }

  async appendToStream(
    streamId: string,
    expectedVersion: number,
    events: OrderEvent[],
    metadata: Record<string, string> = {},
  ): Promise<void> {
    if (events.length === 0) {
      return;
    }

    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');

      const current = await client.query<{ max_version: number | null }>(
        `
          SELECT MAX(version) AS max_version
          FROM events
          WHERE stream_id = $1
          FOR UPDATE
        `,
        [streamId],
      );

      const maxVersion = current.rows[0]?.max_version ?? 0;
      if (maxVersion !== expectedVersion) {
        throw new ConcurrencyError(
          `バージョン競合: expected=${expectedVersion}, actual=${maxVersion}`,
        );
      }

      let nextVersion = expectedVersion;
      for (const event of events) {
        nextVersion += 1;
        await client.query(
          `
            INSERT INTO events (stream_id, version, event_type, payload, metadata, occurred_at)
            VALUES ($1, $2, $3, $4, $5, $6)
          `,
          [
            streamId,
            nextVersion,
            event.type,
            JSON.stringify(event.payload),
            JSON.stringify(metadata),
            event.occurredAt,
          ],
        );
      }

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

  async loadGlobalEventsAfter(
    lastEventId: number,
    limit: number,
  ): Promise<Array<{ id: number; streamId: string; event: OrderEvent }>> {
    const result = await this.pool.query<{
      id: number;
      stream_id: string;
      event_type: string;
      payload: OrderEvent['payload'];
      occurred_at: Date;
    }>(
      `
        SELECT id, stream_id, event_type, payload, occurred_at
        FROM events
        WHERE id > $1
        ORDER BY id ASC
        LIMIT $2
      `,
      [lastEventId, limit],
    );

    return result.rows.map((row) => {
      const occurredAt = row.occurred_at.toISOString();
      let event: OrderEvent;
      if (row.event_type === 'OrderCreated') {
        event = {
          type: 'OrderCreated',
          payload: row.payload as Extract<OrderEvent, { type: 'OrderCreated' }>['payload'],
          occurredAt,
        };
      } else if (row.event_type === 'PaymentCaptured') {
        event = {
          type: 'PaymentCaptured',
          payload: row.payload as Extract<OrderEvent, { type: 'PaymentCaptured' }>['payload'],
          occurredAt,
        };
      } else if (row.event_type === 'OrderCancelled') {
        event = {
          type: 'OrderCancelled',
          payload: row.payload as Extract<OrderEvent, { type: 'OrderCancelled' }>['payload'],
          occurredAt,
        };
      } else {
        throw new Error(`未知のイベント型: ${row.event_type}`);
      }
      return { id: row.id, streamId: row.stream_id, event };
    });
  }
}

expectedVersion が DB 上の最新 version と一致しない場合、別リクエストが先に append したことを意味します。クライアントには 409 Conflict を返し、最新状態の再読み込みとリトライを促します。これは べき等性設計 の「競合検出」とは別軸で、集約単位の楽観的並行制御 です。

コマンドハンドラとべき等性

HTTP 層では Idempotency-Key により 同じコマンドの再送 を吸収します。イベントストアの version 競合とは役割が異なります。

// src/application/command-handlers.ts

import type { Pool } from 'pg';
import { OrderAggregate, ConcurrencyError, DomainError } from '../domain/order/order-aggregate.js';
import type { OrderLineItem } from '../domain/order/events.js';
import { EventStore } from '../infrastructure/event-store.js';

interface CreateOrderCommand {
  customerId: string;
  items: OrderLineItem[];
}

interface CreateOrderResult {
  orderId: string;
  status: 'pending';
  totalAmount: number;
}

export class OrderCommandService {
  constructor(
    private readonly pool: Pool,
    private readonly eventStore: EventStore,
  ) {}

  async createOrder(
    command: CreateOrderCommand,
    idempotencyKey: string,
  ): Promise<{ statusCode: number; body: CreateOrderResult }> {
    const cached = await this.pool.query<{
      result_body: CreateOrderResult;
      http_status: number;
    }>(
      `SELECT result_body, http_status FROM processed_commands WHERE idempotency_key = $1`,
      [idempotencyKey],
    );

    if (cached.rowCount && cached.rowCount > 0) {
      return {
        statusCode: cached.rows[0].http_status,
        body: cached.rows[0].result_body,
      };
    }

    const aggregate = OrderAggregate.createNew(command.customerId, command.items);
    const orderId = aggregate.getOrderId()!;
    const streamId = this.eventStore.streamIdForOrder(orderId);
    const events = [...aggregate.getUncommittedEvents()];

    try {
      await this.eventStore.appendToStream(streamId, 0, events, {
        command: 'CreateOrder',
        idempotencyKey,
      });
    } catch (error) {
      if (error instanceof ConcurrencyError) {
        return { statusCode: 409, body: { orderId: '', status: 'pending', totalAmount: 0 } };
      }
      throw error;
    }

    const totalAmount = command.items.reduce(
      (sum, item) => sum + item.unitPrice * item.quantity,
      0,
    );

    const result: CreateOrderResult = {
      orderId,
      status: 'pending',
      totalAmount,
    };

    await this.pool.query(
      `
        INSERT INTO processed_commands (idempotency_key, command_type, result_body, http_status)
        VALUES ($1, $2, $3, $4)
      `,
      [idempotencyKey, 'CreateOrder', JSON.stringify(result), 201],
    );

    aggregate.clearUncommittedEvents();
    return { statusCode: 201, body: result };
  }

  async capturePayment(
    orderId: string,
    paymentId: string,
    amount: number,
    idempotencyKey: string,
  ): Promise<{ statusCode: number; body: { orderId: string; status: string } }> {
    const cached = await this.pool.query<{
      result_body: { orderId: string; status: string };
      http_status: number;
    }>(
      `SELECT result_body, http_status FROM processed_commands WHERE idempotency_key = $1`,
      [idempotencyKey],
    );

    if (cached.rowCount && cached.rowCount > 0) {
      return {
        statusCode: cached.rows[0].http_status,
        body: cached.rows[0].result_body,
      };
    }

    const history = await this.eventStore.loadOrderEvents(orderId);
    if (history.length === 0) {
      throw new DomainError('注文が存在しません');
    }

    const aggregate = OrderAggregate.rehydrate(history);
    aggregate.capturePayment(paymentId, amount);

    const streamId = this.eventStore.streamIdForOrder(orderId);
    const expectedVersion = aggregate.getVersion();
    const newEvents = [...aggregate.getUncommittedEvents()];

    await this.eventStore.appendToStream(streamId, expectedVersion, newEvents, {
      command: 'CapturePayment',
      idempotencyKey,
    });

    const result = { orderId, status: 'paid' };
    await this.pool.query(
      `
        INSERT INTO processed_commands (idempotency_key, command_type, result_body, http_status)
        VALUES ($1, $2, $3, $4)
      `,
      [idempotencyKey, 'CapturePayment', JSON.stringify(result), 200],
    );

    aggregate.clearUncommittedEvents();
    return { statusCode: 200, body: result };
  }
}

べき等性の詳細な状態遷移（in-flight 処理・パラメータ不一致エラー）は APIべき等性設計ガイド にまとめています。Event Sourcing では コマンド受付層 と イベント追記層 の両方で重複を意識するのが安全です。

投影（Projection）ワーカー

読み取りモデル order_summaries はイベントから 導出される派生データ です。正はあくまで events テーブルであり、投影が壊れてもリプレイで再構築できます。

// src/infrastructure/order-summary-projector.ts

import type { Pool } from 'pg';
import type { OrderEvent } from '../domain/order/events.js';
import { EventStore } from './event-store.js';

const PROJECTION_NAME = 'order_summaries_v1';

export class OrderSummaryProjector {
  constructor(
    private readonly pool: Pool,
    private readonly eventStore: EventStore,
  ) {}

  async runBatch(batchSize: number = 100): Promise<number> {
    const client = await this.pool.connect();
    let processed = 0;

    try {
      await client.query('BEGIN');

      const checkpoint = await client.query<{ last_event_id: number }>(
        `SELECT last_event_id FROM projection_checkpoints WHERE projection_name = $1 FOR UPDATE`,
        [PROJECTION_NAME],
      );

      let lastEventId = checkpoint.rows[0]?.last_event_id ?? 0;

      const events = await this.eventStore.loadGlobalEventsAfter(lastEventId, batchSize);

      for (const record of events) {
        await this.applyEvent(client, record.event);
        lastEventId = record.id;
        processed += 1;
      }

      if (processed > 0) {
        await client.query(
          `
            INSERT INTO projection_checkpoints (projection_name, last_event_id, updated_at)
            VALUES ($1, $2, NOW())
            ON CONFLICT (projection_name)
            DO UPDATE SET last_event_id = EXCLUDED.last_event_id, updated_at = NOW()
          `,
          [PROJECTION_NAME, lastEventId],
        );
      }

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

    return processed;
  }

  private async applyEvent(
    client: import('pg').PoolClient,
    event: OrderEvent,
  ): Promise<void> {
    switch (event.type) {
      case 'OrderCreated': {
        const totalAmount = event.payload.items.reduce(
          (sum, item) => sum + item.unitPrice * item.quantity,
          0,
        );
        await client.query(
          `
            INSERT INTO order_summaries (
              order_id, customer_id, status, total_amount, item_count, created_at, updated_at
            )
            VALUES ($1, $2, 'pending', $3, $4, $5, $5)
            ON CONFLICT (order_id) DO NOTHING
          `,
          [
            event.payload.orderId,
            event.payload.customerId,
            totalAmount,
            event.payload.items.length,
            event.occurredAt,
          ],
        );
        break;
      }
      case 'PaymentCaptured': {
        await client.query(
          `
            UPDATE order_summaries
            SET status = 'paid', updated_at = $2
            WHERE order_id = $1
          `,
          [event.payload.orderId, event.occurredAt],
        );
        break;
      }
      case 'OrderCancelled': {
        await client.query(
          `
            UPDATE order_summaries
            SET status = 'cancelled', updated_at = $2
            WHERE order_id = $1
          `,
          [event.payload.orderId, event.occurredAt],
        );
        break;
      }
      default: {
        const _exhaustive: never = event;
        return _exhaustive;
      }
    }
  }
}

投影は at-least-once で動く前提です。OrderCreated の ON CONFLICT DO NOTHING や、ステータス更新の単調性チェックにより、同じイベントを二度処理しても安全（冪等な投影）にします。ワーカーは setInterval や BullMQ、PostgreSQL の LISTEN/NOTIFY で起動し、バッチごとに接続を返却してプールを圧迫しないようにします。

HTTP API（Fastify）— Command と Query の分離

// src/api/server.ts

import Fastify from 'fastify';
import { Pool } from 'pg';
import { EventStore } from '../infrastructure/event-store.js';
import { OrderCommandService } from '../application/command-handlers.js';
import { DomainError } from '../domain/order/order-aggregate.js';

const pool = new Pool({
  connectionString: process.env.DATABASE_URL,
  max: 10,
  idleTimeoutMillis: 30_000,
  connectionTimeoutMillis: 5_000,
});

const eventStore = new EventStore(pool);
const commandService = new OrderCommandService(pool, eventStore);

const app = Fastify({ logger: true });

app.post('/commands/orders', async (request, reply) => {
  const idempotencyKey = request.headers['idempotency-key'];
  if (typeof idempotencyKey !== 'string' || idempotencyKey.length === 0) {
    return reply.status(400).send({ error: 'Idempotency-Key ヘッダーが必須です' });
  }

  const body = request.body as {
    customerId: string;
    items: Array<{ sku: string; quantity: number; unitPrice: number }>;
  };

  try {
    const result = await commandService.createOrder(
      { customerId: body.customerId, items: body.items },
      idempotencyKey,
    );
    return reply.status(result.statusCode).send(result.body);
  } catch (error) {
    if (error instanceof DomainError) {
      return reply.status(422).send({ error: error.message });
    }
    request.log.error(error);
    return reply.status(500).send({ error: '内部エラー' });
  }
});

app.get('/orders/:orderId', async (request, reply) => {
  const { orderId } = request.params as { orderId: string };

  const result = await pool.query(
    `
      SELECT order_id, customer_id, status, total_amount, item_count, created_at, updated_at
      FROM order_summaries
      WHERE order_id = $1
    `,
    [orderId],
  );

  if (result.rowCount === 0) {
    return reply.status(404).send({ error: '注文が見つかりません' });
  }

  return reply.send(result.rows[0]);
});

app.get('/orders/:orderId/history', async (request, reply) => {
  const { orderId } = request.params as { orderId: string };
  const events = await eventStore.loadOrderEvents(orderId);

  if (events.length === 0) {
    return reply.status(404).send({ error: '注文が見つかりません' });
  }

  return reply.send({ orderId, events });
});

const port = Number(process.env.PORT ?? 3000);
app.listen({ port, host: '0.0.0.0' });

GET /orders/:id は投影テーブルのみを参照し、イベントストアに触れません。これが CQRS の Query 側です。監査用に GET /orders/:id/history でイベント列を返すエンドポイントを別途用意するのが一般的です。

スナップショットでリプレイコストを抑える

イベント数が数千を超えると、集約の load のたびに全イベントを再生するコストが問題になります。スナップショット はある version 時点の集約状態を保存し、それ以降のイベントだけ再生します。

CREATE TABLE aggregate_snapshots (
  stream_id     TEXT        NOT NULL,
  version       INTEGER     NOT NULL,
  state         JSONB       NOT NULL,
  created_at    TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  PRIMARY KEY (stream_id, version)
);

// src/infrastructure/snapshot-store.ts

import type { Pool } from 'pg';
import type { OrderLineItem, OrderStatus } from '../domain/order/events.js';

interface OrderSnapshotState {
  orderId: string;
  customerId: string;
  status: OrderStatus;
  items: OrderLineItem[];
  totalAmount: number;
}

export class SnapshotStore {
  constructor(private readonly pool: Pool) {}

  async saveSnapshot(
    streamId: string,
    version: number,
    state: OrderSnapshotState,
  ): Promise<void> {
    await this.pool.query(
      `
        INSERT INTO aggregate_snapshots (stream_id, version, state)
        VALUES ($1, $2, $3)
      `,
      [streamId, version, JSON.stringify(state)],
    );
  }

  async loadLatestSnapshot(
    streamId: string,
  ): Promise<{ version: number; state: OrderSnapshotState } | null> {
    const result = await this.pool.query<{
      version: number;
      state: OrderSnapshotState;
    }>(
      `
        SELECT version, state
        FROM aggregate_snapshots
        WHERE stream_id = $1
        ORDER BY version DESC
        LIMIT 1
      `,
      [streamId],
    );

    if (result.rowCount === 0) {
      return null;
    }

    return {
      version: result.rows[0].version,
      state: result.rows[0].state,
    };
  }
}

スナップショットは 100 イベントごと などの閾値で非同期に作成するのが一般的です。スナップショット自体が欠損しても、イベント列から再構築できるため、キャッシュと同様の位置づけです。

イベントスキーマの進化（バージョニング）

本番では OrderCreated の payload にフィールドを追加することがあります。後方互換を保つルールの例です。

変更	推奨	例
フィールド追加	旧イベントにデフォルト値を補う	shippingAddress 省略時は null
フィールド削除	読み取り側で無視	新コードは使わないが旧 payload に残る
意味変更	新イベント型を定義	OrderCreatedV2 を別 type に

event_type ごとに upcaster 関数を用意し、古い payload を読み込み時に新形式へ変換する Upcasting パターンが定番です。破壊的変更を同一 type に上書きすると、リプレイ結果が過去と一致しなくなり、会計・監査で致命的です。

運用・監視チェックリスト

接続プールの waitingCount が増え続ける場合は、投影ワーカーのバッチサイズ縮小・インスタンス分離・接続プール枯渇の対処 の手順に沿って調査します。

アンチパターン

アンチパターン	問題
イベントストアの UPDATE / DELETE	監査不能・リプレイ結果が壊れる
集約を跨ぐ単一トランザクション	分散境界を無視し、結合度が上がる
投影なしで毎回全イベント再生して Query	読み取りレイテンシが線形悪化
Idempotency なしのコマンド API	再送で二重注文・二重課金
巨大 payload（PDF バイナリ等）をイベントに格納	ストア肥大・リプレイ遅延。参照は S3 URL のみ
全ドメインを一度に ES 化	学習コストと障害点が一気に増える

よくある質問（FAQ）

フロントマターの FAQ に加え、実装時によく出る論点を補足します。

イベントとメッセージの違いは？

イベント は「起きた事実」で、通常は過去形・不変です。コマンド は「してほしい依頼」で、拒否されることがあります。Command → Aggregate 検証 → Event 発行、という流れを混同しないことが重要です。

Kafka は必須ですか？

必須ではありません。本記事のように PostgreSQL の events テーブルだけでも小〜中規模は成立します。複数サービスへの fan-out や高スループットが必要になったら、Outbox パターン（トランザクション内で outbox テーブルに書き、別プロセスが Kafka に publish）で段階的に移行するチームが多いです。

テストはどう書く？

集約の単体テストは イベント列の Given-When-Then が書きやすいのが利点です。「pending 状態で capturePayment すると PaymentCaptured が 1 件」など、インメモリで完結します。統合テストは Testcontainers で PostgreSQL を立ち上げ、append → 投影 → Query まで一通り検証します。

関連記事

この記事は API・バックエンド テーマの一環です。あわせて読むと理解が深まる関連記事をまとめました。

トピック	記事
APIキャッシュ設計ガイド	APIキャッシュ設計ガイド｜ETag・Cache-Control・304・CDNキャッシュキーの実務
Web APIのエラーハンドリング設計	Web APIのエラーハンドリング設計｜RFC 9457・error_code・400/422/409の使い分け
API設計におけるページネーションの使い分け：オフセット型とカーソル型の比較	API設計におけるページネーションの使い分け：オフセット型とカーソル型の比較
APIのレート制限（Rate Limiting）実装ガイド	APIのレート制限（Rate Limiting）実装ガイド｜アルゴリズム選定と429エラーの適切なハンドリング
APIレート制限と429対応	APIレート制限と429対応｜X-RateLimit・Retry-Afterを使ったクライアント実装
WebHook設計のベストプラクティス	WebHook設計のベストプラクティス｜署名・再試行・べき等性で信頼性を担保する
ZodでAPI境界のランタイム検証を設計する	ZodでAPI境界のランタイム検証を設計する｜z.infer・Expressミドルウェア・zod-to-openapi（2026）
OpenAPI 3.1仕様のCIバリデーション完全ガイド	OpenAPI 3.1仕様のCIバリデーション完全ガイド｜Spectral Lint・Breaking Change検出・GitHub Actions
GraphQL N+1 問題の解決	GraphQL N+1 問題の解決｜DataLoader バッチング実装と Apollo Server 完全ガイド
Circuit Breaker + Retry パターン実装ガイド	Circuit Breaker + Retry パターン実装ガイド｜Node.js opossum/cockatiel・指数バックオフ・Jitter
Sagaパターン完全ガイド	Sagaパターン完全ガイド｜Orchestration vs Choreography と補償トランザクション（Node.js）
Redisキャッシュスタンピード防止ガイド	Redisキャッシュスタンピード防止ガイド｜Singleflight・Mutex・SET NX・確率的早期失効

まとめ

Event Sourcing は状態をイベントの追記ログとして保存し、CQRS は書き込みと読み取りを分離してそれぞれ最適化します。 Node.js + PostgreSQL では、events テーブルに stream_id と version で楽観的並行制御を行い、集約がビジネスルールを守りながらイベントを発行、投影ワーカーが order_summaries を更新する流れが基本形です。

実務では APIべき等性設計 によるコマンド再送対策と、PostgreSQL 接続プール枯渇の対処 による接続設計を最初から組み込んでください。スナップショット・投影ラグ監視・イベントスキーマの進化ルールを整えたうえで、1 集約から段階的に導入 するのが、Event Sourcing / CQRS を安全に運用する近道です。