CQRS

npm install --save @nestjs/cqrs

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

@Module({
  imports: [CqrsModule.forRoot()],
  controllers: [LoansController],
  exports: [Journal],
  providers: [
    LoansRepository,
    LoanViews,
    Journal,
    BorrowBookHandler,
    ReturnBookHandler,
    GetLoanHandler,
    LoanOpenedProjection,
    BookReturnedProjection,
    SendLateNoticeHandler,
    LateReturnSagas,
  ],
})
export class LoansModule {}

import { ConfigModule, ConfigService } from '@nestjs/config';
import { Module } from '@nestjs/common';
import { CqrsModule } from '@nestjs/cqrs';

@Module({
  imports: [
    ConfigModule.forRoot(),
    CqrsModule.forRootAsync({
      imports: [ConfigModule],
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        rethrowUnhandled: config.get<string>('CQRS_RETHROW') === 'true',
      }),
    }),
  ],
})
export class ConfiguredCqrsModule {}

import { Command, Query } from '@nestjs/cqrs';

export class BorrowBookCommand extends Command<{ loanId: string }> {
  constructor(
    public readonly bookId: string,
    public readonly memberId: string,
  ) {
    super();
  }
}

export class ReturnBookCommand extends Command<void> {
  constructor(
    public readonly loanId: string,
    public readonly daysLate = 0,
  ) {
    super();
  }
}

import { CommandHandler, EventPublisher, ICommandHandler } from '@nestjs/cqrs';
import { randomUUID } from 'node:crypto';

@CommandHandler(BorrowBookCommand)
export class BorrowBookHandler implements ICommandHandler<BorrowBookCommand> {
  constructor(
    private readonly repository: LoansRepository,
    private readonly publisher: EventPublisher,
  ) {}

  // ICommandHandler<BorrowBookCommand> forces the return type declared by Command<T>.
  async execute(command: BorrowBookCommand): Promise<{ loanId: string }> {
    const loan = this.publisher.mergeObjectContext(
      Loan.open(randomUUID(), command.bookId, command.memberId),
    );
    this.repository.save(loan);
    loan.commit(); // hands the queued events to the EventBus
    return { loanId: loan.id };
  }
}

import { Body, Controller, Get, HttpCode, Param, Post } from '@nestjs/common';
import { CommandBus, QueryBus } from '@nestjs/cqrs';

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

  @Post()
  borrow(@Body() body: { bookId: string; memberId: string }) {
    // The result type is inferred from Command<{ loanId: string }>.
    return this.commandBus.execute(new BorrowBookCommand(body.bookId, body.memberId));
  }

  @Post(':id/return')
  @HttpCode(204)
  async giveBack(@Param('id') id: string, @Body() body: { daysLate?: number }): Promise<void> {
    await this.commandBus.execute(new ReturnBookCommand(id, body.daysLate));
  }

  @Get(':id')
  find(@Param('id') id: string) {
    return this.queryBus.execute(new GetLoanQuery(id)); // inferred as Promise<LoanView>
  }
}

curl -s -X POST localhost:3000/loans -H 'content-type: application/json' \
  -d '{"bookId":"b1","memberId":"m1"}'

{"loanId":"e368bf64-f4ba-4a13-b20c-d9e57a47723e"}

{"message":"Already returned","error":"Conflict","statusCode":409}

import { IQueryHandler, Query, QueryHandler } from '@nestjs/cqrs';

export class GetLoanQuery extends Query<LoanView> {
  constructor(public readonly loanId: string) {
    super();
  }
}

@QueryHandler(GetLoanQuery)
export class GetLoanHandler implements IQueryHandler<GetLoanQuery> {
  constructor(private readonly views: LoanViews) {}

  async execute({ loanId }: GetLoanQuery): Promise<LoanView> {
    const view = this.views.views.get(loanId);
    if (!view) throw new NotFoundException(`Loan ${loanId} not found`);
    return view;
  }
}

{"loanId":"e368bf64-f4ba-4a13-b20c-d9e57a47723e","bookId":"b1","memberId":"m1","status":"open"}

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

constructor(private eventBus: EventBus) {
  this.eventBus.pipe(takeUntil(this.destroy$)).subscribe((event) => {
    // save the event, log it, forward it
  });
}

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

export class LoanOpenedEvent {
  constructor(
    public readonly loanId: string,
    public readonly bookId: string,
    public readonly memberId: string,
  ) {}
}

export class BookReturnedEvent {
  constructor(
    public readonly loanId: string,
    public readonly daysLate: number,
  ) {}
}

import { EventsHandler, IEventHandler } from '@nestjs/cqrs';

@EventsHandler(LoanOpenedEvent)
export class LoanOpenedProjection implements IEventHandler<LoanOpenedEvent> {
  constructor(private readonly views: LoanViews) {}
  handle(event: LoanOpenedEvent): void {
    this.views.views.set(event.loanId, {
      loanId: event.loanId,
      bookId: event.bookId,
      memberId: event.memberId,
      status: 'open',
    });
  }
}

@EventsHandler(BookReturnedEvent)
export class BookReturnedProjection implements IEventHandler<BookReturnedEvent> {
  constructor(private readonly views: LoanViews) {}
  handle(event: BookReturnedEvent): void {
    const view = this.views.views.get(event.loanId);
    if (view) view.status = 'returned';
  }
}

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

export class Loan extends AggregateRoot {
  private returned = false;

  constructor(
    readonly id: string,
    readonly bookId: string,
    readonly memberId: string,
  ) {
    super();
  }

  static open(id: string, bookId: string, memberId: string): Loan {
    const loan = new Loan(id, bookId, memberId);
    loan.apply(new LoanOpenedEvent(id, bookId, memberId)); // queued until commit()
    return loan;
  }

  giveBack(daysLate: number): void {
    if (this.returned) {
      throw new Error(`Loan ${this.id} is already returned`);
    }
    this.returned = true;
    this.apply(new BookReturnedEvent(this.id, daysLate));
  }

  get isReturned(): boolean {
    return this.returned;
  }
}

import { Injectable } from '@nestjs/common';
import { ICommand, IEvent, ofType, Saga } from '@nestjs/cqrs';
import { filter, map, Observable } from 'rxjs';

@Injectable()
export class LateReturnSagas {
  @Saga()
  lateReturn = (events$: Observable<IEvent>): Observable<ICommand> =>
    events$.pipe(
      ofType(BookReturnedEvent),
      filter((event) => event.daysLate > 0),
      map((event) => new SendLateNoticeCommand(event.loanId)),
    );
}

npm install --save @nestjs/workflows

import { Module } from '@nestjs/common';
import { CommandBus } from '@nestjs/cqrs';
import {
  Workflow,
  WorkflowSignal,
  WorkflowsModule,
  type WorkflowContext,
  type WorkflowRunner,
} from '@nestjs/workflows';
import { SignalOn, StartOn, WorkflowsCqrsModule } from '@nestjs/workflows/cqrs';

// The signal carries the payload type to both ends.
export const bookReturned = new WorkflowSignal<BookReturnedEvent>('book.returned');

// Opening a loan starts one durable instance; returning the book wakes it.
@Workflow('loan-due-date')
@StartOn(LoanOpenedEvent, {
  id: (event) => `loan-${event.loanId}`, // a repeated event finds the same instance
  input: (event) => ({ loanId: event.loanId }),
})
@SignalOn(BookReturnedEvent, {
  signal: bookReturned,
  key: (event) => event.loanId,
})
export class LoanDueDateWorkflow implements WorkflowRunner<{ loanId: string }, 'returned' | 'late-notice-sent'> {
  constructor(private readonly commandBus: CommandBus) {}

  async run(ctx: WorkflowContext, { loanId }: { loanId: string }) {
    // Parks the instance in the store for up to 14 days: no memory, timer or worker is held.
    const returned = await ctx.waitForSignal('await-return', bookReturned, {
      key: loanId,
      timeout: '14d',
    });
    if (returned) return 'returned' as const;

    await ctx.step('late-notice', ({ idempotencyKey }) =>
      this.commandBus.execute(new SendLateNoticeCommand(loanId)).then(() => idempotencyKey),
    );
    return 'late-notice-sent' as const;
  }
}

@Module({
  imports: [
    LoansModule, // already imports CqrsModule.forRoot()
    // allowInMemoryStorage is for demos: register a PostgreSQL or MySQL store in production.
    WorkflowsModule.forRoot({ allowInMemoryStorage: true }),
    WorkflowsCqrsModule,
  ],
  providers: [LoanDueDateWorkflow],
})
export class DurableLoansModule {}

import { Injectable, OnModuleDestroy } from '@nestjs/common';
import { EventsHandler, IEventHandler, UnhandledExceptionBus, UnhandledExceptionInfo } from '@nestjs/cqrs';
import { Subject, takeUntil } from 'rxjs';

export class MailServerDownError extends Error {}

// An event handler that fails: the HTTP request that published the event never sees this error.
@EventsHandler(LoanOpenedEvent)
export class WelcomeMailHandler implements IEventHandler<LoanOpenedEvent> {
  handle(event: LoanOpenedEvent): void {
    if (event.memberId === 'flaky') throw new MailServerDownError('SMTP unreachable');
  }
}

@Injectable()
export class FailureReporter implements OnModuleDestroy {
  readonly seen: UnhandledExceptionInfo[] = [];
  private readonly destroy$ = new Subject<void>();

  constructor(unhandledExceptionBus: UnhandledExceptionBus) {
    unhandledExceptionBus
      .pipe(
        takeUntil(this.destroy$),
        UnhandledExceptionBus.ofType(MailServerDownError), // matches the thrown exception with instanceof
      )
      .subscribe((info) => this.seen.push(info));
  }

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

import { Inject, Scope } from '@nestjs/common';
import { REQUEST } from '@nestjs/core';
import { AsyncContext, Command, CommandHandler, ICommandHandler } from '@nestjs/cqrs';

// The "request" of a CQRS handler is whatever you pass as the second argument of execute().
export class MemberContext extends AsyncContext {
  constructor(public readonly memberId: string) {
    super();
  }
}

export class RenewLoanCommand extends Command<{ renewedBy: string; instance: number }> {
  constructor(public readonly loanId: string) {
    super();
  }
}

let instances = 0;

@CommandHandler(RenewLoanCommand, { scope: Scope.REQUEST })
export class RenewLoanHandler implements ICommandHandler<RenewLoanCommand> {
  private readonly instance = ++instances;

  constructor(
    @Inject(REQUEST) private readonly context: MemberContext,
    private readonly journal: Journal,
  ) {}

  async execute({ loanId }: RenewLoanCommand) {
    this.journal.entries.push(`${this.context.memberId} renewed ${loanId}`);
    return { renewedBy: this.context.memberId, instance: this.instance };
  }
}

const [a, b] = await Promise.all([
  commandBus.execute(new RenewLoanCommand('l1'), new MemberContext('ann')),
  commandBus.execute(new RenewLoanCommand('l1'), new MemberContext('bob')),
]);
// a: { renewedBy: 'ann', instance: 1 }   b: { renewedBy: 'bob', instance: 2 }