Microservices Architecture & Patterns

import { INestApplication, Type } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';

// A modular monolith that already answers on a transport: extract it later by moving the module.
export async function startHybrid(module: Type<unknown>, httpPort: number, tcpPort: number): Promise<INestApplication> {
  const app = await NestFactory.create(module);
  app.connectMicroservice<MicroserviceOptions>(
    { transport: Transport.TCP, options: { host: '127.0.0.1', port: tcpPort } },
    { inheritAppConfig: true },
  );
  await app.startAllMicroservices();
  await app.listen(httpPort);
  return app;
}

nest g app catalog
nest g app reviews
nest g app notifications
nest g lib contracts      # prompts for the alias prefix, default @app
nest build notifications  # builds one app plus the libraries it imports
nest start notifications

{
  "$schema": "https://json.schemastore.org/nest-cli",
  "collection": "@nestjs/schematics",
  "sourceRoot": "apps/gateway/src",
  "monorepo": true,
  "root": "apps/gateway",
  "compilerOptions": {
    "deleteOutDir": true,
    "builder": "tsc",
    "tsConfigPath": "apps/gateway/tsconfig.app.json"
  },
  "projects": {
    "catalog": {
      "type": "application",
      "root": "apps/catalog",
      "entryFile": "main",
      "sourceRoot": "apps/catalog/src",
      "compilerOptions": {
        "tsConfigPath": "apps/catalog/tsconfig.app.json"
      }
    },
    "contracts": {
      "type": "library",
      "root": "libs/contracts",
      "entryFile": "index",
      "sourceRoot": "libs/contracts/src",
      "compilerOptions": {
        "tsConfigPath": "libs/contracts/tsconfig.lib.json"
      }
    },
    "gateway": {
      "type": "application",
      "root": "apps/gateway",
      "entryFile": "main",
      "sourceRoot": "apps/gateway/src",
      "compilerOptions": {
        "tsConfigPath": "apps/gateway/tsconfig.app.json"
      }
    },
    "notifications": {
      "type": "application",
      "root": "apps/notifications",
      "entryFile": "main",
      "sourceRoot": "apps/notifications/src",
      "compilerOptions": {
        "tsConfigPath": "apps/notifications/tsconfig.app.json"
      }
    },
    "reviews": {
      "type": "application",
      "root": "apps/reviews",
      "entryFile": "main",
      "sourceRoot": "apps/reviews/src",
      "compilerOptions": {
        "tsConfigPath": "apps/reviews/tsconfig.app.json"
      }
    }
  }
}

// libs/contracts/src/contracts.ts: the only code the services share.
export const Patterns = {
  BookGet: { cmd: 'book.get' },
  BookCreate: { cmd: 'book.create' },
  BookDelete: { cmd: 'book.delete' },
  ReviewList: { cmd: 'review.list' },
  ReviewCreate: { cmd: 'review.create' },
  DigestSchedule: { cmd: 'digest.schedule' },
  DigestCancel: { cmd: 'digest.cancel' },
  ReviewCreated: 'review.created',
} as const;

export interface BookDto {
  id: string;
  title: string;
}

export interface ReviewCreatedEvent {
  eventId: string; // unique per business event, reused on every redelivery
  reviewId: string;
  bookId: string;
  rating: number;
}

export const CATALOG_CLIENT = 'CATALOG_CLIENT';
export const REVIEWS_CLIENT = 'REVIEWS_CLIENT';
export const NOTIFICATIONS_CLIENT = 'NOTIFICATIONS_CLIENT';

import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { Patterns } from '@app/contracts';
import { NotificationsModule } from './notifications.module.js';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(NotificationsModule, {
    transport: Transport.TCP,
    options: { host: '0.0.0.0', port: Number(process.env.PORT ?? 3002) },
  });
  app.enableShutdownHooks();
  await app.listen();
  console.log(Object.keys(Patterns).length);
}
await bootstrap();

import { Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { TypeOrmModule } from '@nestjs/typeorm';

// Each service wires exactly one database, and it is the only service that holds credentials for it.
@Module({
  imports: [
    ConfigModule.forRoot({ isGlobal: true }),
    TypeOrmModule.forRootAsync({
      inject: [ConfigService],
      useFactory: (config: ConfigService) => ({
        type: 'better-sqlite3' as const, // 'postgres' in production, with config.getOrThrow('NOTIFICATIONS_DATABASE_URL')
        database: config.getOrThrow<string>('NOTIFICATIONS_DB_FILE'),
      }),
    }),
  ],
})
export class NotificationsDatabaseModule {}

import { Global, Module } from '@nestjs/common';
import { ConfigModule, ConfigService } from '@nestjs/config';
import { ClientsModule, Transport } from '@nestjs/microservices';
import { CATALOG_CLIENT, NOTIFICATIONS_CLIENT, REVIEWS_CLIENT } from './contracts.js';

function tcpClient(name: string, prefix: string) {
  return {
    imports: [ConfigModule],
    name,
    inject: [ConfigService],
    useFactory: (config: ConfigService) => ({
      transport: Transport.TCP as const,
      options: {
        // Kubernetes: 'catalog.bookshelf.svc.cluster.local'; Compose: 'catalog'
        host: config.getOrThrow<string>(`${prefix}_HOST`),
        port: Number(config.getOrThrow<string>(`${prefix}_PORT`)), // env values are strings
      },
    }),
  };
}

@Global()
@Module({
  imports: [
    ClientsModule.registerAsync([
      tcpClient(CATALOG_CLIENT, 'CATALOG'),
      tcpClient(REVIEWS_CLIENT, 'REVIEWS'),
      tcpClient(NOTIFICATIONS_CLIENT, 'NOTIFICATIONS'),
    ]),
  ],
  exports: [ClientsModule],
})
export class ServiceClientsModule {}

import {
  BadGatewayException,
  Body,
  Controller,
  Get,
  GatewayTimeoutException,
  Inject,
  InternalServerErrorException,
  NotFoundException,
  Param,
  Post,
  ServiceUnavailableException,
} from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom, TimeoutError, timeout } from 'rxjs';
import { BookDto, CATALOG_CLIENT, Patterns, REVIEWS_CLIENT } from './contracts.js';
import { PublishBookSaga } from './publish-book.saga.js';
import { SagaFailedError } from './saga.js';

@Controller()
export class GatewayController {
  constructor(
    @Inject(CATALOG_CLIENT) private readonly catalog: ClientProxy,
    @Inject(REVIEWS_CLIENT) private readonly reviews: ClientProxy,
    private readonly publishBook: PublishBookSaga,
  ) {}

  /** One round trip per service, in parallel, translated into HTTP semantics. */
  @Get('books/:id')
  async getBook(@Param('id') id: string) {
    const [book, reviews] = await Promise.all([
      this.call<BookDto>(this.catalog, Patterns.BookGet, { id }),
      this.call<unknown[]>(this.reviews, Patterns.ReviewList, { bookId: id }),
    ]);
    return { ...book, reviews };
  }

  @Post('reviews')
  addReview(@Body() body: { bookId: string; rating: number }) {
    return this.call(this.reviews, Patterns.ReviewCreate, body);
  }

  @Post('books')
  async publish(@Body() { title }: { title: string }) {
    try {
      return { id: await this.publishBook.publish(title) };
    } catch (error) {
      if (error instanceof SagaFailedError) {
        throw new ServiceUnavailableException({ failedStep: error.failedStep, compensated: error.compensated });
      }
      throw error;
    }
  }

  private async call<T>(client: ClientProxy, pattern: object, data: unknown): Promise<T> {
    try {
      return await firstValueFrom(client.send<T>(pattern, data).pipe(timeout(2000)));
    } catch (error) {
      if (error instanceof TimeoutError) throw new GatewayTimeoutException();
      const status = (error as { status?: number }).status;
      if (status === 404) throw new NotFoundException((error as { message?: string }).message);
      if (typeof status === 'number') throw new BadGatewayException();
      throw new InternalServerErrorException();
    }
  }
}

import { Controller, Module } from '@nestjs/common';
import { MessagePattern, Payload, RpcException } from '@nestjs/microservices';
import { randomUUID } from 'node:crypto';
import { type BookDto, Patterns } from './contracts.js';

@Controller()
export class CatalogController {
  // The catalog owns this data: no other service reads it directly.
  private readonly books = new Map<string, BookDto>([['b1', { id: 'b1', title: 'Dune' }]]);

  @MessagePattern(Patterns.BookGet)
  get(@Payload() { id }: { id: string }): BookDto {
    const book = this.books.get(id);
    if (!book) throw new RpcException({ status: 404, message: `Book ${id} not found` });
    return book;
  }

  @MessagePattern(Patterns.BookCreate)
  create(@Payload() { title }: { title: string }): BookDto {
    const book = { id: randomUUID(), title };
    this.books.set(book.id, book);
    return book;
  }

  @MessagePattern(Patterns.BookDelete)
  delete(@Payload() { id }: { id: string }): { deleted: boolean } {
    return { deleted: this.books.delete(id) };
  }

  size(): number {
    return this.books.size;
  }
}

@Module({ controllers: [CatalogController] })
export class CatalogModule {}

curl -s localhost:3000/books/b1
curl -s -o /dev/null -w '%{http_code}\n' localhost:3000/books/nope

{"id":"b1","title":"Dune","reviews":[{"id":"r1","bookId":"b1","rating":5}]}

import { Injectable, OnApplicationBootstrap, OnApplicationShutdown } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';

/** Registers this instance in Consul's agent HTTP API; Kubernetes needs none of this. */
@Injectable()
export class ConsulRegistration implements OnApplicationBootstrap, OnApplicationShutdown {
  private readonly id: string;

  constructor(private readonly config: ConfigService) {
    this.id = `${config.getOrThrow<string>('SERVICE_NAME')}-${config.getOrThrow<string>('SERVICE_INSTANCE')}`;
  }

  async onApplicationBootstrap(): Promise<void> {
    const consul = this.config.getOrThrow<string>('CONSUL_URL');
    const port = Number(this.config.getOrThrow<string>('SERVICE_PORT'));
    const res = await fetch(`${consul}/v1/agent/service/register`, {
      method: 'PUT',
      headers: { 'content-type': 'application/json' },
      body: JSON.stringify({
        ID: this.id,
        Name: this.config.getOrThrow<string>('SERVICE_NAME'),
        Address: this.config.getOrThrow<string>('SERVICE_HOST'),
        Port: port,
        Check: { TCP: `${this.config.getOrThrow<string>('SERVICE_HOST')}:${port}`, Interval: '10s', DeregisterCriticalServiceAfter: '1m' },
      }),
    });
    if (!res.ok) throw new Error(`Consul registration failed: ${res.status}`);
  }

  async onApplicationShutdown(): Promise<void> {
    const consul = this.config.getOrThrow<string>('CONSUL_URL');
    await fetch(`${consul}/v1/agent/service/deregister/${this.id}`, { method: 'PUT' });
  }
}

import { Controller, Inject, Module } from '@nestjs/common';
import { ConfigModule } from '@nestjs/config';
import { ClientProxy, MessagePattern, Payload } from '@nestjs/microservices';
import { randomUUID } from 'node:crypto';
import { NOTIFICATIONS_CLIENT, Patterns, ReviewCreatedEvent } from './contracts.js';
import { ServiceClientsModule } from './service-clients.module.js';

interface Review {
  id: string;
  bookId: string;
  rating: number;
}

@Controller()
export class ReviewsController {
  private readonly reviews: Review[] = [{ id: 'r1', bookId: 'b1', rating: 5 }];

  constructor(@Inject(NOTIFICATIONS_CLIENT) private readonly bus: ClientProxy) {}

  @MessagePattern(Patterns.ReviewList)
  list(@Payload() { bookId }: { bookId: string }): Review[] {
    return this.reviews.filter((r) => r.bookId === bookId);
  }

  @MessagePattern(Patterns.ReviewCreate)
  create(@Payload() { bookId, rating }: { bookId: string; rating: number }): Review {
    const review = { id: randomUUID(), bookId, rating };
    this.reviews.push(review);
    // Choreography: announce the fact, do not tell anyone what to do about it.
    const event: ReviewCreatedEvent = { eventId: randomUUID(), reviewId: review.id, bookId, rating };
    this.bus.emit(Patterns.ReviewCreated, event).subscribe();
    return review;
  }
}

@Module({ imports: [ConfigModule.forRoot({ isGlobal: true }), ServiceClientsModule], controllers: [ReviewsController] })
export class ReviewsModule {}

export interface SagaStep<C> {
  name: string;
  invoke(context: C): Promise<void>;
  compensate?(context: C): Promise<void>;
}

export class SagaFailedError extends Error {
  constructor(
    readonly failedStep: string,
    readonly compensated: string[],
    readonly compensationFailures: string[],
    cause: unknown,
  ) {
    super(`Saga failed at "${failedStep}"`, { cause });
  }
}

/** Orchestrated saga: run steps in order, on failure compensate completed steps in reverse. */
export async function runSaga<C>(steps: SagaStep<C>[], context: C): Promise<void> {
  const done: SagaStep<C>[] = [];
  for (const step of steps) {
    try {
      await step.invoke(context);
      done.push(step);
    } catch (error) {
      const compensated: string[] = [];
      const failures: string[] = [];
      for (const completed of done.reverse()) {
        if (!completed.compensate) continue;
        try {
          await completed.compensate(context);
          compensated.push(completed.name);
        } catch {
          failures.push(completed.name); // park it for a retry queue or a human
        }
      }
      throw new SagaFailedError(step.name, compensated, failures, error);
    }
  }
}

import { Inject, Injectable } from '@nestjs/common';
import { ClientProxy } from '@nestjs/microservices';
import { firstValueFrom, timeout } from 'rxjs';
import { BookDto, CATALOG_CLIENT, NOTIFICATIONS_CLIENT, Patterns } from './contracts.js';
import { runSaga, SagaStep } from './saga.js';

interface PublishContext {
  title: string;
  bookId?: string;
}

@Injectable()
export class PublishBookSaga {
  constructor(
    @Inject(CATALOG_CLIENT) private readonly catalog: ClientProxy,
    @Inject(NOTIFICATIONS_CLIENT) private readonly notifications: ClientProxy,
  ) {}

  private call<T>(client: ClientProxy, pattern: object, data: unknown): Promise<T> {
    return firstValueFrom(client.send<T>(pattern, data).pipe(timeout(2000)));
  }

  async publish(title: string): Promise<string> {
    const context: PublishContext = { title };
    const steps: SagaStep<PublishContext>[] = [
      {
        name: 'create-book',
        invoke: async (c) => {
          c.bookId = (await this.call<BookDto>(this.catalog, Patterns.BookCreate, { title: c.title })).id;
        },
        compensate: async (c) => {
          await this.call(this.catalog, Patterns.BookDelete, { id: c.bookId });
        },
      },
      {
        name: 'schedule-digest',
        invoke: async (c) => {
          await this.call(this.notifications, Patterns.DigestSchedule, { bookId: c.title === 'Boom' ? 'fail' : c.bookId });
        },
        compensate: async (c) => {
          await this.call(this.notifications, Patterns.DigestCancel, { bookId: c.bookId });
        },
      },
    ];
    await runSaga(steps, context);
    return context.bookId!;
  }
}

{"failedStep":"schedule-digest","compensated":["create-book"]}

import { ConfigService } from '@nestjs/config';
import { RmqOptions, Transport } from '@nestjs/microservices';

export function notificationsRmqOptions(config: ConfigService): RmqOptions {
  return {
    transport: Transport.RMQ,
    options: {
      urls: [config.getOrThrow<string>('RABBITMQ_URL')],
      queue: 'notifications',
      noAck: false, // manual ack: required for dead-lettering
      prefetchCount: 10,
      persistent: true,
      queueOptions: {
        durable: true,
        arguments: {
          'x-dead-letter-exchange': 'bookshelf.dlx',
          'x-dead-letter-routing-key': 'notifications.dead',
        },
      },
    },
  };
}

  @EventPattern<string>('review.moderated')
  async onModerated(@Payload() event: ReviewCreatedEvent, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const message = context.getMessage();
    try {
      if (event.rating < 1) throw new Error('poison message');
      this.mailer.send('author', `Review ${event.reviewId} moderated`);
      channel.ack(message);
    } catch {
      channel.nack(message, false, false); // requeue = false -> dead-letter exchange
    }
  }

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

@Injectable()
export class ProcessedMessages {
  constructor(private readonly dataSource: DataSource) {}

  async init(): Promise<void> {
    await this.dataSource.query(
      `CREATE TABLE IF NOT EXISTS processed_messages (
         consumer TEXT NOT NULL,
         message_id TEXT NOT NULL,
         PRIMARY KEY (consumer, message_id)
       )`,
    );
  }

  /** Resolves true for the first caller of a (consumer, message) pair, false afterwards. */
  async firstTime(consumer: string, messageId: string): Promise<boolean> {
    // Named parameters, rewritten by the driver into its own placeholders: `?` for SQLite, `$1, $2` for PostgreSQL.
    const [sql, parameters] = this.dataSource.driver.escapeQueryWithParameters(
      `INSERT INTO processed_messages (consumer, message_id) VALUES (:consumer, :messageId)
       ON CONFLICT DO NOTHING RETURNING message_id`,
      { consumer, messageId },
    );
    const rows: unknown[] = await this.dataSource.query(sql, parameters);
    return rows.length === 1;
  }
}

import { Controller, Injectable, Logger, Module, OnModuleInit } from '@nestjs/common';
import {
  BaseRpcContext,
  Ctx,
  EventPattern,
  MessagePattern,
  Payload,
  RmqContext,
  RpcException,
} from '@nestjs/microservices';
import { ConfigModule } from '@nestjs/config';
import { DataSource } from 'typeorm';
import { Patterns, type ReviewCreatedEvent } from './contracts.js';
import { ProcessedMessages } from './processed-messages.js';

// With noAck: false on RabbitMQ every message must be acked or nacked, or it stays unacknowledged and the
// prefetch window fills up. Over TCP (tests, local runs) there is nothing to acknowledge, so both are no-ops.
function ack(context: BaseRpcContext): void {
  if (context instanceof RmqContext) context.getChannelRef().ack(context.getMessage());
}

function deadLetter(context: BaseRpcContext): void {
  // requeue = false -> the queue's dead-letter exchange
  if (context instanceof RmqContext) context.getChannelRef().nack(context.getMessage(), false, false);
}

@Injectable()
export class Mailer {
  readonly sent: string[] = [];
  send(to: string, subject: string): void {
    this.sent.push(`${to}: ${subject}`);
  }
}

@Controller()
export class NotificationsController implements OnModuleInit {
  private readonly logger = new Logger(NotificationsController.name);
  readonly digests = new Set<string>();

  constructor(
    private readonly processed: ProcessedMessages,
    private readonly mailer: Mailer,
  ) {}

  onModuleInit() {
    return this.processed.init();
  }

  // Idempotent consumer: at-least-once delivery + a dedupe key = effectively-once effects.
  @EventPattern<string>(Patterns.ReviewCreated)
  async onReviewCreated(@Payload() event: ReviewCreatedEvent, @Ctx() context: BaseRpcContext): Promise<void> {
    try {
      if (!(await this.processed.firstTime('notifications', event.eventId))) {
        this.logger.log(`Skipping duplicate ${event.eventId}`);
        ack(context); // a skipped duplicate is still consumed, or the broker redelivers it again
        return;
      }
      this.mailer.send('author', `New ${event.rating}-star review on ${event.bookId}`);
      ack(context);
    } catch (error) {
      deadLetter(context);
      throw error;
    }
  }

  // Dead-letter handling on RabbitMQ: reject without requeue so the broker routes the message.
  @EventPattern<string>('review.moderated')
  async onModerated(@Payload() event: ReviewCreatedEvent, @Ctx() context: RmqContext) {
    const channel = context.getChannelRef();
    const message = context.getMessage();
    try {
      if (event.rating < 1) throw new Error('poison message');
      this.mailer.send('author', `Review ${event.reviewId} moderated`);
      channel.ack(message);
    } catch {
      channel.nack(message, false, false); // requeue = false -> dead-letter exchange
    }
  }

  // Request-response commands: the reply (a result or an RpcException) is the outcome the saga
  // waits for, so the request is acked either way instead of being dead-lettered.
  @MessagePattern(Patterns.DigestSchedule)
  schedule(@Payload() { bookId }: { bookId: string }, @Ctx() context: BaseRpcContext): { scheduled: string } {
    try {
      if (bookId === 'fail') throw new RpcException({ status: 503, message: 'mail provider down' });
      this.digests.add(bookId);
      return { scheduled: bookId };
    } finally {
      ack(context);
    }
  }

  @MessagePattern(Patterns.DigestCancel)
  cancel(@Payload() { bookId }: { bookId: string }, @Ctx() context: BaseRpcContext): { cancelled: boolean } {
    try {
      return { cancelled: this.digests.delete(bookId) };
    } finally {
      ack(context);
    }
  }
}

@Module({
  imports: [ConfigModule.forRoot({ isGlobal: true })],
  providers: [
    Mailer,
    ProcessedMessages,
    { provide: DataSource, useFactory: async () => new DataSource({ type: 'better-sqlite3', database: ':memory:' }).initialize() },
  ],
  controllers: [NotificationsController],
})
export class NotificationsModule {}

import { INestMicroservice, Type } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';

// apps/notifications/src/main.ts, parameterised so the e2e spec can start it on any port.
export async function startMicroservice(module: Type<unknown>, port: number): Promise<INestMicroservice> {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(module, {
    transport: Transport.TCP,
    options: { host: '127.0.0.1', port },
  });
  app.enableShutdownHooks();
  await app.listen();
  return app;
}

import { type INestMicroservice } from '@nestjs/common';
import { ConfigService } from '@nestjs/config';
import { NestFactory } from '@nestjs/core';
import { type AsyncMicroserviceOptions } from '@nestjs/microservices';
import { notificationsRmqOptions } from './rabbit.options.js';
import { NotificationsModule } from './notifications.service.js';

// apps/notifications/src/main.ts in production: transport options come from ConfigService.
export async function bootstrapNotifications(): Promise<INestMicroservice> {
  const app = await NestFactory.createMicroservice<AsyncMicroserviceOptions>(NotificationsModule, {
    useFactory: (config: ConfigService) => notificationsRmqOptions(config),
    inject: [ConfigService],
  });
  app.enableShutdownHooks();
  await app.listen();
  return app;
}