Stone.js - Queue

Agnostic job queue for Stone.js. Dispatch work now or later, process it with a worker, retry with backoff, dead-letter failures. One connection contract, pluggable drivers (memory now, Redis via ioredis, provider queues next), a @JobHandler decorator, and a queueManager injected in the container.

Installation

npm install @stone-js/queue

# for the Redis connection (optional):
npm install ioredis

Peer dependency: @stone-js/core. ioredis is an optional peer, imported lazily only when a Redis connection is used.

Enable it

Declarative (single connection):

import { StoneApp } from '@stone-js/core'
import { Queue } from '@stone-js/queue'

@Queue({ driver: 'redis', url: 'redis://localhost:6379' })
@StoneApp({ name: 'app' })
export class Application {}

Imperative / multi-connection via stone.queue:

import { defineConfig } from '@stone-js/core'

export const AppConfig = defineConfig((blueprint) => blueprint.set('stone.queue', {
  default: 'redis',
  connections: [
    { name: 'redis', driver: 'redis', url: 'redis://localhost:6379', prefix: 'jobs' },
    { name: 'sync', driver: 'memory' }
  ]
}))

Dispatch jobs

Inject the default connection (queue) or the manager (queueManager):

export class OrderService {
  constructor (private readonly queue) {}

  async checkout (order) {
    await this.queue.dispatch('send-receipt', { orderId: order.id }, { delay: 5, maxAttempts: 3, backoff: 10 })
  }
}

Handle jobs

import { JobHandler } from '@stone-js/queue'

@JobHandler('send-receipt')
export class SendReceipt {
  constructor (private readonly mailer) {}      // dependency-injected
  async handle (payload: { orderId: string }) { await this.mailer.receipt(payload.orderId) }
}

One class can handle several jobs, one method each, with @OnJob (a name-less @JobHandler() marks the class for scanning):

import { JobHandler, OnJob } from '@stone-js/queue'

@JobHandler()
export class Jobs {
  @OnJob('resize') async resize (payload) { /* … */ }
  @OnJob('purge')  async purge (payload) { /* … */ }
}

Register handlers imperatively too: blueprint.set('stone.queue.handlers', [defineJobHandler('send-receipt', SendReceipt, { isClass: true }) ]).

Process jobs

Run the worker in a long-running process (Node). On serverless, the provider adapter invokes per message instead, routing to the same handlers.

export class Consumer {
  constructor (private readonly worker) {}
  async start () { await this.worker.run({ queues: ['default'], sleep: 1000 }) }
}

The worker reserves each job, runs its handler, and acknowledges it; on failure it retries with linear backoff up to maxAttempts, then dead-letters.

Drivers

| driver | backend | notes | | --- | --- | --- | | memory | in-process | zero-config default; delay, retries, dead-letter; single process | | redis | ioredis | shared/reliable; ready LIST + delayed ZSET + processing LIST | | provider | SQS / Pub/Sub / Azure | coming next, wired to the FaaS adapters |

Documentation

See the official documentation for the full guide.

License

MIT