Queue Jobs

Everything about defining work: schema-defined jobs, typed handlers, the retry policy that kicks in when a handler throws, recurring triggers via cron, one-shot and delayed runs, and how handlers cooperate with timeouts through the abort signal.

#defineJobs

defineJobs takes a kind → TypeBox-schema map. The result is the single source of truth for both inferred payload types — every enqueue(kind, payload) call and every handler is auto-typed, no hand-written job map, no generics — and runtime validation — payloads are validated at enqueue and dequeue, so a buggy caller can't push junk into the durable store and break the worker.

import { defineJobs, t } from '@absolutejs/queue';

const jobs = defineJobs({
  'email.send': t.Object({
    to: t.String({ format: 'email' }),
    subject: t.String(),
    body: t.String(),
    attachments: t.Optional(
      t.Array(t.Object({ name: t.String(), url: t.String() })),
    ),
  }),
});

// `typeof jobs` propagates through every store, handler, and
// queue.enqueue() in the rest of the app.
type Jobs = typeof jobs;

Use the re-exported t

The package re-exports TypeBox's t so every schema shares one TypeBox instance — mixing TypeBox versions silently breaks type narrowing.

#Handlers & Registry

createJobRegistry(jobs) is a fluent registry — chain .on(kind, handler) per kind. Each handler receives the typed payload and a JobContext with id, kind, attempts, maxAttempts, and signal.

const registry = createJobRegistry(jobs)
  .on('email.send', async (payload, ctx) => {
    // payload: { to, subject, body } — inferred from the schema
    // ctx: { id, kind, attempts, maxAttempts, signal }
    await sendEmail(payload);
  })
  .on('image.resize', async ({ src, width }, { attempts }) => {
    // attempts — which try this is
    await resize(src, width);
  });

A claimed job whose kind has no registered handler is dead-lettered rather than silently dropped — register a handler for every kind in the definition before the worker starts.

#Retries & Dead-letter

Throwing from a handler triggers the retry policy — exponential backoff (configurable via the worker's backoff option) up to maxAttempts (set per job at enqueue, default 5). After the final attempt the job moves to dead, surfaced separately in the worker metrics and listable via store.listByKind.

import { exponentialBackoff } from '@absolutejs/queue';

// The attempt ceiling is per job — set it at enqueue (default 5):
await queue.enqueue('webhook.deliver', payload, { maxAttempts: 8 });

// The delay between retries comes from the worker's backoff strategy.
// Default: exponentialBackoff() → 1s, 2s, 4s, … capped at 5 minutes.
createQueueWorker({
  backoff: exponentialBackoff({ baseMs: 500, factor: 2, maxMs: 60_000 }),
  registry,
  store,
});

// ctx.attempts / ctx.maxAttempts — change behavior on the last try:
registry.on('webhook.deliver', async (payload, ctx) => {
  try {
    await deliver(payload);
  } catch (error) {
    if (ctx.attempts >= ctx.maxAttempts - 1) {
      await alertOps(payload, error);
    }
    throw error; // rethrow — the worker schedules retry / dead-letter
  }
});

Schema drift dead-letters too

The worker re-validates each claimed payload against the registry's schema before running the handler. A persisted job that no longer matches (stale data after a schema change) is dead-lettered instead of crashing the handler.

#Recurring Jobs (cron)

The queue deliberately does not reinvent cron — pair it with @elysiajs/cron for recurring triggers. Cron decides when; the queue guarantees the work happens — once, surviving restarts, with retries and dead-lettering.

// src/jobs/index.ts — module-scoped store so the cron trigger and the
// queue plugin's worker share the same backing state (the cron run
// callback has no Elysia Context, so it closes over the store).
import { cron } from '@elysiajs/cron';

export const store = createInMemoryJobStore(jobs);
export const registry = createJobRegistry(jobs).on(
  'email.send',
  async (payload) => {
    await sendDigest(payload);
  },
);

export const backgroundJobs = new Elysia({ name: 'background-jobs' })
  .use(queue({ registry, store }))
  .use(
    cron({
      name: 'weekly-digest',
      pattern: '0 8 * * 1', // Mondays at 08:00
      run: () =>
        store.enqueue({
          idempotencyKey: `weekly-digest:${new Date()
            .toISOString()
            .slice(0, 10)}`,
          kind: 'email.send',
          payload: {
            subject: 'Weekly digest',
            to: 'team@example.com',
          },
        }),
    }),
  );

Tag recurring enqueues with an idempotencyKey

A per-day idempotencyKey means a cron misfire (or an extra process running the same schedule) doesn't double-run the job — the store returns the existing job's id instead of enqueuing a duplicate.

#One-shot Triggers

Delayed one-shots are just enqueues with a future runAt — the job persists immediately and the worker claims it once due:

// Delayed one-shot — deliver the webhook in one hour. The job is
// persisted immediately and claimed once runAt is due.
app.post('/notify', ({ body, queue }) =>
  queue.enqueue(
    'webhook.deliver',
    { body, url: 'https://example.com/hook' },
    { runAt: Date.now() + 60 * 60 * 1000 },
  ),
);

For manual backfills, admin re-runs, unit tests, or bun scripts/foo.ts wrappers that share logic with a cron, use runHandlerOnce — it invokes a registered handler directly, without spinning up the worker or the plugin:

// scripts/runWeeklyDigest.ts — invoke a handler directly: no worker,
// no store writes. The payload is validated through the registry's
// schema and a JobContext is synthesized for you.
import { runHandlerOnce } from '@absolutejs/queue';
import { registry } from '../src/jobs/registry';

await runHandlerOnce(registry, 'email.send', {
  subject: 'Weekly digest (manual trigger)',
  to: 'team@example.com',
});

// Overrides: options.context ({ attempts, maxAttempts, id, … }) and
// options.validators — false to skip validation, or pre-compiled
// JobValidators for hot loops.

Don't import the plugin barrel in scripts

Importing the module that exports your Elysia plugin pulls in @elysiajs/cron, whose timers keep the process alive and prevent a one-shot script from exiting. Export the registry from a separate file, or process.exit(0) at the end of the script.

#Timeouts & Abort Signal

handlerTimeoutMs bounds the wall-clock time a handler may run before the worker aborts its ctx.signal and fails the job through the normal retry / dead-letter path — so a hung handler frees its worker slot instead of holding it for the full lease.

createQueueWorker({
  // Wall-clock budget per handler. A number applies to every kind; a
  // function sets one per kind — return undefined for no limit on that
  // kind. Unset = no timeout.
  handlerTimeoutMs: (kind) =>
    kind === 'ai.synthesize' ? 300_000 : 15_000,
  registry,
  store,
});

registry.on('ai.synthesize', async (payload, { signal }) => {
  // On timeout the worker aborts `signal` and fails the job through
  // the normal retry / dead-letter path. Pass it to in-flight I/O so
  // the work actually stops — the timeout only bounds how long the
  // WORKER waits.
  await fetch(payload.url, { signal });
});

Honor the signal

The timeout only bounds how long the worker waits — check ctx.signal on cooperative paths and pass it to in-flight HTTP calls so the handler's work actually stops when it fires.