TypeScript로 CQRS + Event Sourcing 직접 구현하기 — EventStore, 이벤트 리플레이, Projection까지

interface DomainEvent<T = unknown> {
  readonly eventId: string;
  readonly eventType: string;
  readonly aggregateId: string;  // 이 시스템에서는 streamId와 동일 (예: 'account-uuid')
  readonly aggregateVersion: number;
  readonly occurredAt: Date;
  readonly payload: T;
}
 
interface AccountCreatedPayload {
  ownerId: string;
  initialBalance: number;
}
 
interface MoneyDepositedPayload {
  amount: number;
  description: string;
}
 
interface MoneyWithdrawnPayload {
  amount: number;
  description: string;
}
 
// 계좌 도메인 이벤트 discriminated union
// accountReducer에서 캐스팅 없이 payload 타입이 자동으로 좁혀집니다
type AccountEvent =
  | DomainEvent<AccountCreatedPayload> & { eventType: 'AccountCreated' }
  | DomainEvent<MoneyDepositedPayload> & { eventType: 'MoneyDeposited' }
  | DomainEvent<MoneyWithdrawnPayload> & { eventType: 'MoneyWithdrawn' };

CREATE TABLE events (
  global_position   BIGSERIAL,
  id                UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  stream_id         VARCHAR(255) NOT NULL,
  event_type        VARCHAR(255) NOT NULL,
  aggregate_version INTEGER NOT NULL,
  occurred_at       TIMESTAMPTZ NOT NULL DEFAULT NOW(),
  payload           JSONB NOT NULL,
  UNIQUE (stream_id, aggregate_version)
);
 
CREATE INDEX idx_events_stream ON events (stream_id, aggregate_version);
CREATE INDEX idx_events_global_position ON events (global_position);

interface EventStore {
  append(streamId: string, events: DomainEvent[], expectedVersion: number): Promise<void>;
  readStream(streamId: string, fromVersion?: number): Promise<DomainEvent[]>;
  readAllBatch(
    fromPosition: number,
    limit: number
  ): Promise<{ events: DomainEvent[]; nextPosition: number }>;
  subscribeToAll(handler: (event: DomainEvent) => Promise<void>): void;
}

import { Pool } from 'pg';
import { randomUUID } from 'crypto';
 
class OptimisticConcurrencyError extends Error {
  constructor(streamId: string, expectedVersion: number) {
    super(`Stream ${streamId}의 예상 버전 ${expectedVersion}이 맞지 않습니다`);
    this.name = 'OptimisticConcurrencyError';
  }
}
 
class PostgresEventStore implements EventStore {
  constructor(private pool: Pool) {}
 
  async append(
    streamId: string,
    events: DomainEvent[],
    expectedVersion: number
  ): Promise<void> {
    const client = await this.pool.connect();
    try {
      await client.query('BEGIN');
 
      // 빠른 실패를 위한 버전 확인 — 실제 동시성 가드는 UNIQUE 제약이 담당합니다
      // PostgreSQL은 집계 함수와 FOR UPDATE를 같은 쿼리에서 허용하지 않으므로
      // SELECT MAX에는 행 잠금을 걸지 않습니다
      const { rows } = await client.query(
        `SELECT MAX(aggregate_version) AS current FROM events WHERE stream_id = $1`,
        [streamId]
      );
      const current = rows[0].current ?? -1;
 
      if (current !== expectedVersion) {
        throw new OptimisticConcurrencyError(streamId, expectedVersion);
      }
 
      for (const event of events) {
        await client.query(
          `INSERT INTO events
             (id, stream_id, event_type, aggregate_version, occurred_at, payload)
           VALUES ($1, $2, $3, $4, $5, $6)`,
          [
            event.eventId,          // 핸들러가 생성한 ID를 그대로 사용
            streamId,
            event.eventType,
            event.aggregateVersion,
            event.occurredAt,
            JSON.stringify(event.payload),
          ]
        );
      }
 
      await client.query('COMMIT');
    } catch (err) {
      await client.query('ROLLBACK');
      // UNIQUE 위반 = 동시 쓰기 충돌 → OptimisticConcurrencyError로 변환
      if ((err as { code?: string }).code === '23505') {
        throw new OptimisticConcurrencyError(streamId, expectedVersion);
      }
      throw err;
    } finally {
      client.release();
    }
  }
 
  async readStream(streamId: string, fromVersion = 0): Promise<DomainEvent[]> {
    const { rows } = await this.pool.query(
      `SELECT * FROM events
       WHERE stream_id = $1 AND aggregate_version >= $2
       ORDER BY aggregate_version ASC`,
      [streamId, fromVersion]
    );
    return rows.map(this.rowToEvent);
  }
 
  async readAllBatch(
    fromPosition: number,
    limit: number
  ): Promise<{ events: DomainEvent[]; nextPosition: number }> {
    const { rows } = await this.pool.query(
      `SELECT * FROM events
       WHERE global_position > $1
       ORDER BY global_position ASC
       LIMIT $2`,
      [fromPosition, limit]
    );
    const events = rows.map(this.rowToEvent);
    const nextPosition =
      rows.length > 0 ? rows[rows.length - 1].global_position : fromPosition;
    return { events, nextPosition };
  }
 
  subscribeToAll(handler: (event: DomainEvent) => Promise<void>): void {
    let lastPosition = 0;
    setInterval(async () => {
      try {
        const { events, nextPosition } = await this.readAllBatch(lastPosition, 100);
        for (const event of events) {
          await handler(event);
        }
        lastPosition = nextPosition;
      } catch (err) {
        console.error('이벤트 구독 처리 중 오류:', err);
      }
    }, 100);
  }
 
  private rowToEvent(row: Record<string, unknown>): DomainEvent {
    return {
      eventId: row.id as string,
      eventType: row.event_type as string,
      aggregateId: row.stream_id as string,
      aggregateVersion: row.aggregate_version as number,
      occurredAt: row.occurred_at as Date,
      payload: row.payload,
    };
  }
}

interface AccountState {
  id: string;
  ownerId: string;
  balance: number;
  version: number;
  isClosed: boolean;
}
 
// AccountEvent 유니언을 매개변수로 받으면 switch 분기마다 payload가 자동으로 좁혀집니다
const accountReducer = (
  state: AccountState | null,
  event: AccountEvent
): AccountState => {
  switch (event.eventType) {
    case 'AccountCreated':
      return {
        id: event.aggregateId,
        ownerId: event.payload.ownerId,        // AccountCreatedPayload로 자동 추론
        balance: event.payload.initialBalance,
        version: event.aggregateVersion,
        isClosed: false,
      };
    case 'MoneyDeposited':
      if (!state) throw new Error('잘못된 이벤트 순서입니다');
      return {
        ...state,
        balance: state.balance + event.payload.amount,  // MoneyDepositedPayload로 자동 추론
        version: event.aggregateVersion,
      };
    case 'MoneyWithdrawn':
      if (!state) throw new Error('잘못된 이벤트 순서입니다');
      return {
        ...state,
        balance: state.balance - event.payload.amount,
        version: event.aggregateVersion,
      };
  }
};
 
async function rehydrateAccount(
  eventStore: EventStore,
  accountId: string
): Promise<{ state: AccountState; version: number }> {
  const streamId = `account-${accountId}`;
  const events = await eventStore.readStream(streamId);
 
  if (events.length === 0) {
    throw new Error(`계좌 ${accountId}를 찾을 수 없습니다`);
  }
 
  // EventStore가 반환하는 DomainEvent[]를 AccountEvent[]로 경계 캐스팅
  const state = (events as AccountEvent[]).reduce(
    (acc, event) => accountReducer(acc, event),
    null as AccountState | null
  ) as AccountState;
 
  return { state, version: state.version };
}

interface Snapshot {
  streamId: string;
  state: AccountState;
  version: number;
}
 
interface SnapshotStore {
  save(snapshot: Snapshot): Promise<void>;
  getLatest(streamId: string): Promise<Snapshot | null>;
}
 
async function rehydrateWithSnapshot(
  eventStore: EventStore,
  snapshotStore: SnapshotStore,
  accountId: string
): Promise<{ state: AccountState; version: number }> {
  const streamId = `account-${accountId}`;
 
  const snapshot = await snapshotStore.getLatest(streamId);
  const fromVersion = snapshot ? snapshot.version + 1 : 0;
  const events = await eventStore.readStream(streamId, fromVersion);
 
  const initialState = snapshot ? snapshot.state : null;
  const state = (events as AccountEvent[]).reduce(
    (acc, event) => accountReducer(acc, event),
    initialState
  ) as AccountState;
 
  return { state, version: state.version };
}

interface CreateAccountCommand {
  type: 'CreateAccount';
  accountId: string;
  ownerId: string;
  initialBalance: number;
}
 
interface WithdrawMoneyCommand {
  type: 'WithdrawMoney';
  accountId: string;
  amount: number;
  description: string;
}
 
class CreateAccountHandler {
  constructor(private eventStore: EventStore) {}
 
  async handle(command: CreateAccountCommand): Promise<void> {
    const { accountId, ownerId, initialBalance } = command;
    const streamId = `account-${accountId}`;
 
    const event: DomainEvent<AccountCreatedPayload> = {
      eventId: randomUUID(),
      eventType: 'AccountCreated',
      aggregateId: streamId,
      aggregateVersion: 0,  // 첫 이벤트의 버전은 0
      occurredAt: new Date(),
      payload: { ownerId, initialBalance },
    };
 
    // 새 스트림: expectedVersion = -1 (이벤트가 없는 상태)
    await this.eventStore.append(streamId, [event], -1);
  }
}
 
class WithdrawMoneyHandler {
  constructor(private eventStore: EventStore) {}
 
  async handle(command: WithdrawMoneyCommand): Promise<void> {
    const { accountId, amount, description } = command;
 
    // 1. 이벤트 리플레이로 현재 상태 복원
    const { state, version } = await rehydrateAccount(this.eventStore, accountId);
 
    // 2. 도메인 규칙 검증
    if (state.isClosed) throw new Error('닫힌 계좌에서는 출금할 수 없습니다');
    if (state.balance < amount) throw new Error(`잔액 부족: 현재 잔액 ${state.balance}원`);
    if (amount <= 0) throw new Error('출금액은 0보다 커야 합니다');
 
    const streamId = `account-${accountId}`;
    const event: DomainEvent<MoneyWithdrawnPayload> = {
      eventId: randomUUID(),
      eventType: 'MoneyWithdrawn',
      aggregateId: streamId,
      aggregateVersion: version + 1,
      occurredAt: new Date(),
      payload: { amount, description },
    };
 
    // 3. EventStore에 저장 — expectedVersion = 현재 버전
    await this.eventStore.append(streamId, [event], version);
  }
}

CREATE TABLE account_read_models (
  account_id           VARCHAR(255) PRIMARY KEY,
  owner_id             VARCHAR(255) NOT NULL,
  balance              BIGINT NOT NULL DEFAULT 0,
  last_applied_version INTEGER NOT NULL DEFAULT -1,
  last_updated_at      TIMESTAMPTZ NOT NULL
);

class AccountProjection {
  constructor(private db: Pool) {}
 
  async handle(event: DomainEvent): Promise<void> {
    switch (event.eventType) {
      case 'AccountCreated': return this.onAccountCreated(event);
      case 'MoneyDeposited': return this.onMoneyDeposited(event);
      case 'MoneyWithdrawn': return this.onMoneyWithdrawn(event);
    }
  }
 
  private async onAccountCreated(event: DomainEvent): Promise<void> {
    const payload = event.payload as AccountCreatedPayload;
    // ON CONFLICT DO NOTHING: 같은 이벤트가 두 번 처리돼도 중복 삽입 없음
    await this.db.query(
      `INSERT INTO account_read_models
         (account_id, owner_id, balance, last_applied_version, last_updated_at)
       VALUES ($1, $2, $3, $4, $5)
       ON CONFLICT (account_id) DO NOTHING`,
      [
        event.aggregateId,
        payload.ownerId,
        payload.initialBalance,
        event.aggregateVersion,
        event.occurredAt,
      ]
    );
  }
 
  private async onMoneyDeposited(event: DomainEvent): Promise<void> {
    const payload = event.payload as MoneyDepositedPayload;
    // last_applied_version < $2: 같은 이벤트 중복 처리 방지
    await this.db.query(
      `UPDATE account_read_models
       SET balance = balance + $1,
           last_applied_version = $2,
           last_updated_at = $3
       WHERE account_id = $4
         AND last_applied_version < $2`,
      [payload.amount, event.aggregateVersion, event.occurredAt, event.aggregateId]
    );
  }
 
  private async onMoneyWithdrawn(event: DomainEvent): Promise<void> {
    const payload = event.payload as MoneyWithdrawnPayload;
    await this.db.query(
      `UPDATE account_read_models
       SET balance = balance - $1,
           last_applied_version = $2,
           last_updated_at = $3
       WHERE account_id = $4
         AND last_applied_version < $2`,
      [payload.amount, event.aggregateVersion, event.occurredAt, event.aggregateId]
    );
  }
}

const pool = new Pool({ connectionString: process.env.DATABASE_URL });
const eventStore = new PostgresEventStore(pool);
const projection = new AccountProjection(pool);
 
// EventStore 구독 → Projection 연결
eventStore.subscribeToAll(async (event) => {
  await projection.handle(event);
});

class ProjectionRebuilder {
  constructor(private eventStore: EventStore, private db: Pool) {}
 
  async rebuild(projection: AccountProjection): Promise<void> {
    console.log('읽기 모델 재구성 시작...');
    await this.db.query('TRUNCATE TABLE account_read_models');
 
    let position = 0;
    let processed = 0;
    const BATCH_SIZE = 1000;
 
    while (true) {
      const { events, nextPosition } = await this.eventStore.readAllBatch(
        position,
        BATCH_SIZE
      );
      if (events.length === 0) break;
 
      for (const event of events) {
        await projection.handle(event);
        processed++;
      }
 
      position = nextPosition;
      console.log(`${processed}개 이벤트 처리 완료`);
    }
 
    console.log(`재구성 완료: 총 ${processed}개 이벤트 처리`);
  }
}