@nathapp/nestjs-queue

Multi-provider queue abstraction for NestJS — BullMQ, Kafka, and RabbitMQ behind one IQueueProvider interface, with decorator-based processors, idempotency, circuit breaking, and delayed-job scheduling.

Installation

npm install @nathapp/nestjs-queue

Install the client library for whichever provider you use (e.g. bullmq + ioredis, kafkajs, or amqplib) as a peer dependency. This package is part of the @nathapp/nestjs-* peer-layer stack and builds on @nathapp/nestjs-common.

What it provides

• QueueModule — register() / registerAsync() for full processing mode (wires QueueService + ProcessorExplorer for @Processor/@Process decorators); forProducer() / forProducerAsync() for producer-only mode (just QueueService, no decorator scanning).
• QueueService — add, addBulk, process, getJob, getJobs, removeJob, pause, resume, and cleanup methods, delegating to the configured IQueueProvider.
• QueueProviderType enum — BULLMQ, KAFKA, RABBITMQ, CUSTOM.
• Decorators — @Processor(queueName | options) (class) and @Process(name? | options) (method) to declare job handlers; @OnQueueEvent, @BatchProcess for event hooks and batch processing.
• Interfaces — IJob<T>, JobOptions, IQueueProvider, QueueModuleOptions, QueueModuleAsyncOptions, HealthCheckResult, IdempotencyStoreOptions.
• Utilities — DelayedJobScheduler, QueueError, ShutdownOptions, sanitization/validation helpers, an InMemoryIdempotencyStore (under store), and a Redis-backed idempotency store (under utils).
• Circuit breaking — QueueModuleOptions.circuitBreaker (global) and JobOptions.circuitBreaker (per-processor, via @Processor) accept a circuit-breaker config object to guard against cascading failures. The breaker implementation and its option/state/metrics types are internal and not part of the package's exported surface.

Provider implementations (BullMQ/Kafka/RabbitMQ) are lazy-loaded internally and are not part of the package's static export surface — select one via provider: QueueProviderType.<X> in the module options.

Usage

Register a queue and declare a processor:

import { Module } from '@nestjs/common';
import { QueueModule, Processor, Process, IJob } from '@nathapp/nestjs-queue';
import { QueueProviderType } from '@nathapp/nestjs-queue';

@Module({
  imports: [
    QueueModule.register({
      provider: QueueProviderType.BULLMQ,
      options: { connection: { host: 'redis', port: 6379 } },
    }),
  ],
  providers: [EmailProcessor],
})
export class AppModule {}

@Processor('email-queue')
export class EmailProcessor {
  @Process('send')
  async handleSend(job: IJob<{ to: string }>) {
    // ... send email
  }
}

To only enqueue jobs (no local processing), use producer-only mode:

QueueModule.forProducer({
  provider: QueueProviderType.BULLMQ,
  options: { connection: { host: 'redis', port: 6379 } },
});

import { Injectable } from '@nestjs/common';
import { QueueService } from '@nathapp/nestjs-queue';

@Injectable()
export class EmailQueueClient {
  constructor(private readonly queue: QueueService) {}

  enqueue(to: string) {
    return this.queue.add('email-queue', 'send', { to });
  }
}