Background jobs. Typed, retried, durable.

Define a job #

defineJob<Payload>(name, handler) returns a typed definition. The handler receives the decoded payload and a ctx carrying ctx.attempt (1 on the first run).

import { createQueue, defineJob } from '@apex-stack/core/server'

const email = defineJob<{ to: string }>('email', async (payload, ctx) => {
  await sendEmail(payload.to)   // ctx.attempt === 1 on the first try
})

Enqueue & process #

Register a job, then enqueue work against it by name. size() reports pending jobs; process() drains everything due and returns a tally. An enqueued job whose handler isn't registered is skipped and left pending for a worker that knows it.

const queue = createQueue()   // memory driver (default)
queue.register(email)

const id = await queue.enqueue('email', { to: 'a@b.c' })
await queue.size()      // → 1 (pending)

await queue.process()   // → { processed: 1, done: 1, retried: 0, failed: 0, skipped: 0 }
await queue.size()      // → 0

Retries & backoff #

A throwing handler is retried with exponential backoff — rescheduled to now + backoffBaseMs · 2attempt — and dead-lettered once attempts reach maxAttempts. A job that isn't due yet simply isn't picked up. Pass delaySeconds to defer a job, and override maxAttempts per enqueue.

const queue = createQueue({ maxAttempts: 3, backoffBaseMs: 100 })

// Attempt 1 fails → retry at +200ms; attempt 2 fails → +400ms; attempt 3 fails → dead-letter.
await queue.enqueue('email', { to: 'a@b.c' })

await queue.enqueue('email', { to: 'a@b.c' }, { delaySeconds: 5 })  // runs no earlier than +5s
await queue.enqueue('email', { to: 'a@b.c' }, { maxAttempts: 1 })  // per-job override

The worker loop #

work() polls process() on a timer and returns a handle; stop() ends the loop and clears the interval.

const handle = queue.work({ intervalMs: 1000 })  // drain due jobs every second
// … on shutdown:
handle.stop()

Durable database driver #

The database driver persists jobs so they survive a restart and can be processed by many workers. Pass a QueueDbHandle (exec/query over ? placeholders) and run the migration once. Table names are validated as identifiers — no injection.

const queue = createQueue({ driver: 'database', handle, table: 'jobs' })

// Columns: id, name, payload, attempts, max_attempts, run_at, status, last_error
await handle.exec(queue.migrationSql())

queue.register(email)
await queue.enqueue('email', { to: 'a@b.c' })
await queue.process()   // same retry/backoff/delay semantics as memory

The server barrel also exports the pieces directly — createQueueMemoryDriver, createQueueDatabaseDriver, and buildQueueMigrationSql — if you want to wire the driver up by hand.