Kafka와 PostgreSQL로 Event Sourcing + CQRS 실전 구현하기

기존 방식: [현재 상태] balance = 5000
ES 방식:   [이벤트 로그]
             AccountOpened  { initialBalance: 3000 }  // version=1
             DepositMade    { amount: 2500 }           // version=2
             WithdrawalMade { amount: 500 }            // version=3

flowchart TD
    클라이언트 --> CMD[Command Handler]
    CMD --> AGG[Aggregate]
    AGG --> CMD
    CMD --> PG[PostgreSQL events 테이블]
    CMD --> KFK[Kafka - account-events]
    PG -->|재수화 조회| CMD
    KFK --> PJ1[Projection - 계좌 잔액뷰]
    KFK --> PJ2[Projection - 감사 로그뷰]
    KFK --> PJ3[Projection - 거래 내역뷰]
    PJ1 --> RDB[PostgreSQL - 읽기 모델]
    PJ2 --> RDB
    PJ3 --> RDB
    RDB --> QH[Query Handler]
    QH --> 클라이언트

// types/domain-events.ts
import { IEvent } from '@nestjs/cqrs';
 
// BaseEvent가 IEvent를 확장하면 NestJS AggregateRoot.apply()에 as any 없이 넘길 수 있습니다.
export interface BaseEvent extends IEvent {
  aggregateId: string;
  timestamp: Date;
  version: number;
}
 
export interface AccountOpenedEvent extends BaseEvent {
  type: 'AccountOpened';
  payload: { ownerId: string; initialBalance: number };
}
 
export interface DepositMadeEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}
 
export interface WithdrawalMadeEvent extends BaseEvent {
  type: 'WithdrawalMade';
  payload: { amount: number; description: string };
}
 
// 판별 유니온 — applyEvent의 switch에서 TypeScript가 각 case 안의 타입을 좁혀줍니다.
export type DomainEvent = AccountOpenedEvent | DepositMadeEvent | WithdrawalMadeEvent;

// types/commands.ts
export interface OpenAccountCommand {
  accountId: string;
  ownerId: string;
  initialBalance: number;
}
 
export interface DepositCommand {
  accountId: string;
  amount: number;
  description: string;
}
 
export interface WithdrawCommand {
  accountId: string;
  amount: number;
  description: string;
}

// account.aggregate.ts
import { AggregateRoot } from '@nestjs/cqrs';
import { DomainEvent } from './types/domain-events';
import { OpenAccountCommand, DepositCommand, WithdrawCommand } from './types/commands';
 
export type AccountStatus = 'ACTIVE' | 'FROZEN' | 'CLOSED';
 
export interface AccountState {
  id: string;
  balance: number;
  status: AccountStatus;
  version: number;
}
 
export class AccountAggregate extends AggregateRoot<DomainEvent> {
  private state: AccountState = { id: '', balance: 0, status: 'ACTIVE', version: 0 };
 
  // NestJS loadFromHistory의 기본 구현은 uncommitted 목록 관리만 합니다.
  // 도메인 상태를 변경하는 reduceState 호출을 여기서 연결합니다.
  loadFromHistory(events: DomainEvent[]): void {
    events.forEach(event => this.reduceState(event));
  }
 
  static fromSnapshot(state: AccountState): AccountAggregate {
    const agg = new AccountAggregate();
    agg.state = { ...state };
    return agg;
  }
 
  openAccount(command: OpenAccountCommand): void {
    if (this.state.id) {
      throw new Error('계좌가 이미 존재합니다');
    }
    this.applyAndRecord({
      type: 'AccountOpened',
      aggregateId: command.accountId,
      payload: { ownerId: command.ownerId, initialBalance: command.initialBalance },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  deposit(command: DepositCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('활성 상태의 계좌에만 입금할 수 있습니다');
    }
    if (command.amount <= 0) {
      throw new Error('입금 금액은 0보다 커야 합니다');
    }
    this.applyAndRecord({
      type: 'DepositMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  withdraw(command: WithdrawCommand): void {
    if (this.state.status !== 'ACTIVE') {
      throw new Error('활성 상태의 계좌에만 출금할 수 있습니다');
    }
    if (command.amount <= 0) {
      throw new Error('출금 금액은 0보다 커야 합니다');
    }
    if (this.state.balance < command.amount) {
      throw new Error('잔액이 부족합니다');
    }
    this.applyAndRecord({
      type: 'WithdrawalMade',
      aggregateId: this.state.id,
      payload: { amount: command.amount, description: command.description },
      timestamp: new Date(),
      version: this.state.version + 1,
    });
  }
 
  private applyAndRecord(event: DomainEvent): void {
    this.reduceState(event);
    this.apply(event); // NestJS AggregateRoot — uncommitted 목록에 추가
  }
 
  private reduceState(event: DomainEvent): void {
    switch (event.type) {
      case 'AccountOpened':
        this.state.id = event.aggregateId;
        this.state.balance = event.payload.initialBalance;
        this.state.status = 'ACTIVE';
        break;
      case 'DepositMade':
        this.state.balance += event.payload.amount;
        break;
      case 'WithdrawalMade':
        this.state.balance -= event.payload.amount;
        break;
    }
    this.state.version = event.version;
  }
 
  getState(): AccountState {
    return { ...this.state };
  }
}

// event-store/event-store.service.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { InjectDataSource } from '@nestjs/typeorm';
import { DataSource } from 'typeorm';
import { Kafka, Producer, CompressionTypes } from 'kafkajs';
import { DomainEvent } from '../types/domain-events';
 
export class OptimisticConcurrencyException extends Error {
  constructor(aggregateId: string, expectedVersion: number) {
    super(
      `[${aggregateId}] 버전 충돌: 예상 버전 ${expectedVersion}은 이미 다른 커맨드가 사용했습니다`,
    );
  }
}
 
@Injectable()
export class EventStoreService implements OnModuleInit {
  private producer: Producer;
 
  constructor(@InjectDataSource() private readonly dataSource: DataSource) {
    const kafka = new Kafka({
      clientId: 'account-service',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.producer = kafka.producer();
  }
 
  async onModuleInit(): Promise<void> {
    await this.producer.connect();
    await this.ensureEventsTable();
  }
 
  async appendEvents(
    aggregateId: string,
    events: DomainEvent[],
    expectedVersion: number,
  ): Promise<void> {
    await this.dataSource.transaction(async (manager) => {
      for (const event of events) {
        try {
          // UNIQUE (aggregate_id, version) 제약이 낙관적 잠금을 강제합니다.
          // 두 커맨드가 동시에 같은 버전으로 쓰려 하면 두 번째 INSERT가 실패합니다.
          await manager.query(
            `INSERT INTO events (aggregate_id, version, type, payload, timestamp)
             VALUES ($1, $2, $3, $4, $5)`,
            [
              event.aggregateId,
              event.version,
              event.type,
              JSON.stringify(event.payload),
              event.timestamp,
            ],
          );
        } catch (err: any) {
          if (err.code === '23505') {
            throw new OptimisticConcurrencyException(aggregateId, expectedVersion);
          }
          throw err;
        }
      }
    });
 
    // PostgreSQL 트랜잭션 커밋 후 Kafka 발행
    // 프로덕션에서는 Transactional Outbox + CDC로 원자성을 보장하세요.
    await this.producer.send({
      topic: 'account-events',
      compression: CompressionTypes.GZIP,
      messages: events.map(event => ({
        key: event.aggregateId, // 같은 Aggregate → 같은 파티션 → 순서 보장
        value: JSON.stringify(event),
        headers: {
          eventType: event.type,
          version: String(event.version),
        },
      })),
    });
  }
 
  async readEvents(
    aggregateId: string,
    options?: { maxTimestamp?: Date; fromVersion?: number },
  ): Promise<DomainEvent[]> {
    const params: unknown[] = [aggregateId];
    let query = `
      SELECT type, payload, aggregate_id AS "aggregateId", version, timestamp
      FROM events
      WHERE aggregate_id = $1
    `;
 
    if (options?.fromVersion !== undefined) {
      params.push(options.fromVersion);
      query += ` AND version > $${params.length}`;
    }
    if (options?.maxTimestamp) {
      params.push(options.maxTimestamp.toISOString());
      query += ` AND timestamp <= $${params.length}`;
    }
 
    query += ' ORDER BY version ASC';
 
    const rows: Array<{
      type: string;
      payload: Record<string, unknown>;
      aggregateId: string;
      version: number;
      timestamp: string;
    }> = await this.dataSource.query(query, params);
 
    return rows.map(row => ({
      type: row.type,
      aggregateId: row.aggregateId,
      version: row.version,
      timestamp: new Date(row.timestamp),
      payload: row.payload,
    } as DomainEvent));
  }
 
  private async ensureEventsTable(): Promise<void> {
    await this.dataSource.query(`
      CREATE TABLE IF NOT EXISTS events (
        id            BIGSERIAL    PRIMARY KEY,
        aggregate_id  TEXT         NOT NULL,
        version       INT          NOT NULL,
        type          TEXT         NOT NULL,
        payload       JSONB        NOT NULL,
        timestamp     TIMESTAMPTZ  NOT NULL,
        CONSTRAINT uq_aggregate_version UNIQUE (aggregate_id, version)
      );
      CREATE INDEX IF NOT EXISTS idx_events_aggregate_id
        ON events (aggregate_id, version);
    `);
  }
}

// commands/deposit.command-handler.ts
import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { DepositCommand } from '../types/commands';
import { AccountAggregate } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
import { DomainEvent } from '../types/domain-events';
 
@CommandHandler(DepositCommand)
export class DepositCommandHandler implements ICommandHandler<DepositCommand> {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async execute(command: DepositCommand): Promise<void> {
    // 1. PostgreSQL에서 이 Aggregate의 전체 이벤트 조회
    const events = await this.eventStore.readEvents(command.accountId);
 
    // 2. 이벤트 재생으로 Aggregate 재조립
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    const expectedVersion = account.getState().version;
 
    // 3. 도메인 로직 실행 (유효성 검사 포함)
    account.deposit(command);
 
    // 4. uncommitted 이벤트를 꺼내 저장 — getUncommittedEvents는 NestJS 내장
    const newEvents = account.getUncommittedEvents() as DomainEvent[];
    await this.eventStore.appendEvents(command.accountId, newEvents, expectedVersion);
 
    account.uncommit();
  }
}

// projections/account-balance.projection.ts
import { Injectable, OnModuleInit } from '@nestjs/common';
import { Kafka, Consumer, KafkaMessage } from 'kafkajs';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { AccountBalanceView } from './account-balance-view.entity';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class AccountBalanceProjection implements OnModuleInit {
  private consumer: Consumer;
 
  constructor(
    @InjectRepository(AccountBalanceView)
    private readonly viewRepo: Repository<AccountBalanceView>,
  ) {
    const kafka = new Kafka({
      clientId: 'balance-projection',
      brokers: [process.env.KAFKA_BROKER ?? 'localhost:9092'],
    });
    this.consumer = kafka.consumer({ groupId: 'account-balance-projection' });
  }
 
  async onModuleInit(): Promise<void> {
    await this.consumer.connect();
    await this.consumer.subscribe({ topic: 'account-events', fromBeginning: true });
    await this.consumer.run({
      eachMessage: async ({ message }) => this.handle(message),
    });
  }
 
  private async handle(message: KafkaMessage): Promise<void> {
    const event = JSON.parse(message.value!.toString()) as DomainEvent;
 
    try {
      switch (event.type) {
        case 'AccountOpened': {
          await this.viewRepo
            .createQueryBuilder()
            .insert()
            .values({
              accountId: event.aggregateId,
              balance: event.payload.initialBalance,
              ownerId: event.payload.ownerId,
              status: 'ACTIVE',
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .orIgnore() // 멱등성: 이미 존재하면 무시
            .execute();
          break;
        }
 
        case 'DepositMade': {
          // WHERE projectionVersion < version — 중복 소비 시 UPDATE 조건 불일치로 건너뜀
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance + ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
 
        case 'WithdrawalMade': {
          await this.viewRepo
            .createQueryBuilder()
            .update()
            .set({
              balance: () => `balance - ${event.payload.amount}`,
              lastUpdatedAt: event.timestamp,
              projectionVersion: event.version,
            })
            .where('accountId = :id AND projectionVersion < :version', {
              id: event.aggregateId,
              version: event.version,
            })
            .execute();
          break;
        }
      }
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error);
      } else {
        throw error; // 일시적 오류는 re-throw → Kafka가 재시도
      }
    }
  }
 
  private async sendToDeadLetterQueue(message: KafkaMessage, error: Error): Promise<void> {
    // 'account-events-dlq' 토픽으로 이동해 수동 검토 후 재처리
  }
}

// time-travel/time-travel.service.ts
import { Injectable, NotFoundException, BadRequestException } from '@nestjs/common';
import { EventStoreService } from '../event-store/event-store.service';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { DomainEvent } from '../types/domain-events';
 
@Injectable()
export class TimeTravelService {
  constructor(private readonly eventStore: EventStoreService) {}
 
  async getAccountStateAt(accountId: string, timestamp: Date): Promise<AccountState> {
    const events = await this.eventStore.readEvents(accountId, {
      maxTimestamp: timestamp,
    });
 
    if (events.length === 0) {
      throw new NotFoundException(
        `${timestamp.toISOString()} 시점에 계좌 ${accountId}가 존재하지 않습니다`,
      );
    }
 
    const account = new AccountAggregate();
    account.loadFromHistory(events);
    return account.getState();
  }
 
  async getEventsBetween(
    accountId: string,
    from: Date,
    to: Date,
  ): Promise<DomainEvent[]> {
    const allEvents = await this.eventStore.readEvents(accountId);
    return allEvents.filter(e => {
      const t = new Date(e.timestamp);
      return t >= from && t <= to;
    });
  }
}

// time-travel/time-travel.controller.ts
import { Controller, Get, Param, Query, BadRequestException } from '@nestjs/common';
import { TimeTravelService } from './time-travel.service';
 
@Controller('accounts')
export class TimeTravelController {
  constructor(private readonly timeTravelService: TimeTravelService) {}
 
  @Get(':id/state')
  async getStateAt(
    @Param('id') accountId: string,
    @Query('at') at: string,
  ) {
    if (!at || isNaN(Date.parse(at))) {
      throw new BadRequestException('at 파라미터는 ISO 8601 형식이어야 합니다');
    }
    return this.timeTravelService.getAccountStateAt(accountId, new Date(at));
  }
}
 
// GET /accounts/acc-123/state?at=2025-03-15T09:00:00Z

sequenceDiagram
    participant 클라이언트
    participant API서버
    participant PostgreSQL
    participant Aggregate
 
    클라이언트->>API서버: GET /accounts/123/state?at=2025-03-15T09:00:00Z
    API서버->>PostgreSQL: SELECT ... WHERE aggregate_id='123' AND timestamp <= '2025-03-15...' ORDER BY version
    PostgreSQL-->>API서버: version 1~N 이벤트 반환
    API서버->>Aggregate: new AccountAggregate(), loadFromHistory(events)
    loop 이벤트 재생
        Aggregate->>Aggregate: reduceState 순서대로 적용
    end
    Aggregate-->>API서버: 과거 시점의 AccountState
    API서버-->>클라이언트: balance=3000, status=ACTIVE

// snapshot/snapshot.service.ts
import { Injectable } from '@nestjs/common';
import { InjectRepository } from '@nestjs/typeorm';
import { Repository } from 'typeorm';
import { SnapshotEntity } from './snapshot.entity';
import { AccountAggregate, AccountState } from '../account.aggregate';
import { EventStoreService } from '../event-store/event-store.service';
 
@Injectable()
export class SnapshotService {
  private readonly SNAPSHOT_INTERVAL = 100;
 
  constructor(
    @InjectRepository(SnapshotEntity)
    private readonly snapshotRepo: Repository<SnapshotEntity>,
  ) {}
 
  async loadAggregate(
    aggregateId: string,
    eventStore: EventStoreService,
  ): Promise<AccountAggregate> {
    const snapshot = await this.snapshotRepo.findOne({
      where: { aggregateId },
      order: { version: 'DESC' },
    });
 
    let account: AccountAggregate;
    let fromVersion: number | undefined;
 
    if (snapshot) {
      // fromSnapshot은 AccountAggregate.fromSnapshot(state)으로 정의한 static 메서드
      account = AccountAggregate.fromSnapshot(snapshot.state as AccountState);
      fromVersion = snapshot.version;
    } else {
      account = new AccountAggregate();
    }
 
    const remainingEvents = await eventStore.readEvents(aggregateId, { fromVersion });
    account.loadFromHistory(remainingEvents);
 
    return account;
  }
 
  async maybeSaveSnapshot(account: AccountAggregate): Promise<void> {
    const state = account.getState();
    if (state.version > 0 && state.version % this.SNAPSHOT_INTERVAL === 0) {
      await this.snapshotRepo.save({
        aggregateId: state.id,
        version: state.version,
        state: state,
        createdAt: new Date(),
      });
    }
  }
}

flowchart TD
    A[새 도메인 설계] --> B{이벤트 히스토리가 핵심 요건인가\n감사 추적 시간여행 버그재현}
    B -->|아니오| C[단순 CRUD]
    B -->|예| D{이벤트 자체에 비즈니스 가치가 있는가\n금융 의료 컴플라이언스 커머스}
    D -->|아니오| E[CDC나 감사 테이블\n또는 Outbox 패턴 고려]
    D -->|예| F[Event Sourcing 적용]
    F --> G{읽기 부하가 쓰기보다 압도적으로 크거나\n다양한 읽기 모델이 필요한가}
    G -->|아니오| H[Event Sourcing 단독 적용도 충분]
    G -->|예| I[CQRS 분리 추가]
    I --> J[Bounded Context 단위로 선택적 적용]

// 나쁜 예: 파생 데이터가 섞임
interface BadDepositEvent {
  type: 'DepositMade';
  amount: number;
  balanceAfter: number;          // ❌ Aggregate가 계산하면 되는 파생값
  allTransactions: unknown[];    // ❌ 이벤트에 담기엔 너무 많음
}
 
// 좋은 예: 발생한 사실만
interface GoodDepositEvent extends BaseEvent {
  type: 'DepositMade';
  payload: { amount: number; description: string };
}

// 나쁜 예
'DepositAccount', 'CreateOrder', 'UpdateUser'
 
// 좋은 예 — 구체적으로 쪼갤수록 감사 로그의 가치가 높아집니다
'DepositMade', 'OrderPlaced', 'UserEmailChanged', 'UserPhoneNumberChanged'

await this.consumer.run({
  eachMessage: async ({ message }) => {
    try {
      await this.handle(message);
    } catch (error) {
      if (error instanceof KnownBusinessError) {
        await this.sendToDeadLetterQueue(message, error); // 격리 후 수동 재처리
      } else {
        throw error; // 일시적 오류 → Kafka가 재시도
      }
    }
  },
});