CQRS 패턴 구현

지난 절에서는 NestJS로 실시간 웹 애플리케이션 핵심 기술인 WebSocket을 구현하는 방법을 알아봤습니다.

이제 12장 세 번째 절에서는 대규모 엔터프라이즈 애플리케이션의 확장성과 성능을 높이는 고급 아키텍처 패턴, CQRS (Command Query Responsibility Segregation)와 이를 NestJS에서 구현하는 방법을 살펴보겠습니다.

기존 애플리케이션은 대부분 CRUD(Create, Read, Update, Delete) 모델로 데이터를 처리합니다.

즉 데이터를 읽는(Query) 작업과 쓰는(Command) 작업이 동일한 데이터 모델과 로직을 공유합니다.

하지만 복잡한 시스템에서는 읽기와 쓰기의 빈도와 요구사항이 크게 달라 성능 병목이나 확장성 문제가 생길 수 있습니다.

CQRS는 이런 문제를 해결하기 위해 등장한 패턴입니다.

CQRS란?

CQRS는 애플리케이션의 작업(Operation)을 명령(Command)과 조회(Query)로 나누고, 이 두 가지 책임에 대해 서로 다른 모델을 사용하는 아키텍처 패턴입니다.

• 명령 (Command): 데이터 상태를 변경하는 작업입니다 (생성, 업데이트, 삭제). 명령은 비즈니스 로직을 포함하며, 일반적으로 결과 값을 반환하지 않고 작업의 성공/실패 여부만 알립니다.
• 조회 (Query): 데이터를 읽는 작업입니다. 조회는 데이터의 상태를 변경하지 않으며, 복잡한 비즈니스 로직 없이 데이터를 가져와 소비자에게 반환합니다.

전통적인 CRUD 모델에서는 동일한 데이터 모델(예: ORM 엔티티)을 읽기(조회)와 쓰기(명령) 모두에 사용합니다.

이로 인해 다음과 같은 문제가 발생할 수 있습니다.

• 성능 병목: 읽기 작업이 쓰기 작업보다 훨씬 많을 때, 쓰기 작업에 최적화된 데이터 모델이 읽기 성능을 저하시킬 수 있습니다.
• 복잡성 증가: 읽기와 쓰기 로직이 얽히면서 코드 복잡도가 증가하고 유지보수가 어려워집니다.
• 확장성 제약: 읽기 스케일링과 쓰기 스케일링 요구사항이 다를 때, 동일한 모델로는 효율적인 확장이 어렵습니다.

CQRS는 이를 분리하여 각 책임에 최적화된 아키텍처를 구축할 수 있도록 합니다.

• 확장성 (Scalability): 읽기 모델과 쓰기 모델을 독립적으로 확장할 수 있습니다. 예를 들어, 읽기 작업이 많으면 조회 모델의 인스턴스를 늘리고, 쓰기 작업이 많으면 명령 모델의 인스턴스를 늘릴 수 있습니다.
• 성능 최적화
  • 쓰기 모델: 쓰기 작업에 최적화된 데이터베이스(예: NoSQL)나 데이터 구조를 사용할 수 있습니다.
  • 읽기 모델: 읽기 작업에 최적화된 데이터베이스(예: 검색 엔진, Read Replica)나 캐싱 전략을 사용할 수 있습니다. 복잡한 JOIN 없이 쿼리에 필요한 데이터만 미리 평탄화(denormalization)하여 저장할 수 있습니다.
• 유연성: 각 모델에 독립적인 기술 스택을 적용할 수 있습니다.
• 복잡성 분리: 읽기/쓰기 로직을 분리하여 코드의 응집도를 높이고 유지보수를 용이하게 합니다.
• 이벤트 소싱(Event Sourcing)과의 시너지: CQRS는 종종 이벤트 소싱과 함께 사용됩니다. 모든 상태 변경을 이벤트로 기록하고, 이 이벤트를 통해 읽기 모델을 구축하는 방식입니다.

• 복잡성 증가: 읽기/쓰기 모델을 분리하므로 시스템 아키텍처가 더 복잡해지고, 초기 설정 및 관리에 더 많은 노력이 필요합니다.
• 데이터 동기화: 쓰기 모델에서 변경된 데이터가 읽기 모델에 반영되는 과정에서 데이터 일관성(Eventual Consistency) 문제가 발생할 수 있습니다. 즉, 쓰기 후 읽기까지 약간의 지연이 발생할 수 있습니다.
• 학습 곡선: 새로운 아키텍처 패턴에 대한 이해와 구현 경험이 필요합니다.

NestJS에서 CQRS 구현

NestJS는 공식 @nestjs/cqrs 패키지로 명령·조회·이벤트 버스와 핸들러를 지원합니다.

주로 @nestjs/cqrs 패키지를 사용하여 CQRS의 핵심 구성 요소를 통합합니다.

@nestjs/cqrs 패키지 설치

npm install @nestjs/cqrs @nestjs/event-emitter # event-emitter는 이벤트 처리에 유용

CQRS의 주요 구성 요소

@nestjs/cqrs는 다음과 같은 핵심 구성 요소를 제공합니다.

• Command (명령): 시스템의 상태를 변경하는 작업을 정의하는 클래스입니다.
  • CommandBus: 명령을 발행(dispatch)하고 해당 명령 핸들러로 라우팅합니다.
  • CommandHandler: 특정 명령을 처리하는 비즈니스 로직을 포함하는 클래스입니다.
• Query (조회): 시스템의 상태를 조회하는 작업을 정의하는 클래스입니다.
  • QueryBus: 조회를 발행하고 해당 조회 핸들러로 라우팅합니다.
  • QueryHandler: 특정 조회를 처리하고 데이터를 반환하는 클래스입니다.
• Event (이벤트): 시스템에서 중요한 상태 변경이 발생했음을 나타내는 메시지입니다.
  • EventBus: 이벤트를 발행하고 모든 관련 이벤트 핸들러로 브로드캐스트합니다.
  • EventHandler: 특정 이벤트를 구독하고 처리하는 클래스입니다. (예: 읽기 모델 업데이트)
• Saga: 여러 명령과 이벤트를 조율하여 복잡한 비즈니스 프로세스를 관리하는 클래스입니다.

NestJS에서 CQRS 구현 예시

사용자 생성(명령) 및 사용자 조회(조회) 기능을 CQRS 패턴으로 구현하는 예시입니다.

export class CreateUserCommand {
  constructor(
    public readonly name: string,
    public readonly email: string,
  ) {}
}

import { CommandHandler, ICommandHandler } from '@nestjs/cqrs';
import { CreateUserCommand } from '../impl/create-user.command';
import { Logger } from '@nestjs/common';
import { UserRepository } from '../../user.repository'; // 사용자 데이터 저장소 (쓰기 모델)
import { UserCreatedEvent } from '../../events/impl/user-created.event';
import { EventBus } from '@nestjs/cqrs'; // EventBus 주입

@CommandHandler(CreateUserCommand)
export class CreateUserHandler implements ICommandHandler<CreateUserCommand> {
  private readonly logger = new Logger(CreateUserHandler.name);

  constructor(
    private readonly userRepository: UserRepository,
    private readonly eventBus: EventBus, // EventBus 주입
  ) {}

  async execute(command: CreateUserCommand) {
    this.logger.log(`Handling CreateUserCommand: ${command.email}`);
    // 실제 데이터베이스에 사용자 생성 로직
    const newUser = await this.userRepository.createUser(command.name, command.email);

    // 사용자 생성 후 이벤트 발행
    this.eventBus.publish(new UserCreatedEvent(newUser.id, newUser.email, newUser.name));

    return newUser; // 또는 생성 성공 여부만 반환
  }
}

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

import { IQueryHandler, QueryHandler } from '@nestjs/cqrs';
import { GetUserByIdQuery } from '../impl/get-user-by-id.query';
import { Logger } from '@nestjs/common';
import { UserReadModelRepository } from '../../user-read-model.repository'; // 조회 모델 데이터 저장소

@QueryHandler(GetUserByIdQuery)
export class GetUserByIdHandler implements IQueryHandler<GetUserByIdQuery> {
  private readonly logger = new Logger(GetUserByIdHandler.name);

  constructor(private readonly userReadModelRepository: UserReadModelRepository) {} // 읽기 모델 저장소 주입

  async execute(query: GetUserByIdQuery) {
    this.logger.log(`Handling GetUserByIdQuery: ${query.userId}`);
    // 읽기 전용 데이터베이스에서 사용자 정보 조회 로직
    const user = await this.userReadModelRepository.findById(query.userId);
    return user;
  }
}

export class UserCreatedEvent {
  constructor(
    public readonly userId: string,
    public readonly email: string,
    public readonly name: string,
  ) {}
}

import { EventsHandler, IEventHandler } from '@nestjs/cqrs';
import { UserCreatedEvent } from '../impl/user-created.event';
import { Logger } from '@nestjs/common';
import { UserReadModelRepository } from '../../user-read-model.repository'; // 조회 모델 데이터 저장소

@EventsHandler(UserCreatedEvent)
export class UserCreatedHandler implements IEventHandler<UserCreatedEvent> {
  private readonly logger = new Logger(UserCreatedHandler.name);

  constructor(private readonly userReadModelRepository: UserReadModelRepository) {} // 읽기 모델 저장소 주입

  async handle(event: UserCreatedEvent) {
    this.logger.log(`Handling UserCreatedEvent: User ID ${event.userId}`);
    // 사용자가 생성되면, 읽기 모델을 업데이트하는 로직
    // 예: 별도의 읽기 전용 DB(Elasticsearch, Redis 등)에 사용자 정보 저장
    await this.userReadModelRepository.createOrUpdate({
      id: event.userId,
      name: event.name,
      email: event.email,
      // 필요한 추가적인 읽기 전용 필드
    });
  }
}

import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';
import { UserController } from './user.controller';
import { UserRepository } from './user.repository';
import { UserReadModelRepository } from './user-read-model.repository';

// Command Handlers
import { CreateUserHandler } from './commands/handlers/create-user.handler';

// Query Handlers
import { GetUserByIdHandler } from './queries/handlers/get-user-by-id.handler';

// Event Handlers
import { UserCreatedHandler } from './events/handlers/user-created.handler';

// CQRS 관련 프로바이더들을 배열로 정의
export const CommandHandlers = [CreateUserHandler];
export const QueryHandlers = [GetUserByIdHandler];
export const EventHandlers = [UserCreatedHandler];

@Module({
  imports: [CqrsModule], // CqrsModule 임포트
  controllers: [UserController],
  providers: [
    UserRepository, // 쓰기 모델 리포지토리
    UserReadModelRepository, // 읽기 모델 리포지토리
    ...CommandHandlers, // 모든 Command Handler 등록
    ...QueryHandlers,   // 모든 Query Handler 등록
    ...EventHandlers,   // 모든 Event Handler 등록
  ],
})
export class UserModule {}

import { Module } from '@nestjs/common';
import { AppController } from './app.controller';
import { AppService } from './app.service';
import { UserModule } from './user/user.module';

@Module({
  imports: [UserModule], // UserModule 임포트
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

import { Controller, Post, Get, Body, Param, Logger } from '@nestjs/common';
import { CommandBus, QueryBus } from '@nestjs/cqrs'; // CQRS 버스 주입
import { CreateUserCommand } from './commands/impl/create-user.command';
import { GetUserByIdQuery } from './queries/impl/get-user-by-id.query';

@Controller('users')
export class UserController {
  private readonly logger = new Logger(UserController.name);

  constructor(
    private readonly commandBus: CommandBus,
    private readonly queryBus: QueryBus,
  ) {}

  @Post()
  async createUser(@Body() body: { name: string; email: string }) {
    this.logger.log('Received createUser request');
    // Command 발행
    const user = await this.commandBus.execute(new CreateUserCommand(body.name, body.email));
    return { message: 'User created successfully', user };
  }

  @Get(':id')
  async getUser(@Param('id') id: string) {
    this.logger.log(`Received getUser request for ID: ${id}`);
    // Query 발행
    const user = await this.queryBus.execute(new GetUserByIdQuery(id));
    if (!user) {
      // NestJS 예외 처리 (NotFoundException 등)
      throw new Error('User not found');
    }
    return user;
  }
}

CQRS에서 명령의 책임은 상태 변경이고 조회의 책임은 읽기입니다. 명령이 생성 ID나 처리 결과를 반환하는 것은 이 책임 분리와 모순되지 않습니다. 원 CreateUserHandler도 newUser를 반환하며, Nest의 CommandBus 계약은 handler의 반환 값을 Promise 결과로 전달합니다. 앞의 “일반적으로 결과 값을 반환하지 않는다”는 표현을 반환 값 금지 규칙으로 읽지 않습니다.

원문의 빈 저장소에 name=민, email=[email protected]을 한 번 생성하면 쓰기 배열에는 ID 1의 사용자가 추가되고 같은 ID의 이벤트가 발행됩니다. 명령 함수에는 읽기 Map 갱신의 완료를 기다리는 연결이 없습니다. 이벤트 handler의 갱신을 지연시킨 조건에서는 생성 결과가 돌아온 뒤 조회가 아직 사용자를 못 찾을 수 있습니다. 반영 뒤의 Map에는 키 1이 있습니다. 같은 이메일을 다시 생성하면 원 저장소의 중복 검사 없이 ID 2가 추가됩니다. 사용자 생성 실패, 이메일 중복 방지, 읽기 반영 완료는 서로 다른 조건입니다.

독립 확인: 두 생성이 성공하고 읽기 Map에는 키 1·2가 각각 반영된 조건입니다. ID 1의 동일한 이벤트를 읽기 handler에 한 번 더 전달합니다. Map의 항목 수와 키 1의 값을 설명하세요. 이어 ID 3을 조회하면 어떤 코드 경로로 가나요? publish()를 호출했다는 사실만으로 DB commit·outbox 저장·정확히 한 번 전달·읽기 완료를 증명할 수 있나요?

CQRS 구현 시 고려사항

• 데이터 일관성 (Eventual Consistency): 쓰기 모델과 읽기 모델 간의 데이터 동기화 지연을 허용해야 합니다. 사용자에게 “지금은 데이터가 바로 보이지 않을 수 있지만, 잠시 후 반영됩니다”와 같은 메시지를 제공해야 할 수도 있습니다.
• 트랜잭션 관리: 쓰기 모델에서 복잡한 트랜잭션이 필요한 경우, 도메인 주도 설계(DDD)의 애그리게이트(Aggregate) 패턴과 함께 사용하는 것이 효과적입니다.
• 메시지 큐/브로커: 쓰기 모델에서 읽기 모델로 이벤트(데이터 변경 알림)를 전송할 때, Kafka, RabbitMQ, AWS SQS/SNS 등 신뢰할 수 있는 메시지 큐/브로커를 사용하는 것이 일반적입니다. 이는 비동기 통신을 통해 시스템의 결합도를 낮추고 확장성을 높입니다.
• 데이터베이스 선택: 읽기/쓰기 모델에 각각 최적화된 데이터베이스를 선택합니다.
  • 쓰기 모델: 관계형 DB(PostgreSQL, MySQL), 문서 DB(MongoDB) 등
  • 읽기 모델: 검색 엔진(Elasticsearch), 인메모리 캐시(Redis), Read Replica, NoSQL 등
• 로깅 및 모니터링: CQRS는 시스템을 더 분산시키므로, 각 컴포넌트 간의 데이터 흐름과 병목 지점을 추적할 수 있도록 분산 로깅 및 트레이싱 시스템(예: OpenTelemetry, Jaeger)을 구축하는 것이 중요합니다.

CQRS를 도입할 때는 쓰기 성공 이후 조회 반영까지의 지연과 운영 추적을 함께 설계해야 합니다.

CQRS 패턴은 모든 애플리케이션에 필요한 것은 아닙니다.

시스템 복잡성이 낮거나 읽기/쓰기 비율이 크게 다르지 않다면, 전통적인 CRUD 모델이 더 단순하고 관리하기 쉽습니다.

높은 확장성, 성능 최적화, 복잡한 도메인 모델이 필요한 대규모 엔터프라이즈 시스템에서는 CQRS가 구조적 이점을 제공할 수 있습니다.

NestJS의 모듈성과 @nestjs/cqrs 패키지는 이 복잡한 패턴을 구조적으로 구현하는 데 도움을 줍니다.

아래 그림은 원문을 영속 outbox 방식으로 확장할 때의 설계안입니다. 원문의 쓰기 저장소는 배열, 읽기 저장소는 Map이며 CreateUserHandler는 사용자 생성 후 EventBus.publish()를 호출하고 newUser를 반환합니다. DB transaction·outbox·멱등 projector·pending receipt는 구현하지 않았습니다.

조회 모델의 값과 갱신 순서를 추적하기

앞의 Map 기반 조회 모델에는 이벤트 버전·중복 ID·영구 저장이 없습니다. createOrUpdate라는 메서드 이름은 같은 사용자 ID의 값을 마지막 호출로 교체한다는 뜻이며, 최신 업무 버전을 자동으로 고른다는 뜻이 아닙니다. CQRS에서 명령과 조회의 책임을 나누더라도 실제 저장소의 갱신·응답 계약은 따로 정해야 합니다.

예제: 순서가 바뀐 전체 스냅샷

사용자 ID u1의 전체 조회 정보를 가진 사건 두 개를 생각해 봅시다. 버전 4는 이름이 New, 버전 3은 이름이 Old입니다. 이 버전 번호는 개선한 설계를 설명하기 위한 추가 필드이며, 앞의 UserCreatedEvent 클래스에 이미 있는 필드는 아닙니다.

전체 스냅샷은 해당 버전의 완전한 상태를 제공한다는 가정에서 최신 버전으로 교체할 수 있습니다. 반대로 “잔액에 10을 더하라” 같은 델타 사건은 이전 변경을 포함하지 않습니다. 예를 들어 잔액 100에서 버전 2의 +10, 버전 3의 +20이 순서대로 적용되면 130입니다. 버전 3이 먼저 왔다고 120을 저장하고 버전 2를 버리면 10이 누락됩니다. 델타에는 연속 버전 확인·누락 복구·순서 제어가 필요합니다.

같은 객체를 조회해도 상태가 바뀔 수 있다

현재 findById는 Map에 저장한 객체를 그대로 반환합니다. 직접 메서드를 호출해 얻은 객체의 name을 바꾸면, 같은 객체를 가리키는 Map의 조회 결과도 바뀝니다. 조회 책임을 분리했다는 설명만으로 JavaScript 객체가 불변이 되지는 않습니다.

HTTP로 직렬화된 JSON을 클라이언트가 수정하는 것은 서버 Map의 객체를 직접 수정하는 것과 다른 경계입니다. 내부 서비스의 직접 호출까지 읽기 전용 계약으로 만들려면 반환용 복사본이나 불변 데이터 정책을 정해야 합니다. 중첩 객체가 있다면 얕은 복사만으로 모든 내부 참조를 분리하지 못한다는 점도 확인합니다.

조건을 바꿔 보는 추가 연습

1. 현재 UserReadModelRepository에 같은 ID로 New를 저장한 뒤 Old를 저장했습니다. 다음 조회의 이름은 무엇인가요? 전체 스냅샷 버전 4와 3이라는 정보를 추가한다면 어떤 조건을 검사해야 하나요?
2. createOrUpdate({ id: 'u1', name: 'A', email: '[email protected]' }) 이후 findById('u1')로 얻은 객체의 이름을 B로 바꿨습니다. 직접 다시 조회하면 무엇을 얻나요? 이 실험만으로 브라우저에서 받은 JSON을 바꾸면 서버 저장소도 바뀐다고 결론 내릴 수 있나요?
3. 쓰기 저장소의 사용자 생성은 성공했지만 프로세스가 종료되어 조회 모델이 갱신되지 않았습니다. 새 프로세스에서 조회 값이 없다는 사실만으로 사용자가 존재하지 않는다고 판정할 수 있을까요? 영구 DB와 outbox를 도입한다면 저장·재시도·중복 처리의 경계를 설명하세요.

풀이

1. 현재 코드는 마지막 호출의 Old를 반환합니다. 전체 스냅샷 설계에서는 사용자별 적용 버전을 함께 저장하고, 더 높은 버전만 원자적으로 적용해야 이전 값으로 돌아가지 않습니다. 같은 버전 재전송도 중복 처리 정책에 포함합니다. 델타 사건에는 이 규칙만 적용해 버전 간격을 건너뛰면 안 됩니다.
2. 직접 조회는 저장된 같은 객체를 반환하므로 이름이 B가 됩니다. 이는 서버 내부 객체 참조에 관한 실험입니다. 브라우저가 받은 JSON은 직렬화된 데이터이므로 그쪽의 변경이 서버 객체에 직접 연결되지 않습니다. 내부 호출자의 변경을 막으려면 반환 값의 복사·불변 정책을 별도로 구현해야 합니다.
3. 조회 모델의 부재는 투영 지연·실패·실제 부재 중 무엇인지 구분하지 못합니다. 업무에 맞춰 대기 상태를 응답하거나 쓰기 상태·처리 상태를 확인하는 정책이 필요합니다. 앞의 배열·Map 예제는 프로세스 재시작 후 상태 보존도 하지 않습니다. 영구 DB 기반 outbox를 도입한다면 사용자 변경과 사건 기록을 같은 DB 트랜잭션으로 저장하고, 별도 전달자가 재시도하도록 합니다. 전달 후 확인 전에 종료될 수 있으므로 수신 측의 사건 ID 중복 제거와 상태 갱신도 원자적으로 처리해야 합니다. 이런 저장·재시도 계약은 현재 예제의 단순 publish 호출에 구현되어 있지 않습니다.

Nest의 이벤트 핸들러는 비동기로 실행될 수 있으며 예외를 별도로 처리해야 합니다. 명령 반환과 모든 조회 투영의 성공을 같은 완료 시점으로 묶으려면 그 계약을 직접 설계해야 합니다. Nest CQRS 문서