CQRS & Event Sourcing

분류: Layer 9 - 아키텍처 & 설계 패턴

CQRS / Event Sourcing

1. 한 줄 정의

CQRS는 읽기(Query)와 쓰기(Command) 책임을 분리하는 아키텍처 패턴이고, Event Sourcing은 상태 대신 이벤트(변경 이력)를 저장해 현재 상태를 재현하는 패턴이다. 두 패턴은 독립적으로 쓸 수 있지만, 결합하면 Event Sourcing의 이벤트가 CQRS 읽기 모델 동기화 트리거가 되어 시너지가 극대화된다.

2. 왜 중요한가

CQRS가 필요한 이유

대부분의 시스템에서 읽기(Read)와 쓰기(Write) 요청의 특성은 근본적으로 다르다. 읽기는 빠른 응답과 복잡한 조인(JOIN)이 필요하고, 쓰기는 데이터 정합성과 트랜잭션이 중요하다. 단일 모델로 두 가지를 모두 만족시키려 하면 어느 쪽도 최적화하기 어렵다.

트래픽이 많은 서비스에서 읽기와 쓰기를 독립적으로 스케일링할 수 있다
읽기 모델과 쓰기 모델을 서로 다른 DB로 분리할 수 있다 (예: 쓰기는 PostgreSQL, 읽기는 Elasticsearch)
DDD(Domain-Driven Design)와 잘 맞아떨어진다

Event Sourcing이 필요한 이유

전통적인 CRUD는 현재 상태만 저장한다. “언제, 왜, 어떻게 이 상태가 됐는가”를 알 수 없다. 금융, 주문 이력, 감사(Audit) 시스템처럼 변경 이력이 중요한 도메인에서는 이벤트 자체를 저장해 완벽한 히스토리를 보장해야 한다.

프론트엔드 개발자를 위한 브릿지

React에서 useMutation으로 데이터를 변경(Command)한 후 invalidateQueries로 캐시를 무효화(Query 읽기 모델 재조회)하는 패턴은 사실상 미니 CQRS다. useMutation = Command 실행, invalidateQueries = 읽기 모델 무효화 후 최신 데이터 재조회. 규모가 커지면 이 패턴이 서버 아키텍처 레벨의 CQRS로 발전한다.

CRUD 단일 모델에서 책임 분리와 이벤트 저장으로 넘어가는 이유

전통적인 CRUD는 하나의 모델이 쓰기 검증, 조인 기반 조회, 권한 필터링, 감사 추적을 모두 맡는다. Microsoft Azure의 CQRS 설명도 단순 CRUD에는 이 방식이 적합하지만, 시스템이 커지면 읽기/쓰기 요구사항의 비대칭 때문에 단일 모델 최적화가 어려워지고 lock contention, 복잡한 쿼리, 보안 경계 혼선이 생긴다고 설명한다. 즉 이 토픽의 출발점은 “새로운 패턴을 쓰자”가 아니라 단일 모델이 서로 다른 책임을 동시에 최적화하지 못한다는 운영상의 압력이다. (출처: Microsoft Azure - CQRS Pattern)

CQRS는 이 한계를 읽기 모델과 쓰기 모델의 분리로 푼다. 쓰기 모델은 Aggregate 불변식, 트랜잭션, Command 검증에 집중하고, 읽기 모델은 화면이 필요한 DTO나 materialized view를 미리 만들어 조인 비용을 줄인다. Event Sourcing은 여기서 한 단계 더 나아가 현재 상태만 저장하는 CRUD의 약점, 즉 “왜 이 상태가 되었는가”를 알기 어렵다는 문제를 이벤트 로그로 해결한다. Martin Fowler가 말하듯 현재 상태를 묻는 것만으로 충분하지 않고 “어떻게 여기까지 왔는가”가 필요한 순간이 있으며, 이벤트 로그는 과거 상태 재구성, temporal query, replay의 기반이 된다. (출처: Martin Fowler - Event Sourcing)

두 패턴을 결합할 때는 이벤트를 DB 저장과 함께 안전하게 남기는 메커니즘이 필요하다. 메시지 브로커와 데이터베이스를 항상 하나의 분산 트랜잭션으로 묶을 수 없기 때문에 Outbox가 등장한다. Transactional Outbox는 2PC를 쓰지 않으면서도 DB 트랜잭션이 커밋된 경우에만 메시지가 발행되도록 보장하지만, relay가 같은 메시지를 두 번 발행할 수 있으므로 소비자는 멱등적이어야 한다. 이 구조가 frontmatter의 “CRUD 단일 모델의 읽기/쓰기 트레이드오프와 감사 추적 필요성 → 책임 분리 + 이벤트 저장”이라는 계보를 실제 메커니즘으로 연결한다. (출처: microservices.io - Transactional outbox)

3. 핵심 개념

3-1. CQRS (Command Query Responsibility Segregation)

비유: 은행 창구

은행에는 두 종류의 창구가 있다.

입출금 창구 (Command): 돈을 넣거나 빼는 작업 — 상태를 변경한다. 빠른 처리보다 정확성이 중요하다.
잔액 조회 창구 (Query): 현재 잔액을 확인하는 작업 — 상태를 읽기만 한다. 빠른 응답이 중요하다.

만약 하나의 창구에서 모든 업무를 처리하면 입출금 작업 때문에 조회도 느려지고, 조회 때문에 입출금도 영향을 받는다. CQRS는 이 두 창구를 완전히 분리하는 설계다.

원리

클라이언트 요청
    │
    ├─ 쓰기 요청 (Command) ──▶ Command 모델 ──▶ 쓰기 DB (정규화, 무결성 중심)
    │                                              │
    │                                              │ 이벤트/동기화
    │                                              ▼
    └─ 읽기 요청 (Query)  ──▶ Query 모델  ──▶ 읽기 DB (역정규화, 조회 최적화)

Command 모델: 데이터 정합성이 우선. 정규화된 테이블 구조. 트랜잭션 처리.
Query 모델: 응답 속도가 우선. 역정규화(Denormalization)된 구조. 복잡한 JOIN 없이 바로 조회.

왜 이렇게 동작하는가 — 읽기와 쓰기의 근본적 차이

전통적인 CRUD 아키텍처에서는 하나의 도메인 모델(Entity)이 읽기와 쓰기를 모두 담당한다. 그런데 실무에서 읽기와 쓰기는 요구사항이 전혀 다르다.

쓰기(Command)의 특성:

비즈니스 규칙 검증이 필수다 (재고가 0 이하로 갈 수 없음, 잔액보다 많이 출금 불가)
정규화된 스키마로 데이터 무결성을 보장해야 한다
동시성 제어(Lock, 낙관적 제어)가 필요하다
트래픽 비중이 상대적으로 적은 경우가 많다. 예를 들어 주문 상세 조회가 주문 생성보다 5배 이상 많고, 조회 p95가 조인 때문에 500ms를 넘는 상황이면 읽기 모델 분리 후보가 된다. 이 5:1·500ms 기준은 공식 표준이 아니라 팀 내부 의사결정용 초기 가드레일(추정)로 둔다.

읽기(Query)의 특성:

여러 테이블을 JOIN해서 한 화면에 필요한 데이터를 조합해야 한다
비즈니스 규칙 검증이 필요 없다 (읽기만 하므로)
빠른 응답 속도가 핵심이다 (UX 직결)
트래픽 비중이 쓰기보다 큰 경우가 많다. 단, 단순 관리자 CRUD처럼 목록·상세 조회가 느리지 않고 권한/필터 조건도 단순하면 CQRS의 분리 비용이 이득보다 커질 수 있다.

하나의 모델로 양쪽을 만족시키려면 정규화 ↔ 역정규화 사이에서 타협해야 한다. 정규화하면 쓰기는 안전하지만 읽기가 느려지고(JOIN 폭발), 역정규화하면 읽기는 빠르지만 쓰기 시 데이터 동기화 부담이 생긴다. CQRS는 이 딜레마를 “그냥 둘 다 만들자” 로 해결한다.

CQRS를 도입하면 안 되는 경우:

단순 CRUD 앱 (읽기/쓰기 복잡도 차이가 거의 없음)
팀 규모가 작고 운영 복잡도를 감당하기 어려운 경우
읽기 모델 동기화 지연(Eventual Consistency)을 비즈니스가 허용하지 않는 경우

도입 판단은 “분리하면 빨라지는가”보다 “분리 후 생기는 지연과 운영 지표를 감당할 수 있는가”를 먼저 본다. 예를 들어 관리자 주문 목록이 6개 테이블 조인 때문에 p95 800ms이고, 주문 생성은 p95 80ms이며, 조회 요청이 쓰기보다 10배 많다면 Query 모델을 Elasticsearch나 별도 PostgreSQL projection으로 분리할 근거가 있다(수치는 예시, 추정). 반대로 “결제 직후 영수증 조회는 1초 안에 반드시 최신이어야 한다”처럼 read-your-writes SLA가 강하면, 비동기 projection만으로는 부족하므로 Command 응답에 필요한 조회 데이터를 함께 반환하거나 해당 화면만 Primary DB를 읽게 해야 한다.

Event Sourcing을 도입하면 안 되는 경우:

GDPR·개인정보 삭제 요건이 있는 도메인: Event Sourcing은 append-only 원칙 때문에 이벤트 삭제가 불가능하다. 개인정보 삭제(Right to Erasure) 요건을 충족하려면 Crypto Shredding(민감 필드를 별도 키로 암호화 후 키 폐기)이나 PII 분리 저장 같은 추가 설계가 필요하다 — 구현 복잡도가 크게 높아진다. (출처: EventStoreDB — Protecting Sensitive Data with Crypto Shredding)
외부 시스템과 결합된 이벤트 재생이 필요한 도메인: 이벤트 재생 결과가 외부 API 호출·결제 처리 등 사이드 이펙트에 의존하면, 과거 이벤트를 재생할 때 동일한 결과를 보장할 수 없다. (출처: Martin Fowler — Event Sourcing)
현재 상태만 필요한 단순 도메인: 변경 이력이 비즈니스 가치를 갖지 않는 경우 (예: 세션 상태, 오늘의 날씨). Kurrent의 4C 프레임워크 기준으로 Collaborative·Compliant·Complex·Changing 중 3개 이상 해당하지 않으면 도입을 재고해야 한다. (출처: Is Event Sourcing Right for Your Project? — Kurrent)

📖 더 보기: Microsoft Azure - CQRS Pattern — CQRS 적용 시나리오/비적용 시나리오를 클라우드 아키텍처 관점에서 상세 설명 (입문)

NestJS `@nestjs/cqrs` 구현 예시

설치

npm install @nestjs/cqrs

모듈 설정

import { Module } from "@nestjs/common";
import { CqrsModule } from "@nestjs/cqrs";
import { TaskController } from "./task.controller";
import { TaskRepository } from "./task.repository";
import { CreateTaskHandler } from "./commands/create-task.handler";
import { GetAllTasksHandler } from "./queries/get-all-tasks.handler";
import { TaskCreatedHandler } from "./events/task-created.handler";

@Module({
  imports: [CqrsModule],
  controllers: [TaskController],
  providers: [
    TaskRepository,
    CreateTaskHandler, // Command Handler 등록
    GetAllTasksHandler, // Query Handler 등록
    TaskCreatedHandler, // Event Handler 등록
  ],
})
export class TaskModule {}

Command 정의 및 Handler

export class CreateTaskCommand {
  constructor(
    public readonly title: string,
    public readonly description: string,
    public readonly userId: string,
  ) {}
}

// commands/create-task.handler.ts
import { CommandHandler, ICommandHandler, EventBus } from "@nestjs/cqrs";
import { CreateTaskCommand } from "./create-task.command";
import { TaskCreatedEvent } from "../events/task-created.event";

@CommandHandler(CreateTaskCommand)
export class CreateTaskHandler implements ICommandHandler<CreateTaskCommand> {
  constructor(
    private readonly taskRepository: TaskRepository,
    private readonly eventBus: EventBus,
  ) {}

  async execute(command: CreateTaskCommand): Promise<void> {
    const { title, description, userId } = command;

    // 1. 쓰기 DB에 저장 (Command 모델)
    const task = await this.taskRepository.create({
      title,
      description,
      userId,
    });

    // 2. 이벤트 발행 → 읽기 DB 동기화 트리거
    this.eventBus.publish(new TaskCreatedEvent(task.id, title, userId));
  }
}

Query 정의 및 Handler

export class GetAllTasksQuery {
  constructor(public readonly userId: string) {}
}

// queries/get-all-tasks.handler.ts
import { QueryHandler, IQueryHandler } from "@nestjs/cqrs";
import { GetAllTasksQuery } from "./get-all-tasks.query";

@QueryHandler(GetAllTasksQuery)
export class GetAllTasksHandler implements IQueryHandler<GetAllTasksQuery> {
  constructor(private readonly taskReadRepository: TaskReadRepository) {}

  async execute(query: GetAllTasksQuery): Promise<TaskDto[]> {
    // 읽기 전용 DB에서 조회 (역정규화된 뷰)
    return this.taskReadRepository.findAllByUser(query.userId);
  }
}

Controller에서 CommandBus / QueryBus 사용

import { Controller, Get, Post, Body } from "@nestjs/common";
import { CommandBus, QueryBus } from "@nestjs/cqrs";
import { CreateTaskCommand } from "./commands/create-task.command";
import { GetAllTasksQuery } from "./queries/get-all-tasks.query";

@Controller("tasks")
export class TaskController {
  constructor(
    private readonly commandBus: CommandBus,
    private readonly queryBus: QueryBus,
  ) {}

  @Post()
  async createTask(@Body() dto: CreateTaskDto) {
    // Command 실행 → 쓰기 모델
    await this.commandBus.execute(
      new CreateTaskCommand(dto.title, dto.description, dto.userId),
    );
    return { message: "태스크가 생성되었습니다." };
  }

  @Get()
  async getAllTasks(@Query("userId") userId: string) {
    // Query 실행 → 읽기 모델
    return this.queryBus.execute(new GetAllTasksQuery(userId));
  }
}

예상 출력

POST /tasks
Body: { "title": "CQRS 학습", "description": "패턴 이해하기", "userId": "user-1" }

→ Response: { "message": "태스크가 생성되었습니다." }
  (내부적으로 TaskCreatedEvent 발행, 읽기 DB 비동기 업데이트)

GET /tasks?userId=user-1
→ Response: [
    { "id": "uuid-1", "title": "CQRS 학습", "status": "open", "createdAt": "..." }
  ]

CQRS 고급: Outbox 패턴 — 이벤트 유실 방지

Command Handler에서 DB에 저장하고 이벤트를 발행(EventBus.publish)하는 사이에 서버가 다운되면 이벤트가 유실될 수 있다. Outbox 패턴은 이 문제를 해결한다.

비유: 편지를 보내려면 먼저 발신함(Outbox)에 넣어두고, 우체부(별도 프로세스)가 주기적으로 발신함을 확인해서 배달한다. 편지를 직접 던지는 것보다 훨씬 안전하다.

왜 이렇게 동작하는가:

DB 저장과 이벤트를 같은 트랜잭션에 묶는다 (atomic 보장)
이벤트 발행은 DB 커밋이 완료된 후 별도 프로세스가 처리한다
서버 다운 후 재시작 시 미발행 이벤트를 다시 처리할 수 있다

// Outbox 테이블 스키마 (PostgreSQL)
// CREATE TABLE outbox (
//   id UUID PRIMARY KEY,
//   aggregate_type VARCHAR(50),
//   aggregate_id VARCHAR(100),
//   event_type VARCHAR(100),
//   payload JSONB,
//   created_at TIMESTAMP DEFAULT NOW(),
//   published_at TIMESTAMP NULL  -- NULL이면 미발행
// );

// Command Handler with Outbox Pattern
@CommandHandler(CreateTaskCommand)
export class CreateTaskHandler implements ICommandHandler<CreateTaskCommand> {
  constructor(private readonly dataSource: DataSource) {}

  async execute(command: CreateTaskCommand): Promise<void> {
    // 단일 트랜잭션 내에서 도메인 저장 + Outbox 저장
    await this.dataSource.transaction(async (manager) => {
      // 1. 도메인 데이터 저장
      const task = manager.create(Task, {
        title: command.title,
        userId: command.userId,
      });
      await manager.save(task);

      // 2. Outbox에 이벤트 저장 (같은 트랜잭션)
      await manager.insert(Outbox, {
        id: randomUUID(),
        aggregateType: "Task",
        aggregateId: task.id,
        eventType: "TaskCreated",
        payload: { taskId: task.id, title: task.title },
      });
      // 트랜잭션 커밋 → 둘 다 저장 또는 둘 다 롤백
    });
    // 실제 이벤트 발행은 별도 OutboxProcessor가 담당
  }
}

📖 더 보기: Designing Scalable Systems with CQRS and Event Sourcing — 프로덕션 수준의 CQRS/Event Sourcing 설계 패턴 (Outbox, Snapshot 포함, 중급)

CQRS 고급: 낙관적 동시성 제어 (Optimistic Concurrency)

두 사용자가 동시에 같은 Command를 보내면 경쟁 조건(Race Condition)이 발생할 수 있다. 비관적 락(SELECT FOR UPDATE)은 성능을 희생하는 반면, 낙관적 동시성 제어는 충돌이 드문 경우에 성능을 유지하면서 정합성을 보장한다.

// 낙관적 동시성 제어를 위한 버전 필드
@Entity()
export class Task {
  @PrimaryGeneratedColumn("uuid")
  id: string;

  @Column()
  title: string;

  @VersionColumn() // TypeORM이 자동으로 버전 관리
  version: number; // UPDATE 시 version을 WHERE 조건에 추가
}

// 동시 수정 시 동작:
// 트랜잭션 A: task.version=1 로 읽고 → version=2 로 UPDATE 시도 (성공)
// 트랜잭션 B: task.version=1 로 읽고 → version=2 로 UPDATE 시도 (실패! version이 이미 2)
// → TypeORM이 OptimisticLockVersionMismatchError 발생
// → 애플리케이션이 재시도 로직 실행

흐름 정리: CQRS는 읽기/쓰기 모델을 분리하는 구조를 만든다. 그런데 쓰기 모델 변경 사항을 읽기 모델에 어떻게 전달할까? 이 질문의 답이 Event Sourcing이다. 아래에서 Event Sourcing의 원리를 이해하고, 이어지는 3-3절에서 두 패턴이 결합되는 방식을 살펴본다.

3-2. Event Sourcing (이벤트 소싱)

비유: 은행 통장 거래 내역

전통적인 CRUD는 통장 잔액(현재 상태)만 저장한다. 잔액이 100만원이라는 사실만 알 뿐, 어떻게 100만원이 됐는지는 모른다.

Event Sourcing은 거래 내역(이벤트)을 모두 저장한다.

입금 +50만원 (2024-01-01)
출금 -10만원 (2024-01-05)
입금 +30만원 (2024-01-10)
출금 -20만원 (2024-01-15)
입금 +50만원 (2024-01-20)
────────────────────────────
현재 잔액: 100만원 (이벤트 재생으로 계산)

거래 내역을 순서대로 재생(replay)하면 어느 시점의 잔액이든 계산할 수 있다.

원리

Command (입금 요청)
    │
    ▼
이벤트 생성 (MoneyDepositedEvent { amount: 500000, at: ... })
    │
    ▼
이벤트 스토어 (Event Store) — append-only, 절대 수정 불가
    │
    ├─ 이벤트 재생(Replay) ──▶ 현재 상태(Aggregate) 복원
    │
    └─ 이벤트 발행(Publish) ──▶ Read Model 업데이트 (CQRS와 결합)

왜 이렇게 동작하는가 — Append-Only의 근본 이유

왜 이벤트를 수정하거나 삭제하면 안 되는가? 이벤트 소싱에서 이벤트는 “이미 발생한 사실” 이다. “1월 5일에 10만원을 출금했다”는 사실은 변경할 수 없다. 잘못된 이벤트가 있다면 삭제하는 것이 아니라 보상 이벤트(Compensating Event) 를 새로 추가한다. 예를 들어 잘못된 출금이 있었다면 “1월 6일에 10만원 출금 취소” 이벤트를 append한다.

이 원칙 덕분에 다음이 보장된다:

완벽한 감사 추적: 어떤 이벤트도 사라지지 않으므로 모든 변경의 전체 이력이 남는다
동시성 안전: 이벤트는 추가만 하므로 UPDATE 충돌이 발생하지 않는다
디버깅 용이: 버그 발생 시점의 이벤트 시퀀스를 그대로 재생하면 정확한 버그 원인을 파악할 수 있다

전통 CRUD vs Event Sourcing의 근본적 차이:

전통 CRUD (현재 상태 저장):
  accounts 테이블: { id: 1, balance: 100만원, updated_at: 1/20 }
  → "왜 100만원인가?" 알 수 없음
  → UPDATE가 이전 값을 덮어씀

Event Sourcing (이벤트 이력 저장):
  events 테이블:
    { type: DEPOSIT,  amount: 50만, at: 1/01 }
    { type: WITHDRAW, amount: 10만, at: 1/05 }
    { type: DEPOSIT,  amount: 30만, at: 1/10 }
    { type: WITHDRAW, amount: 20만, at: 1/15 }
    { type: DEPOSIT,  amount: 50만, at: 1/20 }
  → 100만원이 된 경위를 완벽하게 알 수 있음
  → INSERT만 발생, UPDATE/DELETE 없음

핵심 특성

Append-Only: 이벤트는 추가만 가능하다. 수정이나 삭제는 없다. 과거는 바꿀 수 없다.
이벤트 재생(Replay): 처음부터 모든 이벤트를 순서대로 적용하면 어느 시점이든 상태를 복원할 수 있다.
시간 여행(Time Travel): 특정 시점의 상태를 재현할 수 있다. “3일 전 오후 2시의 주문 상태는?”에 답할 수 있다.
스냅샷(Snapshot): 이벤트가 너무 많아지면 재생 시간이 길어진다. 특정 시점의 상태를 스냅샷으로 저장해 두고, 그 이후 이벤트만 재생하면 성능을 개선할 수 있다.

📖 더 보기: Martin Fowler - Event Sourcing — Event Sourcing 패턴의 원형 정의. “왜 상태가 아니라 이벤트를 저장하는가”에 대한 근본 철학 (중급)

TypeScript 구현 예시

// 이벤트 타입 정의
interface DomainEvent {
  eventId: string;
  occurredAt: Date;
  aggregateId: string;
}

interface MoneyDepositedEvent extends DomainEvent {
  type: "MONEY_DEPOSITED";
  amount: number;
}

interface MoneyWithdrawnEvent extends DomainEvent {
  type: "MONEY_WITHDRAWN";
  amount: number;
}

type BankEvent = MoneyDepositedEvent | MoneyWithdrawnEvent;

// 이벤트 스토어 (간단한 메모리 구현)
class EventStore {
  private events: Map<string, BankEvent[]> = new Map();

  append(accountId: string, event: BankEvent): void {
    const existing = this.events.get(accountId) ?? [];
    this.events.set(accountId, [...existing, event]);
    console.log(
      `[EventStore] 이벤트 저장: ${event.type} (accountId: ${accountId})`,
    );
  }

  getEvents(accountId: string): BankEvent[] {
    return this.events.get(accountId) ?? [];
  }
}

// Aggregate: 이벤트 재생으로 상태 복원
class BankAccount {
  private balance: number = 0;
  private accountId: string;

  constructor(accountId: string) {
    this.accountId = accountId;
  }

  // 이벤트를 순서대로 적용해서 상태 복원
  static rehydrate(accountId: string, events: BankEvent[]): BankAccount {
    const account = new BankAccount(accountId);
    for (const event of events) {
      account.apply(event);
    }
    console.log(
      `[Rehydrate] ${events.length}개 이벤트 재생 완료. 잔액: ${account.balance}원`,
    );
    return account;
  }

  private apply(event: BankEvent): void {
    switch (event.type) {
      case "MONEY_DEPOSITED":
        this.balance += event.amount;
        break;
      case "MONEY_WITHDRAWN":
        this.balance -= event.amount;
        break;
    }
  }

  getBalance(): number {
    return this.balance;
  }
}

// 사용 예시
const store = new EventStore();
const accountId = "account-1";

// 이벤트 저장 (append-only)
store.append(accountId, {
  eventId: "e1",
  type: "MONEY_DEPOSITED",
  amount: 500000,
  aggregateId: accountId,
  occurredAt: new Date("2024-01-01"),
});
store.append(accountId, {
  eventId: "e2",
  type: "MONEY_WITHDRAWN",
  amount: 100000,
  aggregateId: accountId,
  occurredAt: new Date("2024-01-05"),
});
store.append(accountId, {
  eventId: "e3",
  type: "MONEY_DEPOSITED",
  amount: 300000,
  aggregateId: accountId,
  occurredAt: new Date("2024-01-10"),
});

// 이벤트 재생으로 현재 상태 복원
const account = BankAccount.rehydrate(accountId, store.getEvents(accountId));
console.log(`현재 잔액: ${account.getBalance()}원`);

예상 출력

[EventStore] 이벤트 저장: MONEY_DEPOSITED (accountId: account-1)
[EventStore] 이벤트 저장: MONEY_WITHDRAWN (accountId: account-1)
[EventStore] 이벤트 저장: MONEY_DEPOSITED (accountId: account-1)
[Rehydrate] 3개 이벤트 재생 완료. 잔액: 700000원
현재 잔액: 700000원

NestJS에서 Event Sourcing 라이브러리 활용

직접 구현하는 대신 커뮤니티 라이브러리를 활용하면 보일러플레이트를 크게 줄일 수 있다.

# @ocoda/event-sourcing — NestJS 전용 Event Sourcing 라이브러리
npm install @ocoda/event-sourcing

# 지원하는 이벤트 스토어 백엔드
# - In-Memory (개발/테스트용)
# - PostgreSQL
# - MongoDB
# - DynamoDB (AWS 환경에 적합)

// @ocoda/event-sourcing 기본 사용 예시
import { Aggregate, AggregateRoot, EventHandler } from "@ocoda/event-sourcing";

@Aggregate({ streamName: "bank-account" })
export class BankAccount extends AggregateRoot {
  private balance: number = 0;

  deposit(amount: number): void {
    this.apply(new MoneyDepositedEvent(amount));
  }

  @EventHandler(MoneyDepositedEvent)
  onMoneyDeposited(event: MoneyDepositedEvent): void {
    this.balance += event.amount;
  }

  getBalance(): number {
    return this.balance;
  }
}

// 예상 출력: 이벤트 스토어에 append-only 저장 + 자동 rehydrate

📖 더 보기: GitHub - ocoda/event-sourcing — NestJS 전용 Event Sourcing 라이브러리. PostgreSQL/DynamoDB 백엔드 지원. AWS 환경에 바로 적용 가능 (입문)

Event Sourcing 장단점

장점	단점
완벽한 감사 로그(Audit Log)	구현 복잡도가 높음
시간 여행(Time Travel) 가능	이벤트 스키마 변경이 어려움
버그 재현 및 디버깅 용이	이벤트 스토어 크기가 계속 증가
CQRS와 자연스럽게 결합	쿼리가 불편함 (재생 필요)
이벤트 기반 마이크로서비스와 궁합 좋음	러닝 커브가 가파름

CQRS + Event Sourcing + Kafka 결합 심화: Projection 패턴

CQRS에서 이벤트를 Kafka로 발행하면 여러 읽기 모델(Projection)을 독립적으로 구성할 수 있다. 이것이 “한 번 쓰고, 여러 곳에서 읽는(Write Once, Read Many Views)” 아키텍처다.

비유: 회사 공지사항 게시판(Event Store/Kafka)에 공지가 올라가면, 팀별 전담 담당자(각 Projection Consumer)가 그 공지를 자기 팀 형식에 맞게 해석해 별도 메모(각 읽기 모델)를 만든다.

왜 이렇게 동작하는가:

Kafka는 이벤트를 append-only 로그(토픽)로 저장 — Event Store와 동일한 철학
각 Consumer Group은 독립적으로 오프셋을 관리 → 읽기 모델 재구성 시 오프셋을 0으로 초기화해 전체 재생 가능
새 읽기 모델이 필요하면 코드 추가 후 처음부터 이벤트를 재생 → 기존 시스템 무중단

이벤트 (OrderCreated, OrderShipped...)
    │
    ▼
Kafka 토픽 (영구 보존, append-only)
    │
    ├─ Consumer 1: Redis Projection (실시간 주문 현황 캐시)
    ├─ Consumer 2: PostgreSQL Projection (주문 이력 검색 테이블)
    ├─ Consumer 3: Elasticsearch Projection (전문 검색 인덱스)
    └─ Consumer 4: 통계 집계 서비스 (배치 집계 테이블)

// NestJS에서 Kafka Consumer를 Projection으로 구성
import { Controller } from "@nestjs/common";
import { EventPattern, Payload } from "@nestjs/microservices";

@Controller()
export class OrderProjectionHandler {
  constructor(
    private readonly redisClient: Redis,
    private readonly orderReadRepository: OrderReadRepository,
  ) {}

  // Kafka 'order.created' 토픽 구독 → 읽기 모델 업데이트
  @EventPattern("order.created")
  async handleOrderCreated(@Payload() event: OrderCreatedEvent): Promise<void> {
    // 1. Redis에 실시간 현황 캐싱
    await this.redisClient.setex(
      `order:${event.orderId}`,
      3600, // 1시간 TTL
      JSON.stringify({ status: "CREATED", ...event }),
    );

    // 2. PostgreSQL 읽기 전용 뷰 테이블에 역정규화 저장
    await this.orderReadRepository.upsert({
      orderId: event.orderId,
      customerId: event.customerId,
      customerName: event.customerName, // 역정규화: JOIN 불필요
      totalAmount: event.totalAmount,
      status: "CREATED",
      createdAt: event.occurredAt,
    });
  }
}

// 예상 동작:
// order.created 이벤트 수신 → Redis 캐시 업데이트 (수ms)
// → PostgreSQL 읽기 테이블 업데이트 (수십ms)
// → GET /orders?customerId=xxx 조회 시 JOIN 없이 즉시 반환

Kafka Offset 재설정으로 읽기 모델 재구성하기:

# 읽기 모델을 처음부터 다시 만들어야 할 때 (예: 새 컬럼 추가)
# Consumer Group의 오프셋을 처음으로 리셋
kafka-consumer-groups.sh \
  --bootstrap-server localhost:9092 \
  --group order-projection-consumer \
  --topic order.created \
  --reset-offsets --to-earliest --execute

# 예상 출력:
# GROUP                    TOPIC           PARTITION  NEW-OFFSET
# order-projection-consumer order.created  0          0    ← 처음부터 재생
# order-projection-consumer order.created  1          0
# order-projection-consumer order.created  2          0
# → Consumer 재시작하면 전체 이벤트를 처음부터 재처리해 읽기 모델 재구성

📖 더 보기: Event Sourcing and CQRS with Apache Kafka — Confluent — Kafka를 Event Store로 활용하는 CQRS 구현 패턴. Projection Consumer 설계 방법 포함 (입문~중급)

3-3. CQRS + Event Sourcing 결합: WAL 기반 CDC 동기화

CQRS와 Event Sourcing이 결합할 때 핵심 과제는 “쓰기 모델의 변경을 어떻게 읽기 모델에 전파하는가” 다. 애플리케이션 코드에서 직접 이벤트를 발행하는 방식(Outbox 패턴) 외에, DB 레이어에서 CDC(Change Data Capture)로 자동 동기화하는 방식도 널리 쓰인다.

WAL(Write-Ahead Log)의 상세 원리(Redo Log, Undo Log, MVCC)는 transaction-basics.md 를 참고하세요.

여기서는 WAL이 CQRS 아키텍처에서 어떻게 활용되는지, 즉 CDC 기반 읽기 모델 동기화 관점에 집중한다.

WAL 기반 CDC가 CQRS 읽기 모델 동기화에 쓰이는 방식:

PostgreSQL의 WAL은 모든 DB 변경을 순서대로 기록한 append-only 로그다. Debezium 같은 CDC 커넥터가 이 WAL을 읽어 변경 이벤트를 Kafka로 스트리밍하면, CQRS의 읽기 모델(Elasticsearch, Redis 등)이 이 이벤트를 구독해 자신의 뷰를 업데이트한다. 애플리케이션 코드에 동기화 로직을 넣지 않아도 DB 변경이 자동으로 읽기 모델에 전파된다.

PostgreSQL (쓰기 DB, WAL 생성)
    │
    │  WAL logical decoding
    ▼
Debezium (CDC 커넥터) — DB 변경을 이벤트 스트림으로 변환
    │
    ▼
Kafka 토픽 (변경 이벤트)
    │
    ├─ CQRS 읽기 모델 1: Elasticsearch (전문 검색)
    ├─ CQRS 읽기 모델 2: Redis (실시간 캐시)
    └─ CQRS 읽기 모델 3: 집계 테이블 (대시보드)

이 구조는 쓰기 모델(정규화된 PostgreSQL)과 읽기 모델(역정규화된 각 뷰)의 동기화를 DB 레이어에서 자동으로 처리한다. 애플리케이션 Command Handler는 쓰기 DB에만 집중하면 되고, 읽기 모델 동기화는 CDC 파이프라인이 담당한다.

3-4. AWS + TypeORM 관점에서의 CQRS / Event Sourcing

AWS RDS + TypeORM으로 CQRS 읽기/쓰기 DB 분리

실제 프로덕션 환경에서 CQRS의 물리적 구현은 RDS Primary(쓰기) + Aurora Read Replica 또는 ElastiCache(읽기) 조합이 가장 일반적이다. CQRS 맥락에서 중요한 점은 Command Handler는 항상 Master(Primary)에, Query Handler는 항상 Replica(Reader)에 연결되도록 라우팅을 고정하는 것이다.

Command 경로(쓰기): RDS Primary → 데이터 정합성 및 트랜잭션 보장
Query 경로(읽기): Aurora Reader Endpoint → Auto Scaling으로 읽기 부하를 수평 분산

TypeORM의 replication 설정 코드 및 Aurora Reader Endpoint 활용법은 db-replication-sharding.md 를 참조하세요. CQRS 관점에서 이 설정의 핵심은 애플리케이션 코드 변경 없이 읽기 트래픽을 Replica로 자동 분산한다는 점이다.

TypeORM DataSource로 Command/Query Handler에서 트랜잭션 제어

CQRS Command Handler에서 Outbox 패턴을 TypeORM DataSource.transaction()과 결합하면 데이터 저장과 이벤트 발행을 원자적으로 보장할 수 있다.

import { CommandHandler, ICommandHandler } from "@nestjs/cqrs";
import { InjectDataSource } from "@nestjs/typeorm";
import { DataSource } from "typeorm";

@CommandHandler(CreateOrderCommand)
export class CreateOrderHandler implements ICommandHandler<CreateOrderCommand> {
  constructor(@InjectDataSource() private readonly dataSource: DataSource) {}

  async execute(command: CreateOrderCommand): Promise<string> {
    return this.dataSource.transaction(async (manager) => {
      // 1. 주문 저장 (Master DB)
      const order = manager.create(Order, {
        customerId: command.customerId,
        items: command.items,
        status: "PENDING",
      });
      await manager.save(order);

      // 2. Outbox 이벤트 저장 (같은 트랜잭션 — Master DB)
      await manager.insert(OutboxEvent, {
        aggregateId: order.id,
        aggregateType: "Order",
        eventType: "OrderCreated",
        payload: JSON.stringify({
          orderId: order.id,
          customerId: command.customerId,
        }),
        occurredAt: new Date(),
      });

      // 3. 트랜잭션 성공 → 둘 다 커밋
      // 트랜잭션 실패 → 둘 다 롤백 (이벤트 유실 없음)
      return order.id;
    });
    // OutboxProcessor가 별도로 이벤트를 폴링하여 읽기 모델(ElastiCache/Elasticsearch) 업데이트
  }
}

예상 출력

POST /orders
Body: { "customerId": "user-42", "items": [{"productId": "p-1", "qty": 2}] }

→ Response: { "orderId": "uuid-xxx" }

[DB 내부]
  orders 테이블: id=uuid-xxx, status=PENDING ← Master DB에 저장
  outbox_events: aggregateId=uuid-xxx, eventType=OrderCreated ← 같은 트랜잭션

[OutboxProcessor 실행 후]
  ElastiCache(Redis): order:uuid-xxx 캐시 갱신 ← 읽기 모델 업데이트

AWS DynamoDB를 Event Store로 사용 (대규모 이벤트 처리)

대용량 이벤트(수억 건)를 처리해야 하거나 운영 부담을 최소화하고 싶을 때 DynamoDB를 Event Store로 사용할 수 있다. AWS SDK for JavaScript v3 기반 예시:

import {
  DynamoDBClient,
  PutItemCommand,
  QueryCommand,
} from "@aws-sdk/client-dynamodb";
import { marshall, unmarshall } from "@aws-sdk/util-dynamodb";

// DynamoDB Event Store 테이블 설계
// PK: aggregateId (파티션 키)
// SK: version (정렬 키) — 이벤트 순서 보장
// GSI: eventType-occurredAt-index — 이벤트 타입별 조회

const client = new DynamoDBClient({ region: "ap-northeast-2" });

export class DynamoDBEventStore {
  private readonly tableName = process.env.EVENT_STORE_TABLE!; // "order-events"

  async appendEvent(event: DomainEvent): Promise<void> {
    await client.send(
      new PutItemCommand({
        TableName: this.tableName,
        Item: marshall({
          aggregateId: event.aggregateId,
          version: event.version,
          eventType: event.type,
          payload: JSON.stringify(event),
          occurredAt: event.occurredAt.toISOString(),
        }),
        // 낙관적 동시성: version이 이미 존재하면 에러 (중복 이벤트 방지)
        ConditionExpression: "attribute_not_exists(version)",
      }),
    );
  }

  async getEvents(
    aggregateId: string,
    fromVersion = 0,
  ): Promise<DomainEvent[]> {
    const result = await client.send(
      new QueryCommand({
        TableName: this.tableName,
        KeyConditionExpression: "aggregateId = :id AND version >= :v",
        ExpressionAttributeValues: marshall({
          ":id": aggregateId,
          ":v": fromVersion,
        }),
      }),
    );
    return (result.Items ?? []).map((item) =>
      JSON.parse(unmarshall(item).payload),
    );
  }
}

// 예상 동작:
// appendEvent({ aggregateId: "order-1", version: 1, type: "OrderCreated", ... })
//   → DynamoDB에 저장 성공
// appendEvent({ aggregateId: "order-1", version: 1, type: "OrderCreated", ... })
//   → ConditionalCheckFailedException (중복 버전 방지)
// getEvents("order-1") → [OrderCreated, OrderPaid, OrderShipped]

DynamoDB Event Store의 장단점:

항목	DynamoDB	PostgreSQL
운영 부담	없음 (서버리스)	높음 (백업, 튜닝)
확장성	무제한	제한적 (수직 확장)
이벤트 수	수십억 건 부담 없음	수억 건부터 관리 필요
트랜잭션	DynamoDB Transactions 지원 (최대 100개 아이템)	완전한 ACID
비용	요청 수 기반 (대용량에서 비쌀 수 있음)	인스턴스 크기 기반

다만 DynamoDB를 Event Store로 쓸 때의 “무제한 확장”은 설계 제약이 없다는 뜻이 아니다. AWS 공식 제한 기준으로 한 아이템은 최대 400KB이고, 트랜잭션은 최대 100개 unique item 또는 4MB까지이며, 글로벌 테이블의 트랜잭션 ACID 보장은 원본 쓰기 리전 안에서만 성립한다. 따라서 이벤트 payload가 커지는 결제/배송 도메인은 원문 payload를 S3에 두고 이벤트에는 포인터만 저장하거나, Aggregate당 이벤트를 너무 큰 단일 아이템에 누적하지 않고 PK=aggregateId, SK=version 형태로 쪼개야 한다. (출처: AWS DynamoDB Constraints)

📖 더 보기: GitHub - ocoda/event-sourcing — NestJS 전용 Event Sourcing 라이브러리. PostgreSQL/MongoDB/DynamoDB 백엔드를 선택적으로 사용 가능 (입문~중급)

4. 실무에서 어떻게 쓰이나

CQRS 실무 적용

대형 이커머스: 주문 생성(Command)과 주문 목록 조회(Query)를 분리. 조회 DB는 Elasticsearch로 전문 검색 최적화.
실시간 대시보드: Command로 데이터를 쓰고, 읽기 모델은 Redis에 캐싱해서 빠른 응답.
결제 시스템: Command로 결제를 처리하고, Query 모델에 영수증 조회용 역정규화 뷰를 별도 관리.

Event Sourcing 실무 적용

금융 거래: 모든 입출금을 이벤트로 저장. 어느 시점의 잔액도 재현 가능.
주문 상태 추적: OrderPlaced → OrderPaid → OrderShipped → OrderDelivered 이벤트 스트림.
감사 로그(Audit Log): 누가 언제 무엇을 변경했는지 완벽하게 추적.

Event Store 선택 가이드

Event Sourcing을 도입할 때 이벤트를 어디에 저장할지 결정해야 한다. 주요 선택지와 비교:

저장소	특성	적합한 상황
PostgreSQL	ACID, 친숙한 스키마, 트랜잭션 내 Outbox 패턴 가능	팀이 PostgreSQL에 익숙, 기존 인프라 활용
EventStoreDB / KurrentDB	이벤트 소싱에 최적화, Persistent Subscriptions, 스트림 단위 관리	이벤트 소싱이 핵심 아키텍처, 높은 이벤트 처리량
DynamoDB	서버리스, 무한 확장, AWS 네이티브	AWS 환경, 운영 부담 최소화
Kafka	높은 처리량, 스트리밍, Consumer Group 기반 여러 Projection	대용량 이벤트 스트리밍이 동시에 필요

EventStoreDB(KurrentDB) vs Kafka의 근본 차이:

두 기술은 경쟁 관계가 아니라 역할이 다르다.

EventStoreDB(KurrentDB): "데이터베이스" — 이벤트를 저장하고 조회하는 것이 핵심
Kafka: "메시지 브로커" — 이벤트를 전달하고 스트리밍하는 것이 핵심

→ EventStoreDB에서 이벤트를 저장하고, Kafka로 외부 서비스에 전파하는
  "함께 쓰는(hybrid)" 패턴이 실무에서 자주 사용된다.

NestJS에서 EventStoreDB 연동:

// @nestjs/cqrs의 IEventPublisher를 커스터마이징해서
// EventStoreDB에 이벤트를 영속화
import { IEventPublisher, IMessageSource } from "@nestjs/cqrs";
import { EventStoreDBClient, jsonEvent } from "@eventstore/db-client";

export class EventStorePublisher implements IEventPublisher {
  constructor(private readonly client: EventStoreDBClient) {}

  async publish<T extends IEvent>(event: T): Promise<void> {
    const streamName = `${event.constructor.name}-${event["aggregateId"]}`;
    await this.client.appendToStream(streamName, [
      jsonEvent({ type: event.constructor.name, data: event }),
    ]);
  }
}
// → EventBus.publisher를 위 클래스로 교체하면
//   모든 이벤트가 자동으로 EventStoreDB에 영속화됨

WAL 기반 CDC 실무 활용 (CQRS 맥락)

WAL의 내부 동작 원리(Redo/Undo, MVCC)는 transaction-basics.md 참조.

CQRS 맥락에서 WAL의 핵심 역할은 CDC(Change Data Capture) 파이프라인의 데이터 원천이다.

Debezium + Kafka: Debezium이 PostgreSQL WAL을 읽어 변경 이벤트를 Kafka로 스트리밍 → CQRS 읽기 모델(Elasticsearch, Redis 등)이 자동 동기화
애플리케이션 코드 분리: Command Handler는 쓰기 DB에만 집중하고, 읽기 모델 동기화는 CDC 파이프라인이 담당 → 관심사 분리 극대화

CDC의 가장 위험한 실패는 에러가 크게 터지는 장애보다 읽기 모델이 조용히 뒤처지거나 일부 변경을 건너뛰는 silent failure다. Debezium 공식 문서는 PostgreSQL 업그레이드 과정에서 replication slot이 제거되면 커넥터가 기대한 위치에서 재개하지 못하고 기존 로그의 변경 이벤트를 건너뛸 수 있으며, 이 경우 skipped event를 알리는 레코드가 나오지 않는 silent data loss가 발생할 수 있다고 경고한다. 운영에서는 배포·DB 승격·업그레이드 전후에 slot과 LSN을 확인한다. (출처: Debezium PostgreSQL Connector)

-- PostgreSQL에서 Debezium replication slot 상태 확인
SELECT
  slot_name,
  active,
  restart_lsn,
  confirmed_flush_lsn
FROM pg_replication_slots
WHERE slot_name = 'debezium_orders';

예상 출력

slot_name       | active | restart_lsn | confirmed_flush_lsn
debezium_orders | t      | 0/16B6C50   | 0/16B6D10

slot이 없거나 active=false가 장시간 지속되면 쓰기 중단 없이 커넥터만 재시작하지 말고, 먼저 Kafka의 마지막 처리 offset과 PostgreSQL WAL 보존 범위를 대조한다. 이미 slot이 사라진 뒤 쓰기가 진행되었다면 projection을 부분 복구하려 하지 말고 Event Store 또는 원본 DB 스냅샷에서 읽기 모델을 전체 재생성하는 쪽이 안전하다.

5. 내 업무와의 연결고리

BackOps 엔지니어 관점 (Nest.js / AWS)

@nestjs/cqrs 모듈을 도입하면 기존 Service 중심 코드보다 Command/Query 경계가 명확해진다. 새 기능 추가 시 어디에 코드를 넣을지 고민이 줄어든다.
운영 중인 서비스에서 특정 API가 DB에 너무 큰 부담을 준다면, CQRS로 읽기 모델을 분리하고 Redis 캐시를 Query 모델로 활용할 수 있다.
AWS 환경에서는 RDS(쓰기) + ElastiCache or Aurora Read Replica(읽기) 조합이 CQRS의 물리적 구현이다.
PostgreSQL WAL 기반의 CDC를 Debezium + MSK(Kafka)로 구성하면 마이크로서비스 간 데이터 동기화를 이벤트 기반으로 처리할 수 있다.
관리자 시스템에서 “누가 이 데이터를 언제 바꿨는가”를 추적해야 할 때 Event Sourcing의 감사 로그 기능이 직접적으로 필요하다.

6. 비슷한 개념과 비교

CQRS vs 전통적인 CRUD

항목	전통 CRUD	CQRS
모델	읽기/쓰기 하나의 모델	읽기/쓰기 분리
복잡도	낮음	높음
최적화	어느 쪽도 최적화 어려움	각자 독립 최적화
적합한 규모	소규모 CRUD 앱	읽기/쓰기 불균형이 큰 서비스

Event Sourcing vs 전통적인 CRUD

항목	전통 CRUD	Event Sourcing
저장 대상	현재 상태	이벤트 이력 전체
과거 상태 조회	불가능	가능 (이벤트 재생)
저장 공간	적음	계속 증가
쿼리	간단	복잡 (재생 필요)
적합한 도메인	일반적인 서비스	금융, 주문, 감사 로그

CQRS와 Event Sourcing의 관계

CQRS와 Event Sourcing은 별개의 패턴이지만 자주 함께 쓰인다.

CQRS 없이 Event Sourcing만 쓸 수 있다. (이벤트로 상태 저장, 단일 모델로 조회)
Event Sourcing 없이 CQRS만 쓸 수 있다. (쓰기 DB와 읽기 DB만 분리)
둘을 함께 쓰면 Event Sourcing의 이벤트가 자연스럽게 CQRS의 읽기 모델 동기화 트리거가 된다.

Saga 패턴과 CQRS — 분산 트랜잭션 처리

CQRS와 마이크로서비스를 함께 사용하면 분산 트랜잭션 문제가 발생한다. “주문 생성 → 결제 처리 → 재고 차감” 처럼 여러 서비스에 걸친 작업이 원자적으로 성공하거나 실패해야 할 때, 단일 DB 트랜잭션(2PC)은 마이크로서비스 환경에서 사용하기 어렵다.

Saga 패턴은 이 문제를 해결하는 대표적인 방법이다. 분산 트랜잭션을 일련의 로컬 트랜잭션과 보상 트랜잭션(Compensating Transaction) 으로 분해한다.

주문 Saga 예시:

성공 경로:
  1. OrderService: 주문 생성 (OrderCreated 이벤트 발행)
  2. PaymentService: 결제 처리 (PaymentCompleted 이벤트 발행)
  3. InventoryService: 재고 차감 (InventoryReserved 이벤트 발행)
  4. ShippingService: 배송 준비 (ShippingScheduled 이벤트 발행)

실패 경로 (재고 부족으로 3단계 실패):
  1. OrderService: 주문 생성 ✅
  2. PaymentService: 결제 완료 ✅
  3. InventoryService: 재고 부족 ❌ → InventoryReservationFailed 이벤트
  4. PaymentService: 결제 취소 (보상 트랜잭션) ← 보상 이벤트로 롤백
  5. OrderService: 주문 취소 (보상 트랜잭션) ← 보상 이벤트로 롤백

Saga에는 두 가지 구현 방식이 있다.

Choreography Saga (이벤트 기반): 각 서비스가 이벤트를 구독하고 직접 다음 단계를 처리한다. 중앙 조율자 없이 분산되어 결합도가 낮지만, 전체 흐름을 파악하기 어렵다.

Orchestration Saga (중앙 오케스트레이터): 하나의 Saga Orchestrator가 각 서비스에 Command를 보내고 결과를 기다린다. NestJS에서는 @nestjs/cqrs의 Saga 기능이 이 방식을 지원한다.

// NestJS Saga (Orchestration 방식)
// Saga는 Observable 스트림으로 이벤트를 받아 다음 Command를 발행
import { Saga, ICommand, ofType } from "@nestjs/cqrs";
import { Observable } from "rxjs";
import { map, filter } from "rxjs/operators";

@Injectable()
export class OrderSaga {
  @Saga()
  orderCreated = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(OrderCreatedEvent), // OrderCreated 이벤트 감지
      map((event) => new ProcessPaymentCommand(event.orderId, event.amount)),
      // → PaymentService에 결제 처리 Command 발행
    );
  };

  @Saga()
  paymentCompleted = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(PaymentCompletedEvent),
      map((event) => new ReserveInventoryCommand(event.orderId, event.items)),
    );
  };

  @Saga()
  paymentFailed = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(PaymentFailedEvent),
      map((event) => new CancelOrderCommand(event.orderId)), // 보상 트랜잭션
    );
  };
}

CQRS와 Saga의 관계:

Command가 로컬 트랜잭션을 수행하고 Domain Event를 발행한다
Saga가 이벤트를 구독하고 다음 서비스에 Command를 발행한다
실패 시 보상 Command를 발행해 이미 성공한 단계를 되돌린다

📖 더 보기: NestJS CQRS 공식 문서 — Sagas — NestJS에서 Saga를 구현하는 공식 가이드. RxJS Observable 기반 Saga 패턴 설명 (입문)

6.5. 트러블슈팅

문제 1: CQRS 읽기 모델 동기화 지연 (Eventual Consistency)

증상

POST /tasks → 201 Created (태스크 생성 성공)
GET  /tasks → [] (빈 배열 반환 — 방금 만든 태스크가 안 보임)

사용자가 데이터를 저장했는데 바로 조회하면 없다고 나오는 상황.

원인

Command가 쓰기 DB에 저장하고 이벤트를 발행한 뒤, 읽기 모델 업데이트는 비동기로 처리된다. 이벤트 핸들러가 읽기 DB를 업데이트하기 전에 클라이언트가 Query를 보내면 이전 상태가 반환된다.

해결 방법

클라이언트 측 낙관적 업데이트(Optimistic Update): 클라이언트가 응답을 받으면 로컬 상태에 즉시 반영하고, 다음 조회 시에는 서버 상태로 동기화.
일관성 보장이 필요한 경우 동기 처리: 이벤트 핸들러를 동기로 실행하거나, Command Handler 내에서 읽기 DB도 함께 업데이트.
클라이언트에게 지연 안내: “저장되었습니다. 반영까지 수초가 걸릴 수 있습니다.”
읽기 모델 폴링 또는 WebSocket 알림: 읽기 모델 업데이트 완료 시 클라이언트에 Push.

// 동기 처리 예시 (성능 vs 일관성 트레이드오프)
@CommandHandler(CreateTaskCommand)
export class CreateTaskHandler implements ICommandHandler<CreateTaskCommand> {
  async execute(command: CreateTaskCommand): Promise<void> {
    const task = await this.writeRepo.create(command);
    // 읽기 DB도 즉시 업데이트 (동기)
    await this.readRepo.upsert(task);
    // 이후 이벤트는 외부 시스템 알림용으로만 사용
    this.eventBus.publish(new TaskCreatedEvent(task.id));
  }
}

문제 2: 이벤트 스토어 크기 폭증

증상

ERROR: disk quota exceeded on event-store volume
이벤트 스토어 DB 용량: 500GB (서비스 오픈 1년 만에)
BankAccount.rehydrate() 응답 시간: 12초 (이벤트 수 과다)

원인

Event Sourcing은 append-only 특성상 이벤트가 계속 쌓인다. 활성 사용자가 많고 이벤트 발생 빈도가 높으면 스토어 용량이 빠르게 증가한다. 또한 이벤트가 많아질수록 Aggregate 재수화(rehydrate) 시간이 선형으로 증가한다.

해결 방법

스냅샷(Snapshot) 도입: 일정 이벤트 수마다 현재 상태를 스냅샷으로 저장. 재수화 시 가장 최근 스냅샷 + 그 이후 이벤트만 재생.

class EventStore {
  async getEventsAfterSnapshot(aggregateId: string): Promise<{
    snapshot: AggregateSnapshot | null;
    events: DomainEvent[];
  }> {
    const snapshot = await this.snapshotRepo.findLatest(aggregateId);
    const events = await this.eventRepo.findAfter(
      aggregateId,
      snapshot?.version ?? 0,
    );
    return { snapshot, events };
  }
}

// 스냅샷 생성 기준 — Greg Young(EventStoreDB 창시자) 권고:
// "이벤트가 ~1,000개를 넘기 전까지는 스냅샷을 고려하지 마라.
//  수십 개의 소형 이벤트 재생은 성능 문제가 아니다."
// (출처: https://www.kurrent.io/blog/snapshots-in-event-sourcing)
//
// 실무 기준 (구체적 수치):
//  - 이벤트 수 < 500개    → 스냅샷 불필요
//  - 재수화 시간 > 200ms  → 스냅샷 도입 검토 (P99 기준)
//  - 이벤트 수 > 1,000개  → 스냅샷 필수 (예: N=500마다 저장)
if (aggregate.version % 500 === 0) {
  await snapshotRepo.save(aggregate.toSnapshot());
}

오래된 이벤트 아카이빙: 일정 기간(예: 2년) 이상의 이벤트를 S3 등 저비용 스토리지로 이동.
이벤트 압축(Compaction): 최종 결과가 동일한 연속 이벤트를 하나로 합치기.

문제 3: 이벤트 스키마 변경 — 하위 호환성 파손

증상

TypeError: Cannot read property 'recipientEmail' of undefined
  at OrderShippedEvent.apply (order.aggregate.ts:45)

// 기존 이벤트에는 'recipientEmail' 필드가 없었는데
// 새 코드에서 이 필드를 필수로 읽으려 함

원인

Event Sourcing의 이벤트는 영구 저장된다. 새 버전의 코드가 기존 이벤트를 재생할 때, 예전 이벤트에 없는 필드를 읽으려 하면 오류가 발생한다.

해결 방법

이벤트 버전 관리: 이벤트에 버전 필드 추가, Handler에서 버전별 분기 처리.

interface OrderShippedEvent {
  version: number;
  orderId: string;
  shippedAt: Date;
  // v2부터 추가
  recipientEmail?: string;
}

// Aggregate에서 버전별 처리
private applyOrderShipped(event: OrderShippedEvent): void {
  this.status = 'SHIPPED';
  this.shippedAt = event.shippedAt;
  // 하위 호환: v2 이상에만 존재하는 필드
  if (event.version >= 2 && event.recipientEmail) {
    this.notificationEmail = event.recipientEmail;
  }
}

Upcaster 패턴: 이벤트를 읽을 때 구버전을 최신 버전으로 변환하는 레이어.
필드 추가는 Optional로: 새 필드는 반드시 Optional(?)로 추가. 기존 이벤트에는 해당 필드가 없을 수 있음을 항상 고려.

문제 4: CQRS에서 Command 실패 시 읽기 모델 불일치

증상

POST /orders → 500 Internal Server Error (Command 실패)
그런데 읽기 모델(Elasticsearch)에는 이미 주문 데이터가 반영됨
→ 사용자가 주문 목록에서 존재하지 않는 주문을 보게 됨

원인

Command Handler에서 쓰기 DB 저장과 이벤트 발행을 별개의 작업으로 처리하는 경우, 이벤트가 먼저 발행된 후 DB 저장이 실패하면 읽기 모델만 업데이트되는 불일치가 발생한다. 또는 반대로, DB 저장 후 이벤트 발행 전에 서버가 다운되면 읽기 모델이 업데이트되지 않는다.

해결 방법

// 핵심: Outbox 패턴으로 원자적 보장
// DB 저장과 이벤트를 같은 트랜잭션에 묶는다

@CommandHandler(CreateOrderCommand)
export class CreateOrderHandler implements ICommandHandler<CreateOrderCommand> {
  async execute(command: CreateOrderCommand): Promise<void> {
    await this.dataSource.transaction(async (manager) => {
      // 1. 도메인 데이터 저장
      const order = manager.create(Order, { ...command });
      await manager.save(order);

      // 2. Outbox에 이벤트 저장 (같은 트랜잭션!)
      await manager.insert(Outbox, {
        aggregateId: order.id,
        eventType: "OrderCreated",
        payload: JSON.stringify(order),
      });
      // → 트랜잭션 실패 시 둘 다 롤백
      // → 트랜잭션 성공 시 둘 다 커밋
    });
    // 실제 이벤트 발행은 별도 OutboxProcessor가 폴링하여 처리
  }
}

이 패턴의 핵심은 “이벤트 발행”을 직접 하지 않고 “이벤트를 DB에 저장”하는 것이다. DB 트랜잭션이 보장하므로 데이터와 이벤트가 항상 함께 존재하거나 함께 없다.

문제 4-1: OutboxProcessor 중단 및 중복 발행 멱등성

증상

-- 미발행 이벤트 폭증 (OutboxProcessor가 멈춘 신호)
SELECT COUNT(*) FROM outbox WHERE published_at IS NULL;
→ 15,430건 (정상 운영 시 < 100건)

-- 또는 반대로: 같은 이벤트가 두 번 처리됨
[Consumer] OrderCreated orderId=uuid-xxx → 이메일 발송 2회

원인

OutboxProcessor는 Outbox 테이블을 주기적으로 폴링해서 이벤트를 브로커에 발행하고, 발행 성공 후 published_at을 갱신한다. 이때 두 가지 실패 모드가 존재한다:

Processor 중단: 서버 재배포, 메모리 부족, 브로커 연결 실패 등으로 폴러 자체가 중단되면 이벤트가 발행되지 않고 쌓인다.
중복 발행(at-least-once): 브로커 발행 직후, published_at 갱신 전에 Processor가 크래시하면 재시작 시 같은 이벤트를 다시 발행한다. Outbox 패턴은 구조적으로 at-least-once를 보장하므로, 소비자 쪽에서 멱등성을 별도로 구현해야 한다.

해결 방법

// 소비자 측 멱등성 보장 — 처리된 이벤트 ID를 별도 테이블에 기록
// (출처: https://microservices.io/patterns/communication-style/idempotent-consumer.html)

@EventPattern("order.created")
async handleOrderCreated(@Payload() event: OrderCreatedEvent): Promise<void> {
  const alreadyProcessed = await this.processedEvents.exists(event.eventId);
  if (alreadyProcessed) {
    return; // 중복 이벤트 무시
  }

  await this.dataSource.transaction(async (manager) => {
    // 비즈니스 로직 처리
    await this.orderReadRepository.upsert(event);

    // 처리된 이벤트 ID 기록 (같은 트랜잭션)
    await manager.insert(ProcessedEvent, { eventId: event.eventId });
  });
}

// OutboxProcessor 중단 감지 — 알람 기준
// published_at IS NULL인 이벤트가 5분 이상 100건 초과 → PagerDuty 알람

운영 알람 기준: Outbox 미발행 이벤트(published_at IS NULL) 수가 5분 이상 100건을 초과하면 Processor 중단으로 간주하고 알람을 발생시킨다. 이 수치는 정상 처리 처리량과 폴링 주기에 따라 조정한다.

문제 4-2: Choreography Saga 실패 추적 어려움

증상

OrderService 로그: OrderCreated orderId=ord-1 ✅
PaymentService 로그: PaymentCompleted orderId=ord-1 ✅
InventoryService 로그: 없음 (이벤트 수신 자체가 안 됨?)
ShippingService 로그: 없음

→ 주문이 결제까지 됐는데 배송 준비가 안 됨. 어느 서비스에서 멈췄는지 불명확.

원인

Choreography Saga는 중앙 오케스트레이터가 없고 각 서비스가 이벤트를 독립적으로 구독한다. Saga 전체 흐름이 코드 한 곳에 명시되지 않아, 실패가 발생해도 어느 단계에서 멈췄는지 파악하기 위해 여러 서비스 로그를 따로 확인해야 한다. (출처: microservices.io — Implementing a choreography-based saga)

해결 방법

// 1. Correlation ID를 모든 이벤트에 포함 — 분산 추적의 기본
interface SagaEvent {
  eventId: string;
  correlationId: string; // Saga 전체를 묶는 ID (보통 최초 Command ID)
  causationId: string; // 이 이벤트를 유발한 직전 이벤트 ID
  occurredAt: Date;
}

// OrderCreated 발행 시
const event: OrderCreatedEvent = {
  eventId: randomUUID(),
  correlationId: command.orderId, // 주문 ID로 전체 Saga 추적
  causationId: command.commandId,
  orderId: command.orderId,
  // ...
};

// 2. Saga 상태를 별도 테이블로 추적 (보완 수단)
// CREATE TABLE saga_state (
//   correlation_id UUID PRIMARY KEY,
//   current_step   VARCHAR(50),   -- 'ORDER_CREATED', 'PAYMENT_DONE', ...
//   status         VARCHAR(20),   -- 'IN_PROGRESS', 'COMPLETED', 'FAILED'
//   updated_at     TIMESTAMP
// );

Choreography vs Orchestration 선택 기준: 서비스가 3개 이하이고 Saga 흐름이 단순하면 Choreography로 충분하다. 서비스가 4개 이상이거나 보상 트랜잭션이 복잡하면 Orchestration으로 전환해 Saga 상태를 중앙에서 관리하는 것이 운영상 유리하다.

문제 5: TypeORM `@VersionColumn` 낙관적 락 충돌 — OptimisticLockVersionMismatchError

증상

OptimisticLockVersionMismatchError: The optimistic lock on entity Order failed,
  version 3 was expected, but is actually 4.
  at UpdateQueryBuilder.updateEntity (typeorm/query-builder/UpdateQueryBuilder.ts)

동시 요청이 몰릴 때 간헐적으로 발생. 특히 마감 시간대(예: 공연 티켓 예매, 한정 수량 판매)에 집중적으로 나타난다.

원인

TypeORM @VersionColumn은 UPDATE 시 WHERE version = :expected_version 조건을 자동으로 추가한다. 두 트랜잭션이 같은 버전(3)을 읽고 동시에 업데이트를 시도하면, 먼저 성공한 트랜잭션이 버전을 4로 올리고, 나중 트랜잭션은 WHERE version = 3이 매칭되지 않아 에러가 발생한다.

확인 방법

@Entity()
export class Order {
  @PrimaryGeneratedColumn("uuid")
  id: string;

  @Column()
  status: string;

  @VersionColumn()
  version: number; // TypeORM이 UPDATE마다 자동으로 1씩 증가
}

// TypeORM이 내부적으로 실행하는 SQL:
// UPDATE order SET status='PAID', version=4
// WHERE id='xxx' AND version=3   ← 이 조건이 실패하면 에러 발생

해결 방법

// 방법 1: 재시도 로직 (낙관적 락의 표준 패턴)
@CommandHandler(PayOrderCommand)
export class PayOrderHandler implements ICommandHandler<PayOrderCommand> {
  private readonly MAX_RETRIES = 3;

  async execute(command: PayOrderCommand): Promise<void> {
    let attempt = 0;
    while (attempt < this.MAX_RETRIES) {
      try {
        const order = await this.orderRepository.findOneOrFail({
          where: { id: command.orderId },
        });
        order.status = "PAID";
        await this.orderRepository.save(order); // VersionColumn 자동 검증
        return;
      } catch (e) {
        if (
          e instanceof OptimisticLockVersionMismatchError &&
          attempt < this.MAX_RETRIES - 1
        ) {
          attempt++;
          await new Promise((resolve) => setTimeout(resolve, 50 * attempt)); // 지수 백오프
          continue;
        }
        throw e; // 최대 재시도 초과 또는 다른 에러
      }
    }
  }
}

// 방법 2: 재고처럼 절대 음수가 안 되는 경우 비관적 락 사용
const order = await this.orderRepository.findOne({
  where: { id: command.orderId },
  lock: { mode: "pessimistic_write" }, // SELECT ... FOR UPDATE
});
// → 한 트랜잭션이 끝날 때까지 다른 트랜잭션이 대기
// → 충돌 자체를 방지하지만 동시성 처리량 감소

예상 출력 (재시도 로직 적용 후)

1차 시도: OptimisticLockVersionMismatchError → 50ms 대기 후 재시도
2차 시도: 성공 (다른 트랜잭션이 이미 완료됨)
→ 사용자에게는 정상 응답 반환 (에러 없음)

문제 6: NestJS CQRS + AWS Lambda — Cold Start 시 CommandBus 초기화 지연

증상

Lambda Cold Start 시 첫 번째 API 요청 응답 시간: 8~12초
그 이후 요청은 정상 (100~200ms)
CloudWatch 로그: "NestFactory.create" 시간이 7초 이상

원인

@nestjs/cqrs는 모듈 초기화 시 모든 CommandHandler, QueryHandler, EventHandler를 스캔하고 버스에 등록한다. Handler 수가 많아질수록(50개 이상) 초기화 시간이 증가한다. Lambda는 컨테이너를 재사용하지 않으면 매번 이 초기화를 반복한다.

해결 방법

// 방법 1: Lambda 워밍업 — CloudWatch Events로 주기적 호출
// EventBridge Rule: 매 5분마다 /health 엔드포인트 호출
// → 컨테이너 재사용 비율 증가 → Cold Start 빈도 감소

// 방법 2: Provisioned Concurrency 설정 (AWS Lambda)
// 항상 N개의 실행 환경을 미리 초기화 상태로 유지
// 비용은 증가하지만 Cold Start 완전 제거 가능

// 방법 3: Handler 지연 등록 — 초기화 비용 분산
// 대신 NestJS의 LazyModuleLoader 사용
import { LazyModuleLoader } from "@nestjs/core";

@Injectable()
export class OrderCommandService {
  constructor(private lazyModuleLoader: LazyModuleLoader) {}

  async processLargeOrder(command: ProcessLargeOrderCommand) {
    // 무거운 Handler 모듈은 실제 필요 시 로드
    const { LargeOrderModule } = await import("./large-order.module");
    const moduleRef = await this.lazyModuleLoader.load(() => LargeOrderModule);
    const handler = moduleRef.get(ProcessLargeOrderHandler);
    return handler.execute(command);
  }
}

// 방법 4: ECS Fargate나 EC2로 마이그레이션 검토
// CQRS 패턴의 무거운 Handler 초기화는 서버리스보다 상시 실행 환경에 더 적합

CloudWatch 지표로 Cold Start 측정

CloudWatch Metrics → Lambda → FUNCTION_NAME → Init Duration
→ Init Duration이 지속적으로 높으면 Handler 수 또는 의존성 최적화 필요

권장 임계값:
  Init Duration < 3초 → 정상
  Init Duration 3~10초 → 최적화 권장
  Init Duration > 10초 → 아키텍처 재검토 (ECS/EC2 고려)

7. 체크리스트

CQRS 이해 체크리스트

Command와 Query의 차이를 한 문장으로 설명할 수 있다
@nestjs/cqrs의 CommandBus, QueryBus, EventBus의 역할을 구분할 수 있다
@CommandHandler, @QueryHandler 데코레이터를 사용해 핸들러를 구현할 수 있다
CQRS에서 읽기 모델과 쓰기 모델이 왜 분리되는지 설명할 수 있다
Eventual Consistency가 무엇인지, CQRS에서 왜 발생하는지 설명할 수 있다

Event Sourcing 이해 체크리스트

Event Store가 왜 append-only인지 설명할 수 있다
이벤트 재생(Replay)으로 현재 상태를 복원하는 코드를 직접 작성할 수 있다
스냅샷이 왜 필요한지, 언제 사용하는지 설명할 수 있다
CQRS와 Event Sourcing이 어떻게 결합되는지 설명할 수 있다
이벤트 스키마 변경 시 하위 호환성 문제와 해결 방법을 알고 있다

8. 핵심 키워드

키워드	설명
CQRS	Command Query Responsibility Segregation — 읽기/쓰기 책임 분리
CommandBus	Command를 적절한 Handler로 라우팅하는 버스
QueryBus	Query를 적절한 Handler로 라우팅하는 버스
EventBus	도메인 이벤트를 발행하고 Handler들에게 전달
Event Sourcing	상태 대신 이벤트 이력을 저장하는 패턴
Event Store	이벤트를 append-only로 저장하는 저장소
Aggregate	관련 도메인 객체의 묶음, 이벤트를 적용해 상태 관리
Replay	이벤트를 순서대로 재적용해 현재 상태를 복원
Snapshot	특정 시점의 Aggregate 상태를 저장해 재생 시간 단축
Time Travel	이벤트 재생으로 과거 임의 시점의 상태를 재현하는 기능
Eventual Consistency	즉각적 일관성 대신 결과적으로 일관성을 보장
CDC	Change Data Capture — WAL 기반으로 DB 변경을 캡처해 읽기 모델에 전파
WAL	Write-Ahead Log — CDC 파이프라인의 데이터 원천. 상세 원리는 `transaction-basics.md` 참조

8.5. 추천 리소스

📚 추천 리소스

📖 NestJS CQRS 공식 문서 — @nestjs/cqrs CommandHandler, QueryHandler, EventHandler, Saga 기본 사용법 공식 가이드 (입문)
📖 Microsoft Azure - CQRS Pattern — CQRS 적용 시나리오, 장단점, Event Sourcing과의 결합 방식을 클라우드 아키텍처 관점에서 설명 (입문)
📖 Martin Fowler - Event Sourcing — Event Sourcing 패턴의 원형 정의. “왜 상태가 아니라 이벤트를 저장하는가”에 대한 근본 철학 (중급)
📖 Designing Scalable Systems with CQRS and Event Sourcing — Outbox 패턴, 낙관적 동시성 제어, Snapshot 전략 등 프로덕션 수준 설계 패턴 종합 (중급)
📖 Event Sourcing and CQRS with Apache Kafka — Confluent — Kafka를 Event Store로 활용하는 CQRS 구현 패턴. Projection Consumer 설계 방법 포함 (입문~중급)

9. 직접 확인해보기

NestJS CQRS 모듈 설치 및 동작 확인

# 새 NestJS 프로젝트 생성
npm i -g @nestjs/cli
nest new cqrs-demo && cd cqrs-demo

# CQRS 모듈 설치
npm install @nestjs/cqrs

# 프로젝트 실행
npm run start:dev

예상 출력

[Nest] LOG  Starting Nest application...
[Nest] LOG  TaskModule dependencies initialized +1ms
[Nest] LOG  Application is running on: http://[::1]:3000

Event Sourcing 이벤트 재생 테스트

# TypeScript 코드 직접 실행 (ts-node 사용)
npx ts-node --esm << 'EOF'
// 간단한 Event Sourcing 테스트
interface Event { type: string; amount?: number }

function rehydrate(events: Event[]): number {
  return events.reduce((balance, event) => {
    if (event.type === 'DEPOSIT') return balance + (event.amount ?? 0);
    if (event.type === 'WITHDRAW') return balance - (event.amount ?? 0);
    return balance;
  }, 0);
}

const events: Event[] = [
  { type: 'DEPOSIT', amount: 500000 },
  { type: 'WITHDRAW', amount: 100000 },
  { type: 'DEPOSIT', amount: 300000 },
];

console.log('현재 잔액:', rehydrate(events), '원');
// 시간 여행: 처음 2개 이벤트만 재생
console.log('1월 5일 잔액:', rehydrate(events.slice(0, 2)), '원');
EOF

예상 출력

현재 잔액: 700000 원
1월 5일 잔액: 400000 원

CQRS API 테스트 (curl)

# Command: 태스크 생성
curl -X POST http://localhost:3000/tasks \
  -H "Content-Type: application/json" \
  -d '{"title": "CQRS 학습", "description": "패턴 이해", "userId": "user-1"}'

# Query: 태스크 목록 조회
curl http://localhost:3000/tasks?userId=user-1

예상 출력

// POST 응답
{ "message": "태스크가 생성되었습니다." }

// GET 응답
[
  {
    "id": "550e8400-e29b-41d4-a716-446655440000",
    "title": "CQRS 학습",
    "status": "open",
    "createdAt": "2024-01-15T10:23:45.000Z"
  }
]

Outbox 테이블 확인 (PostgreSQL)

-- Outbox 테이블 생성 후 미발행 이벤트 확인 쿼리
-- (CQRS + Outbox 패턴 운영 시 주기적으로 확인)

SELECT
  id,
  aggregate_type,
  event_type,
  created_at,
  published_at,
  CASE
    WHEN published_at IS NULL THEN '미발행'
    ELSE '발행 완료'
  END AS status
FROM outbox
WHERE published_at IS NULL
ORDER BY created_at ASC
LIMIT 10;

-- 예상 출력 (미발행 이벤트가 있는 경우):
-- id       | aggregate_type | event_type   | created_at          | published_at | status
-- uuid-101 | Task           | TaskCreated  | 2026-04-07 10:23:45 | NULL         | 미발행
-- uuid-102 | Order          | OrderPlaced  | 2026-04-07 10:24:01 | NULL         | 미발행
-- → published_at이 NULL인 행이 계속 쌓이면 OutboxProcessor가 멈춘 것

-- 미발행 이벤트 수 추이 모니터링
SELECT
  date_trunc('hour', created_at) AS hour,
  COUNT(*) FILTER (WHERE published_at IS NULL) AS pending,
  COUNT(*) FILTER (WHERE published_at IS NOT NULL) AS published
FROM outbox
WHERE created_at > NOW() - INTERVAL '24 hours'
GROUP BY hour
ORDER BY hour DESC;

10. 한 줄 요약

CQRS는 읽기와 쓰기 창구를 분리해 각자 최적화하고, Event Sourcing은 현재 상태 대신 변경 이력(이벤트)을 쌓아 언제든 과거로 돌아갈 수 있게 하며, WAL은 데이터를 바꾸기 전에 로그를 먼저 써두어 서버가 죽어도 데이터를 잃지 않게 보장하는 세 가지 패턴은 모두 “신뢰할 수 있는 시스템”을 만들기 위한 핵심 설계 원칙이다.

CQRS & Event Sourcing

CQRS / Event Sourcing

1. 한 줄 정의

2. 왜 중요한가

CQRS가 필요한 이유

Event Sourcing이 필요한 이유

프론트엔드 개발자를 위한 브릿지

CRUD 단일 모델에서 책임 분리와 이벤트 저장으로 넘어가는 이유

3. 핵심 개념

3-1. CQRS (Command Query Responsibility Segregation)

비유: 은행 창구

원리

왜 이렇게 동작하는가 — 읽기와 쓰기의 근본적 차이

NestJS @nestjs/cqrs 구현 예시

CQRS 고급: Outbox 패턴 — 이벤트 유실 방지

CQRS 고급: 낙관적 동시성 제어 (Optimistic Concurrency)

3-2. Event Sourcing (이벤트 소싱)

비유: 은행 통장 거래 내역

원리

왜 이렇게 동작하는가 — Append-Only의 근본 이유

핵심 특성

TypeScript 구현 예시

NestJS에서 Event Sourcing 라이브러리 활용

Event Sourcing 장단점

CQRS + Event Sourcing + Kafka 결합 심화: Projection 패턴

3-3. CQRS + Event Sourcing 결합: WAL 기반 CDC 동기화

3-4. AWS + TypeORM 관점에서의 CQRS / Event Sourcing

AWS RDS + TypeORM으로 CQRS 읽기/쓰기 DB 분리

TypeORM DataSource로 Command/Query Handler에서 트랜잭션 제어

AWS DynamoDB를 Event Store로 사용 (대규모 이벤트 처리)

4. 실무에서 어떻게 쓰이나

CQRS 실무 적용

Event Sourcing 실무 적용

Event Store 선택 가이드

WAL 기반 CDC 실무 활용 (CQRS 맥락)

5. 내 업무와의 연결고리

6. 비슷한 개념과 비교

CQRS vs 전통적인 CRUD

Event Sourcing vs 전통적인 CRUD

CQRS와 Event Sourcing의 관계

Saga 패턴과 CQRS — 분산 트랜잭션 처리

6.5. 트러블슈팅

문제 1: CQRS 읽기 모델 동기화 지연 (Eventual Consistency)

문제 2: 이벤트 스토어 크기 폭증

문제 3: 이벤트 스키마 변경 — 하위 호환성 파손

문제 4: CQRS에서 Command 실패 시 읽기 모델 불일치

문제 4-1: OutboxProcessor 중단 및 중복 발행 멱등성

문제 4-2: Choreography Saga 실패 추적 어려움

문제 5: TypeORM @VersionColumn 낙관적 락 충돌 — OptimisticLockVersionMismatchError

문제 6: NestJS CQRS + AWS Lambda — Cold Start 시 CommandBus 초기화 지연

7. 체크리스트

CQRS 이해 체크리스트

Event Sourcing 이해 체크리스트

8. 핵심 키워드

8.5. 추천 리소스

📚 추천 리소스

9. 직접 확인해보기

NestJS CQRS 모듈 설치 및 동작 확인

Event Sourcing 이벤트 재생 테스트

CQRS API 테스트 (curl)

Outbox 테이블 확인 (PostgreSQL)

10. 한 줄 요약

NestJS `@nestjs/cqrs` 구현 예시

문제 5: TypeORM `@VersionColumn` 낙관적 락 충돌 — OptimisticLockVersionMismatchError