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) ```typescript 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) ```typescript 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, context)` if handle is a function (`context` is what `.withContext()` attached, on every driver) 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 ```typescript 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) ```typescript 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) ```typescript 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 ```typescript 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) ```typescript 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 ```typescript 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. ```typescript 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) ```typescript 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) ```json { "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. ```typescript 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 ```typescript const unsub = onQueueEvent('job:failed', (payload) => { ... }) const unsub2 = onQueueEvent('*', (event, payload) => { ... }) await emitQueueEvent('job:completed', { jobId, result, duration }) ``` ### Event Types