CQRS와 이벤트 소싱 기초

[전통적 CRUD]
- 같은 모델로 읽기·쓰기
- 읽기 최적화 vs 쓰기 정합성 충돌

[CQRS]
- Command (쓰기): 트랜잭션·정합성
- Query (읽기): 비정규화·빠름

→ 같은 데이터, 다른 모델

// Command — 쓰기
async function placeOrder(input: PlaceOrderCommand) {
  return db.$transaction(async (tx) => {
    const order = await tx.order.create({ data: input });
    await tx.orderItem.createMany({ data: input.items });
    return order;
  });
}


// Query — 읽기 (Materialized View)
async function getOrderListView() {
  return db.$queryRaw`
    SELECT
      o.id,
      o.created_at,
      u.name as customer_name,
      COUNT(oi.id) as item_count,
      o.total,
      o.status
    FROM orders o
    JOIN users u ON u.id = o.user_id
    JOIN order_items oi ON oi.order_id = o.id
    GROUP BY o.id, u.name
    ORDER BY o.created_at DESC
  `;
}


// 또는 별도 read DB
async function getOrderListView() {
  return readDb.orderListView.findMany({
    orderBy: { createdAt: 'desc' },
  });
}

[일반 DB]
- 현재 상태만 저장
- 변경 이력 별도 audit log

[Event Sourcing]
- 모든 변경을 이벤트로 저장
- 현재 상태 = 모든 이벤트 재생


예: 계좌
[일반]
balance: 50000

[Event Sourcing]
2026-01-01: AccountOpened (initial 0)
2026-01-15: Deposited (10000)
2026-01-20: Deposited (50000)
2026-02-01: Withdrew (10000)

balance = 50000 (재계산)

CREATE TABLE events (
  id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
  aggregate_id UUID NOT NULL,    -- 어떤 엔티티
  aggregate_type VARCHAR NOT NULL, -- 'Account', 'Order'
  event_type VARCHAR NOT NULL,    -- 'Deposited', 'Withdrew'
  event_data JSONB NOT NULL,
  event_version INT NOT NULL,
  created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),

  UNIQUE (aggregate_id, event_version)  -- 동시성 제어
);

CREATE INDEX ON events(aggregate_id, event_version);
CREATE INDEX ON events(aggregate_type, created_at);

class Account {
  private state = {
    id: '',
    balance: 0,
    version: 0,
  };

  // 이벤트로부터 상태 복원
  static replay(events: Event[]): Account {
    const account = new Account();
    for (const event of events) {
      account.apply(event);
    }
    return account;
  }

  apply(event: Event) {
    switch (event.type) {
      case 'AccountOpened':
        this.state.id = event.data.id;
        this.state.balance = 0;
        break;
      case 'Deposited':
        this.state.balance += event.data.amount;
        break;
      case 'Withdrew':
        if (this.state.balance < event.data.amount) {
          throw new Error('Insufficient funds');
        }
        this.state.balance -= event.data.amount;
        break;
    }
    this.state.version = event.version;
  }

  // Command → 새 이벤트 생성
  deposit(amount: number): Event[] {
    return [{
      type: 'Deposited',
      data: { amount, at: new Date() },
      version: this.state.version + 1,
    }];
  }
}


// 사용
async function deposit(accountId: string, amount: number) {
  const events = await eventStore.load(accountId);
  const account = Account.replay(events);
  const newEvents = account.deposit(amount);
  await eventStore.append(accountId, newEvents);
}

// 이벤트 → 읽기 모델 자동 갱신
async function projectAccount(event: Event) {
  switch (event.type) {
    case 'AccountOpened':
      await readDb.accountSummary.create({
        data: { id: event.aggregateId, balance: 0 },
      });
      break;

    case 'Deposited':
      await readDb.accountSummary.update({
        where: { id: event.aggregateId },
        data: { balance: { increment: event.data.amount } },
      });
      break;

    case 'Withdrew':
      await readDb.accountSummary.update({
        where: { id: event.aggregateId },
        data: { balance: { decrement: event.data.amount } },
      });
      break;
  }
}


// Kafka consumer
consumer.run({
  eachMessage: async ({ message }) => {
    const event = JSON.parse(message.value!.toString());
    await projectAccount(event);
  },
});

✅ 장점:
- 완전한 audit log (자동)
- 시점 복구 (특정 날짜로 재생)
- 여러 read model (다른 관점)
- 이벤트 분석 (BI)

❌ 단점:
- 학습 곡선 가파름
- Snapshot 필요 (큰 aggregate)
- 이벤트 스키마 변경 어려움
- 디버깅 어려움


→ 금융·감사 필수 도메인에 적합
→ 단순 CRUD에는 과함

[적합]
- 금융 (계좌·거래)
- 이커머스 (주문 흐름)
- 의료 (진료 기록)
- 법무 (계약·이력)

[부적합]
- 단순 콘텐츠 (블로그)
- 이미 존재하는 데이터의 캐시
- 트래픽 적은 내부 도구