Overview

The Queue module runs work outside the request that asked for it. You publish a job with one call, handle it with one decorator, and the module takes care of everything that should not depend on your broker: routing a job to its handler, validating its payload, retrying it, timing it out and reporting what happened.

Four transports ship with it, and they can be mounted side by side:

Installation

$ pnpm add @vercube/queue

$ npm install @vercube/queue

$ bun install @vercube/queue

The transport clients are optional dependencies, so install the one you use:

$ pnpm add bullmq

$ pnpm add amqplib

$ pnpm add kafkajs

MemoryStrategy needs nothing at all.

Quick Start

import { defineConfig } from '@vercube/core';
import { defineQueueStrategy, QueuePlugin } from '@vercube/queue';
import { BullMQStrategy } from '@vercube/queue/strategies/BullMQStrategy';

export default defineConfig({
  plugins: [
    [QueuePlugin, {
      strategies: [
        defineQueueStrategy({
          strategy: BullMQStrategy,
          initOptions: { connection: { host: '127.0.0.1', port: 6379 } },
        }),
      ],
    }],
  ],
});

import { Inject } from '@vercube/di';
import { Consumer, Job } from '@vercube/queue';
import { MailerService } from '../services/MailerService';

@Consumer({ queue: 'emails', concurrency: 5 })
export class EmailConsumer {
  @Inject(MailerService)
  private gMailer!: MailerService;

  @Job('welcome')
  public async welcome(payload: { userId: string }): Promise<void> {
    await this.gMailer.sendWelcome(payload.userId);
  }
}

import { EmailConsumer } from './consumers/EmailConsumer';
import type { Container } from '@vercube/di';

export function setupContainer(container: Container): void {
  container.bind(EmailConsumer);
}

import { Controller, Post } from '@vercube/core';
import { Inject } from '@vercube/di';
import { QueueManager } from '@vercube/queue';

@Controller('/users')
export class UsersController {
  @Inject(QueueManager)
  private gQueue!: QueueManager;

  @Post('/')
  public async create(): Promise<{ ok: true }> {
    const user = await this.gUsers.create();

    await this.gQueue.add({ queue: 'emails', job: 'welcome', payload: { userId: user.id } });

    return { ok: true };
  }
}

That is the whole setup. The response goes out immediately, and the welcome mail is sent by a worker.

How a job travels

1. add() hands the job to the mounted strategy, together with the headers the module needs to route and retry it.
2. The transport delivers it back, possibly in another process, to the strategy consuming that queue.
3. The manager finds the handler registered for the job name, validates the payload if the handler declares a schema, and runs the handler.
4. Returning marks the job done. Throwing marks the attempt failed and applies the retry policy.

Everything in step 3 and 4 behaves identically on every transport, which is the point of the module: switching from MemoryStrategy to BullMQ or Kafka changes where jobs live, not how your code reads.

Retries, delays and timeouts

Every transport declares what it can do natively, and the manager fills in the rest:

@Job('digest', {
  attempts: 3,
  backoff: { type: 'exponential', delay: 1000 },
  timeout: 30_000,
})
public async digest(payload: Digest): Promise<void> {}

• On BullMQ, attempts and backoff belong to the broker, so pass them when adding the job: add({ ..., options: { attempts: 3, backoff: 1000 } }).
• On RabbitMQ, Kafka and memory, the manager retries the job itself by publishing it again with a higher attempt counter, waiting for the backoff first.
• A job that runs out of attempts is reported as failed and handed back to the transport, which is where a RabbitMQ dead letter exchange picks it up.
• timeout is enforced by the manager on every transport. A timed out handler keeps running, it is only no longer waited for.

Type-safe queues

Queue and job names are plain strings by default. Augment the registry and every add() is checked against it, while unregistered queues keep working as before.

declare module '@vercube/queue' {
  namespace QueueTypes {
    interface Registry {
      emails: {
        welcome: { userId: string };
        digest: { userId: string; period: 'daily' | 'weekly' };
      };
    }
  }
}

// checked: the job exists on the queue and the payload matches
await queue.add({ queue: 'emails', job: 'welcome', payload: { userId: '1' } });

// a compile error: 'wellcome' is not a job of 'emails'
await queue.add({ queue: 'emails', job: 'wellcome', payload: { userId: '1' } });

Tracing

The module traces itself. add() opens a producer span and writes the active trace context into the job's headers; the consumer reads those headers back and opens its span as a child of it. A request that queues a job and a worker that runs it half a minute later, in another process, end up in one trace.

Nothing has to be passed around for this. Turn telemetry on and it works:

export default defineConfig({
  telemetry: true,
});

Every attempt reuses the headers of the original publish, so a retry stays in the trace that queued the job rather than starting one of its own. Two counters come with it: vercube.queue.published and vercube.queue.processed, the latter split by outcome (completed, failed, retried, unhandled).

See Telemetry for exporting any of it.

Producers without consumers

A web process usually publishes jobs while a separate worker process handles them. Turn the consumers off where they do not belong, and the same code keeps publishing.

app.addPlugin(QueuePlugin, {
  autoStart: false,
  strategies: [defineQueueStrategy({ strategy: BullMQStrategy, initOptions: { connection } })],
});

What next

• Consumers covers the decorators, the job context, payload validation and how to consume a queue somebody else fills.
• Strategies covers every transport, its options and its limits.
• API is the reference for QueueManager and the types around it.
• Devtools has a Queues panel showing the mounted transports, the counters and what a queue is holding.