Queues

Queues complement commands, streams, and subscriptions when you need pull-based workloads (CQRS projections, AI agent pools, delayed jobs). A queue definition describes payload/parameter schemas, lifecycle defaults (visibility timeout, retries, heartbeats), and optional dead-letter routing. Workers lease jobs from a queue bridge, acknowledge progress, and emit telemetry just like commands.

The difference between a subscription and a queue comes down to who controls delivery timing. With subscriptions the broker pushes messages to your handler as they arrive, which is the right model for reactive side effects with bounded retry windows. With queues, your worker pulls the next job when it is ready, which is the right model when the work itself is expensive, long-running, or needs operator-visible backlog management. A checkout command might trigger a paymentReceived event (subscription territory) and separately enqueue a PDF receipt generation job (queue territory) — both happen after the same business event but with very different delivery and scaling characteristics.

Queue bridges are separate from event bridges, which is an intentional design choice. It means you can evolve your async job infrastructure independently of your event-driven messaging. In development or tests you can rely on the in-memory DefaultQueueBridge without any external dependencies, then swap in a durable bridge for production by changing a single constructor call.

CLI-first workflow

npm run add:queue
npm run add:queue-worker

The CLI scaffolds queue definitions, workers, optional producer commands, schemas, and Vitest specs. Run npm run add:queue to define payload/contracts + lifecycle defaults, then npm run add:queue-worker each time you need a new worker profile (sequential, interval, or continuous).

Architectural context

  • Queue definitions live next to commands/streams/subscriptions on the service builder and participate in the same schema + guard system (.canEnqueue mirrors .canInvoke).
  • Queue bridges are independent from event bridges. Mix RabbitMQ/MQTT/NATS/Dapr for push traffic with Redis queues (or the default in-memory bridge) for pull workloads.
  • Security/access control reuse the same guards/hooks as commands; if your command requires scopes or tenant checks, the .canEnqueue declaration enforces them before a payload ever leaves the handler.
  • Inject the queue bridge alongside the event bridge when you instantiate a service:
const eventBridge = new AmqpBridge({ /* ... */ })
const queueBridge = new RedisQueueBridge({ keyPrefix: 'acme:queue:' })

const myService = await myV1Service.getInstance(eventBridge, {
  queueBridge,
  logger,
})
await myService.start()

Leave queueBridge undefined to fall back to the in-memory DefaultQueueBridge.

Lifecycle defaults & safety

Each queue has a lifecycle state machine (pending → inFlight → completed | failed | deadLetter) with robust defaults:

  • visibilityTimeoutMs = 15m, heartbeatIntervalMs = 5m (renew leases automatically, but override for long jobs).
  • maxAttempts = 10, exponential backoff with jitter, and provider-specific DLQ naming (override per queue if needed).
  • Workers call context.job.complete, retry, extendLease, or moveToDeadLetter so no job stays stuck even if a worker crashes.

Override any of these via .setLifecycleConfig(...) on the queue builder; defaults keep things safe for most deployments.

Async HTTP endpoints

Commands exposed with { mode: 'async' } enqueue work, return 202 Accepted, and respond with { queueId, jobId, statusUrl }. A separate status command (usually GET /status/:jobId) reads queue metrics so clients know when to retry or surface errors. Use context.queue.scheduleAt.<queueId>() for delayed starts (cron-style tasks). See Async HTTP exposure for end-to-end examples.

Key topics

Quick facts

  • Queue bridges are independent from event bridges, so you can pair RabbitMQ for events with Redis queues (or vice versa).
  • Default behavior falls back to the in-memory DefaultQueueBridge; inject @purista/redis-queue-bridge (or future adapters) for production durability.
  • .canEnqueue('queueId', payloadSchema, parameterSchema) generates typed helpers (context.queue.enqueue.queueId) for commands, streams, and subscriptions.
  • Queue test helpers/mocks run on the default bridge for deterministic unit tests.

When to use queues

  • You need pull-based scaling (workers decide when to fetch jobs).
  • Clients must receive a fast response while heavy work continues asynchronously (202 Accepted pattern).
  • You implement CQRS write → async projection flows or AI agent pools that process one task at a time.

Common pitfalls

  • Skipping .canEnqueue(...) so enqueue attempts throw at runtime.
  • Returning entire payloads instead of a compact { queueId, jobId, statusUrl } document.
  • Forgetting to configure dead-letter routing / retry budgets.
  • Expecting queue bridges to reuse event bridge settings (they are independent resources).

Checklist

  • Queue payload/parameter schemas defined and validated.
  • .canEnqueue declared for every producer (command/subscription/stream).
  • At least one worker registered with the desired mode (sequential, interval, or continuous).
  • Queue bridge (default vs Redis) chosen intentionally and injected where needed.
  • Status endpoint implemented for async HTTP exposure.
  • Tests cover enqueue + worker flow (use DefaultQueueBridge for deterministic unit tests).