Queue jobs (messaging)

This feature copies broker-agnostic bases (Cloudflare Queues, SQS, RabbitMQ, etc.). You implement the infra adapter later; the consume loop and port are ready.

Environment variables

When the feature is installed, the CLI injects into env.ts / .env.example / .env. Defining them in .env is optional — the Zod schema applies defaults if the keys are missing. QueueBase reads these via EnvService (do not pass them in the constructor).

No provider-specific variables (Cloudflare IDs, SQS credentials, etc.) — those belong in your IQueueService implementation.

Create a handler

1. Define the message DTO + RequestValidatorBase.
2. Extend QueueBase<TMessage> and implement processMessage.
3. Pass only queueName in queueOptions — concurrency and delays come from env:

@Injectable()
export class ConsumeExampleQueueHandler extends QueueBase<ExampleMessageDto> {
  constructor(queue: IQueueService, env: EnvService) {
    super(queue, env, {
      loggerName: ConsumeExampleQueueHandler.name,
      validator: ExampleMessageValidator,
      queueOptions: {
        queueName: QueueName.example,
      },
    });
  }

  protected async processMessage(message: ExampleMessageDto): Promise<void> {
    // business logic
  }
}

4. Register the handler in the feature module and call start() during bootstrap (main.ts or an OnModuleInit service).
5. Replace QueueService with a real adapter in InfraModule ({ provide: IQueueService, useClass: MyAdapter }).

Tests

Use QueueFakeService in unit/integration tests: enqueue, push/pull/ack/retry, and setConfigured(false) to simulate an unconfigured queue.