# @wrnexus/queue

> A background job queue with delays, retries + exponential backoff, recurring jobs, and concurrent workers.

Part of the **WrNexus** framework — an SSR-first, Bun-native full-stack web framework.

## Overview

`@wrnexus/queue` is a server-side durable job queue. You register named
workers, enqueue jobs (optionally delayed or recurring), and let the queue poll
and run them on a timer — with per-job retry limits and doubling backoff between
attempts. `defineQueue` persists to `.wrnexus/queue.sqlite` by default, while
low-level `createQueue` remains an intentionally in-memory primitive. Reach for it when you
need to defer work (emails, webhooks, cleanup) off the request path without a
heavyweight external broker. Tests can drive it deterministically via `drain()`.

## Installation

```bash
bun add @wrnexus/queue
```

> Private package — the machine must be authenticated to the `wrnexus` npm org
> (a read token in `~/.npmrc`). Requires **Bun** (Node is not supported).

## API

The package exports a single factory plus its supporting types.

### `createQueue(options?): Queue`

Creates a new queue instance.

```ts
function createQueue(options?: QueueOptions): Queue;
```

#### `QueueOptions`

| Option        | Type                                 | Default    | Description                                              |
| ------------- | ------------------------------------ | ---------- | -------------------------------------------------------- |
| `maxAttempts` | `number`                             | `3`        | Default max attempts per job before it is dead-lettered. |
| `backoffMs`   | `number`                             | `1000`     | Base retry backoff in ms; doubles per attempt.           |
| `pollMs`      | `number`                             | `250`      | Poll interval used once `start()` is called (ms).        |
| `onFailed`    | `(job: Job, error: unknown) => void` | —          | Called when a job exhausts its attempts.                 |
| `concurrency` | `number`                             | unlimited  | Maximum jobs claimed by one `drain()` call.              |
| `capacity`    | `number`                             | unlimited  | Maximum queued plus active jobs before adds reject.      |
| `now`         | `() => number`                       | `Date.now` | Clock injection for deterministic tests.                 |

### `Queue`

The object returned by `createQueue`.

| Method     | Signature                                                      | Description                                                    |
| ---------- | -------------------------------------------------------------- | -------------------------------------------------------------- |
| `add`      | `add<T>(name, data: T, options?: AddOptions): Promise<Job<T>>` | Enqueue a job under a worker name. Returns the created job.    |
| `process`  | `process<T>(name, handler: JobHandler<T>): void`               | Register the worker that runs jobs of the given name.          |
| `drain`    | `drain(now?: number): Promise<number>`                         | Run every job whose `runAt ≤ now`, once. Returns how many ran. |
| `start`    | `start(): void`                                                | Begin polling every `pollMs`. No-op if already started.        |
| `stop`     | `stop(): void`                                                 | Stop the poll timer.                                           |
| `shutdown` | `shutdown({ force? }): Promise<void>`                          | Stop accepting jobs and await active work; force aborts it.    |
| `size`     | `size(): number`                                               | Number of jobs currently queued.                               |
| `get/list` | `get(id)` / `list(name?)`                                      | Inspect defensive copies of pending jobs.                      |
| `cancel`   | `cancel(id): boolean`                                          | Remove queued work or abort an active handler.                 |
| `failed`   | `failed(): Job[]`                                              | Inspect exhausted jobs in the dead-letter collection.          |
| `retry`    | `retry(id): Promise<boolean>`                                  | Reset and requeue a dead-lettered job.                         |

#### `AddOptions`

| Option           | Type     | Description                                                                |
| ---------------- | -------- | -------------------------------------------------------------------------- |
| `delayMs`        | `number` | Delay before the job becomes runnable (ms).                                |
| `maxAttempts`    | `number` | Max attempts before dead-lettering. Defaults to the queue's `maxAttempts`. |
| `repeat`         | `number` | Re-enqueue this job this many ms after each successful run (recurring).    |
| `priority`       | `number` | Higher values are selected first among due jobs.                           |
| `idempotencyKey` | `string` | Return the matching pending job instead of enqueueing a duplicate.         |

#### `JobHandler<T>`

```ts
type JobHandler<T = unknown> = (
  job: Job<T>,
  context: { signal: AbortSignal },
) => void | Promise<void>;
```

#### `Job<T>`

```ts
interface Job<T = unknown> {
  id: string; // e.g. "job_1"
  name: string;
  data: T;
  attempts: number;
  maxAttempts: number;
  runAt: number; // epoch ms; job runs when now ≥ runAt
  repeat?: number; // if set, re-enqueue this many ms after each success
}
```

## Usage

Application queues belong in `app/queues/*.ts` and default-export `defineQueue(...)`.
Configure persistence once in `wrnexus.config.ts`:

```ts
export default {
  queue: { storage: "database", databaseName: "default" },
  // Or omit queue entirely for .wrnexus/queue.sqlite.
};
```

Use `storage: "memory"` only for disposable work. A selected database must exist;
WrNexus fails startup/first use with a clear error instead of silently falling back.
Each job may declare `success(data, job)` and `failed(data, error, job)` lifecycle hooks.

Register workers, enqueue jobs, then start the poller:

```ts
import { createQueue } from "@wrnexus/queue";

const queue = createQueue({ maxAttempts: 3, backoffMs: 1000 });

// Register a worker for the "email" job name.
queue.process<{ to: string }>("email", async (job) => {
  await send(job.data.to);
});

// Enqueue a delayed job with up to 3 attempts.
await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });

queue.start(); // begin polling; queue.stop() to halt
```

### Application queues with `defineQueue`

WrNexus applications can place typed queue modules in `app/queues`. The dev and
production servers discover them, start them after runtime initialization, and
shut them down gracefully:

```ts
import { defineQueue } from "@wrnexus/queue";

export default defineQueue({
  name: "email",
  jobs: {
    send: {
      maxAttempts: 3,
      idempotency: ({ messageId }: { messageId: number }) => `message:${messageId}`,
      run: async ({ messageId }) => sendMessage(messageId),
    },
  },
});
```

Callers get a typed producer and inspection helpers:

```ts
const job = await email.send.add({ messageId: 42 });
await email.send.status(job.id);
await email.send.cancel(job.id);
await email.send.retry(job.id);
```

`createDurableQueue` also exposes `start()`, `stop()`, `isRunning()`, `health()`
and `get()` so applications do not need to build their own polling middleware.

### SQLite

Use `sqliteQueueStore(db)` from `@wrnexus/queue/sqlite` with a WrNexus-compatible
SQLite client. `installSqliteQueueSchema(db)` installs the queue table and its
due-job and idempotency indexes.

Use `context.signal` in network/database calls so forced shutdown and active
cancellation finish promptly. For process termination, prefer
`await queue.shutdown()`; use `{ force: true }` only after your grace period.

### Durable queue

`createDurableQueue({ store })` retains jobs until their handler succeeds and
supports atomic leases when a driver implements `QueueStore.claim`. It exposes
the same cancellation/shutdown behavior plus `list`, `failed`, and `retry`.
The included `memoryQueueStore()` is useful for tests; production Redis/SQL
drivers should make `claim()` atomic to prevent two workers executing one job.

### Recurring jobs

Pass `repeat` to re-enqueue a job a fixed interval after each successful run:

```ts
queue.process("heartbeat", async () => ping());
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute
```

### Handling permanent failures

When a job's `attempts` reaches `maxAttempts`, it is dropped and `onFailed`
fires instead of retrying:

```ts
const queue = createQueue({
  onFailed: (job, error) => {
    console.error(`job ${job.id} (${job.name}) gave up`, error);
  },
});
```

### Deterministic testing

Instead of `start()`, inject a clock and drive the queue with `drain()`:

```ts
let clock = 0;
const queue = createQueue({ now: () => clock });

queue.process("task", async () => {
  /* ... */
});
await queue.add("task", {}, { delayMs: 5000 });

clock = 5000;
const ran = await queue.drain(); // => 1
```

### Durable workflows and approvals

`createWorkflowEngine(store)` executes dependency-ordered steps and persists every transition,
result, progress update, failure, cancellation, and approval record. Approval steps pause safely
and can resume after a process restart because the snapshot lives in the supplied `WorkflowStore`.

```ts
const workflow = defineDurableWorkflow({
  name: "publish-report",
  steps: [
    { name: "build", run: buildReport },
    { name: "approve", dependsOn: ["build"], approval: true, run: (report) => report },
    { name: "publish", dependsOn: ["approve"], run: publishReport },
  ],
});

const run = await engine.start(workflow, input);
await engine.approve(workflow, run.id, "approve", currentUser.id);
```

Use `memoryWorkflowStore()` for tests. Production stores implement the small `get`, `put`, and
`list` contract using the same transactional database or durable service as the application.

## Retry & backoff behavior

- On a thrown handler error, the job is retried while `attempts < maxAttempts`.
- The next `runAt` is set to `now + backoffMs * 2^(attempts - 1)` (exponential
  backoff): with `backoffMs: 1000` the delays are 1s, 2s, 4s, …
- A job whose worker name has no registered handler stays queued until one is
  registered (it is not counted as runnable by `drain`).
- `drain` is re-entrant-safe: overlapping calls are skipped while one is running.

## Requirements / Notes

- **Bun-only** runtime (Node is not supported), consistent with the rest of the
  WrNexus framework. The queue itself relies only on standard timers
  (`setInterval`/`clearInterval`) and has no runtime dependencies.
- The default store is in-process, so queued jobs do not survive a restart; a
  pluggable driver is intended for backing it with Redis/SQL for durability.
- Works alongside `@wrnexus/core` for offloading work from the request path.