PostgreSQL 이벤트 스토어로 구현하는 CQRS와 Read Model on-demand 재구성

flowchart LR
  Client[클라이언트]
  CMD[커맨드 핸들러]
  ES[이벤트 스토어]
  DBT[DB 트리거]
  PW[프로젝션 워커]
  RM[Read Model DB]
  QH[쿼리 핸들러]
 
  Client -->|커맨드 전송| CMD
  CMD --> ES
  ES --> DBT
  DBT -->|pg_notify| PW
  PW --> RM
  Client -->|쿼리 요청| QH
  QH --> RM

-- 기존 방식: 상태를 덮어씀
UPDATE accounts SET balance = 950 WHERE id = 'acc-001';
-- 50이 빠진 이유는? 누가 뺐지? 언제? → 알 수 없음

-- Event Sourcing: 이벤트를 추가 (Append-Only)
INSERT INTO events (aggregate_id, event_type, event_data, sequence_number)
VALUES ('acc-001', 'MoneyWithdrawn', '{"amount": 50, "reason": "ATM"}', 5);
-- 현재 잔액 = 모든 입출금 이벤트의 합산
-- 누가, 언제, 왜 빠졌는지 → 이벤트에 다 있음

CREATE TABLE events (
  event_id        UUID        DEFAULT gen_random_uuid() PRIMARY KEY,
  aggregate_id    UUID        NOT NULL,
  aggregate_type  TEXT        NOT NULL,    -- 'Account', 'Order' 등
  event_type      TEXT        NOT NULL,    -- 'MoneyWithdrawn' 등
  event_data      JSONB       NOT NULL,
  sequence_number INTEGER     NOT NULL,
  global_position BIGSERIAL   NOT NULL,    -- 전역 순서: 폴링 복구·재구성 체크포인트용
  occurred_at     TIMESTAMPTZ DEFAULT now()
);
 
-- 낙관적 동시성 제어의 핵심
CREATE UNIQUE INDEX idx_events_aggregate_seq
  ON events (aggregate_id, sequence_number);
 
-- 조회 최적화
CREATE INDEX idx_events_aggregate_id ON events (aggregate_id, occurred_at);
CREATE INDEX idx_events_event_type   ON events (event_type);
CREATE INDEX idx_events_global_pos   ON events (global_position);

sequenceDiagram
  participant App as 애플리케이션
  participant ES as 이벤트 스토어
 
  App->>ES: aggregate_id로 이벤트 조회
  ES-->>App: 이벤트 목록 seq 1~N 반환
  Note over App: Aggregate는 in-memory 객체
  loop 각 이벤트를 순서대로
    App->>App: applyEvent로 상태 누적
  end
  Note over App: 현재 Aggregate 상태 완성

type AccountEvent =
  | { type: 'AccountOpened'; ownerId: string; initialBalance: number }
  | { type: 'MoneyDeposited'; amount: number; description: string }
  | { type: 'MoneyWithdrawn'; amount: number; description: string }
  | { type: 'AccountClosed'; reason: string };
 
async function appendEvents(
  aggregateId: string,
  events: AccountEvent[],
  expectedVersion: number,  // 낙관적 동시성 제어
  db: Pool
): Promise<void> {
  const client = await db.connect();
  try {
    await client.query('BEGIN');
 
    // 프로덕션에서는 단건 INSERT 반복 대신 멀티-row INSERT로 최적화 권장
    // (이벤트 수가 늘수록 단건 루프는 병목이 됨)
    for (let i = 0; i < events.length; i++) {
      const event = events[i];
      const nextSeq = expectedVersion + i + 1;
 
      await client.query(
        `INSERT INTO events
           (aggregate_id, aggregate_type, event_type, event_data, sequence_number)
         VALUES ($1, 'Account', $2, $3, $4)`,
        [aggregateId, event.type, JSON.stringify(event), nextSeq]
      );
    }
 
    await client.query('COMMIT');
  } catch (err) {
    await client.query('ROLLBACK');
    // PostgreSQL 유니크 제약 위반 = 동시성 충돌
    if ((err as any).code === '23505') {
      throw new ConcurrencyError('동시 수정 충돌. 재시도가 필요합니다.');
    }
    throw err;
  } finally {
    client.release();
  }
}

async function withdrawMoney(
  aggregateId: string,
  amount: number,
  description: string,
  db: Pool
): Promise<void> {
  // ① 이벤트를 재생해 현재 상태 복원
  const state = await loadAccount(aggregateId, db);
 
  // ② 비즈니스 규칙 검증 (Write 모델에서 수행)
  if (state.isClosed) {
    throw new Error('닫힌 계좌에는 출금할 수 없습니다.');
  }
  if (state.balance < amount) {
    throw new Error(`잔액 부족: 현재 ${state.balance}, 요청 ${amount}`);
  }
 
  // ③ 이벤트 저장. state.version = 마지막 sequence_number → 동시성 충돌 감지
  await appendEvents(
    aggregateId,
    [{ type: 'MoneyWithdrawn', amount, description }],
    state.version,  // expectedVersion
    db
  );
}

interface AccountState {
  id: string;
  ownerId: string;      // AccountOpened 이벤트에서 추출
  balance: number;
  isClosed: boolean;
  version: number;      // 마지막으로 처리한 sequence_number
}
 
function applyEvent(state: AccountState, event: AccountEvent): AccountState {
  switch (event.type) {
    case 'AccountOpened':
      return { ...state, ownerId: event.ownerId, balance: event.initialBalance };
    case 'MoneyDeposited':
      return { ...state, balance: state.balance + event.amount };
    case 'MoneyWithdrawn':
      return { ...state, balance: state.balance - event.amount };
    case 'AccountClosed':
      return { ...state, isClosed: true };
  }
}
 
async function loadAccount(aggregateId: string, db: Pool): Promise<AccountState> {
  const result = await db.query(
    `SELECT event_type, event_data, sequence_number
     FROM events
     WHERE aggregate_id = $1 AND aggregate_type = 'Account'
     ORDER BY sequence_number ASC`,
    [aggregateId]
  );
 
  const initial: AccountState = {
    id: aggregateId,
    ownerId: '',
    balance: 0,
    isClosed: false,
    version: 0,
  };
 
  return result.rows.reduce((state, row) => ({
    ...applyEvent(state, row.event_data as AccountEvent),
    version: row.sequence_number,
  }), initial);
}

CREATE OR REPLACE FUNCTION notify_event_inserted()
RETURNS TRIGGER AS $$
BEGIN
  PERFORM pg_notify(
    'new_event',
    json_build_object(
      'event_id',        NEW.event_id,
      'aggregate_id',    NEW.aggregate_id,
      'event_type',      NEW.event_type,
      'global_position', NEW.global_position
    )::text  -- NOTIFY payload는 8,000바이트 제한. 메타데이터만 전송하고 데이터는 별도 조회
  );
  RETURN NEW;
END;
$$ LANGUAGE plpgsql;
 
CREATE TRIGGER event_inserted_trigger
  AFTER INSERT ON events
  FOR EACH ROW EXECUTE FUNCTION notify_event_inserted();

// 프로젝션 워커: LISTEN + 폴링 폴백 조합
async function startProjectionWorker(db: Pool): Promise<void> {
  const workerClient = await db.connect();
  let lastProcessedPosition = await getLastProcessedPosition(db);
 
  await workerClient.query('LISTEN new_event');
 
  workerClient.on('notification', async (msg) => {
    const payload = JSON.parse(msg.payload!);
    await updateAccountSummaryReadModel(payload.aggregate_id, db);
    lastProcessedPosition = payload.global_position;
    await saveLastProcessedPosition(lastProcessedPosition, db);
  });
 
  // 연결 복구 후 NOTIFY 유실분을 폴링으로 보정
  // NOTIFY는 워커가 연결되어 있을 때만 전달됨. 연결 단절 구간의 이벤트는 여기서 처리
  workerClient.on('connect', async () => {
    await catchUpFromPosition(lastProcessedPosition, db);
  });
}
 
async function catchUpFromPosition(fromPosition: number, db: Pool): Promise<void> {
  const missed = await db.query(
    `SELECT DISTINCT aggregate_id
     FROM events
     WHERE aggregate_type = 'Account' AND global_position > $1
     ORDER BY MIN(global_position) ASC`,
    [fromPosition]
  );
 
  for (const row of missed.rows) {
    await updateAccountSummaryReadModel(row.aggregate_id, db);
  }
}

-- Read Model 테이블: 조회에 최적화된 비정규화 뷰
CREATE TABLE account_summary (
  account_id   UUID           PRIMARY KEY,
  owner_id     TEXT           NOT NULL,
  balance      NUMERIC(15, 2) NOT NULL DEFAULT 0,
  is_closed    BOOLEAN        NOT NULL DEFAULT false,
  last_updated TIMESTAMPTZ,
  version      INTEGER        NOT NULL DEFAULT 0
);

async function updateAccountSummaryReadModel(
  aggregateId: string,
  db: Pool
): Promise<void> {
  const state = await loadAccount(aggregateId, db);
 
  await db.query(
    `INSERT INTO account_summary (account_id, owner_id, balance, is_closed, last_updated, version)
     VALUES ($1, $2, $3, $4, now(), $5)
     ON CONFLICT (account_id) DO UPDATE
       SET owner_id     = EXCLUDED.owner_id,
           balance      = EXCLUDED.balance,
           is_closed    = EXCLUDED.is_closed,
           last_updated = EXCLUDED.last_updated,
           version      = EXCLUDED.version`,
    [state.id, state.ownerId, state.balance, state.isClosed, state.version]
  );
}

async function rebuildReadModel(db: Pool): Promise<void> {
  const client = await db.connect();
  try {
    // pg_advisory_lock은 세션 레벨 락. 연결이 끊기면 자동 해제되므로
    // finally 블록에서 명시적으로 unlock해 다음 재구성 요청이 즉시 진입할 수 있게 함
    await client.query('SELECT pg_advisory_lock(12345)');
 
    // CREATE TABLE IF NOT EXISTS: 중간 실패 후 재시도 시 오류 방지
    await client.query(
      'CREATE TABLE IF NOT EXISTS account_summary_new (LIKE account_summary INCLUDING ALL)'
    );
    await client.query('TRUNCATE account_summary_new');
 
    // 체크포인트: 재구성 시작 시점의 전역 위치를 기록
    const { rows: [{ checkpoint }] } = await client.query(
      'SELECT COALESCE(MAX(global_position), 0) AS checkpoint FROM events'
    );
 
    // 체크포인트까지의 이벤트로 새 Read Model 구성
    const aggregates = await client.query(
      `SELECT DISTINCT aggregate_id FROM events
       WHERE aggregate_type = 'Account' AND global_position <= $1`,
      [checkpoint]
    );
 
    for (const row of aggregates.rows) {
      const state = await loadAccountUpTo(row.aggregate_id, checkpoint, db);
      await client.query(
        `INSERT INTO account_summary_new
           (account_id, owner_id, balance, is_closed, last_updated, version)
         VALUES ($1, $2, $3, $4, now(), $5)`,
        [state.id, state.ownerId, state.balance, state.isClosed, state.version]
      );
    }
 
    // 체크포인트 이후 유입된 이벤트 보정 (재구성 중 누락분)
    const lateAggregates = await client.query(
      `SELECT DISTINCT aggregate_id FROM events
       WHERE aggregate_type = 'Account' AND global_position > $1`,
      [checkpoint]
    );
 
    for (const row of lateAggregates.rows) {
      const state = await loadAccount(row.aggregate_id, db);
      await client.query(
        `INSERT INTO account_summary_new
           (account_id, owner_id, balance, is_closed, last_updated, version)
         VALUES ($1, $2, $3, $4, now(), $5)
         ON CONFLICT (account_id) DO UPDATE
           SET owner_id     = EXCLUDED.owner_id,
               balance      = EXCLUDED.balance,
               is_closed    = EXCLUDED.is_closed,
               last_updated = EXCLUDED.last_updated,
               version      = EXCLUDED.version`,
        [state.id, state.ownerId, state.balance, state.isClosed, state.version]
      );
    }
 
    // 원자적 테이블 교체: 읽기 서비스 무중단
    await client.query('BEGIN');
    await client.query('ALTER TABLE account_summary RENAME TO account_summary_old');
    await client.query('ALTER TABLE account_summary_new RENAME TO account_summary');
    await client.query('DROP TABLE account_summary_old');
    await client.query('COMMIT');
 
  } finally {
    await client.query('SELECT pg_advisory_unlock(12345)');
    client.release();
  }
}
 
// global_position 상한을 지정해 특정 체크포인트까지만 재생
async function loadAccountUpTo(
  aggregateId: string,
  maxPosition: number,
  db: Pool
): Promise<AccountState> {
  const result = await db.query(
    `SELECT event_type, event_data, sequence_number
     FROM events
     WHERE aggregate_id = $1 AND aggregate_type = 'Account'
       AND global_position <= $2
     ORDER BY sequence_number ASC`,
    [aggregateId, maxPosition]
  );
 
  const initial: AccountState = {
    id: aggregateId, ownerId: '', balance: 0, isClosed: false, version: 0,
  };
 
  return result.rows.reduce((state, row) => ({
    ...applyEvent(state, row.event_data as AccountEvent),
    version: row.sequence_number,
  }), initial);
}

CREATE TABLE snapshots (
  aggregate_id    UUID        NOT NULL,
  aggregate_type  TEXT        NOT NULL,
  state_data      JSONB       NOT NULL,
  version         INTEGER     NOT NULL,  -- 반영된 마지막 sequence_number
  created_at      TIMESTAMPTZ DEFAULT now(),
  PRIMARY KEY (aggregate_id, version)
);

const SNAPSHOT_THRESHOLD = 100; // 이벤트 100개마다 스냅샷 저장
 
async function saveSnapshot(state: AccountState, db: Pool): Promise<void> {
  await db.query(
    `INSERT INTO snapshots (aggregate_id, aggregate_type, state_data, version)
     VALUES ($1, 'Account', $2, $3)
     ON CONFLICT (aggregate_id, version) DO NOTHING`,
    [state.id, JSON.stringify(state), state.version]
  );
}
 
async function loadAccountWithSnapshot(aggregateId: string, db: Pool): Promise<AccountState> {
  // 1. 최신 스냅샷 조회
  const snapResult = await db.query(
    `SELECT state_data, version FROM snapshots
     WHERE aggregate_id = $1 AND aggregate_type = 'Account'
     ORDER BY version DESC LIMIT 1`,
    [aggregateId]
  );
 
  let state: AccountState;
  let fromVersion: number;
 
  if (snapResult.rows.length > 0) {
    state = snapResult.rows[0].state_data as AccountState;
    fromVersion = snapResult.rows[0].version;
  } else {
    state = { id: aggregateId, ownerId: '', balance: 0, isClosed: false, version: 0 };
    fromVersion = 0;
  }
 
  // 2. 스냅샷 이후 이벤트만 재생
  const events = await db.query(
    `SELECT event_type, event_data, sequence_number FROM events
     WHERE aggregate_id = $1 AND sequence_number > $2
     ORDER BY sequence_number ASC`,
    [aggregateId, fromVersion]
  );
 
  const finalState = events.rows.reduce((s, row) => ({
    ...applyEvent(s, row.event_data as AccountEvent),
    version: row.sequence_number,
  }), state);
 
  // 3. 스냅샷 저장 여부 판단: 마지막 스냅샷 이후 이벤트가 임계치 초과 시 저장
  if (events.rows.length >= SNAPSHOT_THRESHOLD) {
    await saveSnapshot(finalState, db);
  }
 
  return finalState;
}