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

interface DomainEvent<T = unknown> {
  readonly eventId: string;
  readonly eventType: string;
  readonly aggregateId: string;  // same as streamId in this system (e.g., '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 of account domain events
// payload type is narrowed automatically in accountReducer without casting
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(`Expected version ${expectedVersion} for stream ${streamId} did not match`);
    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');
 
      // Version check for fast failure — the actual concurrency guard is the UNIQUE constraint
      // PostgreSQL does not allow aggregate functions and FOR UPDATE in the same query,
      // so no row lock is applied to the 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,          // use the ID generated by the handler as-is
            streamId,
            event.eventType,
            event.aggregateVersion,
            event.occurredAt,
            JSON.stringify(event.payload),
          ]
        );
      }
 
      await client.query('COMMIT');
    } catch (err) {
      await client.query('ROLLBACK');
      // UNIQUE violation = concurrent write conflict → convert to 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('Error processing event subscription:', 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;
}
 
// Taking the AccountEvent union as a parameter narrows payload automatically in each switch branch
const accountReducer = (
  state: AccountState | null,
  event: AccountEvent
): AccountState => {
  switch (event.eventType) {
    case 'AccountCreated':
      return {
        id: event.aggregateId,
        ownerId: event.payload.ownerId,        // inferred as AccountCreatedPayload automatically
        balance: event.payload.initialBalance,
        version: event.aggregateVersion,
        isClosed: false,
      };
    case 'MoneyDeposited':
      if (!state) throw new Error('Invalid event order');
      return {
        ...state,
        balance: state.balance + event.payload.amount,  // inferred as MoneyDepositedPayload automatically
        version: event.aggregateVersion,
      };
    case 'MoneyWithdrawn':
      if (!state) throw new Error('Invalid event order');
      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(`Account ${accountId} not found`);
  }
 
  // Boundary cast of DomainEvent[] returned by EventStore to 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,  // version of the first event is 0
      occurredAt: new Date(),
      payload: { ownerId, initialBalance },
    };
 
    // New stream: expectedVersion = -1 (no events yet)
    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. Restore current state via event replay
    const { state, version } = await rehydrateAccount(this.eventStore, accountId);
 
    // 2. Validate domain rules
    if (state.isClosed) throw new Error('Cannot withdraw from a closed account');
    if (state.balance < amount) throw new Error(`Insufficient balance: current balance is ${state.balance}`);
    if (amount <= 0) throw new Error('Withdrawal amount must be greater than 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. Save to EventStore — expectedVersion = current version
    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: no duplicate insert even if the same event is processed twice
    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: prevents duplicate processing of the same event
    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);
 
// Subscribe to EventStore → connect to 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('Starting read model rebuild...');
    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 ${processed} events`);
    }
 
    console.log(`Rebuild complete: processed ${processed} events total`);
  }
}