Stacks Queue

Stacks Queue System

Key Paths

• Queue package: storage/framework/core/queue/src/
• Configuration: config/queue.ts
• Application jobs: app/Jobs/
• Job model: storage/framework/defaults/app/Models/Job.ts
• Failed job model: storage/framework/defaults/app/Models/FailedJob.ts

Source Files

queue/src/
├── action.ts          # Job class with dispatch/dispatchIf/dispatchAfter/dispatchNow
├── job.ts             # JobBuilder fluent API + job() helper + jobBatch() + runJob()
│                     # also `Jobs` (the JobName registry) and `resolveJobFile`
├── discovery.ts       # Job auto-discovery from app/Jobs/ via Bun.Glob
├── scheduler.ts       # Cron-based job scheduling with overlap prevention
├── worker.ts          # Queue worker (database polling & Redis processing)
├── batch.ts           # Job batching (PendingBatch, DispatchedBatch, Batch)
├── events.ts          # QueueEvents emitter, QueueMetrics, WorkerTracker
├── health.ts          # Queue health checks (checkQueueHealth, HTTP handler)
├── notifications.ts   # Failed job notifications (email/slack/discord/webhook)
├── testing.ts         # FakeQueue, QueueTester, runJob, expectJobToFail
├── utils.ts           # storeJob helper
├── drivers/redis.ts   # RedisQueue (bun-queue wrapper), StacksQueueManager
└── index.ts           # Re-exports everything

Job Class (action.ts)

export class Job {
  name: string
  description: string
  action?: string | Function  // action name or function
  handle?: Function           // direct handler function
  queue?: string              // queue name (default: 'default')
  rate?: string               // cron expression for scheduling
  tries?: number              // max retry attempts
  timeout?: number            // timeout in seconds
  backoff?: number | number[] // backoff delay(s) in seconds
  backoffConfig?: { strategy, initialDelay, factor, maxDelay, jitter }
  enabled?: boolean

  async dispatch(payload?): Promise<void>
  async dispatchIf(condition: boolean, payload?): Promise<void>
  async dispatchUnless(condition: boolean, payload?): Promise<void>
  async dispatchAfter(delaySeconds: number, payload?): Promise<void>
  async dispatchNow(payload?): Promise<void>  // bypasses queue, runs immediately
}

Dispatch Routing

1. Checks isFaked() -- if testing, dispatches to FakeQueue
2. sync driver: calls dispatchNow() (immediate execution)
3. redis driver: creates RedisQueue, calls queue.add() with bun-queue options
4. database driver: inserts into jobs table with JSON payload
5. Fallback: dispatchNow()

dispatchNow() Execution Order

1. If handle is a function: await this.handle(payload)
2. If action is a string: await runAction(this.action) via @stacksjs/actions
3. If action is a function: await this.action()
4. Otherwise: throws error

JobBuilder Fluent API (job.ts)

import { job } from '@stacksjs/queue'

// The name is checked: `job('SendWelcomeEmial')` is a compile error, not a
// queue row no worker can resolve. `dispatch()` does not look the job up - only
// the worker does - so an unresolvable name used to enqueue successfully and
// fail later, out of sight of the caller.
await job('SendWelcomeEmail', { email, name })
  .onQueue('emails')
  .delay(60)           // seconds
  .tries(5)
  .timeout(30)         // seconds
  .backoff([10, 30, 60])
  .withContext({ userId })
  .dispatch()

// Conditional dispatch
await job('ProcessOrder', data).dispatchIf(isValid)
await job('ProcessOrder', data).dispatchUnless(isDuplicate)

// Immediate execution
await job('SendNotification', data).dispatchNow()

The job() helper loads job modules from app/Jobs/{name}.ts via runJob().

runJob(name, options)

Dynamically imports app/Jobs/{name}.ts and executes:

1. jobConfig.handle(payload) if handle is a function
2. runAction(jobConfig.action) if action is a string
3. jobConfig.action() if action is a function
4. jobConfig(payload, context) if default export is a function

jobBatch() Helper

import { jobBatch } from '@stacksjs/queue'
const batch = await jobBatch([SendEmail, ProcessOrder])
  .name('Onboard User')
  .allowFailures()
  .then(async (b) => console.log('All done!'))
  .dispatch()

Job Batching (batch.ts)

PendingBatch (not yet dispatched)

const batch = Batch.create([
  { job: SendEmail, payload: { to: 'a@b.com' } },
  { job: ProcessImage, payload: { id: 1 } },
])
  .name('welcome-batch')
  .onQueue('default')
  .allowFailures()           // continue processing on failure
  .then(async (batch) => {}) // all succeeded (or allowFailures + all done)
  .catch(async (batch, error) => {}) // first failure
  .finally(async (batch) => {})      // always runs
  .progress(async (batch) => {})     // each job completion

const dispatched = await batch.dispatch()

DispatchedBatch (inspectable)

const b = await Batch.find(batchId)  // returns DispatchedBatch | null
await b.progress()        // 0-100 percentage
await b.pendingJobs()     // count
await b.failedJobs()      // count
await b.completedJobs()   // total - pending
await b.failedJobIds()    // string[]
await b.finished()        // boolean (finished_at !== null)
await b.cancelled()       // boolean
await b.hasFailures()     // boolean
await b.cancel()          // prevents remaining jobs, fires finally callbacks
await b.add([moreJobs])   // add jobs to running batch
await b.delete()          // remove batch record + callbacks
await b.fresh()           // get fresh BatchRecord from storage

Batch Static Methods

Batch.create(jobs): PendingBatch
await Batch.find(id): DispatchedBatch | null
await Batch.all(): DispatchedBatch[]
await Batch.prune(olderThanHours = 24): number  // delete finished batches

Batch Storage

• Database: job_batches table with columns: id, name, total_jobs, pending_jobs, failed_jobs, failed_job_ids (JSON), options (JSON), cancelled_at, created_at, finished_at
• Redis: hash at stacks:batch:{id}, set index at stacks:batches
• Callbacks stored in in-memory Map (not serializable -- only work in dispatching process)

Batch Lifecycle

• On job completion: decrements pending_jobs, fires progress callbacks. When pending_jobs === 0: fires then + finally callbacks, emits batch:completed
• On job failure: increments failed_jobs, appends to failed_job_ids. If allowFailures is false: cancels batch immediately. Fires catch callbacks, emits batch:failed
• Worker checks _batchId in payload to skip cancelled batch jobs

Job Discovery (discovery.ts)

const jobs = await discoverJobs()         // scans app/Jobs/**/*.{ts,js} via Bun.Glob
const job = getJob('SendWelcomeEmail')    // by name from registry
const scheduled = getScheduledJobs()      // jobs with rate/schedule
const all = getAllJobs()
await executeJob('SendWelcomeEmail', payload)  // execute by name

Skips .test., .spec., and index files. Supports both class-based (has handle on prototype) and function-based (object with handle method) jobs.

JobRegistry

Global singleton jobRegistry:

• register(job), get(name), all(), byQueue(queue), scheduled(), has(name), clear()

BackoffConfig

interface BackoffConfig {
  strategy?: 'fixed' | 'exponential' | 'linear'
  initialDelay?: number
  factor?: number
  maxDelay?: number
  jitter?: { enabled?: boolean, factor?: number, minDelay?: number, maxDelay?: number }
}

Job Scheduler (scheduler.ts)

Cron-based scheduling with 1-minute check interval.

await startScheduler({ checkInterval: 60000, timezone: 'UTC', preventOverlapping: true })
await stopScheduler()
isSchedulerRunning()
getSchedulerStatus()  // { isRunning, jobCount, jobs: [{ name, schedule, lastRun, nextRun, isRunning }] }
getRegisteredJobs()   // Map<string, ScheduledJobState>
await triggerJob('SendReport')  // manually trigger a scheduled job

Cron Support

• Standard 5-part cron: minute hour day month dayOfWeek
• 6-part cron (with seconds): seconds are stripped, runs at minute granularity
• Predefined: @yearly, @monthly, @weekly, @daily, @hourly, @midnight
• Every expressions: Every.minute, Every.fiveMinutes, Every.hour, Every.day, etc.
• Supports lists (1,2,3), ranges (1-5), steps (*/5, 1-5/2), wildcards (*)

Overlap Prevention

• Global preventOverlapping: true (default) skips if job is still running
• Per-job withoutOverlapping: true on job config

Queue Worker (worker.ts)

await startProcessor('default', { concurrency: 5 })  // start worker
await stopProcessor()                    // graceful shutdown
await executeFailedJobs()                // requeue all failed jobs
await retryFailedJob(id)                 // requeue specific failed job
getActiveJobCount()                      // currently processing count
isWorkerRunning()                        // boolean

Database Worker Details

• Polls jobs table every 1 second (sleep(1000))
• Refreshes queue list every 10 seconds via getAllQueues() (SELECT DISTINCT queue FROM jobs)
• Atomic claim: SELECT pending job + UPDATE with CAS pattern (WHERE reserved_at IS NULL)
  • Increments attempts, sets reserved_at to current timestamp
  • If numUpdatedRows === 0, another worker claimed it -- tries next
• On success: DELETE from jobs, emit job:completed
• On failure: compare attempts vs maxAttempts (from payload options.tries, default 1)
  • Exceeded: move to failed_jobs table (uuid, connection, queue, payload, exception, failed_at), emit job:failed
  • Retrying: release job by setting reserved_at = null, available_at = now + backoff, emit job:retrying
• Backoff: uses options.backoff from payload -- array (indexed by attempt), number (fixed), or 30s default
• Worker ID: worker-{pid}-{timestamp}
• Handles SIGINT / SIGTERM for graceful shutdown

Redis Worker Details

• Uses RedisQueue.process(concurrency, handler) from bun-queue
• Handler receives BunJob objects
• Throws on failure for bun-queue's built-in retry handling
• Keep-alive loop: polls sleep(1000) until workerRunning === false
• Closes queue on shutdown

Job Payload Format (database)

{
  "jobName": "SendWelcomeEmail",
  "payload": { "email": "user@example.com" },
  "options": { "queue": "emails", "tries": 3, "timeout": 30, "backoff": [10, 30, 60] }
}

Queue Events (events.ts)

QueueEvents Class

Custom event emitter (not mitt-based) with Map<QueueEventType, Set<Handler>> + wildcard Set.

const events = getQueueEvents()  // global singleton
const unsub = events.on('job:failed', handler)
const unsub2 = events.onAny((event, payload) => { ... })
events.once('job:completed', handler)
await events.emit('job:completed', payload)
events.off('job:completed')       // remove all handlers for event
events.removeAllListeners()

Convenience Functions

const unsub = onQueueEvent('job:failed', (payload) => { ... })
const unsub2 = onQueueEvent('*', (event, payload) => { ... })
await emitQueueEvent('job:completed', { jobId, result, duration })

Event Types

'job:added' | 'job:processing' | 'job:completed' | 'job:failed' | 'job:retrying' | 'job:stalled' | 'job:progress' | 'queue:paused' | 'queue:resumed' | 'queue:error' | 'worker:started' | 'worker:stopped' | 'batch:added' | 'batch:completed' | 'batch:failed'

QueueEventPayload

interface QueueEventPayload {
  jobId?: string, queueName?: string, jobName?: string, data?: any,
  result?: any, error?: Error, progress?: number, timestamp: number,
  attemptsMade?: number, duration?: number
}

Event-Aware Wrapper

const wrappedHandler = withEvents('emails', originalHandler)
// Automatically emits job:processing, job:completed, job:failed

OnQueueEvent Decorator

Subscribes an instance method on new, with this bound to that instance. Declaring the class subscribes nothing, so the class has to be constructed — and the instance has to stay referenced, because the global emitter holds listeners weakly (a collected listener stops receiving events).

class MyHandler {
  @OnQueueEvent('job:failed')
  handleFailed(payload: QueueEventPayload) { ... }
}

// Required: subscription happens here, and this binding keeps it alive.
export const myHandler = new MyHandler()

// Deterministic teardown (otherwise it lasts as long as the instance does)
getQueueEvents().unsubscribeListener(myHandler)

Rejected with a TypeError at class-definition time: static methods, fields, accessors and whole classes (no instance to bind to — use onQueueEvent(...)), and legacy experimentalDecorators decoration (no construction-time hook).

Listener Introspection

const events = getQueueEvents()
events.subscribeListener(obj, 'job:failed', fn)  // weak, `this` === obj
events.unsubscribeListener(obj)                  // → count removed
events.listenerCount('job:failed')               // live handlers ('*' for wildcards)

QueueMetrics

const metrics = getGlobalMetrics()  // global singleton
metrics.getThroughputPerMinute()    // completions in last 60 seconds
metrics.getAverageProcessingTime()  // avg duration in last 60 seconds
metrics.getMetrics()  // { counts: {added,completed,failed,processing}, averageDuration, recentErrors, throughputPerMinute }
metrics.reset()
metrics.stop()  // unsubscribe from events

Tracks last 1000 completions and last 100 errors in sliding windows.

WorkerTracker

const tracker = getWorkerTracker()  // global singleton (not lazy)
tracker.register(id, queue)
tracker.markActive(id) / tracker.markIdle(id)
tracker.recordCompletion(id) / tracker.recordFailure(id)
tracker.unregister(id)  // marks as 'stopped'
tracker.getAll(): TrackedWorker[]
tracker.clear()

interface TrackedWorker {
  id: string, status: 'active' | 'idle' | 'stopped', queue: string,
  processedCount: number, failedCount: number,
  lastActivityAt: string, startedAt: string
}

Queue Health (health.ts)

const result = await checkQueueHealth({
  maxPendingWarning: 1000,    maxPendingCritical: 5000,
  maxFailedWarning: 10,       maxFailedCritical: 100,
  maxJobAgeWarning: 3600,     maxJobAgeCritical: 86400,    // seconds
  maxErrorRateWarning: 0.1,   maxErrorRateCritical: 0.5,
  queues?: ['default', 'emails']  // filter to specific queues
})
// result: { status: 'healthy'|'degraded'|'unhealthy', queues, workers, metrics, alerts }

• Queries jobs and failed_jobs tables directly
• Groups by queue name, calculates pending/processing/delayed/failed counts
• Oldest job age calculated from created_at
• Error rate = totalFailed / totalJobs
• Worker statuses from getWorkerTracker().getAll()
• Metrics from getGlobalMetrics()
• Returns real TrackedWorker data even if DB query fails

const healthy = await isQueueHealthy()  // returns boolean
const handler = createHealthCheckHandler()  // HTTP handler (200/207/503)

Failed Job Notifications (notifications.ts)

configureFailedJobNotifications({
  channels: ['email', 'slack', 'discord', 'webhook', 'log'],
  email: { to: ['admin@example.com'], from?: string, subject?: string },
  slack: { webhookUrl: '...', channel?: '#alerts', username?: 'Queue Monitor', iconEmoji?: ':warning:' },
  discord: { webhookUrl: '...', username?: 'Queue Monitor', avatarUrl?: string },
  webhook: { url: '...', headers?: Record<string,string>, secret?: 'hmac-key' },
  rateLimit: 100,        // max per hour
  batch: true,           // batch notifications
  batchInterval: 60000,  // ms (default 60s)
  filter: (job) => job.queue !== 'low-priority',
})

• Email: uses @stacksjs/email (mail.send()) with HTML table format
• Slack: sends Block Kit payload (header + sections, max 10 jobs per message)
• Discord: sends embeds (red color, max 10 per message)
• Webhook: JSON payload with HMAC SHA-256 signature in X-Signature header (via crypto.subtle)
• Log: writes to application log via @stacksjs/logging
• Rate limiter resets every 3,600,000ms (1 hour)
• Batch: accumulates jobs, flushes on interval via setTimeout

const notifier = getFailedJobNotifier()  // global singleton
await notifyJobFailed(jobInfo)           // uses global notifier

Queue Testing (testing.ts)

import { fake, restore, runJob as runTestJob, expectJobToFail } from '@stacksjs/queue'

const fakeQueue = fake()  // sets global FakeQueue singleton

// Dispatch goes to fake queue (not real driver)
await job('SendEmail', data).dispatch()
await SendWelcomeEmail.dispatch(data)

// Assertions (throw on failure)
fakeQueue.assertDispatched('SendEmail')
fakeQueue.assertDispatched('SendEmail', (j) => j.data.to === 'test@example.com')
fakeQueue.assertNotDispatched('ProcessPayment')
fakeQueue.assertDispatchedTimes('SendEmail', 3)
fakeQueue.assertNothingDispatched()
fakeQueue.assertPushed('Reminder')
fakeQueue.assertPushedWithDelay('Reminder', 3600)
fakeQueue.assertPushedOn('emails', 'SendEmail')

// Run job synchronously for testing
const result = await runTestJob({ handle: async (data) => 'ok' }, testData)

// Assert job failure
const error = await expectJobToFail(BadJob, data, /expected error/)
// or: await expectJobToFail(BadJob, data, 'substring match')

restore()  // clears FakeQueue singleton, re-enables real queue

QueueTester Class

const tester = createQueueTester()  // auto-calls fake()
tester.dispatch('Job', data).assertDispatched('Job').reset()
tester.cleanup()  // calls restore()

FakeQueue Internals

• dispatchedJobs[] -- recorded dispatches
• pushedJobs[] -- recorded delayed pushes
• processedJobs[] -- simulated processing results
• failedJobs[] -- simulated failures

Redis Driver (drivers/redis.ts)

Wraps bun-queue's Queue class with Stacks-compatible API.

const queue = new RedisQueue<T>('emails', redisConfig)
await queue.add(data, options)          // delay/attempts/priority/timeout/backoff
queue.process(concurrency, handler)     // start processing (one-time call)
await queue.getJob(jobId)
await queue.getJobs('waiting' | 'active' | 'completed' | 'failed' | 'delayed')
await queue.getJobCounts()              // { waiting, active, completed, failed, delayed }
await queue.removeJob(jobId)
await queue.pause() / queue.resume()
await queue.empty() / queue.close()
await queue.getMetrics()
await queue.ping()                      // health check
await queue.scheduleCron({ cron, data, tz?, name? })
await queue.unscheduleCron(jobId)
await queue.getDeadLetterJobs()
await queue.republishDeadLetterJob(jobId)
await queue.clearDeadLetterQueue()
await queue.bulkRemove(jobIds): number
await queue.getClusterInfo()
queue.isLeader(): boolean               // for horizontal scaling
queue.getQueue(): BunQueue<T>           // underlying bun-queue instance
queue.on(event, handler)                // subscribe to bun-queue events

StacksQueueManager

const manager = new StacksQueueManager(connectionConfigs)
manager.queue('emails')     // get or create RedisQueue by name
manager.setDefaultConnection('high-priority')
await manager.closeAll()

RedisQueue Events

jobAdded, jobCompleted, jobFailed, jobProgress, jobActive, jobStalled, jobDelayed, ready, error

bun-queue Re-exports

batch, chain, dispatch, dispatchAfter, dispatchSync, getQueueManager, setQueueManager, QueueManager

Creating a Job

// app/Jobs/SendWelcomeEmail.ts
import { Job } from '@stacksjs/queue'

export default new Job({
  name: 'SendWelcomeEmail',
  description: 'Send welcome email to new user',
  queue: 'emails',
  tries: 3,
  backoff: [10, 30, 60],
  timeout: 30,

  async handle(payload: { email: string; name: string }) {
    console.log(`Sending to ${payload.email}`)
    return { sent: true }
  }
})

Dispatching

import SendWelcomeEmail from '~/app/Jobs/SendWelcomeEmail'

await SendWelcomeEmail.dispatch({ email, name })
await SendWelcomeEmail.dispatchIf(isNewUser, { email, name })
await SendWelcomeEmail.dispatchUnless(isDuplicate, { email, name })
await SendWelcomeEmail.dispatchAfter(60, { email, name })  // 60s delay
await SendWelcomeEmail.dispatchNow({ email, name })        // bypass queue

config/queue.ts

{
  default: 'sync',  // 'sync' | 'database' | 'redis' | 'sqs' | 'memory'
  connections: {
    sync: { driver: 'sync' },
    database: { driver: 'database', table: 'jobs', queue: 'default', retryAfter: 90 },
    redis: {
      driver: 'redis',
      redis: { url, host: 'localhost', port: 6379, password, db: 0 },
      prefix: 'stacks:queue',
      concurrency: 5,
      logLevel: 'info',
      distributedLock: true,
      stalledJobCheckInterval: 30000,
      maxStalledJobRetries: 3,
      limiter: { max: 100, duration: 1000 },  // if QUEUE_RATE_LIMIT_ENABLED
      metrics: { enabled: false, collectInterval: 10000 },
      defaultDeadLetterOptions: { enabled: true, maxRetries: 3, queueSuffix: '-dead-letter' },
      horizontalScaling: { enabled, maxWorkersPerInstance, jobsPerWorker, leaderElection: { heartbeatInterval: 5000, leaderTimeout: 15000 } },
      defaultJobOptions: { attempts: 3, removeOnComplete: true, removeOnFail: false, backoff: { type: 'exponential', delay: 1000 } }
    },
    sqs: { driver: 'sqs', key, secret, prefix, suffix, queue: 'default', region: 'us-east-1' },
    memory: { driver: 'memory', maxSize: 10000 }
  },
  failed: { driver: 'database', table: 'failed_jobs', prefix: 'stacks:failed' },
  batching: { driver: 'redis', prefix: 'stacks:batches' },
  worker: { concurrency: 5, shutdownTimeout: 30000 }
}

Gotchas

• Default driver is sync -- jobs run immediately in the same process (no background processing)
• Database driver polls every 1 second, refreshes queue list every 10 seconds
• Atomic claim prevents race conditions: SELECT + UPDATE WHERE reserved_at IS NULL (CAS pattern)
• Backoff supports: fixed number, array of delays (indexed by attempt), or BackoffConfig object with strategy
• Job batches store callbacks in-memory (Map) -- callbacks only work in the dispatching process, not across restarts
• Redis driver wraps bun-queue -- uses bun-queue Queue class for actual queue management
• Failed job notifications use HMAC SHA-256 (via Web Crypto API) for webhook signatures
• Health check queries jobs and failed_jobs tables directly for real counts
• Rate limiter for notifications resets hourly (3,600,000ms)
• Testing: fake() sets global singleton, restore() clears it -- affects ALL code in the process
• Queue events use a custom QueueEvents class (Map + Set), not mitt
• Metrics track last 1000 completions and 100 errors in sliding windows
• Worker ID format: worker-{pid}-{timestamp}
• The QUEUE_DRIVER env var controls which driver is used (reads via @stacksjs/env)
• dispatchAfter() on sync driver uses setTimeout to delay then executes immediately
• Failed jobs table: uuid, connection ('database'), queue, payload, exception (stack trace), failed_at
• process.on('unhandledRejection') and process.on('uncaughtException') are set at module load in worker.ts