nestjs-rabbitmq

name	nestjs-rabbitmq
description	NestJS + RabbitMQ integration methodology — @golevelup/nestjs-rabbitmq library, exchange patterns, consumer error handling, DLX/retry strategies, health checks, and event naming conventions. Load when integrating RabbitMQ with NestJS.
user-invocable	false

NestJS + RabbitMQ Integration

1. When to Use

Load this skill when:

• Adding RabbitMQ messaging to a NestJS application
• Designing exchange/queue topology for domain events
• Implementing consumer error handling or retry strategies
• Setting up DLX/DLQ dead-letter infrastructure
• Writing RabbitMQ health checks for NestJS Terminus
• Choosing event naming conventions or payload structure
• Debugging consumer requeue loops or connection issues

2. Library Choice

Verdict: @golevelup/nestjs-rabbitmq — production standard (153K weekly downloads, 2.7K GitHub stars).

Capabilities: @RabbitSubscribe, @RabbitRPC, AmqpConnection.publish/request, auto-ack on void return, Nack control, per-channel prefetchCount, consumer batching, custom serializer/deserializer, multi-exchange bindings per handler, amqp-connection-manager reconnection built-in.

Why NOT alternatives

Library	Problem
@nestjs/microservices RMQ transport	Direct exchange only by default; proprietary message envelope breaks interop with non-NestJS consumers; topic/fanout require third-party shims
Raw amqplib	No NestJS DI integration, no lifecycle hooks, no decorators, manual connection management

Dependency chain

@golevelup/nestjs-rabbitmq
  └── amqp-connection-manager  (reconnection, channel pooling)
       └── amqplib              (AMQP 0-9-1 protocol)

3. Module Setup

import { RabbitMQModule } from '@golevelup/nestjs-rabbitmq';
import { MessageHandlerErrorBehavior } from '@golevelup/nestjs-rabbitmq';

@Module({
  imports: [
    RabbitMQModule.forRootAsync(RabbitMQModule, {
      useFactory: (config: ConfigService) => ({
        uri: config.getOrThrow<string>('RABBITMQ_URI'),
        connectionInitOptions: { wait: false }, // non-blocking startup
        defaultSubscribeErrorBehavior: MessageHandlerErrorBehavior.NACK,
        prefetchCount: 10,
        exchanges: [
          { name: 'bikes.topic', type: 'topic' },
          { name: 'orders.topic', type: 'topic' },
          { name: 'notifications.fanout', type: 'fanout' },
        ],
      }),
      inject: [ConfigService],
    }),
  ],
})
export class MessagingModule {}

connectionInitOptions: { wait: false } is mandatory. It enables non-blocking startup via amqp-connection-manager — the app starts without waiting for RabbitMQ and reconnects automatically. Requires a health check that reports degraded state while the connection is pending (see section 10).

4. Exchange Pattern Selection

Need	Exchange Type	Example
Exact 1:1 routing	Direct	Notification channel dispatch
Broadcast to ALL consumers	Fanout	Cache invalidation, audit log
Wildcard/hierarchical routing	Topic	bike.status.changed, order.created.eu
Multi-attribute routing	Headers	Avoid unless truly needed (slowest)

Topic is the default for domain events. Performance overhead vs direct is <5% under 10K bindings — negligible.

Decision tree:

1. Does every bound queue need every message? → Fanout
2. Does routing need wildcards (*, #)? → Topic
3. Is routing a single exact key? → Direct (or topic with no wildcards — equivalent)
4. Routing on multiple header attributes? → Headers (last resort)

Use domain-specific exchanges (bikes.topic, orders.topic) — never one mega-exchange.

5. Publishing

@Injectable()
export class BikeEventsPublisher {
  constructor(private readonly amqp: AmqpConnection) {}

  async publishStatusChanged(bike: Bike, oldStatus: string): Promise<void> {
    await this.amqp.publish('bikes.topic', 'bike.status.changed', {
      eventId: randomUUID(),
      eventType: 'bike.status.changed',
      aggregateId: bike.id,
      version: 1,
      occurredAt: new Date().toISOString(),
      payload: { bikeId: bike.id, oldStatus, newStatus: bike.status },
    }, {
      persistent: true, // survives broker restart
      messageId: randomUUID(), // idempotency key
    });
  }
}

Publisher confirms: enable via defaultPublishOptions: { persistent: true } in module config. Every published message should be persistent and carry a messageId for consumer-side idempotency.

6. Consuming

@Injectable()
export class PricingConsumer {
  @RabbitSubscribe({
    exchange: 'bikes.topic',
    routingKey: 'bike.status.changed',
    queue: 'pricing.recalculate',
    queueOptions: {
      arguments: {
        'x-queue-type': 'quorum',
        'x-dead-letter-exchange': 'dlx.bikes',
        'x-delivery-limit': 3,
      },
    },
  })
  async handleStatusChanged(event: BikeStatusChangedEvent): Promise<void> {
    // void return = auto-ack
    await this.pricingService.recalculate(event.payload.bikeId);
  }
}

Ack semantics (handler return value)

Return	Behavior	Use when
void (nothing)	Auto-ack — message removed from queue	Processing succeeded
new Nack(false)	Nack without requeue — routes to DLX/DLQ	Permanent failure (bad data, business rule violation)
new Nack(true)	Nack with requeue — message returns to queue head	Transient failure (but use sparingly — loop risk)

Import Nack from @golevelup/nestjs-rabbitmq.

7. Error Handling

CRITICAL: The default errorBehavior is REQUEUE. If a handler throws an unhandled exception, the message is requeued indefinitely — creating an infinite loop that saturates the consumer and fills logs. This is the #1 production trap with this library.

MUST override globally in module config:

defaultSubscribeErrorBehavior: MessageHandlerErrorBehavior.NACK

This ensures unhandled exceptions send messages to the DLX instead of requeuing. Without this override, a single poison message blocks the consumer forever.

Per-handler error control

@RabbitSubscribe({
  exchange: 'orders.topic',
  routingKey: 'order.created',
  queue: 'fulfillment.process',
  errorHandler: (channel, msg, error) => {
    logger.error({ error, routingKey: msg.fields.routingKey }, 'Consumer error');
    channel.nack(msg, false, false); // nack, no requeue → DLX
  },
})

Retry counter (manual)

Access via amqpMsg.properties.headers['x-retry-count'] — increment on each retry attempt. Quorum queue x-delivery-limit is the preferred automated alternative.

8. DLX + Retry

Basic pattern: Main → DLX → DLQ

Main Queue ──(nack/reject)──→ DLX ──→ DLQ (dead letter queue)

Every queue MUST declare x-dead-letter-exchange. Without it, nacked messages vanish.

Exponential retry pattern

Main Queue ──(nack)──→ Retry Exchange ──→ Retry Queue (TTL 5s)
                                              │
                                              └──(TTL expires)──→ Main Queue
                                                    │
                                              (after N retries)──→ DLQ

Quorum queue native retry (preferred)

Quorum queues support x-delivery-limit natively — after N redeliveries, the message routes to DLX automatically. No application-level retry counter needed.

queueOptions: {
  arguments: {
    'x-queue-type': 'quorum',
    'x-dead-letter-exchange': 'dlx.bikes',
    'x-delivery-limit': 3, // auto-DLQ after 3 attempts
  },
},

9. Connection Management

• One connection per app instance, multiple channels (channels are lightweight)
• connectionInitOptions: { wait: false } enables non-blocking startup + auto-reconnect via amqp-connection-manager
• Heartbeat: handled automatically by amqp-connection-manager

Exponential backoff with jitter

Prevents thundering herd on broker restart:

const delay = Math.min(2 ** attempt + Math.random(), 30) * 1000;

amqp-connection-manager handles reconnection internally — manual backoff is only needed for custom connection wrappers.

10. Health Checks

MicroserviceHealthIndicator.pingCheck('rabbitmq', { transport: Transport.RMQ }) is the standard NestJS Terminus path, but has a known incompatibility with RabbitMQ 4.1+ (frameMax negotiation bug — nestjs/terminus#2672, Aug 2025).

Custom health indicator (recommended)

@Injectable()
export class RabbitMQHealthIndicator extends HealthIndicator {
  constructor(private readonly amqp: AmqpConnection) {
    super();
  }

  async isHealthy(key: string): Promise<HealthIndicatorResult> {
    try {
      // amqp-connection-manager exposes channel readiness
      const channel = this.amqp.channel;
      if (!channel) {
        return this.getStatus(key, false, { message: 'No channel available' });
      }
      // Lightweight check — assert a known queue exists
      await channel.checkQueue('health.check');
      return this.getStatus(key, true);
    } catch (error) {
      return this.getStatus(key, false, { message: error.message });
    }
  }
}

When connectionInitOptions: { wait: false } is used, the health check MUST report degraded (not crashed) while the connection is still pending. The app can serve non-RMQ requests during this window.

11. Event Naming

Resource	Convention	Example
Exchange	{domain}.{type}	bikes.topic, orders.topic, notifications.fanout
Queue	{service}.{action}	pricing.recalculate, fulfillment.process
DLX	dlx.{domain}	dlx.bikes, dlx.orders
DLQ	dlq.{service}.{action}	dlq.pricing.recalculate
Routing key	{entity}.{event}[.{qualifier}]	bike.status.changed, order.created.eu

Versioning

Simplest: version in routing key — order.created.v2. Consumers bind to the version they support.

Alternative: version in event metadata (for schema registry compatibility).

12. Event Payload Structure

interface DomainEvent<T = unknown> {
  eventId: string;          // UUID — unique per event instance
  eventType: string;        // matches routing key: 'bike.status.changed'
  aggregateId: string;      // entity ID this event belongs to
  version: number;          // schema version (1, 2, ...)
  occurredAt: string;       // ISO 8601 UTC timestamp
  correlationId?: string;   // trace across event chains
  causationId?: string;     // direct parent event ID
  payload: T;               // domain-specific data
}

eventId is for deduplication. messageId (AMQP property) is for transport-level idempotency. Both should be UUIDs. correlationId threads through an entire business flow (e.g., order placement → reconditioning → shipping).

13. Production Checklist

Concern	Setting	Why
Queue type	x-queue-type: quorum	Classic mirrored queues deprecated since RMQ 3.13, removed in 4.0
Dead letters	x-dead-letter-exchange on every queue	Nacked messages must go somewhere
Delivery limit	x-delivery-limit: 3 (quorum)	Auto-DLQ after N attempts
Queue bounds	x-max-length + x-message-ttl + x-overflow: reject-publish	Prevent unbounded growth
Prefetch	prefetchCount: 10 (never 0)	0 = unlimited = memory bomb
Persistence	persistent: true on publish	Survives broker restart
Idempotency	messageId header (UUID)	Consumer-side dedup
Correlation	correlationId in payload	Distributed tracing
Error behavior	defaultSubscribeErrorBehavior: NACK	Prevent requeue loops

14. Testing

Unit tests — mock AmqpConnection

import { createMock } from '@golevelup/ts-jest';

const mockAmqp = createMock<AmqpConnection>();
mockAmqp.publish.mockResolvedValue(undefined);

const module = await Test.createTestingModule({
  providers: [
    BikeEventsPublisher,
    { provide: AmqpConnection, useValue: mockAmqp },
  ],
}).compile();

For the module itself: pass undefined config when NODE_ENV === 'test' to skip connection.

Integration tests — Testcontainers

import { RabbitMQContainer } from '@testcontainers/rabbitmq';

let container: StartedRabbitMQContainer;

beforeAll(async () => {
  container = await new RabbitMQContainer(
    'rabbitmq:3.13-management-alpine',
  ).start();
  // Use container.getAmqpUrl() for module config
}, 30_000);

afterAll(async () => {
  await container.stop();
});

Real containers for integration tests — no in-memory fakes per project CLAUDE.md mandate.

15. Anti-Patterns

Anti-Pattern	Consequence	Fix
Default errorBehavior (REQUEUE)	Infinite requeue loop on poison messages	Set defaultSubscribeErrorBehavior: NACK globally
One mega-exchange for all events	Routing key collisions, impossible to reason about bindings	One exchange per domain: bikes.topic, orders.topic
prefetchCount: 0	Consumer pulls all messages into memory — OOM on backlog	Set prefetchCount: 10 (tune per workload)
Classic mirrored queues	Deprecated in RMQ 3.13, removed in 4.0	Use quorum queues (x-queue-type: quorum)
No x-dead-letter-exchange	Nacked messages vanish silently	Every queue declares its DLX
connectionInitOptions: { wait: true }	App blocks startup waiting for RabbitMQ — cascading failure in orchestrated deploys	Use wait: false + health check
Nack(true) in error handlers	Requeue to head of queue — blocks all other messages	Use Nack(false) → DLX for permanent failures
Publishing without persistent: true	Messages lost on broker restart	Always set persistent
No messageId on published messages	Consumers cannot deduplicate on retry	Include UUID messageId
Terminus MicroserviceHealthIndicator with RMQ 4.1+	frameMax negotiation bug causes false negatives	Custom HealthIndicator using AmqpConnection.channel

16. Cross-References

Asset	Relevance
Project CLAUDE.md (Architecture)	RabbitMQ role: fire-and-forget events, pub/sub, lightweight async jobs
Project CLAUDE.md (Testing)	Real containers mandate — no mocks for infrastructure in integration tests
service-governance rule	Health endpoint, preflight, graceful degradation requirements
portability rule	RABBITMQ_URI via env var, no hardcoded connection strings
logging-hygiene rule	Structured logging for consumer errors, correlation ID propagation
fallback-deprecation rule	Default port/URI fallbacks must follow FIXME protocol
Temporal integration	Temporal handles durable workflows; RabbitMQ handles fire-and-forget events. Complementary, not competing.