CQRS 패턴 구현

앞 글에서는 WebSocket으로 실시간 애플리케이션을 만들었습니다. 이 글에서는 큰 애플리케이션의 확장성과 성능을 높이는 아키텍처 패턴인 CQRS(Command Query Responsibility Segregation)와, 이를 NestJS에서 구현하는 방법을 살펴봅니다.

대부분의 애플리케이션은 CRUD(Create, Read, Update, Delete) 모델로 데이터를 처리합니다. 데이터를 읽는 작업과 쓰는 작업이 같은 데이터 모델과 로직을 함께 씁니다. 하지만 복잡한 시스템에서는 읽기와 쓰기의 빈도와 요구사항이 크게 달라 성능 병목이나 확장 문제가 생길 수 있고, CQRS는 이런 문제에 대응하기 위해 나온 패턴입니다.

CQRS란

CQRS는 애플리케이션의 작업을 명령(command)과 조회(query)로 나누고, 둘에 서로 다른 모델을 쓰는 아키텍처 패턴입니다.

• 명령: 데이터의 상태를 바꾸는 작업입니다(생성, 수정, 삭제). 업무 규칙을 담으며, 상태 변경이 주된 목적입니다.
• 조회: 데이터를 읽는 작업입니다. 상태를 바꾸지 않고, 데이터를 가져와 돌려줍니다.

명령이 결과 값을 전혀 돌려주면 안 된다는 뜻은 아닙니다. Nest CQRS 문서에 따르면 CommandBus.execute()는 handler가 반환한 값으로 이행되는 Promise를 돌려주며, 생성한 ID 같은 결과를 반환하는 것은 명령과 조회의 분리와 모순되지 않습니다.

전통적인 CRUD 모델은 ORM 엔티티 같은 하나의 데이터 모델을 읽기와 쓰기에 함께 씁니다. 그래서 읽기가 쓰기보다 훨씬 많을 때 쓰기에 맞춘 모델이 읽기 성능을 떨어뜨리고, 읽기와 쓰기 로직이 얽혀 코드가 복잡해지며, 둘의 확장 요구가 다를 때 효율적으로 확장하기 어렵습니다. CQRS는 둘을 나누어 각자에 맞는 구조를 만들 수 있게 합니다.

CQRS의 장점은 다음과 같습니다.

• 확장성: 읽기 모델과 쓰기 모델을 따로 확장할 수 있습니다.
• 성능 최적화: 쓰기 모델에는 쓰기에 맞는 데이터 구조를, 읽기 모델에는 검색 엔진, 읽기 전용 복제본, 캐시처럼 읽기에 맞는 저장소를 씁니다. 복잡한 JOIN 없이 조회에 필요한 데이터를 미리 펼쳐 저장하는 역정규화(denormalization)를 쓸 수도 있습니다.
• 유연성: 두 모델에 서로 다른 기술 스택을 쓸 수 있습니다.
• 복잡성 분리: 읽기와 쓰기 로직을 나누어 코드의 응집도를 높입니다.
• 이벤트 소싱과의 궁합: 모든 상태 변경을 이벤트로 기록하고 그 이벤트로 읽기 모델을 만드는 이벤트 소싱(event sourcing)과 자주 함께 쓰입니다.

단점도 있습니다. 두 모델을 나누는 만큼 구조가 복잡해지고 처음 설정과 관리에 더 많은 노력이 듭니다. 쓰기 모델의 변경이 읽기 모델에 반영되기까지 지연이 생기므로 최종적 일관성(eventual consistency)을 받아들여야 하고, 새 패턴을 익히는 학습 부담도 있습니다.

NestJS에서 CQRS 구현하기

NestJS는 공식 @nestjs/cqrs 패키지로 명령·조회·이벤트 버스와 handler를 지원합니다. 현재 버전은 12.1입니다.

npm install @nestjs/cqrs

@nestjs/cqrs의 핵심 구성 요소는 다음과 같습니다.

• 명령(Command): 상태를 바꾸는 작업을 나타내는 클래스입니다. CommandBus가 명령을 해당 CommandHandler로 보냅니다.
• 조회(Query): 상태를 읽는 작업을 나타내는 클래스입니다. QueryBus가 조회를 해당 QueryHandler로 보냅니다.
• 이벤트(Event): 중요한 상태 변경이 일어났다는 메시지입니다. EventBus가 이벤트를 발행하면 관련된 모든 EventHandler가 받습니다. 읽기 모델 갱신이 대표적인 예입니다.
• Saga: 여러 명령과 이벤트를 엮어 복잡한 업무 흐름을 관리하는 구성 요소입니다.

사용자 생성과 조회 예시

사용자 생성(명령)과 사용자 조회(조회)를 CQRS 패턴으로 구현해 보겠습니다. 먼저 명령 클래스입니다. 공식 문서처럼 Command를 상속하면 CommandBus.execute()의 반환 타입이 추론됩니다.

import { Command } from '@nestjs/cqrs';
import type { UserRecord } from '../../user.repository';

// 실행 결과로 생성한 사용자를 돌려줍니다.
export class CreateUserCommand extends Command<UserRecord> {
  constructor(
    public readonly name: string,
    public readonly email: string,
  ) {
    super();
  }
}

명령 클래스는 실행에 필요한 값인 name과 email만 담는 메시지이고, 저장 로직은 없습니다. 실제 작업은 이 명령과 짝지은 명령 handler가 합니다. 명령 handler는 쓰기 모델에 사용자를 저장하고, 이벤트를 발행한 뒤 생성한 사용자를 돌려줍니다. 아래 코드에서는 @CommandHandler(CreateUserCommand)가 명령과 handler를 연결하는 부분과, execute() 안에서 저장, eventBus.publish(), 반환이 차례로 일어나는 부분을 봅니다.

import { Logger } from '@nestjs/common';
import { CommandHandler, EventBus, ICommandHandler } from '@nestjs/cqrs';
import { UserCreatedEvent } from '../../events/impl/user-created.event';
import { UserRepository } from '../../user.repository';
import { CreateUserCommand } from '../impl/create-user.command';

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

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

  async execute(command: CreateUserCommand) {
    this.logger.log('Handling CreateUserCommand');
    // 쓰기 모델에 사용자를 저장합니다.
    const newUser = await this.userRepository.createUser(command.name, command.email);

    // 저장한 뒤 이벤트를 발행합니다.
    this.eventBus.publish(new UserCreatedEvent(newUser.id, newUser.email, newUser.name));

    return newUser;
  }
}

조회 클래스와 handler입니다. Query를 상속해 반환 타입을 정합니다.

import { Query } from '@nestjs/cqrs';
import type { UserView } from '../../user-read-model.repository';

export class GetUserByIdQuery extends Query<UserView | undefined> {
  constructor(public readonly userId: string) {
    super();
  }
}

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;
  }
}

조회 handler는 쓰기 모델의 UserRepository가 아니라 읽기 모델의 UserReadModelRepository만 주입받습니다. 그래서 조회 결과는 이벤트가 읽기 모델에 반영해 둔 데이터에 달려 있고, 아직 반영되지 않은 사용자는 undefined로 돌아옵니다. 다음은 이 읽기 모델을 채우는 이벤트 클래스와, 이벤트를 받아 읽기 모델을 갱신하는 handler입니다.

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,
      // 필요한 추가적인 읽기 전용 필드
    });
  }
}

쓰기 모델 저장소와 읽기 모델 저장소는 개념을 보여 주기 위한 메모리 구현입니다. 실제로는 쓰기 모델에 TypeORM이나 Mongoose 같은 저장소를, 읽기 모델에 Redis, Elasticsearch, 읽기 전용 복제본 같은 저장소를 씁니다.

import { Injectable } from '@nestjs/common';

export interface UserRecord {
  id: string;
  name: string;
  email: string;
  createdAt: Date;
}

@Injectable()
export class UserRepository {
  // 실제로는 TypeORM이나 Mongoose 등을 씁니다.
  private users: UserRecord[] = [];
  private nextId = 1;

  async createUser(name: string, email: string): Promise<UserRecord> {
    const newUser = { id: (this.nextId++).toString(), name, email, createdAt: new Date() };
    this.users.push(newUser);
    return newUser;
  }
}

import { Injectable } from '@nestjs/common';

export interface UserView {
  id: string;
  name: string;
  email: string;
}

@Injectable()
export class UserReadModelRepository {
  // 실제로는 Redis나 Elasticsearch 클라이언트 등을 씁니다.
  private readUsers = new Map<string, UserView>();

  async findById(id: string): Promise<UserView | undefined> {
    return this.readUsers.get(id);
  }

  async createOrUpdate(user: UserView): Promise<void> {
    this.readUsers.set(user.id, user);
  }
}

NestJS 12의 공식 문서는 루트 모듈에서 CqrsModule.forRoot()를 한 번 가져오라고 안내합니다. forRoot()는 전역 모듈로 등록되므로, 기능 모듈은 CqrsModule을 다시 가져오지 않고 handler를 provider로 등록하기만 하면 됩니다.

import { Module } from '@nestjs/common';
import { CreateUserHandler } from './commands/handlers/create-user.handler';
import { UserCreatedHandler } from './events/handlers/user-created.handler';
import { GetUserByIdHandler } from './queries/handlers/get-user-by-id.handler';
import { UserReadModelRepository } from './user-read-model.repository';
import { UserController } from './user.controller';
import { UserRepository } from './user.repository';

export const CommandHandlers = [CreateUserHandler];
export const QueryHandlers = [GetUserByIdHandler];
export const EventHandlers = [UserCreatedHandler];

@Module({
  controllers: [UserController],
  providers: [
    UserRepository, // 쓰기 모델 저장소
    UserReadModelRepository, // 읽기 모델 저장소
    ...CommandHandlers,
    ...QueryHandlers,
    ...EventHandlers,
  ],
})
export class UserModule {}

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

@Module({
  imports: [CqrsModule.forRoot(), UserModule],
  controllers: [AppController],
  providers: [AppService],
})
export class AppModule {}

handler를 UserModule의 providers에 빠뜨리면 버스가 명령이나 조회를 받을 handler를 찾지 못하므로 execute()가 실패합니다. controller에서는 CommandBus와 QueryBus를 주입받아 씁니다. controller는 handler 클래스를 직접 알지 못하고, 명령과 조회 객체를 만들어 버스에 넘기기만 합니다. getUser()는 조회 결과가 undefined이면 NotFoundException을 던져 404 응답을 돌려줍니다.

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

@Controller('users')
export class UserController {
  constructor(
    private readonly commandBus: CommandBus,
    private readonly queryBus: QueryBus,
  ) {}

  @Post()
  async createUser(@Body() body: { name: string; email: string }) {
    // 반환 타입이 UserRecord로 추론됩니다.
    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) {
    const user = await this.queryBus.execute(new GetUserByIdQuery(id));
    if (!user) {
      throw new NotFoundException('User not found');
    }
    return user;
  }
}

생성 결과와 읽기 모델의 반영을 따로 보기

빈 저장소에 이름 민, 이메일 [email protected]으로 한 번 생성하면, 쓰기 배열에 ID 1의 사용자가 들어가고 같은 ID의 이벤트가 발행됩니다. 하지만 명령 handler에는 읽기 Map의 갱신이 끝나기를 기다리는 코드가 없습니다. 공식 문서의 설명처럼 이벤트 handler는 비동기로 실행되므로, 갱신이 늦어지면 생성 결과가 돌아온 뒤에도 조회가 아직 사용자를 찾지 못할 수 있습니다. 같은 이메일로 다시 생성하면 중복 검사가 없으므로 ID 2가 추가됩니다. 사용자 생성의 실패, 이메일 중복 방지, 읽기 반영의 완료는 서로 다른 조건입니다.

이벤트 handler에서 처리되지 않은 예외는 EventBus가 로그로 남기고 UnhandledExceptionBus라는 스트림으로 보냅니다. 명령을 호출한 쪽에는 이 예외가 전달되지 않으므로, 읽기 모델 갱신 실패를 놓치지 않으려면 UnhandledExceptionBus를 구독해 처리하거나, forRoot({ rethrowUnhandled: true })로 다시 던지게 설정합니다.

순서가 바뀐 이벤트와 버전

지금의 Map 기반 읽기 모델에는 이벤트 버전, 중복 ID, 영구 저장이 없습니다. createOrUpdate라는 이름은 같은 사용자 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을 클라이언트가 고치는 것은 서버의 객체를 고치는 것과 다른 문제입니다.

위 그림은 이 글의 예제를 영속적인 outbox 방식으로 넓힐 때의 설계입니다. 예제는 쓰기 저장소로 배열을, 읽기 저장소로 Map을 쓰고, CreateUserHandler가 사용자를 만든 뒤 EventBus.publish()를 호출합니다. 그림은 여기에 사용자 변경과 이벤트 기록을 같은 DB 트랜잭션에 저장하는 outbox, 같은 이벤트를 여러 번 받아도 결과가 같은 projector, 읽기 반영이 끝나기 전의 대기 상태 응답을 더합니다. 이 장치들은 이 글의 예제 코드에는 구현되어 있지 않습니다.

더 깊이: 이벤트 버스의 publisher를 바꿀 수 있는 이유

@nestjs/cqrs의 버스는 메시지를 실제로 전달하는 publisher와 분리되어 있습니다. Nest CQRS 문서에 따르면 CqrsModule.forRoot()의 eventPublisher 옵션으로 이벤트를 내보내는 publisher를 바꿀 수 있고, 기본값은 메모리 안에서 전달하는 DefaultPubSub입니다. 명령, 조회, 처리되지 않은 예외에도 각각 commandPublisher, queryPublisher, unhandledExceptionPublisher 옵션이 있습니다. 그래서 명령 handler의 publish() 호출은 그대로 두고, 이벤트를 메시지 브로커에 쓰는 publisher로 바꿀 수 있습니다.

이 차이는 @nestjs/cqrs 12.1에서 모델의 commit()에도 드러납니다. 12.1부터 commit(), publish(), publishAll()은 이벤트 버스가 돌려준 값, 곧 publisher가 돌려준 값을 그대로 돌려줍니다. 기본 메모리 publisher는 handler를 기다리지 않으므로 아무것도 돌려주지 않지만, 브로커에 쓰는 비동기 publisher라면 await commit()이 기록을 기다리고 실패하면 예외를 던집니다. 반면 기본 구성에서는 문서의 경고대로 트랜잭션이 롤백되기 직전에 발행한 이벤트도 handler와 saga에 그대로 도착하므로, 롤백된 변경이 읽기 모델에 반영되는 일을 막으려면 커밋한 뒤에 발행하거나 앞의 outbox처럼 같은 트랜잭션에 기록해야 합니다.

CQRS를 도입할 때 고려할 점

• 최종적 일관성: 쓰기 모델과 읽기 모델 사이의 동기화 지연을 받아들여야 합니다. 사용자에게 “잠시 뒤에 반영됩니다” 같은 안내가 필요할 수도 있습니다.
• 트랜잭션 관리: 쓰기 모델에 복잡한 트랜잭션이 필요하다면, 다음 글에서 다룰 DDD의 애그리게이트(aggregate) 패턴과 함께 쓰는 것이 효과적입니다.
• 메시지 브로커: 쓰기 모델에서 읽기 모델로 변경 알림을 보낼 때는 7장의 Kafka나 RabbitMQ, AWS SQS/SNS 같은 브로커를 흔히 씁니다. 이때도 저장과 발행 사이의 틈을 outbox로 막고, 받는 쪽에서 이벤트 ID로 중복을 거릅니다.
• 데이터베이스 선택: 쓰기 모델에는 PostgreSQL, MySQL, MongoDB 같은 저장소를, 읽기 모델에는 Elasticsearch, Redis, 읽기 전용 복제본 같은 저장소를 고를 수 있습니다.
• 로그와 추적: CQRS는 시스템을 더 나누므로, 구성 요소 사이의 데이터 흐름과 병목을 추적할 수 있도록 OpenTelemetry 같은 분산 추적을 구성합니다.

CQRS가 모든 애플리케이션에 필요한 것은 아닙니다. 시스템이 단순하거나 읽기와 쓰기의 비율이 크게 다르지 않다면 전통적인 CRUD 모델이 더 단순하고 관리하기 쉽습니다. 높은 확장성, 성능 최적화, 복잡한 도메인 모델이 필요한 큰 시스템에서 CQRS는 구조적인 이점을 주며, NestJS의 모듈 구조와 @nestjs/cqrs가 이를 구현하는 데 도움이 됩니다.

정리

• CQRS는 상태를 바꾸는 명령과 상태를 읽는 조회를 나누어 각자에 맞는 모델을 쓰는 패턴이며, 그 대가로 구조가 복잡해지고 최종적 일관성을 받아들여야 합니다.
• NestJS 12에서는 루트 모듈에 CqrsModule.forRoot()를 한 번 가져오고, 기능 모듈에는 handler를 provider로 등록합니다.
• Command<T>와 Query<T>를 상속하면 CommandBus.execute()와 QueryBus.execute()의 반환 타입이 추론되며, 명령이 결과를 돌려주는 것도 괜찮습니다.
• 이벤트 handler는 비동기로 실행되므로 명령의 반환이 읽기 모델의 반영 완료를 뜻하지 않으며, 처리되지 않은 예외는 UnhandledExceptionBus로 갑니다.
• 읽기 모델은 이벤트 버전과 중복 ID로 순서가 바뀌거나 다시 온 이벤트를 걸러야 하고, 저장과 발행 사이의 틈은 outbox로 막습니다.

연습 문제

1. 같은 이벤트가 다시 오면

두 번의 생성이 성공해 읽기 Map에 key 1과 2가 반영되어 있습니다. 여기에 ID 1의 같은 이벤트를 읽기 handler에 한 번 더 전달하면 Map의 항목 수와 key 1의 값은 어떻게 될까요? 이어서 ID 3을 조회하면 어떤 경로로 가나요? publish()를 호출했다는 사실만으로 DB 커밋, outbox 저장, 정확히 한 번 전달, 읽기 반영 완료를 증명할 수 있을까요?

2. 늦게 도착한 이전 버전

지금의 UserReadModelRepository에 같은 ID로 New를 저장한 뒤 Old를 저장했습니다. 다음 조회의 이름은 무엇일까요? 두 값이 각각 전체 상태를 담은 버전 4와 3의 이벤트라는 정보를 더한다면, 저장할 때 무엇을 검사해야 할까요?

3. 프로세스가 종료된 뒤의 조회

쓰기 저장소에 사용자 생성은 성공했지만, 이벤트 handler가 읽기 모델을 갱신하기 전에 프로세스가 종료되었습니다. 이 글의 메모리 예제와 실제 영속 DB를 쓰는 경우를 나누어, 새 프로세스에서 조회 결과가 없다는 사실만으로 사용자가 없다고 판단할 수 있는지 생각해 보세요.

다음 글 NestJS와 함께하는 DDD에서는 업무 영역을 중심으로 모델을 설계하는 도메인 주도 설계를 NestJS의 모듈 구조에 적용합니다.