Strategies

A strategy owns the connection to one broker and translates between its vocabulary and the module's job model. Everything else - routing, validation, retries, timeouts, hooks and counters - lives in the manager, so a strategy stays small and every transport behaves the same way.

What each transport can do

no does not mean unsupported: everything the transport cannot do is either emulated by the manager, or documented below as a real limit. The manager only stops emulating retries when the transport owns them, which is how a BullMQ job keeps its attempts in Redis instead of being republished.

Mounting

Every strategy is mounted by name, so an application can talk to several brokers at once.

app.addPlugin(QueuePlugin, {
  strategies: [
    defineQueueStrategy({ strategy: BullMQStrategy, initOptions: { connection } }),
    defineQueueStrategy({ name: 'events', strategy: KafkaStrategy, initOptions: { client, groupId } }),
  ],
});

Each entry declares the options of its own strategy, and defineQueueStrategy() is what makes TypeScript check that pairing. Without it the entry still works, but initOptions goes unchecked: a plain array gives TypeScript nothing to infer the strategy from.

The first mount has no name, which makes it default. Anything that does not name a strategy uses it:

await queue.add({ queue: 'emails', job: 'welcome', payload });                     // default
await queue.add({ strategy: 'events', queue: 'orders', job: 'placed', payload });  // kafka

@Consumer({ queue: 'orders', strategy: 'events' })
export class OrderConsumer {}

MemoryStrategy

Jobs live in the running process and are gone when it stops. This is the strategy for tests, for examples and for work that may be lost on a restart. It is also the reference implementation: delays, priorities and progress behave exactly as documented.

import { MemoryStrategy } from '@vercube/queue/strategies/MemoryStrategy';

await queueManager.mount({ strategy: MemoryStrategy });

It needs no options. For tests it adds one method the other strategies do not have:

const memory = queueManager.getStrategy() as MemoryStrategy;

await queueManager.add({ queue: 'emails', job: 'welcome', payload: { userId: '1' } });
await memory.idle(); // resolves once nothing is waiting, delayed or running

expect(sent).toHaveLength(1);

BullMQStrategy

Redis keeps the jobs and BullMQ owns their lifecycle: attempts, backoff, delays, priorities, progress and the completed and failed sets.

import { BullMQStrategy } from '@vercube/queue/strategies/BullMQStrategy';

await queueManager.mount({
  strategy: BullMQStrategy,
  initOptions: {
    connection: { host: '127.0.0.1', port: 6379 },
    prefix: 'my-app',
    defaultJobOptions: { removeOnComplete: 1000, removeOnFail: 5000 },
  },
});

Jobs are stored as { payload, headers } so the module's headers survive the round trip. A job added by another BullMQ producer, with plain data, is read as the payload itself.

peek() reads the waiting, delayed and failed sets straight from Redis, which makes the failed set browsable with the data and the stack trace of every job that gave up.

RabbitMQStrategy

Jobs are ordinary AMQP messages: the job name travels in the message type property, the module's bookkeeping in the headers, and the payload as JSON. Queues are declared as durable, and messages are published as persistent.

import { RabbitMQStrategy } from '@vercube/queue/strategies/RabbitMQStrategy';

await queueManager.mount({
  strategy: RabbitMQStrategy,
  initOptions: {
    url: 'amqp://localhost',
    queueOptions: { durable: true, deadLetterExchange: 'failed' },
  },
});

Things worth knowing:

• A job whose attempts are exhausted is nacked without requeue, so it lands wherever the queue's dead letter exchange points. Declare one to keep failures instead of dropping them.
• The job name travels in the type property. A message put on the queue by another application has none, so it arrives as unknown: handle it with @AnyJob(), described in consumers.
• Delays are kept by the manager on a timer, not by the broker, because a plain RabbitMQ has no delayed delivery. A process that stops before the timer fires loses the pending retry.
• Each consumed queue gets its own channel, so its prefetch is its own.
• Priorities only take effect on queues declared with maxPriority.
• The queue cannot be peeked: looking at a message means taking delivery of it. peek() returns nothing rather than perturbing your queue, and the devtools panel says so.

KafkaStrategy

A queue is a topic. The job name travels in the x-job header and the payload is JSON.

import { KafkaStrategy } from '@vercube/queue/strategies/KafkaStrategy';

await queueManager.mount({
  strategy: KafkaStrategy,
  initOptions: {
    client: { clientId: 'orders', brokers: ['localhost:9092'] },
    groupId: 'orders-workers',
  },
});

Things worth knowing:

• Kafka commits offsets rather than acknowledging single messages, so a failed job cannot be singled out. With the default skip, the offset moves on and the failure is logged and counted. With crash, the consumer stops and the partition waits, which is the right choice only when no job may ever be skipped.
• Ordering is per partition. Use the key job option to keep related jobs on the same partition: add({ ..., options: { key: order.id } }).
• Counters report how far the consumer group is behind the end of the topic, which needs a groupId.
• publishMany() produces the whole batch in a single request.
• A topic cannot be peeked without joining it with a throwaway consumer group, so peek() returns nothing.

Writing your own strategy

Extend QueueStrategy and implement four methods. Report what the transport can do through capabilities, and the manager fills in the rest.

import { QueueStrategy } from '@vercube/queue';
import type { QueueTypes } from '@vercube/queue';

export interface HttpStrategyOptions {
  endpoint: string;
}

export class HttpStrategy extends QueueStrategy<HttpStrategyOptions> {
  public readonly transport = 'http';

  private fEndpoint: string = '';

  private fSources: Map<string, EventSource> = new Map();

  public override get capabilities(): QueueTypes.Capabilities {
    return { retries: false, delay: false, priority: false, progress: false, stats: false, peek: false };
  }

  public initialize(options: HttpStrategyOptions): void {
    this.fEndpoint = options.endpoint;
  }

  public async publish(request: QueueTypes.PublishRequest): Promise<QueueTypes.JobRef> {
    const response = await fetch(this.fEndpoint, {
      method: 'POST',
      headers: { 'content-type': 'application/json', ...request.headers },
      body: JSON.stringify(request.payload),
    });

    // fetch() resolves for 4xx and 5xx too, and returning a JobRef for one would
    // report a job that was never accepted as published.
    if (!response.ok) {
      throw new Error(`The endpoint refused the job: ${response.status} ${response.statusText}`);
    }

    return { id: response.headers.get('x-id') ?? '', queue: request.queue, job: request.job, strategy: this.transport };
  }

  public async consume(request: QueueTypes.ConsumeRequest): Promise<QueueTypes.ConsumerHandle> {
    const source = new EventSource(`${this.fEndpoint}/${request.queue}`);

    // call request.dispatch() for every job received
    this.fSources.set(request.queue, source);

    return {
      queue: request.queue,
      stop: async () => {
        this.fSources.delete(request.queue);
        source.close();
      },
    };
  }

  public async close(): Promise<void> {
    for (const source of this.fSources.values()) {
      source.close();
    }

    this.fSources.clear();
  }
}

Two rules keep a custom strategy consistent with the built-in ones:

• dispatch() rejecting means the job failed for good. The manager has already run the hooks and applied the retry policy, so put nothing back on the queue: acknowledge it, discard it or dead-letter it.
• A strategy resolved through the container can use @Inject, so a logger or a config service is available like anywhere else.