---
title: "@playstack/queues"
description: "Typed jobs with progress, scheduling, inspection, cancellation, retries, and NestJS lifecycle bindings."
tags: ["package","infrastructure","queues","bullmq","inngest","cloudflare","free"]
---

{/* package-access:start */}

> **Free.** MIT licensed. Check preview availability before installing. See [package access](/docs/packages#access-policy).

{/* package-access:end */}

`@playstack/queues` defines typed background jobs with enforced application/environment namespaces, runtime payload validation, retry policy, failure hooks, idempotent IDs, and W3C trace propagation.

{/* package-install:start */}

## Install

After confirming [preview access](/docs/packages#access-policy), install the package at your application's shared Playstack version:

```sh
npm install --save-exact @playstack/queues@0.1.0-beta.1
```

Check the peer requirements below before choosing a runtime or provider.

{/* package-install:end */}

## Define and dispatch a job

```ts
import { QueueJob, createQueue } from '@playstack/queues'

class SendWelcomeEmail extends QueueJob<{ userId: string }, void> {
  static readonly jobName = 'emails.send-welcome'
  static readonly defaults = {
    attempts: 5,
    backoff: { type: 'exponential', delay: 1_000 } as const,
  }

  static parse(value: unknown) {
    return welcomePayloadSchema.parse(value)
  }

  async handle({ userId }: { userId: string }) {
    await welcomeMail.send(userId)
  }

  async failed(payload, error, context) {
    await operations.record({ payload, error, context })
  }
}

const emails = createQueue('emails', {
  app: 'billing',
  environment: 'production',
  driver,
  trace: activeTraceProvider,
})

await emails.dispatch(SendWelcomeEmail, { userId }, { jobId: `welcome-${userId}` })
```

Where custom job IDs are supported (for example BullMQ), they are queue-scoped, URL-safe, non-numeric, and contain no colons. Dispatch can report `queued` or `duplicate`; do not infer deduplication for providers that reject custom IDs. Omit `jobId` from the example for Inngest and Cloudflare.

## Queue and dispatch properties

| Property              | Purpose                                                              |
| --------------------- | -------------------------------------------------------------------- |
| Queue `name`          | Logical queue segment.                                               |
| `app` / `environment` | Produce the fixed `{app}:{environment}:{queue}` namespace.           |
| `driver`              | Runtime-neutral enqueue bridge.                                      |
| `trace`               | Optional provider capturing current W3C context.                     |
| Dispatch `jobId`      | Stable idempotency key.                                              |
| Dispatch `trace`      | Explicit context, inherited provider context, or `null` to suppress. |

Payload validation runs before enqueue and again at the worker boundary. Trace metadata stays under envelope `_trace` and is never merged into the typed payload.

## Process jobs

```ts
const processor = createQueueProcessor({
  app: 'billing',
  environment: 'production',
  queue: 'emails',
  jobs: [registerJob(SendWelcomeEmail, () => new SendWelcomeEmail(dependencies))],
  trace: traceRunner,
  onFailureHookError: (error) => errors.captureError(error),
})
```

| Processor property            | Purpose                                                   |
| ----------------------------- | --------------------------------------------------------- |
| `app`, `environment`, `queue` | Rejects deliveries reaching the wrong worker.             |
| `jobs`                        | Unique definitions plus explicit dependency factories.    |
| `trace`                       | Restores W3C metadata around handling.                    |
| `onFailureHookError`          | Isolates errors from the one-time exhausted-failure hook. |

## Report progress and control jobs

Handlers receive a portable progress reporter. It accepts either a percentage or structured `{ completed, total, stage }` data and fails explicitly when a provider cannot persist progress.

```ts
async handle(payload, context) {
  await context.progress.update({
    completed: 28,
    total: 61,
    stage: 'converting',
  })
}
```

Drivers with a portable control plane expose progress subscriptions, interval and cron schedules, bounded job listing, counts, and inactive-job cancellation through `queue.control`. Cancellation reports active and terminal jobs honestly; it does not claim to interrupt an already executing handler.

## BullMQ bridge

`createBullMqQueueDriver` and `createBullMqWorker` accept app, environment, queue, BullMQ connection/options, registered jobs, and worker tuning. The Redis prefix is `{app}:{environment}` and BullMQ queue name is the logical queue. Both expose the exact native Queue or Worker for Queuebert, metrics, pause/resume, and shutdown.

BullMQ `>=5.16.0 <7` is supported. Installation planning validates the application's declared range against that peer contract rather than accepting any existing package with the same name.

`@playstack/nest-queues` adds named queue injection, `@QueueHandler()` discovery, optional worker startup, and application-shutdown handling. Provider-specific concurrency, locking, stalled-job policy, flows, priorities, and rate limiting remain visible BullMQ configuration.

## Inngest bridge

`createInngestQueueDriver` dispatches the portable envelope as a namespaced Inngest event. `createInngestQueueFunction` processes that event with the same registered jobs, payload validation, retry policy, failure hooks, and trace restoration used by other drivers.

```ts
import { Inngest } from 'inngest'
import { createQueue } from '@playstack/queues'
import { createInngestQueueDriver, createInngestQueueFunction } from '@playstack/queues/inngest'

export const inngest = new Inngest({ id: 'billing' })

const driver = createInngestQueueDriver({
  client: inngest,
  app: 'billing',
  environment: 'production',
  queue: 'emails',
})

export const emails = createQueue('emails', {
  app: 'billing',
  environment: 'production',
  driver,
})

export const emailJobs = createInngestQueueFunction({
  client: inngest,
  app: 'billing',
  environment: 'production',
  queue: 'emails',
  jobs: [registerJob(SendWelcomeEmail, () => new SendWelcomeEmail())],
  concurrency: 20,
  throttle: { limit: 100, period: '1m' },
})
```

Serve the generated function from a Next.js App Router endpoint:

```ts
import { serve } from 'inngest/next'
import { emailJobs, inngest } from '@/lib/queues'

export const { GET, POST, PUT } = serve({
  client: inngest,
  functions: [emailJobs],
})
```

NestJS applications using Express mount the same function through `inngest/express`. Inngest supports delayed jobs, up to 21 total attempts, fixed and exponential backoff, and native concurrency and throttling. It does not support portable custom `jobId`, durable progress, or the BullMQ control-plane semantics, so unsupported operations fail before provider I/O.

## Cloudflare Queues bridge

`@playstack/queues/cloudflare` exports `createCloudflareQueueDriver` and `createCloudflareQueueHandler`. Producers use a structural Worker binding; consumers use individual acknowledgements/retries and a required durable terminal-failure callback. The physical queue name is reviewed separately from the logical namespace. Worker integration templates ship with `playstack add queues --adapter=cloudflare --dry-run`; this adapter is not a Nest worker.

Cloudflare supports bounded JSON batches and delayed delivery, but not custom job IDs, progress, schedules, administrative queries or cancellation. At-least-once delivery still requires application idempotency. See [Cloudflare Queues](/integrations/cloudflare-queues) for payload limits, quarantine behavior and deployment notes.

## Testing and boundary

`@playstack/queues/testing` provides a deterministic memory driver. The core does not choose a broker, DI container, concurrency, shutdown order, or monitoring. Applications own Redis or Inngest client lifecycle and should queue only work with explicit retry and terminal-failure policy.

## Existing queues and strict readiness

BullMQ `>=5.16 <7` is supported; selected monitoring integrations retain their own peer restrictions. `assertBullMqRedisNoEviction(queue.client)` is a fail-closed startup/readiness check, distinct from the non-blocking dispatch observer. It fails if the policy is incompatible or cannot be verified.

The optional [Queuebert bridge](/integrations/queuebert) preserves reviewed native queues and processor statistics. Set a per-queue `expectedPrefix` for an existing physical namespace instead of renaming queues. The bridge does not start workers or authorize administrative HTTP. Monitoring compatibility is not payload compatibility: Playstack dispatch wraps data in an envelope, while legacy processors may read raw `job.data`. Coordinate producers, workers and schedulers through an explicit version/drain/cutover/rollback plan. Mixed-payload Redis fixtures cover selected BullMQ 5/6 paths, not an automatic consumer migration.

{/* package-reference:start */}

## API entry points and requirements

Reference snapshot: `@playstack/queues@0.1.0-beta.1`. Import only the entry point your runtime needs. Paths below are relative to the installed package; use **Go to Definition** in your editor to inspect exact parameters, return types and overloads. Do not import the declaration-file paths directly.

| Public entry point | Declaration file |
| --- | --- |
| `@playstack/queues/playstack.integration.json` | No TypeScript declaration (asset or metadata export). |
| `@playstack/queues/cloudflare` | `./dist/cloudflare.d.ts` |
| `@playstack/queues` | `./dist/index.d.ts` |
| `@playstack/queues/errors` | `./dist/errors.d.ts` |
| `@playstack/queues/types` | `./dist/types.d.ts` |
| `@playstack/queues/testing` | `./dist/testing.d.ts` |
| `@playstack/queues/bullmq` | `./dist/bullmq.d.ts` |
| `@playstack/queues/inngest` | `./dist/inngest.d.ts` |
| `@playstack/queues/package.json` | No TypeScript declaration (asset or metadata export). |

Node.js engine requirement: `>=20`. This is not a claim that every entry point works in browsers or Workers.

### Peer dependencies

Keep existing framework versions that satisfy these ranges. Install optional peers only when using the corresponding adapter. The package manager resolves ordinary dependencies separately.

| Peer | Compatible range | When needed |
| --- | --- | --- |
| `bullmq` | `>=5.16.0 <7` | Optional; only for the entry points that use it. |
| `inngest` | `^4.18.1` | Optional; only for the entry points that use it. |
| `ioredis` | `>=5.0.0` | Optional; only for the entry points that use it. |

For a complete first program, start with [Getting started](/docs/getting-started). For API lookup and partial-example conventions, see [Reading the reference](/docs/packages#reading-the-reference). Provider failures, lifecycle requirements and application responsibilities remain described in the guide above; types alone do not establish production safety.

{/* package-reference:end */}