# @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 in-process 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. The default store lives in memory; the design allows a pluggable driver
to back it with Redis/SQL for durability across restarts. 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.                 |
| `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.                                           |
| `size`    | `size(): number`                                               | Number of jobs currently queued.                               |

#### `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).    |

#### `JobHandler<T>`

```ts
type JobHandler<T = unknown> = (job: Job<T>) => 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

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
```

### 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
```

## 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.