@playstack/queues

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

Install

After confirming preview access, install the package at your application's shared Playstack version:

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

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

Define and dispatch a job

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

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

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

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.

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.

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:

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 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 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.

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.

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.

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