Do you need guaranteed message delivery (no lost messages)?
  YES --> RabbitMQ (durable queues) or Kafka (persistent log)
  NO  --> TCP (fastest), Redis pub/sub, or NATS

Do you need to process millions of events per second?
  YES --> Kafka or NATS JetStream
  NO  --> RabbitMQ or Redis

Do consumers need to replay past messages?
  YES --> Kafka (log-based, messages persist) or NATS JetStream
  NO  --> RabbitMQ, Redis, or TCP

Are your services in different languages?
  YES --> gRPC (strongly typed contracts, code generation for all languages)
  NO  --> Any transport works, TCP is simplest for NestJS-to-NestJS

Is it IoT or constrained devices?
  YES --> MQTT
  NO  --> Any of the above

npm install @nestjs/microservices

// main.ts (microservice)
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.TCP,
      options: {
        host: 'localhost',
        port: 3001,
      },
    },
  );

  await app.listen();
  console.log('Microservice is listening on port 3001');
}
bootstrap();

// app.controller.ts
import { Controller } from '@nestjs/common';
import { MessagePattern, Payload } from '@nestjs/microservices';

@Controller()
export class AppController {
  // @MessagePattern defines a request-response handler.
  // When a client sends { cmd: 'sum' } with payload [1, 2, 3],
  // this handler runs and returns the result (6) back to the caller.
  @MessagePattern({ cmd: 'sum' })
  accumulate(@Payload() data: number[]): number {
    return data.reduce((a, b) => a + b, 0);
  }

  // @Payload() extracts the data from the incoming message, similar to
  // @Body() in HTTP controllers. The message format depends on the
  // transport layer (TCP, RabbitMQ, Redis, etc.).
  @MessagePattern({ cmd: 'get_user' })
  getUser(@Payload() data: { id: number }) {
    return { id: data.id, name: 'John Doe' };
  }
}

// app.service.ts
import { Injectable, Inject, OnModuleInit } from '@nestjs/common';
import { ClientProxy, ClientProxyFactory, Transport } from '@nestjs/microservices';

@Injectable()
export class AppService implements OnModuleInit {
  private client: ClientProxy;

  constructor() {
    this.client = ClientProxyFactory.create({
      transport: Transport.TCP,
      options: {
        host: 'localhost',
        port: 3001,
      },
    });
  }

  async onModuleInit() {
    await this.client.connect();
  }

  async sum(data: number[]) {
    return this.client.send({ cmd: 'sum' }, data).toPromise();
  }

  async getUser(id: number) {
    return this.client.send({ cmd: 'get_user' }, { id }).toPromise();
  }
}

// main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  // HTTP application
  const app = await NestFactory.create(AppModule);

  // Microservice
  app.connectMicroservice<MicroserviceOptions>({
    transport: Transport.TCP,
    options: {
      host: 'localhost',
      port: 3001,
    },
  });

  await app.startAllMicroservices();
  await app.listen(3000);
  console.log('HTTP server on port 3000, Microservice on port 3001');
}
bootstrap();

// Handler
@MessagePattern({ cmd: 'get_user' })
getUser(@Payload() data: { id: number }) {
  return this.usersService.findOne(data.id);
}

// Client
const user = await this.client.send({ cmd: 'get_user' }, { id: 1 }).toPromise();

// Handler
@EventPattern('user_created')
handleUserCreated(@Payload() data: { id: number; email: string }) {
  console.log('User created:', data);
  // Send welcome email, etc.
}

// Client
this.client.emit('user_created', { id: 1, email: 'user@example.com' });

@Controller()
export class UsersController {
  @MessagePattern({ cmd: 'create_user' })
  async createUser(@Payload() data: CreateUserDto) {
    return this.usersService.create(data);
  }

  @MessagePattern({ cmd: 'get_user' })
  async getUser(@Payload() data: { id: number }) {
    return this.usersService.findOne(data.id);
  }

  @EventPattern('user_updated')
  async handleUserUpdated(@Payload() data: { id: number }) {
    // Handle event
  }
}

npm install @nestjs/microservices amqplib amqp-connection-manager
npm install --save-dev @types/amqplib

// main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.RMQ,
      options: {
        urls: [process.env.RABBITMQ_URL || 'amqp://localhost:5672'],
        queue: 'orders_queue',
        queueOptions: {
          durable: true,
        },
      },
    },
  );

  await app.listen();
  console.log('RabbitMQ microservice is listening');
}
bootstrap();

// app.service.ts
import { Injectable } from '@nestjs/common';
import { ClientProxy, ClientProxyFactory, Transport } from '@nestjs/microservices';

@Injectable()
export class AppService {
  private client: ClientProxy;

  constructor() {
    this.client = ClientProxyFactory.create({
      transport: Transport.RMQ,
      options: {
        urls: [process.env.RABBITMQ_URL || 'amqp://localhost:5672'],
        queue: 'orders_queue',
        queueOptions: {
          durable: true,
        },
      },
    });
  }

  async createOrder(orderData: any) {
    return this.client.send({ cmd: 'create_order' }, orderData).toPromise();
  }

  async publishOrderCreated(order: any) {
    this.client.emit('order_created', order);
  }
}

Producer Service
    ↓
Publish Event → [RabbitMQ Queue]
    ↓
Consumer Service(s) ← Subscribe to Queue
    ↓
Process Event

// main.ts
import { NestFactory } from '@nestjs/core';
import { MicroserviceOptions, Transport } from '@nestjs/microservices';
import { AppModule } from './app.module';

async function bootstrap() {
  const app = await NestFactory.createMicroservice<MicroserviceOptions>(
    AppModule,
    {
      transport: Transport.REDIS,
      options: {
        host: 'localhost',
        port: 6379,
      },
    },
  );

  await app.listen();
}
bootstrap();

this.client = ClientProxyFactory.create({
  transport: Transport.REDIS,
  options: {
    host: 'localhost',
    port: 6379,
  },
});

@EventPattern('order_created')
async handleOrderCreated(@Payload() event: OrderCreatedEvent) {
  // Store event
  await this.eventStore.append('order_created', event);
  
  // Update read model
  await this.updateReadModel(event);
  
  // Publish to other services
  this.client.emit('order_created_notification', event);
}

@MessagePattern({ cmd: 'process_order' })
async processOrder(@Payload() data: any, @Ctx() context: RmqContext) {
  const correlationId = context.getMessage().properties.correlationId;
  
  // Use correlation ID for tracing
  this.logger.log(`Processing order with correlation ID: ${correlationId}`);
  
  return this.ordersService.process(data);
}

// Choreography-based Saga: each service publishes events, next service reacts
// Order Service publishes "OrderCreated"
// Payment Service listens, charges card, publishes "PaymentCompleted" or "PaymentFailed"
// Inventory Service listens to "PaymentCompleted", reserves stock
// If Payment fails, Order Service listens to "PaymentFailed" and cancels the order

// Orchestration-based Saga: a coordinator service manages the flow
@Injectable()
export class OrderSagaOrchestrator {
  async execute(orderData: CreateOrderDto) {
    let orderId: string;
    let paymentId: string;

    try {
      // Step 1: Create order (pending)
      orderId = await this.orderClient.send({ cmd: 'create_order' }, orderData).toPromise();

      // Step 2: Charge payment
      paymentId = await this.paymentClient.send({ cmd: 'charge' }, {
        orderId, amount: orderData.total,
      }).toPromise();

      // Step 3: Reserve inventory
      await this.inventoryClient.send({ cmd: 'reserve' }, {
        orderId, items: orderData.items,
      }).toPromise();

      // Step 4: Confirm order
      await this.orderClient.send({ cmd: 'confirm_order' }, { orderId }).toPromise();

    } catch (error) {
      // Compensation: undo completed steps in reverse order
      if (paymentId) {
        await this.paymentClient.send({ cmd: 'refund' }, { paymentId }).toPromise();
      }
      if (orderId) {
        await this.orderClient.send({ cmd: 'cancel_order' }, { orderId }).toPromise();
      }
      throw error;
    }
  }
}

import { Injectable, NestInterceptor, ExecutionContext, CallHandler } from '@nestjs/common';
import { Observable } from 'rxjs';
import { trace, context } from '@opentelemetry/api';

@Injectable()
export class TracingInterceptor implements NestInterceptor {
  intercept(context: ExecutionContext, next: CallHandler): Observable<any> {
    const tracer = trace.getTracer('nestjs');
    const span = tracer.startSpan('http_request');
    
    return next.handle().pipe(
      tap(() => span.end()),
      catchError((err) => {
        span.recordException(err);
        span.end();
        throw err;
      }),
    );
  }
}

@Injectable()
export class LoggerService {
  log(message: string, correlationId?: string) {
    console.log(JSON.stringify({
      message,
      correlationId,
      timestamp: new Date().toISOString(),
    }));
  }
}

@MessagePattern({ cmd: 'process_order' })
async processOrder(@Payload() data: any) {
  try {
    return await this.ordersService.process(data);
  } catch (error) {
    // Log error
    this.logger.error('Failed to process order', error);
    
    // Emit error event
    this.client.emit('order_processing_failed', {
      orderId: data.id,
      error: error.message,
    });
    
    throw error;
  }
}

import { retry, catchError, delay } from 'rxjs/operators';

this.client.send({ cmd: 'process_order' }, data).pipe(
  // retry(3) will retry up to 3 times on failure.
  // For production, use retryWhen() with exponential backoff:
  // retry 1 after 1s, retry 2 after 2s, retry 3 after 4s.
  retry(3),
  catchError((error) => {
    // All retries exhausted. Log, send to dead-letter queue, or alert.
    this.logger.error(`Failed after 3 retries: ${error.message}`);
    return throwError(() => error);
  }),
).toPromise();

Message arrives
    |
    +--> Process successfully? --> ACK and continue
    |
    +--> Transient error? --> Retry with backoff (max 3 retries)
    |       |
    |       +--> Still failing? --> Move to dead-letter queue + alert
    |
    +--> Permanent error? --> Move to dead-letter queue immediately
    |
    +--> Partial saga failure? --> Execute compensation logic