CQRS

Luồng xử lý của một ứng dụng CRUD (Create, Read, Update, Delete) đơn giản thường được mô tả như sau:

1. Tầng controller xử lý HTTP request và giao việc lại cho tầng service.
2. Tầng service là nơi chứa phần lớn business logic.
3. Service dùng repository / DAO để thay đổi / lưu trữ entity.
4. Entity đóng vai trò là container chứa giá trị, với các setter và getter.

Pattern này thường là đủ dùng cho các ứng dụng nhỏ và vừa, nhưng có thể không phải là lựa chọn tốt nhất cho các ứng dụng lớn, phức tạp hơn. Trong những trường hợp đó, mô hình CQRS (Command and Query Responsibility Segregation - tách trách nhiệm command và query) có thể phù hợp và mở rộng tốt hơn (tùy vào yêu cầu của ứng dụng). Lợi ích của mô hình này bao gồm:

• Separation of concerns. Mô hình tách riêng các operation đọc (read) và ghi (write) thành hai model riêng biệt.
• Khả năng mở rộng (Scalability). Operation đọc và ghi có thể được scale độc lập với nhau.
• Linh hoạt (Flexibility). Mô hình cho phép dùng các data store khác nhau cho đọc và ghi.
• Hiệu năng (Performance). Mô hình cho phép dùng các data store được tối ưu riêng cho đọc và ghi.

Để hỗ trợ mô hình này, Nest cung cấp một CQRS module nhẹ. Chương này mô tả cách sử dụng nó.

Installation

Đầu tiên, cài package cần thiết:

$ npm install --save @nestjs/cqrs

Sau khi cài xong, vào module gốc của ứng dụng (thường là AppModule), và import CqrsModule.forRoot():

import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';

@Module({
  imports: [CqrsModule.forRoot()],
})
export class AppModule {}

Module này nhận một object cấu hình tùy chọn. Các option sau khả dụng:

Commands (lệnh)

Command dùng để thay đổi state của ứng dụng. Chúng nên hướng theo hành động (task-based), thay vì hướng theo dữ liệu (data centric). Khi một command được dispatch, nó sẽ được xử lý bởi Command Handler tương ứng. Handler chịu trách nhiệm cập nhật state của ứng dụng.

@Injectable()
export class HeroesGameService {
  constructor(private commandBus: CommandBus) {}

  async killDragon(heroId: string, killDragonDto: KillDragonDto) {
    return this.commandBus.execute(
      new KillDragonCommand(heroId, killDragonDto.dragonId)
    );
  }
}

@Injectable()
@Dependencies(CommandBus)
export class HeroesGameService {
  constructor(commandBus) {
    this.commandBus = commandBus;
  }

  async killDragon(heroId, killDragonDto) {
    return this.commandBus.execute(
      new KillDragonCommand(heroId, killDragonDto.dragonId)
    );
  }
}

Trong đoạn code trên, ta khởi tạo class KillDragonCommand và truyền nó vào phương thức execute() của CommandBus. Đây là class command được minh họa:

export class KillDragonCommand extends Command<{
  actionId: string // Type này đại diện cho kết quả thực thi command
}> {
  constructor(
    public readonly heroId: string,
    public readonly dragonId: string,
  ) {
    super();
  }
}

export class KillDragonCommand extends Command {
  constructor(heroId, dragonId) {
    this.heroId = heroId;
    this.dragonId = dragonId;
  }
}

Như bạn thấy, class KillDragonCommand kế thừa từ class Command. Class Command là một class tiện ích đơn giản được export từ package @nestjs/cqrs, cho phép bạn định nghĩa kiểu trả về của command. Trong trường hợp này, kiểu trả về là một object có property actionId. Bây giờ, mỗi khi command KillDragonCommand được dispatch, kiểu trả về của phương thức CommandBus#execute() sẽ được suy ra (infer) là Promise<{ actionId: string }>. Điều này hữu ích khi bạn muốn trả về data nào đó từ command handler.

CommandBus đại diện cho một stream chứa các command. Nó chịu trách nhiệm dispatch command tới handler phù hợp. Phương thức execute() trả về một promise, resolve với giá trị mà handler trả về.

Hãy tạo một handler cho command KillDragonCommand.

@CommandHandler(KillDragonCommand)
export class KillDragonHandler implements ICommandHandler<KillDragonCommand> {
  constructor(private repository: HeroesRepository) {}

  async execute(command: KillDragonCommand) {
    const { heroId, dragonId } = command;
    const hero = this.repository.findOneById(+heroId);

    hero.killEnemy(dragonId);
    await this.repository.persist(hero);

    // "ICommandHandler<KillDragonCommand>" buộc bạn phải trả về giá trị khớp với kiểu trả về của command
    return {
      actionId: crypto.randomUUID(), // Giá trị này sẽ được trả về cho phía gọi
    }
  }
}

@CommandHandler(KillDragonCommand)
@Dependencies(HeroesRepository)
export class KillDragonHandler {
  constructor(repository) {
    this.repository = repository;
  }

  async execute(command) {
    const { heroId, dragonId } = command;
    const hero = this.repository.findOneById(+heroId);

    hero.killEnemy(dragonId);
    await this.repository.persist(hero);

    // "ICommandHandler<KillDragonCommand>" buộc bạn phải trả về giá trị khớp với kiểu trả về của command
    return {
      actionId: crypto.randomUUID(), // Giá trị này sẽ được trả về cho phía gọi
    }
  }
}

Handler này lấy entity Hero từ repository, gọi phương thức killEnemy(), sau đó lưu (persist) thay đổi. Class KillDragonHandler implement interface ICommandHandler, yêu cầu phải implement phương thức execute(). Phương thức execute() nhận command object làm argument.

Lưu ý rằng ICommandHandler<KillDragonCommand> buộc bạn phải trả về giá trị khớp với kiểu trả về của command. Trong trường hợp này, kiểu trả về là một object có property actionId. Điều này chỉ áp dụng cho các command kế thừa từ class Command. Nếu không, bạn có thể trả về bất kỳ giá trị gì.

Cuối cùng, đừng quên đăng ký KillDragonHandler như một provider trong module:

providers: [KillDragonHandler];

Queries (truy vấn)

Query dùng để lấy dữ liệu từ state của ứng dụng. Chúng nên hướng theo dữ liệu (data centric), thay vì hướng theo hành động (task-based). Khi một query được dispatch, nó sẽ được xử lý bởi Query Handler tương ứng. Handler chịu trách nhiệm lấy dữ liệu.

QueryBus theo cùng pattern với CommandBus. Query handler nên implement interface IQueryHandler và được đánh dấu bằng decorator @QueryHandler(). Xem ví dụ sau:

export class GetHeroQuery extends Query<Hero> {
  constructor(public readonly heroId: string) {}
}

Tương tự class Command, class Query là một class tiện ích đơn giản được export từ package @nestjs/cqrs, cho phép bạn định nghĩa kiểu trả về của query. Trong trường hợp này, kiểu trả về là một object Hero. Bây giờ, mỗi khi query GetHeroQuery được dispatch, kiểu trả về của phương thức QueryBus#execute() sẽ được suy ra là Promise<Hero>.

Để lấy thông tin hero, ta cần tạo một query handler:

@QueryHandler(GetHeroQuery)
export class GetHeroHandler implements IQueryHandler<GetHeroQuery> {
  constructor(private repository: HeroesRepository) {}

  async execute(query: GetHeroQuery) {
    return this.repository.findOneById(query.heroId);
  }
}

@QueryHandler(GetHeroQuery)
@Dependencies(HeroesRepository)
export class GetHeroHandler {
  constructor(repository) {
    this.repository = repository;
  }

  async execute(query) {
    return this.repository.findOneById(query.hero);
  }
}

Class GetHeroHandler implement interface IQueryHandler, yêu cầu phải implement phương thức execute(). Phương thức execute() nhận query object làm argument, và phải trả về dữ liệu khớp với kiểu trả về của query (trong trường hợp này, một object Hero).

Cuối cùng, đừng quên đăng ký GetHeroHandler như một provider trong module:

providers: [GetHeroHandler];

Bây giờ, để dispatch query, dùng QueryBus:

const hero = await this.queryBus.execute(new GetHeroQuery(heroId)); // "hero" sẽ được tự động suy ra kiểu "Hero"

Events (sự kiện)

Event dùng để thông báo cho các phần khác của ứng dụng về những thay đổi trong state. Chúng được dispatch bởi model hoặc trực tiếp bằng EventBus. Khi một event được dispatch, nó sẽ được xử lý bởi Event Handler tương ứng. Sau đó handler có thể, ví dụ, cập nhật read model.

Để minh họa, hãy tạo một event class:

export class HeroKilledDragonEvent {
  constructor(
    public readonly heroId: string,
    public readonly dragonId: string,
  ) {}
}

export class HeroKilledDragonEvent {
  constructor(heroId, dragonId) {
    this.heroId = heroId;
    this.dragonId = dragonId;
  }
}

Bây giờ, event có thể được dispatch trực tiếp bằng phương thức EventBus.publish(), nhưng ta cũng có thể dispatch chúng từ model. Hãy cập nhật model Hero để dispatch event HeroKilledDragonEvent khi phương thức killEnemy() được gọi.

export class Hero extends AggregateRoot {
  constructor(private id: string) {
    super();
  }

  killEnemy(enemyId: string) {
    // Business logic
    this.apply(new HeroKilledDragonEvent(this.id, enemyId));
  }
}

export class Hero extends AggregateRoot {
  constructor(id) {
    super();
    this.id = id;
  }

  killEnemy(enemyId) {
    // Business logic
    this.apply(new HeroKilledDragonEvent(this.id, enemyId));
  }
}

Phương thức apply() dùng để dispatch event. Nó nhận một event object làm argument. Tuy nhiên, vì model của ta không biết gì về EventBus, ta cần gắn nó vào model. Ta có thể làm điều đó bằng class EventPublisher.

@CommandHandler(KillDragonCommand)
export class KillDragonHandler implements ICommandHandler<KillDragonCommand> {
  constructor(
    private repository: HeroesRepository,
    private publisher: EventPublisher,
  ) {}

  async execute(command: KillDragonCommand) {
    const { heroId, dragonId } = command;
    const hero = this.publisher.mergeObjectContext(
      await this.repository.findOneById(+heroId),
    );
    hero.killEnemy(dragonId);
    hero.commit();
  }
}

@CommandHandler(KillDragonCommand)
@Dependencies(HeroesRepository, EventPublisher)
export class KillDragonHandler {
  constructor(repository, publisher) {
    this.repository = repository;
    this.publisher = publisher;
  }

  async execute(command) {
    const { heroId, dragonId } = command;
    const hero = this.publisher.mergeObjectContext(
      await this.repository.findOneById(+heroId),
    );
    hero.killEnemy(dragonId);
    hero.commit();
  }
}

Phương thức EventPublisher#mergeObjectContext gộp (merge) event publisher vào object được truyền vào. Object này phải implement interface IAggregateRoot (hoặc kế thừa class AggregateRoot). Sau khi merge, object sẽ có thể publish event lên events stream.

Lưu ý rằng trong ví dụ này ta cũng gọi phương thức commit() trên model. Phương thức này dùng để dispatch các event đang chờ (outstanding). Để tự động dispatch event, ta có thể set property autoCommit thành true:

export class Hero extends AggregateRoot {
  constructor(private id: string) {
    super();
    this.autoCommit = true;
  }
}

Trong trường hợp ta muốn merge event publisher không phải vào một object có sẵn, mà vào một class, ta có thể dùng phương thức EventPublisher#mergeClassContext:

const HeroModel = this.publisher.mergeClassContext(Hero);
const hero = new HeroModel('id'); // <-- HeroModel là một class

Bây giờ mọi instance của class HeroModel sẽ có thể publish event mà không cần dùng phương thức mergeObjectContext().

Flexible Aggregate Roots (Aggregate Root linh hoạt)

Class AggregateRoot là một implementation cụ thể mà bạn có thể kế thừa để thêm khả năng event-driven vào domain model của mình. Tuy nhiên, cách này yêu cầu domain entity phải kế thừa trực tiếp AggregateRoot, điều này có thể là một hạn chế nếu ứng dụng của bạn đã có sẵn một cây kế thừa entity (ví dụ: một class Entity cơ sở, hoặc các class cơ sở theo domain riêng như Monster, Vehicle, v.v.).

Để linh hoạt hơn, package @nestjs/cqrs cung cấp ba cách khác nhau để implement aggregate root:

Cách 1: Truyền thống (Kế thừa class)

Đây là cách chuẩn đã minh họa ở các ví dụ trên. Nó phù hợp hoàn hảo cho các trường hợp đơn giản hoặc project mới hoàn toàn (greenfield).

export class Hero extends AggregateRoot {
  constructor(private id: string) {
    super();
  }

  killEnemy(enemyId: string) {
    this.apply(new HeroKilledDragonEvent(this.id, enemyId));
  }
}

Cách 2: Mixin (Cho các cây kế thừa đã có sẵn)

Nếu bạn đã có một class cơ sở và không thể kế thừa trực tiếp AggregateRoot, bạn có thể dùng hàm mixin WithAggregateRoot<EventBase, TBase>(). Điều này cho phép áp dụng hành vi của aggregate root lên bất kỳ class cơ sở có sẵn nào.

abstract class Monster {
  constructor(protected readonly id: string) {}
  abstract roar(): void;
}

export class Dragon extends WithAggregateRoot(Monster) {
  roar(): void {
    console.log('Roarrrr!');
  }

  die(): void {
    this.roar();
    this.apply(new DragonDiedEvent(this.id)); // Giờ đã khả dụng thông qua mixin!
  }
}

abstract class Monster {
  constructor(id) {
    this.id = id;
  }
}

export class Dragon extends WithAggregateRoot(Monster) {
  roar() {
    console.log('Roarrrr!');
  }

  die() {
    this.roar();
    this.apply(new DragonDiedEvent(this.id));
  }
}

Cách 3: Implementation tùy chỉnh

Để có toàn quyền kiểm soát, hoặc nếu bạn muốn giữ domain layer hoàn toàn không phụ thuộc framework, bạn có thể implement trực tiếp interface IAggregateRoot. EventPublisher chấp nhận bất kỳ object nào implement interface này.

export class CustomEntity implements IAggregateRoot {
  private events: IEvent[] = [];

  getUncommittedEvents() {
    return this.events;
  }

  publish(event: IEvent) {
    // logic tùy chỉnh
  }

  commit() {
    // logic tùy chỉnh
  }

  uncommit() {
    // logic tùy chỉnh
  }

  apply(event: IEvent) {
    this.events.push(event);
  }

  loadFromHistory(history: IEvent[]) {
    // logic tùy chỉnh
  }
}

Cả ba cách trên đều hoạt động liền mạch với EventPublisher, vốn chấp nhận bất kỳ object implement interface IAggregateRoot.

Manual event publishing (publish event thủ công)

this.eventBus.publish(new HeroKilledDragonEvent());

Mỗi event có thể có nhiều Event Handler.

@EventsHandler(HeroKilledDragonEvent)
export class HeroKilledDragonHandler implements IEventHandler<HeroKilledDragonEvent> {
  constructor(private repository: HeroesRepository) {}

  handle(event: HeroKilledDragonEvent) {
    // Business logic
  }
}

Tương tự như command và query, đừng quên đăng ký HeroKilledDragonHandler như một provider trong module:

providers: [HeroKilledDragonHandler];

Sagas

Saga là một tiến trình chạy dài (long-running process), lắng nghe event và có thể trigger command mới. Nó thường được dùng để quản lý các workflow phức tạp trong ứng dụng. Ví dụ, khi một user đăng ký, một saga có thể lắng nghe event UserRegisteredEvent và gửi email chào mừng cho user.

Saga là một tính năng cực kỳ mạnh mẽ. Một saga có thể lắng nghe 1..* event. Dùng thư viện RxJS, ta có thể filter, map, fork và merge các event stream để tạo ra các workflow phức tạp. Mỗi saga trả về một Observable tạo ra một command instance. Command này sau đó được CommandBus dispatch bất đồng bộ (asynchronously).

Hãy tạo một saga lắng nghe event HeroKilledDragonEvent và dispatch command DropAncientItemCommand.

@Injectable()
export class HeroesGameSagas {
  @Saga()
  dragonKilled = (events$: Observable<any>): Observable<ICommand> => {
    return events$.pipe(
      ofType(HeroKilledDragonEvent),
      map((event) => new DropAncientItemCommand(event.heroId, fakeItemID)),
    );
  }
}

@Injectable()
export class HeroesGameSagas {
  @Saga()
  dragonKilled = (events$) => {
    return events$.pipe(
      ofType(HeroKilledDragonEvent),
      map((event) => new DropAncientItemCommand(event.heroId, fakeItemID)),
    );
  }
}

Decorator @Saga() đánh dấu phương thức là một saga. Argument events$ là một Observable stream chứa tất cả event. Operator ofType lọc stream theo loại event được chỉ định. Operator map map event thành một command instance mới.

Trong ví dụ này, ta map event HeroKilledDragonEvent thành command DropAncientItemCommand. Command DropAncientItemCommand sau đó được CommandBus tự động dispatch.

Tương tự như query, command và event handler, đừng quên đăng ký HeroesGameSagas như một provider trong module:

providers: [HeroesGameSagas];

Unhandled exceptions (exception không được xử lý)

Event handler được thực thi bất đồng bộ, vì vậy chúng luôn phải xử lý exception đúng cách để tránh ứng dụng rơi vào trạng thái không nhất quán (inconsistent). Nếu một exception không được xử lý, EventBus sẽ tạo một object UnhandledExceptionInfo và đẩy nó vào stream UnhandledExceptionBus. Stream này là một Observable có thể dùng để xử lý các exception không được xử lý.

private destroy$ = new Subject<void>();

constructor(private unhandledExceptionsBus: UnhandledExceptionBus) {
  this.unhandledExceptionsBus
    .pipe(takeUntil(this.destroy$))
    .subscribe((exceptionInfo) => {
      // Xử lý exception tại đây
      // ví dụ gửi tới external service, terminate process, hoặc publish một event mới
    });
}

onModuleDestroy() {
  this.destroy$.next();
  this.destroy$.complete();
}

Để lọc ra các exception, ta có thể dùng operator ofType, như sau:

this.unhandledExceptionsBus
  .pipe(
    takeUntil(this.destroy$),
    UnhandledExceptionBus.ofType(TransactionNotAllowedException),
  )
  .subscribe((exceptionInfo) => {
    // Xử lý exception tại đây
  });

Trong đó TransactionNotAllowedException là exception ta muốn lọc ra.

Object UnhandledExceptionInfo chứa các property sau:

export interface UnhandledExceptionInfo<
  Cause = IEvent | ICommand,
  Exception = any,
> {
  /**
   * Exception đã được throw.
   */
  exception: Exception;
  /**
   * Nguyên nhân của exception (tham chiếu tới event hoặc command).
   */
  cause: Cause;
}

Subscribing to all events (đăng ký lắng nghe toàn bộ event)

CommandBus, QueryBus và EventBus đều là Observable. Điều này có nghĩa ta có thể subscribe vào toàn bộ stream và, ví dụ, xử lý tất cả event. Ví dụ, ta có thể log tất cả event ra console, hoặc lưu chúng vào event store.

private destroy$ = new Subject<void>();

constructor(private eventBus: EventBus) {
  this.eventBus
    .pipe(takeUntil(this.destroy$))
    .subscribe((event) => {
      // Lưu event vào database
    });
}

onModuleDestroy() {
  this.destroy$.next();
  this.destroy$.complete();
}

Request-scoping (phạm vi theo request)

Với những ai xuất thân từ các ngôn ngữ lập trình khác, có thể sẽ bất ngờ khi biết rằng trong Nest, hầu hết mọi thứ được chia sẻ (shared) xuyên suốt các request đến. Điều này bao gồm connection pool tới database, các service singleton với global state, và nhiều hơn nữa. Hãy nhớ rằng Node.js không theo mô hình request/response đa luồng (multi-threaded) stateless, nơi mỗi request được xử lý bởi một thread riêng. Vì vậy, dùng instance singleton là an toàn cho ứng dụng của chúng ta.

Tuy nhiên, có những trường hợp edge case mà một vòng đời (lifetime) theo request cho handler lại là điều mong muốn. Điều này có thể bao gồm các tình huống như caching theo từng request trong ứng dụng GraphQL, request tracking, hoặc multi-tenancy. Bạn có thể tìm hiểu thêm về cách kiểm soát scope tại đây.

Dùng provider có phạm vi request (request-scoped) cùng với CQRS có thể phức tạp vì CommandBus, QueryBus và EventBus đều là singleton. May mắn là, package @nestjs/cqrs đơn giản hóa việc này bằng cách tự động tạo một instance mới của handler request-scoped cho mỗi command, query, hoặc event được xử lý.

Để làm một handler trở thành request-scoped, bạn có thể:

1. Phụ thuộc vào một provider request-scoped.
2. Đặt scope của nó thành REQUEST một cách rõ ràng bằng decorator @CommandHandler, @QueryHandler, hoặc @EventsHandler, như sau:

@CommandHandler(KillDragonCommand, {
  scope: Scope.REQUEST,
})
export class KillDragonHandler {
  // Implementation ở đây
}

Để inject request payload vào bất kỳ provider request-scoped nào, bạn dùng decorator @Inject(REQUEST). Tuy nhiên, bản chất của request payload trong CQRS tùy vào ngữ cảnh - nó có thể là một HTTP request, một scheduled job, hoặc bất kỳ operation nào khác trigger một command.

Payload phải là một instance của một class kế thừa từ AsyncContext (được cung cấp bởi @nestjs/cqrs), đóng vai trò là request context và chứa dữ liệu có thể truy cập xuyên suốt vòng đời request.

import { AsyncContext } from '@nestjs/cqrs';

export class MyRequest extends AsyncContext {
  constructor(public readonly user: User) {
    super();
  }
}

Khi thực thi một command, truyền request context tùy chỉnh làm argument thứ hai cho phương thức CommandBus#execute:

const myRequest = new MyRequest(user);
await this.commandBus.execute(
  new KillDragonCommand(heroId, killDragonDto.dragonId),
  myRequest,
);

Điều này làm cho instance MyRequest khả dụng như provider REQUEST cho handler tương ứng:

@CommandHandler(KillDragonCommand, {
  scope: Scope.REQUEST,
})
export class KillDragonHandler {
  constructor(
    @Inject(REQUEST) private request: MyRequest, // Inject request context
  ) {}

  // Implementation của handler ở đây
}

Bạn có thể áp dụng cách làm tương tự cho query:

const myRequest = new MyRequest(user);
const hero = await this.queryBus.execute(new GetHeroQuery(heroId), myRequest);

Và trong query handler:

@QueryHandler(GetHeroQuery, {
  scope: Scope.REQUEST,
})
export class GetHeroHandler {
  constructor(
    @Inject(REQUEST) private request: MyRequest, // Inject request context
  ) {}

  // Implementation của handler ở đây
}

Đối với event, dù bạn có thể truyền request provider vào EventBus#publish, cách này ít phổ biến hơn. Thay vào đó, hãy dùng EventPublisher để merge request provider vào một model:

const hero = this.publisher.mergeObjectContext(
  await this.repository.findOneById(+heroId),
  this.request, // Inject request context tại đây
);

Các event handler request-scoped lắng nghe những event này sẽ có quyền truy cập request provider.

Saga luôn là instance singleton vì chúng quản lý các tiến trình chạy dài. Tuy nhiên, bạn có thể lấy request provider từ chính các event object:

@Saga()
dragonKilled = (events$: Observable<any>): Observable<ICommand> => {
  return events$.pipe(
    ofType(HeroKilledDragonEvent),
    map((event) => {
      const request = AsyncContext.of(event); // Lấy request context
      const command = new DropAncientItemCommand(event.heroId, fakeItemID);

      AsyncContext.merge(request, command); // Merge request context vào command
      return command;
    }),
  );
}

Ngoài ra, bạn có thể dùng phương thức request.attachTo(command) để gắn request context vào command.

Example

Một ví dụ hoạt động đầy đủ có tại đây.