Queue

The @talosjs/queue component is a background job layer built on BullMQ. You extend the abstract Queue base class, declare the data your jobs carry, and implement a handler that processes each job. Your subclass provides the BullMQ Queue and Worker; the base class wires the common producer helpers (add, addBulk, removeJob, close) and forwards worker lifecycle events to optional hooks on your class.

Producer and consumer in one class

Extending Queue<T, R> gives you typed add, addBulk, removeJob, and close methods, plus a worker that runs your handler once your subclass constructs it. The job data type T (a record of scalars) and the return type R flow through add and handler, keeping producers and consumers in sync. You can implement optional onCompleted, onFailed, onError, and other hooks, and the base class binds them to the worker when your subclass calls registerEvents(). Jobs persist in Redis through BullMQ, so they survive restarts and scale across worker instances. Register a queue class with a decorator and resolve it from the container.

How it works

A queue class extends Queue<T, R>, where T extends Record<string, ScalarType> is the job data and R is the value the handler returns. Inside the constructor you create a BullMQ Queue (the producer) and a Worker (the consumer) over a shared Redis connection, then call registerEvents() to bind your hooks. The producer side enqueues work:

Method	Purpose
`add(name, data, opts?)`	Enqueue one job; returns the created `Job<T, R>`.
`addBulk(jobs)`	Enqueue many jobs at once; returns the created jobs.
`removeJob(id)`	Remove a job by id; returns the number removed.
`close()`	Close the worker and the queue connections.

The consumer side processes work. You implement the abstract handler:

public abstract handler(job: Job<T, R>): QueueHandlerReturnType<R>;

QueueHandlerReturnType<R> is Promise<R> | R, so the handler may be sync or async. The worker invokes it for each job and the returned value becomes the job result (surfaced to onCompleted). Worker lifecycle events are optional hooks on IQueue. Declare any of them and registerEvents() attaches them:

Hook	Fires when
`onActive(job, prev)`	A job moved to active and started processing.
`onCompleted(job, result, prev)`	A job finished successfully; `result` is the handler’s return value.
`onFailed(job, error, prev)`	A job threw during processing (`job` may be undefined).
`onProgress(job, progress)`	A job reported progress via `job.updateProgress`.
`onStalled(jobId, prev)`	A job was detected as stalled and is about to be retried.
`onDrained()`	The queue had no more jobs to process.
`onPaused()` / `onResumed()`	The worker was paused or resumed.
`onClosing(message)` / `onClosed()`	The worker is closing or has closed.
`onReady()`	The worker connected to Redis and is ready.
`onError(error)`	The worker hit an error unrelated to a specific job. Always attach this to avoid unhandled exceptions.

Decorator and usage

`@decorator.queue()`

Registers a queue class with the container. It accepts an optional scope (defaults to singleton). Apply it to a class that extends Queue.

import { decorator, Queue, type ScalarType } from "@talosjs/queue";
import type { Job } from "bullmq";

export type EmailQueueDataType = { to: string; subject: string };
export type EmailQueueReturnType = unknown;

@decorator.queue()
export class EmailQueue extends Queue<EmailQueueDataType, EmailQueueReturnType> {
  // queue and worker are created in the constructor (see the generated stub).

  public async handler(job: Job<EmailQueueDataType, EmailQueueReturnType>): Promise<EmailQueueReturnType> {
    await sendEmail(job.data.to, job.data.subject);
    return job.data;
  }
}

Resolve it from the container, then enqueue jobs on the producer side. The worker (consumer) runs handler for each:

import { container } from "@talosjs/container";
import { EmailQueue } from "@/queues/EmailQueue";

const emails = container.get(EmailQueue);

// Enqueue one job
await emails.add("welcome", { to: "john@example.com", subject: "Welcome" });

// Enqueue many at once
await emails.addBulk([
  { name: "welcome", data: { to: "a@example.com", subject: "Welcome" } },
  { name: "welcome", data: { to: "b@example.com", subject: "Welcome" } },
]);

Inject the queue where you produce jobs:

import { inject } from "@talosjs/container";

export class SignupService {
  constructor(@inject(EmailQueue) private readonly emails: EmailQueue) {}

  public async signup(email: string): Promise<void> {
    await this.emails.add("welcome", { to: email, subject: "Welcome" });
  }
}

Exceptions

The package exports QueueException as the typed error for queue-related failures in your own subclasses. The abstract Queue base class does not throw one on its own; you choose the keys and messages in the subclass that creates the BullMQ connection and worker.

import { QueueException } from "@talosjs/queue";

try {
  const emails = container.get(EmailQueue);
  await emails.add("welcome", { to: email, subject: "Welcome" });
} catch (error) {
  if (error instanceof QueueException) {
    logger.error(`Queue error [${error.key}]: ${error.message}`, error.data);
  } else {
    throw error;
  }
}

Running queues well

Keep job data small and serializable. It’s a record of scalars (string, number, bigint, boolean), so pass ids and look up the rest in the handler rather than embedding large payloads. Make handlers idempotent too: jobs can be retried after a failure or a stall, so processing the same job twice has to be safe. Always attach onError. Without it, worker-level errors go unhandled, so implement it (and usually onFailed) on every queue. Configure attempts with exponential backoff so flaky downstream calls recover instead of dropping the job, and use removeOnComplete and removeOnFail so finished jobs don’t pile up on a shared instance. Match concurrency and limiter to what the handler’s downstream resources can sustain. Finally, call close() on shutdown to drain the worker and release Redis connections cleanly.

CLI command

Scaffold a queue class and its test file with the generator. It writes the class under modules/<module>/src/queues/<Name>Queue.ts and installs @talosjs/queue if it is missing.

# Interactive: prompts for the name
talos queue:create

# Provide the name
talos queue:create --name=Email

# Target a module and overwrite
talos queue:create --name=Email --module=notifications --override

Option	Description	Default
`--name`	Queue class name. The `Queue` suffix is appended automatically.	Prompted if omitted
`--module`	Target module the class is generated into.	`shared`
`--override`	Overwrite an existing class without prompting.	`false`

The generated class builds the BullMQ Queue and Worker from QUEUE_REDIS_URL, registers its event hooks, and leaves a handler for you to implement.

import { AppEnv } from "@talosjs/app-env";
import { inject } from "@talosjs/container";
import { decorator, Queue, QueueException, type ScalarType } from "@talosjs/queue";
import { RedisClient } from "bun";
import { type BunRedisRawClient, createBunRedisClient, type Job, Queue as BullQueue, Worker } from "bullmq";

export type EmailQueueDataType = Record<string, ScalarType>;
export type EmailQueueReturnType = unknown;

@decorator.queue()
export class EmailQueue extends Queue<EmailQueueDataType, EmailQueueReturnType> {
  private readonly name = "Email";
  protected queue: BullQueue<EmailQueueDataType, EmailQueueReturnType>;
  protected worker: Worker<EmailQueueDataType, EmailQueueReturnType>;

  constructor(@inject(AppEnv) private readonly env: AppEnv) {
    super();

    if (!this.env.QUEUE_REDIS_URL) {
      throw new QueueException(
        "Queue Redis URL is required. Please set the QUEUE_REDIS_URL environment variable.",
        "QUEUE_REDIS_URL_REQUIRED",
      );
    }

    const rawClient = new RedisClient(this.env.QUEUE_REDIS_URL);
    const connection = createBunRedisClient(rawClient as BunRedisRawClient);

    this.queue = new BullQueue<EmailQueueDataType, EmailQueueReturnType>(this.name, {
      connection,
      defaultJobOptions: {
        attempts: 3,
        backoff: { type: "exponential", delay: 1000 },
        removeOnComplete: { age: 3600, count: 1000 },
        removeOnFail: { age: 24 * 3600, count: 5000 },
      },
    });

    this.worker = new Worker<EmailQueueDataType, EmailQueueReturnType>(
      this.name,
      (job) => this.handler(job),
      {
        connection,
        concurrency: 10,
        limiter: { max: 100, duration: 1000 },
        stalledInterval: 30_000,
        maxStalledCount: 3,
        lockDuration: 30_000,
      },
    );

    this.registerEvents();
  }

  public async handler(job: Job<EmailQueueDataType, EmailQueueReturnType>): Promise<EmailQueueReturnType> {
    // TODO: Implement the job processing logic for the "Email" queue.
    return job.data;
  }
}

See queue:create for the full command reference.

Use with Claude and Codex

The generator ships a matching queue:create skill. It runs the scaffold, then guides your AI agent through completing the queue: implementing handler, adding typed add/addBulk wrappers for the jobs it carries, and ensuring QUEUE_REDIS_URL is set. Initialize the skills once for your agent.

Claude
Codex

talos agent:skills:create

Then ask Claude in natural language. It maps the request to the generator, runs it, and fills in the implementation:

Prompt

Create a queue that processes outbound emails.

talos agent:skills:create

Then ask Codex in natural language. It maps the request to the generator, runs it, and fills in the implementation:

Prompt

Create a queue that processes outbound emails.

For example, the prompt above maps to queue:create --name=Email, then implements the handler to send each queued email.

Getting Started

AI

Framework

Command Line

Producer and consumer in one class

How it works

Decorator and usage

`@decorator.queue()`

Exceptions

Running queues well

CLI command

Use with Claude and Codex

​Producer and consumer in one class

​How it works

​Decorator and usage

​@decorator.queue()

​Exceptions

​Running queues well

​CLI command

​Use with Claude and Codex

Producer and consumer in one class

How it works

Decorator and usage

`@decorator.queue()`

Exceptions

Running queues well

CLI command

Use with Claude and Codex