diff --git a/.github/CODEOWNERS b/.github/CODEOWNERS new file mode 100644 index 0000000..e18a3fa --- /dev/null +++ b/.github/CODEOWNERS @@ -0,0 +1,2 @@ +# Every change to this repo needs a review from a code owner (see CONTRIBUTING in mgpl-diagnostics/.github). +* @BackStacked diff --git a/.github/dependabot.yml b/.github/dependabot.yml new file mode 100644 index 0000000..fdd195c --- /dev/null +++ b/.github/dependabot.yml @@ -0,0 +1,21 @@ +version: 2 +updates: + - package-ecosystem: npm + directory: / + schedule: + interval: weekly + open-pull-requests-limit: 5 + groups: + minor-and-patch: + update-types: [minor, patch] + ignore: + # our own packages are released together and bumped by hand + - dependency-name: "@mgpl-diagnostics/*" + - dependency-name: "typescript" + update-types: ["version-update:semver-major"] + - dependency-name: "@types/node" + update-types: ["version-update:semver-major"] + - package-ecosystem: github-actions + directory: / + schedule: + interval: monthly diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..06fa8d1 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,42 @@ +name: CI + +on: + push: + branches: [main] + pull_request: + +permissions: + contents: read + +concurrency: + group: ci-${{ github.ref }} + cancel-in-progress: true + +jobs: + # The job id "ci" is the required status check on main (see mgpl-diagnostics/.github scripts/protect-repos.mjs). + ci: + runs-on: ubuntu-latest + timeout-minutes: 10 + services: + redis: + image: redis:7-alpine + ports: ["6379:6379"] + options: >- + --health-cmd "redis-cli ping" --health-interval 2s --health-timeout 2s --health-retries 15 + env: + REDIS_URL: redis://127.0.0.1:6379 + REQUIRE_REDIS: "1" + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 22 + - run: npm install --no-package-lock + - run: npm run typecheck + - name: Tests (definitions, data rules, in-process, fallback, BullMQ on Redis with the same scenario) + run: npm test + - run: npm run build + - name: The main entry does not import bullmq or ioredis + run: "! grep -E \"from ['\\\"](bullmq|ioredis)['\\\"]\" dist/index.js" + - name: llms-full.txt is up to date (run `npm run llms` and commit if this fails) + run: npm run llms && git diff --exit-code -- llms-full.txt diff --git a/.github/workflows/publish.yml b/.github/workflows/publish.yml new file mode 100644 index 0000000..e1b9509 --- /dev/null +++ b/.github/workflows/publish.yml @@ -0,0 +1,29 @@ +name: Publish to GitHub Packages + +on: + push: + tags: ["v*"] + +permissions: + contents: read + packages: write + +jobs: + publish: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v7 + - uses: actions/setup-node@v7 + with: + node-version: 22 + registry-url: https://npm.pkg.github.com + scope: "@mgpl-diagnostics" + - name: Tag must match package.json version + run: test "v$(node -p "require('./package.json').version")" = "${GITHUB_REF_NAME}" + # installs dependencies (incl. @mgpl-diagnostics/* peers from GitHub Packages) and builds dist/ via "prepare" + - run: npm install --no-package-lock + env: + NODE_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }} + - run: npm publish + env: + NODE_AUTH_TOKEN: ${{ secrets.GITHUB_TOKEN }} diff --git a/AGENTS.md b/AGENTS.md new file mode 100644 index 0000000..2252954 --- /dev/null +++ b/AGENTS.md @@ -0,0 +1,56 @@ +# AGENTS.md — @mgpl-diagnostics/worker-kit + +Background jobs for the MGPL Diagnostics backend: typed job definitions, one queue API over three +backends (BullMQ on Redis, in-process, worker-or-standalone fallback) and the worker runner. +**A library: no server, no GraphQL, no business logic.** The api repo defines the jobs and wires +them (container + `src/worker.ts`); mail-kit queues through `enqueuer()`. + +## Map + +``` +src/types.ts JobDefinition, JobInfo, JobState, JobContext, JobBackend, PreparedJob +src/errors.ts WorkerError (codes), JobFailure (safe message + retry flag), isJobFailure +src/define.ts defineJob, SECRET_KEY, toJsonData / prepareJob (checks in the caller), runAttempt, backoffDelay +src/queue.ts createJobQueue (enqueue, get, enqueuer, close), jobRegistry +src/inline.ts inlineBackend (wait or background, retries, backoff, retention) +src/fallback.ts fallbackBackend (primary when available, else fallback) +src/bullmq.ts "/bullmq" entry: bullmqBackend, startWorker, redisConnection, redisReady, bullJobOptions +scripts/test.ts definitions, data rules, inline, fallback, enqueuer; the same scenario on inline and BullMQ (real Redis) +``` + +## Rules + +1. **Every backend behaves the same.** Data and results cross JSON everywhere; retries, backoff, + states, attempt counts and error messages match. Change behaviour in one backend → change it in + all, and keep the shared scenario in `scripts/test.ts` passing on inline *and* BullMQ. +2. **Failures are safe to show.** Only a `JobFailure`'s message reaches `JobInfo.error`; every other + error goes to `onError` and reads "Job failed". All attempts go through `runAttempt`. +3. **Payloads carry ids, never secrets.** `SECRET_KEY` keys are refused in the caller; do not loosen it + by default. Jobs whose data must not linger use `keepResultsMs: 0`. +4. **Checks happen in the caller** (`prepareJob`): bad data fails the request, not the worker. +5. **Zero runtime dependencies in the main entry.** Only `src/bullmq.ts` imports `bullmq`/`ioredis` + (optional peers); CI greps `dist/index.js` for it. +6. Job ids are random UUIDs; `get` returns a job only under its own definition. Ownership checks + belong to the caller (the api's services). +7. Error codes (`WorkerError.code`) and state names (`JobState`) are public: add, never rename. +8. Backends are structural (`JobBackend`); prefer that over new options when adding a queue technology. +9. Never commit secrets, `.env` files or patient data. Tests use fictional values. + +## Commands + +```bash +npm run typecheck && npm test && npm run build +npm run llms # regenerate llms-full.txt (CI fails when it is stale) +``` + +`npm test` needs Redis for the BullMQ part: `REDIS_URL`, or a local `redis-server` (started on a free +port). CI runs a Redis service with `REQUIRE_REDIS=1`. + +## Repository rules + +`main` takes pull requests only: squash merge, 1 code-owner review (`@BackStacked`), the `ci` check +required. Rules come from `mgpl-diagnostics/.github` (`scripts/protect-repos.mjs`). + +## Licence + +All rights reserved (see LICENSE). diff --git a/CLAUDE.md b/CLAUDE.md new file mode 100644 index 0000000..43c994c --- /dev/null +++ b/CLAUDE.md @@ -0,0 +1 @@ +@AGENTS.md diff --git a/README.md b/README.md new file mode 100644 index 0000000..9faf592 --- /dev/null +++ b/README.md @@ -0,0 +1,146 @@ +# @mgpl-diagnostics/worker-kit + +Background jobs for the MGPL Diagnostics backend. You define a job once (name, processor, retries, +how long its result is kept) and start it from one queue API. Underneath, the jobs run: + +- on **BullMQ over Redis**, with `startWorker` in a separate worker process (production); +- **in-process**, with no Redis (development, tests, a single server); +- or **on the worker when it is available, otherwise standalone** (`fallbackBackend`), the + behaviour mail-kit relies on. + +All three behave the same: the same JSON data, the same retries and backoff, the same states and the +same error messages. `npm test` runs one scenario on each and compares the results. + +- The main entry has **no runtime dependencies**. `bullmq` and `ioredis` are optional peers, used only + by the `/bullmq` entry. +- Job data is checked in the caller: it must be JSON, under a size limit, and free of secret-looking + keys (`password`, `token`, `secret`, …). Payloads carry ids; the processor looks things up. +- Processor failures are safe to show: a `JobFailure` keeps its message, anything else is logged and + shown as "Job failed". + +## Install + +```bash +# .npmrc: @mgpl-diagnostics:registry=https://npm.pkg.github.com +npm install @mgpl-diagnostics/worker-kit +npm install bullmq ioredis # only for the /bullmq entry +``` + +## Define jobs + +```ts +import { defineJob, JobFailure } from "@mgpl-diagnostics/worker-kit"; + +export const renderInvoice = defineJob<{ orgId: string; invoiceId: string }, { base64: string }>({ + name: "invoice", + queue: "documents", // default: the name + attempts: 2, // default 3, exponential backoff from backoffMs (default 2 s) + keepResultsMs: 3600_000, // finished jobs readable for an hour (default); 0 = removed at once + validate: (d) => { if (!d.invoiceId) throw new Error("invoiceId is required"); }, + async process(d, ctx) { // ctx: { id, name, queue, attempt, attempts } + const invoice = await services.billing.find(d.orgId, d.invoiceId); + if (!invoice) throw new JobFailure("Invoice not found", false); // shown to the client, no retry + return { base64: await render(invoice) }; // must be JSON + }, +}); +``` + +Data and results cross JSON on every backend, so a `Date` arrives as an ISO string. + +## Start and read jobs + +```ts +import { createJobQueue, fallbackBackend, inlineBackend } from "@mgpl-diagnostics/worker-kit"; +import { bullmqBackend } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const backend = process.env.REDIS_URL + ? bullmqBackend({ redisUrl: process.env.REDIS_URL, prefix: "mgpl" }) + : inlineBackend(); // { wait: false } runs jobs in the background of this process + +const jobs = createJobQueue({ backend, jobs: [renderInvoice, passwordReset] }); + +const job = await jobs.enqueue(renderInvoice, { orgId, invoiceId }); +// → { id, name, queue, state, attempts, error, result, data, createdAt, finishedAt } +const later = await jobs.get(renderInvoice, job.id); // null if unknown, expired, or another job's id +``` + +Job ids are random UUIDs, but they are not a permission: check that the reader owns the job +(compare `data` with the session) before showing it. + +With `inlineBackend()` (the default `wait: true`), `enqueue` resolves when the job has finished, so +the request waits. With BullMQ it resolves at once with `state: "waiting"`. + +## Run the worker + +```ts +import { startWorker } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const worker = startWorker({ + redisUrl: process.env.REDIS_URL!, + prefix: "mgpl", + jobs: [renderInvoice, passwordReset], // one BullMQ Worker per queue + concurrency: 4, + onError: (e, ctx) => log.error({ err: e, job: ctx }, "job failed"), // errors that are not a JobFailure +}); + +process.once("SIGTERM", () => worker.close()); // finishes the jobs in hand, then stops +``` + +A `JobFailure` with `retry: false` fails the job at once (BullMQ `UnrecoverableError`); every other +failure uses the remaining attempts. + +## Worker when available, standalone otherwise + +```ts +import { redisConnection } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const connection = redisConnection(process.env.REDIS_URL!); +const jobs = createJobQueue({ + backend: fallbackBackend({ + primary: bullmqBackend({ connection, prefix: "mgpl" }), // available() = connection is "ready" + fallback: inlineBackend(), + onFallback: (reason, job) => log.warn({ reason, job: job.def.name }, "ran without the worker"), + // whenFails: "throw" → WorkerError WORKER_UNAVAILABLE instead + }), + jobs: [mailJob], +}); +``` + +### With mail-kit + +`jobs.enqueuer(job)` has mail-kit's `MailQueue` shape, so mail-kit can queue through worker-kit: + +```ts +const mailJob = defineJob({ name: "mail", keepResultsMs: 0, attempts: 5, backoffMs: 10_000, + process: (job) => mailer.deliver(job) }); +const mailer = createMailer({ transport, from, queue: jobs.enqueuer(mailJob) }); +``` + +`keepResultsMs: 0` removes a mail job as soon as it is sent, because it can hold a reset link. +In mail-kit's MailJob the link sits under `data.link`, which is allowed; a key named `token` would be refused. + +## Errors + +`WorkerError.code` is a public contract (add, never rename): + +| Code | When | +| --- | --- | +| `WORKER_UNKNOWN_JOB` | the job is not in this queue's `jobs` (or is a different definition with the same name) | +| `WORKER_BAD_DATA` | data is not JSON, too large, has a secret-looking key, or `validate` threw; bad job/queue name | +| `WORKER_DUPLICATE_JOB` | two definitions share a name | +| `WORKER_UNAVAILABLE` | `fallbackBackend` with `whenFails: "throw"`; an in-process `enqueuer` job failed; no Redis given | +| `WORKER_CLOSED` | the queue was closed | + +## Develop + +```bash +npm run typecheck && npm test && npm run build +npm run llms # regenerate llms-full.txt (CI fails when it is stale) +``` + +`npm test` uses `REDIS_URL`, or starts a local `redis-server` when one is installed; without either the +BullMQ checks are skipped (CI sets `REQUIRE_REDIS=1` and a Redis service, so they always run there). + +## Licence + +All rights reserved (see [LICENSE](LICENSE)). diff --git a/llms-full.txt b/llms-full.txt new file mode 100644 index 0000000..9cb2967 --- /dev/null +++ b/llms-full.txt @@ -0,0 +1,1735 @@ +# @mgpl-diagnostics/worker-kit — full documentation and source + +> Generated by scripts/llms-full.mjs. Do not edit by hand. 14 files. + +## Contents + +- README.md +- AGENTS.md +- llms.txt +- package.json +- scripts/llms-full.mjs +- scripts/test.ts +- src/bullmq.ts +- src/define.ts +- src/errors.ts +- src/fallback.ts +- src/index.ts +- src/inline.ts +- src/queue.ts +- src/types.ts + + +--- + +## File: README.md + +# @mgpl-diagnostics/worker-kit + +Background jobs for the MGPL Diagnostics backend. You define a job once (name, processor, retries, +how long its result is kept) and start it from one queue API. Underneath, the jobs run: + +- on **BullMQ over Redis**, with `startWorker` in a separate worker process (production); +- **in-process**, with no Redis (development, tests, a single server); +- or **on the worker when it is available, otherwise standalone** (`fallbackBackend`), the + behaviour mail-kit relies on. + +All three behave the same: the same JSON data, the same retries and backoff, the same states and the +same error messages. `npm test` runs one scenario on each and compares the results. + +- The main entry has **no runtime dependencies**. `bullmq` and `ioredis` are optional peers, used only + by the `/bullmq` entry. +- Job data is checked in the caller: it must be JSON, under a size limit, and free of secret-looking + keys (`password`, `token`, `secret`, …). Payloads carry ids; the processor looks things up. +- Processor failures are safe to show: a `JobFailure` keeps its message, anything else is logged and + shown as "Job failed". + +## Install + +```bash +# .npmrc: @mgpl-diagnostics:registry=https://npm.pkg.github.com +npm install @mgpl-diagnostics/worker-kit +npm install bullmq ioredis # only for the /bullmq entry +``` + +## Define jobs + +```ts +import { defineJob, JobFailure } from "@mgpl-diagnostics/worker-kit"; + +export const renderInvoice = defineJob<{ orgId: string; invoiceId: string }, { base64: string }>({ + name: "invoice", + queue: "documents", // default: the name + attempts: 2, // default 3, exponential backoff from backoffMs (default 2 s) + keepResultsMs: 3600_000, // finished jobs readable for an hour (default); 0 = removed at once + validate: (d) => { if (!d.invoiceId) throw new Error("invoiceId is required"); }, + async process(d, ctx) { // ctx: { id, name, queue, attempt, attempts } + const invoice = await services.billing.find(d.orgId, d.invoiceId); + if (!invoice) throw new JobFailure("Invoice not found", false); // shown to the client, no retry + return { base64: await render(invoice) }; // must be JSON + }, +}); +``` + +Data and results cross JSON on every backend, so a `Date` arrives as an ISO string. + +## Start and read jobs + +```ts +import { createJobQueue, fallbackBackend, inlineBackend } from "@mgpl-diagnostics/worker-kit"; +import { bullmqBackend } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const backend = process.env.REDIS_URL + ? bullmqBackend({ redisUrl: process.env.REDIS_URL, prefix: "mgpl" }) + : inlineBackend(); // { wait: false } runs jobs in the background of this process + +const jobs = createJobQueue({ backend, jobs: [renderInvoice, passwordReset] }); + +const job = await jobs.enqueue(renderInvoice, { orgId, invoiceId }); +// → { id, name, queue, state, attempts, error, result, data, createdAt, finishedAt } +const later = await jobs.get(renderInvoice, job.id); // null if unknown, expired, or another job's id +``` + +Job ids are random UUIDs, but they are not a permission: check that the reader owns the job +(compare `data` with the session) before showing it. + +With `inlineBackend()` (the default `wait: true`), `enqueue` resolves when the job has finished, so +the request waits. With BullMQ it resolves at once with `state: "waiting"`. + +## Run the worker + +```ts +import { startWorker } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const worker = startWorker({ + redisUrl: process.env.REDIS_URL!, + prefix: "mgpl", + jobs: [renderInvoice, passwordReset], // one BullMQ Worker per queue + concurrency: 4, + onError: (e, ctx) => log.error({ err: e, job: ctx }, "job failed"), // errors that are not a JobFailure +}); + +process.once("SIGTERM", () => worker.close()); // finishes the jobs in hand, then stops +``` + +A `JobFailure` with `retry: false` fails the job at once (BullMQ `UnrecoverableError`); every other +failure uses the remaining attempts. + +## Worker when available, standalone otherwise + +```ts +import { redisConnection } from "@mgpl-diagnostics/worker-kit/bullmq"; + +const connection = redisConnection(process.env.REDIS_URL!); +const jobs = createJobQueue({ + backend: fallbackBackend({ + primary: bullmqBackend({ connection, prefix: "mgpl" }), // available() = connection is "ready" + fallback: inlineBackend(), + onFallback: (reason, job) => log.warn({ reason, job: job.def.name }, "ran without the worker"), + // whenFails: "throw" → WorkerError WORKER_UNAVAILABLE instead + }), + jobs: [mailJob], +}); +``` + +### With mail-kit + +`jobs.enqueuer(job)` has mail-kit's `MailQueue` shape, so mail-kit can queue through worker-kit: + +```ts +const mailJob = defineJob({ name: "mail", keepResultsMs: 0, attempts: 5, backoffMs: 10_000, + process: (job) => mailer.deliver(job) }); +const mailer = createMailer({ transport, from, queue: jobs.enqueuer(mailJob) }); +``` + +`keepResultsMs: 0` removes a mail job as soon as it is sent, because it can hold a reset link. +In mail-kit's MailJob the link sits under `data.link`, which is allowed; a key named `token` would be refused. + +## Errors + +`WorkerError.code` is a public contract (add, never rename): + +| Code | When | +| --- | --- | +| `WORKER_UNKNOWN_JOB` | the job is not in this queue's `jobs` (or is a different definition with the same name) | +| `WORKER_BAD_DATA` | data is not JSON, too large, has a secret-looking key, or `validate` threw; bad job/queue name | +| `WORKER_DUPLICATE_JOB` | two definitions share a name | +| `WORKER_UNAVAILABLE` | `fallbackBackend` with `whenFails: "throw"`; an in-process `enqueuer` job failed; no Redis given | +| `WORKER_CLOSED` | the queue was closed | + +## Develop + +```bash +npm run typecheck && npm test && npm run build +npm run llms # regenerate llms-full.txt (CI fails when it is stale) +``` + +`npm test` uses `REDIS_URL`, or starts a local `redis-server` when one is installed; without either the +BullMQ checks are skipped (CI sets `REQUIRE_REDIS=1` and a Redis service, so they always run there). + +## Licence + +All rights reserved (see [LICENSE](LICENSE)). + + +--- + +## File: AGENTS.md + +# AGENTS.md — @mgpl-diagnostics/worker-kit + +Background jobs for the MGPL Diagnostics backend: typed job definitions, one queue API over three +backends (BullMQ on Redis, in-process, worker-or-standalone fallback) and the worker runner. +**A library: no server, no GraphQL, no business logic.** The api repo defines the jobs and wires +them (container + `src/worker.ts`); mail-kit queues through `enqueuer()`. + +## Map + +``` +src/types.ts JobDefinition, JobInfo, JobState, JobContext, JobBackend, PreparedJob +src/errors.ts WorkerError (codes), JobFailure (safe message + retry flag), isJobFailure +src/define.ts defineJob, SECRET_KEY, toJsonData / prepareJob (checks in the caller), runAttempt, backoffDelay +src/queue.ts createJobQueue (enqueue, get, enqueuer, close), jobRegistry +src/inline.ts inlineBackend (wait or background, retries, backoff, retention) +src/fallback.ts fallbackBackend (primary when available, else fallback) +src/bullmq.ts "/bullmq" entry: bullmqBackend, startWorker, redisConnection, redisReady, bullJobOptions +scripts/test.ts definitions, data rules, inline, fallback, enqueuer; the same scenario on inline and BullMQ (real Redis) +``` + +## Rules + +1. **Every backend behaves the same.** Data and results cross JSON everywhere; retries, backoff, + states, attempt counts and error messages match. Change behaviour in one backend → change it in + all, and keep the shared scenario in `scripts/test.ts` passing on inline *and* BullMQ. +2. **Failures are safe to show.** Only a `JobFailure`'s message reaches `JobInfo.error`; every other + error goes to `onError` and reads "Job failed". All attempts go through `runAttempt`. +3. **Payloads carry ids, never secrets.** `SECRET_KEY` keys are refused in the caller; do not loosen it + by default. Jobs whose data must not linger use `keepResultsMs: 0`. +4. **Checks happen in the caller** (`prepareJob`): bad data fails the request, not the worker. +5. **Zero runtime dependencies in the main entry.** Only `src/bullmq.ts` imports `bullmq`/`ioredis` + (optional peers); CI greps `dist/index.js` for it. +6. Job ids are random UUIDs; `get` returns a job only under its own definition. Ownership checks + belong to the caller (the api's services). +7. Error codes (`WorkerError.code`) and state names (`JobState`) are public: add, never rename. +8. Backends are structural (`JobBackend`); prefer that over new options when adding a queue technology. +9. Never commit secrets, `.env` files or patient data. Tests use fictional values. + +## Commands + +```bash +npm run typecheck && npm test && npm run build +npm run llms # regenerate llms-full.txt (CI fails when it is stale) +``` + +`npm test` needs Redis for the BullMQ part: `REDIS_URL`, or a local `redis-server` (started on a free +port). CI runs a Redis service with `REQUIRE_REDIS=1`. + +## Repository rules + +`main` takes pull requests only: squash merge, 1 code-owner review (`@BackStacked`), the `ci` check +required. Rules come from `mgpl-diagnostics/.github` (`scripts/protect-repos.mjs`). + +## Licence + +All rights reserved (see LICENSE). + + +--- + +## File: llms.txt + +# @mgpl-diagnostics/worker-kit + +> Background jobs for the MGPL Diagnostics backend (Node.js ≥ 20). Typed job definitions and one queue API over BullMQ on Redis, in-process, or worker-when-available-else-standalone; plus the worker runner. The main entry has no runtime dependencies; bullmq and ioredis are optional peers used only by "/bullmq". + +Key facts for agents: +- `defineJob({ name, queue?, process(data, ctx), validate?, attempts = 3, backoffMs = 2000, keepResultsMs = 1h, keepFailedMs, maxBytes, refuseSecrets = true })`. +- `createJobQueue({ backend, jobs })` → `enqueue(job, data, { delayMs? })`, `get(job, id)`, `enqueuer(job)` (mail-kit MailQueue shape), `available()`, `close()`. +- Backends: `inlineBackend({ wait?, onError? })`, `fallbackBackend({ primary, fallback, whenFails?, onFallback? })`, and from "/bullmq" `bullmqBackend({ connection | redisUrl, prefix = "mgpl" })`. +- Worker: `startWorker({ jobs, connection | redisUrl, prefix, concurrency = 4, queues?, onError?, onFailed? })` → `{ queues, workers, close() }`. +- `JobFailure(message, retry = true)`: message shown to clients; `retry: false` fails at once. Other errors → "Job failed" + `onError`. +- `JobInfo`: id, name, queue, state (waiting | delayed | active | completed | failed), attempts, error, result, data, createdAt, finishedAt. +- Errors: `WorkerError.code` = WORKER_UNKNOWN_JOB, WORKER_BAD_DATA, WORKER_DUPLICATE_JOB, WORKER_UNAVAILABLE, WORKER_CLOSED. + +## Docs + +- [README](README.md): usage, backends, worker, mail-kit +- [AGENTS.md](AGENTS.md): map and rules + +## Source + +- [src/define.ts](src/define.ts): definitions, data checks, one attempt +- [src/queue.ts](src/queue.ts): the producer API +- [src/inline.ts](src/inline.ts): in-process backend +- [src/fallback.ts](src/fallback.ts): worker or standalone +- [src/bullmq.ts](src/bullmq.ts): BullMQ backend and worker + + +--- + +## File: package.json + +```json +{ + "name": "@mgpl-diagnostics/worker-kit", + "version": "0.1.0", + "description": "MGPL Diagnostics background jobs: typed job definitions, one queue API over BullMQ on Redis or in-process, a worker runner, and fallback from the worker to standalone. No runtime dependencies; bullmq and ioredis are optional peers used only by the \"/bullmq\" entry.", + "type": "module", + "license": "SEE LICENSE IN LICENSE", + "repository": "github:mgpl-diagnostics/worker-kit", + "sideEffects": false, + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + }, + "./bullmq": { + "types": "./dist/bullmq.d.ts", + "import": "./dist/bullmq.js" + }, + "./package.json": "./package.json" + }, + "files": [ + "dist", + "README.md", + "AGENTS.md", + "llms.txt", + "LICENSE", + "llms-full.txt" + ], + "publishConfig": { + "registry": "https://npm.pkg.github.com" + }, + "scripts": { + "build": "tsup", + "typecheck": "tsc --noEmit", + "test": "tsx scripts/test.ts", + "prepare": "tsup", + "llms": "node scripts/llms-full.mjs @mgpl-diagnostics/worker-kit README.md AGENTS.md llms.txt package.json src scripts" + }, + "peerDependencies": { + "bullmq": ">=5", + "ioredis": ">=5" + }, + "peerDependenciesMeta": { + "bullmq": { + "optional": true + }, + "ioredis": { + "optional": true + } + }, + "devDependencies": { + "@types/node": "^22.20.5", + "bullmq": "^5.81.5", + "ioredis": "^5.11.1", + "tsup": "^8.5.0", + "tsx": "^4.23.15", + "typescript": "^5.9.3" + }, + "engines": { + "node": ">=20" + } +} +``` + + +--- + +## File: scripts/llms-full.mjs + +````js +// Builds llms-full.txt: every doc and source file of the repo in one plain-text document for AI tools. +// node scripts/llms-full.mjs "" <file-or-folder> [file-or-folder …] +// Folders are walked recursively. Output order: README, AGENTS, docs, then the rest alphabetically. +// Skipped: build output, dependencies, lockfiles, generated files, binaries. Files over 150 KB are +// listed (path + size) instead of inlined. CI checks the file is up to date (`npm run llms` + git diff). +import { readdirSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import { extname, join, relative } from "node:path"; + +const [title, ...inputs] = process.argv.slice(2); +if (!title || !inputs.length) { + console.error('usage: node scripts/llms-full.mjs "<title>" <file-or-folder> …'); + process.exit(1); +} + +const TEXT = new Set([".ts", ".tsx", ".mts", ".cts", ".js", ".mjs", ".cjs", ".md", ".mdx", ".json", ".css", ".yml", ".yaml", ".txt", ".graphql", ".gql", ".sql", ".html"]); +const SKIP_DIRS = new Set(["node_modules", "dist", ".next", ".git", ".turbo", "coverage", "examples", ".pg-test"]); +const SKIP_FILES = /(^|\/)(package-lock\.json|pnpm-lock\.yaml|yarn\.lock|llms-full\.txt|[^/]*\.generated\.[^/]+|[^/]*\.log)$/; +const MAX = 150 * 1024; +const LANG = { ".ts": "ts", ".tsx": "tsx", ".mts": "ts", ".cts": "ts", ".js": "js", ".mjs": "js", ".cjs": "js", ".json": "json", ".css": "css", ".yml": "yaml", ".yaml": "yaml", ".graphql": "graphql", ".gql": "graphql", ".sql": "sql", ".html": "html" }; + +const posix = (p) => p.split("\\").join("/"); +const files = new Set(); +function walk(p) { + const st = statSync(p); + if (st.isDirectory()) { + for (const name of readdirSync(p)) if (!SKIP_DIRS.has(name)) walk(join(p, name)); + } else if (TEXT.has(extname(p).toLowerCase()) && !SKIP_FILES.test(posix(p))) { + files.add(posix(relative(".", p))); + } +} +for (const input of inputs) walk(input); + +const rank = (f) => (/^README\.md$/i.test(f) ? 0 : /^AGENTS\.md$/i.test(f) ? 1 : /^docs\//.test(f) ? 2 : 3); +const ordered = [...files].sort((a, b) => rank(a) - rank(b) || (a < b ? -1 : a > b ? 1 : 0)); + +const parts = [`# ${title} — full documentation and source\n\n> Generated by scripts/llms-full.mjs. Do not edit by hand. ${ordered.length} files.\n\n## Contents\n\n${ordered.map((f) => `- ${f}`).join("\n")}\n`]; +const omitted = []; +for (const f of ordered) { + // LF-normalised text, so Windows and Linux checkouts produce byte-identical output + const body = readFileSync(f, "utf8").replace(/\r\n/g, "\n").trim(); + const size = Buffer.byteLength(body); + if (size > MAX) { + omitted.push(`${f} (${(size / 1024).toFixed(0)} KB)`); + parts.push(`\n\n---\n\n## File: ${f}\n\n_Omitted: ${(size / 1024).toFixed(0)} KB. Read the file directly._\n`); + continue; + } + const lang = LANG[extname(f).toLowerCase()]; + const fence = body.includes("```") ? "````" : "```"; + parts.push(`\n\n---\n\n## File: ${f}\n\n${lang ? `${fence}${lang}\n${body}\n${fence}` : body}\n`); +} +const out = parts.join(""); +writeFileSync("llms-full.txt", out); +console.log(`llms-full.txt: ${ordered.length} files, ${(out.length / 1024).toFixed(0)} KB${omitted.length ? `; omitted (too large): ${omitted.join(", ")}` : ""}`); +```` + + +--- + +## File: scripts/test.ts + +```ts +// worker-kit checks: definitions, data rules, the in-process backend, fallback, the producer API, +// and BullMQ on a real Redis, where the same scenario must give the same results as in-process. +// Redis: REDIS_URL, or a local `redis-server` started here; CI sets REQUIRE_REDIS=1 so it never skips. +import assert from "node:assert/strict"; +import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { createServer } from "node:net"; +import { Queue } from "bullmq"; +import { Redis } from "ioredis"; +import { bullJobOptions, bullmqBackend, redisConnection, startWorker } from "../src/bullmq"; +import { + createJobQueue, + defineJob, + fallbackBackend, + inlineBackend, + isJobFailure, + JobFailure, + prepareJob, + WorkerError, + type AnyJob, + type JobBackend, + type JobContext, + type JobInfo, +} from "../src/index"; + +let passed = 0; +const failures: string[] = []; +async function check(name: string, fn: () => unknown): Promise<void> { + try { + await fn(); + passed++; + } catch (e) { + failures.push(`✗ ${name}\n ${(e as Error).stack?.split("\n").slice(0, 4).join("\n ")}`); + } +} +const code = async (fn: () => unknown, expected: string) => { + try { + await fn(); + } catch (e) { + assert.ok(e instanceof WorkerError, `expected WorkerError, got ${String(e)}`); + assert.equal(e.code, expected, e.message); + return e.message; + } + assert.fail(`expected ${expected}`); +}; +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +async function waitFor<T>(fn: () => Promise<T | null | undefined | false>, ms = 10_000): Promise<T> { + const end = Date.now() + ms; + for (;;) { + const v = await fn(); + if (v) return v; + if (Date.now() > end) assert.fail("timed out"); + await sleep(20); + } +} +const quiet = () => undefined; + +// ---- the scenario every backend must run identically -------------------------------------------- + +interface SumData { + a: number; + b: number; + at: Date | string; +} +function scenario() { + const errors: string[] = []; + const calls: Record<string, number[]> = {}; + const seen = (ctx: JobContext) => (calls[ctx.name] ??= []).push(ctx.attempt); + const jobs = { + sum: defineJob<SumData, { sum: number; at: unknown; when: Date }>({ + name: "sum", + queue: "maths", + process: (d, ctx) => { + seen(ctx); + return { sum: d.a + d.b, at: d.at, when: new Date(0) }; + }, + }), + flaky: defineJob<{ n: number }, string>({ + name: "flaky", + queue: "maths", + attempts: 3, + backoffMs: 10, + process: (d, ctx) => { + seen(ctx); + if (ctx.attempt < 3) throw new JobFailure("Busy, try again", true); + return `ok ${d.n} on ${ctx.attempt}/${ctx.attempts}`; + }, + }), + fatal: defineJob<{ invoiceId: string }, never>({ + name: "fatal", + attempts: 5, + backoffMs: 10, + process: (d, ctx) => { + seen(ctx); + throw new JobFailure(`No invoice ${d.invoiceId}`, false); + }, + }), + crash: defineJob<{ id: string }, never>({ + name: "crash", + attempts: 2, + backoffMs: 10, + process: (_d, ctx) => { + seen(ctx); + throw new Error("connect ECONNREFUSED db.internal:5432 user=admin"); + }, + }), + }; + const onError = (e: unknown) => errors.push((e as Error).message); + return { jobs, list: Object.values(jobs) as AnyJob[], errors, calls, onError }; +} + +/** Runs the scenario on a backend; returns comparable results. */ +async function runScenario(backend: JobBackend, s: ReturnType<typeof scenario>) { + const q = createJobQueue({ backend, jobs: s.list }); + const started = [ + await q.enqueue(s.jobs.sum, { a: 2, b: 3, at: new Date("2026-10-14T10:00:00.000Z") }), + await q.enqueue(s.jobs.flaky, { n: 7 }), + await q.enqueue(s.jobs.fatal, { invoiceId: "INV-1" }), + await q.enqueue(s.jobs.crash, { id: "x" }), + ]; + const done: JobInfo[] = []; + for (const j of started) { + const def = q.definition(j.name); + done.push(await waitFor(async () => { + const info = await q.get(def, j.id); + return info && (info.state === "completed" || info.state === "failed") ? info : null; + })); + } + return { q, done: done.map(({ id: _id, createdAt, finishedAt, ...rest }) => ({ ...rest, timed: createdAt instanceof Date && finishedAt instanceof Date && finishedAt >= createdAt })) }; +} + +const expected = [ + { name: "sum", queue: "maths", state: "completed", attempts: 1, error: null, result: { sum: 5, at: "2026-10-14T10:00:00.000Z", when: "1970-01-01T00:00:00.000Z" }, data: { a: 2, b: 3, at: "2026-10-14T10:00:00.000Z" }, timed: true }, + { name: "flaky", queue: "maths", state: "completed", attempts: 3, error: null, result: "ok 7 on 3/3", data: { n: 7 }, timed: true }, + { name: "fatal", queue: "fatal", state: "failed", attempts: 1, error: "No invoice INV-1", result: null, data: { invoiceId: "INV-1" }, timed: true }, + { name: "crash", queue: "crash", state: "failed", attempts: 2, error: "Job failed", result: null, data: { id: "x" }, timed: true }, +]; + +// ---- definitions and data rules ----------------------------------------------------------------- + +await check("defineJob: defaults, queue = name, frozen", () => { + const j = defineJob({ name: "mail", process: () => 1 }); + assert.equal(j.queue, "mail"); + assert.equal(j.attempts, 3); + assert.equal(j.backoffMs, 2000); + assert.equal(j.keepResultsMs, 3600_000); + assert.equal(j.keepFailedMs, 86_400_000); + assert.equal(j.refuseSecrets, true); + assert.ok(Object.isFrozen(j)); + assert.equal(defineJob({ name: "x", keepResultsMs: 2 * 86_400_000, process: () => 1 }).keepFailedMs, 2 * 86_400_000); + assert.equal(defineJob({ name: "x", attempts: 0, process: () => 1 }).attempts, 1); +}); + +await check("defineJob: names are checked (they become Redis keys)", async () => { + await code(() => defineJob({ name: "Bad Name", process: () => 1 }), "WORKER_BAD_DATA"); + await code(() => defineJob({ name: "ok", queue: "a:b", process: () => 1 }), "WORKER_BAD_DATA"); + await code(() => defineJob({ name: "", process: () => 1 }), "WORKER_BAD_DATA"); +}); + +await check("data crosses JSON: Dates become ISO strings, undefined keys drop", () => { + const j = defineJob<Record<string, unknown>, null>({ name: "j", process: () => null }); + assert.deepEqual(prepareJob(j, { at: new Date(0), gone: undefined, n: 1 }).data, { at: "1970-01-01T00:00:00.000Z", n: 1 }); +}); + +await check("data that is not JSON, too large, or carries secrets is refused in the caller", async () => { + const j = defineJob<unknown, null>({ name: "j", maxBytes: 100, process: () => null }); + await code(() => prepareJob(j, { n: 1n }), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, undefined), "WORKER_BAD_DATA"); + const cyclic: Record<string, unknown> = {}; + cyclic.self = cyclic; + await code(() => prepareJob(j, cyclic), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, { s: "x".repeat(200) }), "WORKER_BAD_DATA"); + const m = await code(() => prepareJob(j, { user: { id: "u1", resetToken: "abc" } }), "WORKER_BAD_DATA"); + assert.match(m, /user\.resetToken/); + assert.doesNotMatch(m, /abc/); + await code(() => prepareJob(j, { list: [{ PASSWORD: "x" }] }), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, { headers: { authorization: "Bearer x" } }), "WORKER_BAD_DATA"); + // ids and links are fine; opting out is explicit + prepareJob(j, { userId: "u1", tokenId: "t1", link: "https://x.example/r" }); + prepareJob(defineJob<unknown, null>({ name: "k", refuseSecrets: false, process: () => null }), { token: "x" }); +}); + +await check("validate() runs before enqueueing; its error becomes WORKER_BAD_DATA", async () => { + let ran = false; + const j = defineJob<{ email: string }, null>({ + name: "j", + validate: (d) => { + if (!d.email.includes("@")) throw new Error("email is not valid"); + }, + process: () => { + ran = true; + return null; + }, + }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [j] }); + assert.match(await code(() => q.enqueue(j, { email: "nope" }), "WORKER_BAD_DATA"), /email is not valid/); + assert.equal(ran, false); + assert.equal((await q.enqueue(j, { email: "a@example.com" })).state, "completed"); +}); + +await check("every job id is a fresh UUID", () => { + const j = defineJob({ name: "j", process: () => 1 }); + const ids = new Set(Array.from({ length: 50 }, () => prepareJob(j, {}).id)); + assert.equal(ids.size, 50); + for (const id of ids) assert.match(id, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/); +}); + +await check("JobFailure is recognised across package copies", () => { + const foreign = Object.assign(new Error("x"), { name: "JobFailure", retry: false }); + assert.ok(isJobFailure(foreign)); + assert.ok(isJobFailure(new JobFailure("y"))); + assert.equal(new JobFailure("y").retry, true); + assert.ok(!isJobFailure(new Error("z"))); +}); + +// ---- the producer API --------------------------------------------------------------------------- + +await check("createJobQueue: unknown jobs, duplicate names, foreign definitions", async () => { + const a = defineJob({ name: "a", process: () => 1 }); + await code(() => createJobQueue({ backend: inlineBackend(), jobs: [a, defineJob({ name: "a", process: () => 2 })] }), "WORKER_DUPLICATE_JOB"); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a] }); + assert.deepEqual(q.names(), ["a"]); + assert.equal(q.mode, "inline"); + await code(() => q.enqueue("b", {}), "WORKER_UNKNOWN_JOB"); + await code(() => q.enqueue(defineJob({ name: "a", process: () => 3 }), {}), "WORKER_UNKNOWN_JOB"); + assert.equal((await q.enqueue("a", {})).result, 1); + assert.equal(await q.get(a, ""), null); +}); + +await check("get only returns a job under its own definition", async () => { + const a = defineJob({ name: "a", queue: "shared", process: () => "A" }); + const b = defineJob({ name: "b", queue: "shared", process: () => "B" }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a, b] }); + const job = await q.enqueue(a, {}); + assert.equal((await q.get(a, job.id))?.result, "A"); + assert.equal(await q.get(b, job.id), null); + assert.equal(await q.get(a, "00000000-0000-4000-8000-000000000000"), null); +}); + +await check("close: enqueue refused afterwards, available() false", async () => { + const a = defineJob({ name: "a", process: () => 1 }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a] }); + assert.equal(await q.available(), true); + await q.close(); + await q.close(); + assert.equal(await q.available(), false); + await code(() => q.enqueue(a, {}), "WORKER_CLOSED"); +}); + +// ---- in-process --------------------------------------------------------------------------------- + +await check("inline (waiting): the scenario", async () => { + const s = scenario(); + const { done } = await runScenario(inlineBackend({ onError: s.onError }), s); + assert.deepEqual(done, expected); + assert.deepEqual(s.calls, { sum: [1], flaky: [1, 2, 3], fatal: [1], crash: [1, 2] }); + assert.equal(s.errors.length, 2); +}); + +await check("inline (background): the scenario, with backoff between attempts", async () => { + const s = scenario(); + const t0 = Date.now(); + const { done } = await runScenario(inlineBackend({ wait: false, onError: s.onError }), s); + assert.deepEqual(done, expected); + assert.ok(Date.now() - t0 >= 30, "flaky waited 10 + 20 ms"); +}); + +await check("inline (background): add resolves before the job runs", async () => { + let release!: () => void; + const gate = new Promise<void>((r) => (release = r)); + const j = defineJob({ name: "slow", process: async () => (await gate, "done") }); + const q = createJobQueue({ backend: inlineBackend({ wait: false }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.state, "waiting"); + await sleep(5); + assert.equal((await q.get(j, info.id))?.state, "active"); + release(); + assert.equal((await waitFor(async () => ((await q.get(j, info.id))?.state === "completed" ? q.get(j, info.id) : null)))?.result, "done"); +}); + +await check("inline: errors that are not JobFailure are hidden and passed to onError", async () => { + const logged: unknown[] = []; + const j = defineJob({ name: "j", attempts: 1, process: () => Promise.reject(new Error("secret internals")) }); + const q = createJobQueue({ backend: inlineBackend({ onError: (e, ctx) => logged.push([(e as Error).message, ctx.name, ctx.attempt]) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.state, "failed"); + assert.equal(info.error, "Job failed"); + assert.deepEqual(logged, [["secret internals", "j", 1]]); +}); + +await check("inline: a result that is not JSON fails without retrying", async () => { + let runs = 0; + const j = defineJob({ name: "j", process: () => (runs++, { n: 1n }) }); + const info = await createJobQueue({ backend: inlineBackend({ onError: quiet }), jobs: [j] }).enqueue(j, {}); + assert.equal(info.state, "failed"); + assert.equal(runs, 1); +}); + +await check("inline: delayed jobs wait, then run", async () => { + const j = defineJob({ name: "later", process: () => "ran" }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [j] }); + const info = await q.enqueue(j, {}, { delayMs: 40 }); + assert.equal(info.state, "delayed"); + await sleep(10); + assert.equal((await q.get(j, info.id))?.state, "delayed"); + const done = await waitFor(async () => ((await q.get(j, info.id))?.state === "completed" ? q.get(j, info.id) : null)); + assert.equal(done?.result, "ran"); +}); + +await check("inline: results are kept for keepResultsMs, failures for keepFailedMs", async () => { + const gone = defineJob({ name: "gone", keepResultsMs: 0, process: () => "x" }); + const brief = defineJob({ name: "brief", keepResultsMs: 30, process: () => "y" }); + const bad = defineJob({ name: "bad", attempts: 1, keepFailedMs: 30, process: () => Promise.reject(new JobFailure("no", false)) }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [gone, brief, bad] }); + const g = await q.enqueue(gone, {}); + assert.equal(g.result, "x", "the caller still sees the outcome"); + assert.equal(await q.get(gone, g.id), null); + const b = await q.enqueue(brief, {}); + const f = await q.enqueue(bad, {}); + assert.equal((await q.get(brief, b.id))?.result, "y"); + assert.equal((await q.get(bad, f.id))?.error, "no"); + await sleep(40); + assert.equal(await q.get(brief, b.id), null); + assert.equal(await q.get(bad, f.id), null); +}); + +await check("inline: close finishes jobs in hand and drops pending delays", async () => { + let finished = false; + let delayedRan = false; + const slow = defineJob({ name: "slow", process: async () => (await sleep(30), (finished = true)) }); + const later = defineJob({ name: "later", process: () => (delayedRan = true) }); + const q = createJobQueue({ backend: inlineBackend({ wait: false }), jobs: [slow, later] }); + await q.enqueue(slow, {}); + await q.enqueue(later, {}, { delayMs: 60_000 }); + await sleep(5); + const t0 = Date.now(); + await q.close(); + assert.ok(finished); + assert.ok(!delayedRan); + assert.ok(Date.now() - t0 < 1000, "did not wait for the delay"); +}); + +// ---- fallback ----------------------------------------------------------------------------------- + +function fakeBackend(o: { available?: boolean; fails?: boolean }) { + const added: string[] = []; + const inner = inlineBackend(); + const b: JobBackend = { + mode: "fake", + async add(job) { + if (o.fails) throw new Error("redis went away"); + added.push(job.def.name); + return inner.add(job); + }, + get: (def, id) => inner.get(def, id), + available: () => o.available ?? true, + close: () => inner.close(), + }; + return { b, added }; +} + +await check("fallback: the primary when available", async () => { + const p = fakeBackend({}); + const fellBack: unknown[] = []; + const j = defineJob({ name: "j", process: () => 1 }); + const q = createJobQueue({ backend: fallbackBackend({ primary: p.b, fallback: inlineBackend(), onFallback: (r) => fellBack.push(r) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.deepEqual(p.added, ["j"]); + assert.equal(fellBack.length, 0); + assert.equal((await q.get(j, info.id))?.result, 1); + assert.equal(q.mode, "fallback"); +}); + +await check("fallback: standalone when the primary is unavailable or fails", async () => { + for (const o of [{ available: false }, { fails: true }]) { + const p = fakeBackend(o); + const reasons: string[] = []; + const j = defineJob({ name: "j", process: () => 2 }); + const q = createJobQueue({ backend: fallbackBackend({ primary: p.b, fallback: inlineBackend(), onFallback: (r) => reasons.push((r as Error).message) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.result, 2); + assert.deepEqual(p.added, []); + assert.equal(reasons.length, 1); + assert.equal((await q.get(j, info.id))?.result, 2, "found on the fallback"); + assert.equal(await q.available(), true); + } +}); + +await check("fallback: whenFails 'throw' raises WORKER_UNAVAILABLE", async () => { + const j = defineJob({ name: "j", process: () => 1 }); + for (const o of [{ available: false }, { fails: true }]) { + const q = createJobQueue({ backend: fallbackBackend({ primary: fakeBackend(o).b, fallback: inlineBackend(), whenFails: "throw" }), jobs: [j] }); + await code(() => q.enqueue(j, {}), "WORKER_UNAVAILABLE"); + } +}); + +await check("enqueuer: mail-kit's MailQueue shape", async () => { + // mail-kit: interface MailQueue { enqueue(job): Promise<string | undefined>; available?(): boolean | Promise<boolean> } + interface MailQueueShape { + enqueue(job: { v: 1; template: string }): Promise<string | undefined>; + available?(): boolean | Promise<boolean>; + } + const sent: string[] = []; + const mail = defineJob<{ v: 1; template: string }, null>({ name: "mail", keepResultsMs: 0, process: (d) => (sent.push(d.template), null) }); + const broken = defineJob<{ v: 1; template: string }, null>({ name: "broken", attempts: 1, process: () => Promise.reject(new JobFailure("SMTP refused", false)) }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [mail, broken] }); + const mq: MailQueueShape = q.enqueuer(mail); + const id = await mq.enqueue({ v: 1, template: "password-reset" }); + assert.equal(typeof id, "string"); + assert.deepEqual(sent, ["password-reset"]); + assert.equal(await mq.available?.(), true); + await assert.rejects(q.enqueuer(broken).enqueue({ v: 1, template: "x" }), /SMTP refused/); +}); + +// ---- BullMQ on Redis ---------------------------------------------------------------------------- + +async function freePort(): Promise<number> { + return new Promise((resolve) => { + const s = createServer().listen(0, "127.0.0.1", () => { + const port = (s.address() as { port: number }).port; + s.close(() => resolve(port)); + }); + }); +} + +let redisUrl = process.env.REDIS_URL; +let redisServer: ChildProcess | undefined; +if (!redisUrl && spawnSync("redis-server", ["--version"]).status === 0) { + const port = await freePort(); + redisServer = spawn("redis-server", ["--port", String(port), "--save", "", "--appendonly", "no"], { stdio: "ignore" }); + redisUrl = `redis://127.0.0.1:${port}`; + const probe = new Redis(redisUrl, { lazyConnect: true, maxRetriesPerRequest: 0, retryStrategy: () => 50 }); + probe.on("error", quiet); + await waitFor(async () => (await probe.connect().then(() => true, () => false)) && (await probe.ping()) === "PONG", 5000); + probe.disconnect(); +} + +if (!redisUrl) { + if (process.env.REQUIRE_REDIS) failures.push("✗ BullMQ checks: no Redis (set REDIS_URL or install redis-server)"); + else console.log("- BullMQ checks skipped: no REDIS_URL and no redis-server"); +} else { + const prefix = `wk-test-${process.pid}`; + const admin = new Redis(redisUrl); + + await check("bullmq: the scenario gives the same results as in-process", async () => { + const s = scenario(); + const worker = startWorker({ redisUrl, prefix, jobs: s.list, concurrency: 2, onError: s.onError, onFailed: quiet }); + assert.deepEqual(worker.queues.sort(), ["crash", "fatal", "maths"]); + const backend = bullmqBackend({ redisUrl, prefix }); + await waitFor(async () => backend.available?.()); + const { q, done } = await runScenario(backend, s); + assert.equal(q.mode, "bullmq"); + assert.deepEqual(done, expected); + assert.deepEqual(s.calls, { sum: [1], flaky: [1, 2, 3], fatal: [1], crash: [1, 2] }); + assert.equal(s.errors.length, 2); + await worker.close(); + await q.close(); + }); + + await check("bullmq: keys use the prefix and queue; job options come from the definition", async () => { + const j = defineJob({ name: "keys", queue: "kq", attempts: 4, backoffMs: 500, keepResultsMs: 60_000, keepFailedMs: 120_000, process: () => 1 }); + const prepared = prepareJob(j, { id: "a" }, { delayMs: 1000 }); + assert.deepEqual(bullJobOptions(prepared), { jobId: prepared.id, attempts: 4, backoff: { type: "exponential", delay: 500 }, delay: 1000, removeOnComplete: { age: 60 }, removeOnFail: { age: 120 } }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [j] }); + const info = await q.enqueue(j, { id: "a" }, { delayMs: 60_000 }); + assert.equal(info.state, "delayed"); + assert.equal((await q.get(j, info.id))?.state, "delayed"); + assert.ok((await admin.keys(`${prefix}:kq:*`)).length > 0); + await q.close(); + }); + + await check("bullmq: keepResultsMs 0 removes the job (and its data) once done", async () => { + const j = defineJob({ name: "mail", keepResultsMs: 0, process: () => "sent" }); + const worker = startWorker({ redisUrl, prefix, jobs: [j] }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [j] }); + const info = await q.enqueue(j, { link: "https://x.example/reset/abc" }); + await waitFor(async () => (await q.get(j, info.id)) === null); + assert.equal(await admin.exists(`${prefix}:mail:${info.id}`), 0); + await worker.close(); + await q.close(); + }); + + await check("bullmq: get under another definition returns null", async () => { + const a = defineJob({ name: "a", queue: "same", process: () => 1 }); + const b = defineJob({ name: "b", queue: "same", process: () => 2 }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [a, b] }); + const info = await q.enqueue(a, {}, { delayMs: 60_000 }); + assert.equal((await q.get(a, info.id))?.name, "a"); + assert.equal(await q.get(b, info.id), null); + await q.close(); + }); + + await check("bullmq: a job name the worker does not know fails at once", async () => { + const known = defineJob({ name: "known", queue: "mixed", process: () => 1 }); + const failed: { name: string; final: boolean }[] = []; + const worker = startWorker({ redisUrl, prefix, jobs: [known], onFailed: (f) => failed.push(f) }); + const raw = new Queue("mixed", { connection: redisConnection(redisUrl), prefix }); + await raw.add("stranger", {}, { attempts: 3 }); + await waitFor(async () => failed.length > 0); + assert.deepEqual(failed.map((f) => [f.name, f.final]), [["stranger", true]]); + await raw.close(); + await worker.close(); + }); + + await check("bullmq: a shared connection is not closed by worker-kit", async () => { + const connection = redisConnection(redisUrl); + const j = defineJob({ name: "shared", process: () => "x" }); + const worker = startWorker({ connection, prefix, jobs: [j] }); + const q = createJobQueue({ backend: bullmqBackend({ connection, prefix }), jobs: [j] }); + const info = await q.enqueue(j, {}); + await waitFor(async () => (await q.get(j, info.id))?.state === "completed"); + await worker.close(); + await q.close(); + assert.equal(await connection.ping(), "PONG"); + await connection.quit(); + }); + + await check("bullmq + fallback: Redis down → jobs run standalone, without waiting", async () => { + const port = await freePort(); // nothing listens here + const down = new Redis(`redis://127.0.0.1:${port}`, { maxRetriesPerRequest: null, lazyConnect: true }); + down.on("error", quiet); + const ran: string[] = []; + const j = defineJob<{ to: string }, null>({ name: "mail", process: (d) => (ran.push(d.to), null) }); + const reasons: string[] = []; + const q = createJobQueue({ + backend: fallbackBackend({ primary: bullmqBackend({ connection: down, prefix }), fallback: inlineBackend(), onFallback: (r) => reasons.push((r as Error).message) }), + jobs: [j], + }); + const t0 = Date.now(); + const info = await q.enqueue(j, { to: "a@example.com" }); + assert.ok(Date.now() - t0 < 1000); + assert.equal(info.state, "completed"); + assert.deepEqual(ran, ["a@example.com"]); + assert.match(reasons[0] ?? "", /bullmq backend is not available/); + await q.close(); + down.disconnect(); + }); + + await check("bullmq: needs a connection or a URL", async () => { + await code(() => bullmqBackend({}), "WORKER_UNAVAILABLE"); + }); + + const leftovers = await admin.keys(`${prefix}:*`); + if (leftovers.length) await admin.del(...leftovers); + await admin.quit(); +} +redisServer?.kill(); + +for (const f of failures) console.log(f); +console.log(`\n${passed} passed, ${failures.length} failed`); +process.exit(failures.length ? 1 : 0); +``` + + +--- + +## File: src/bullmq.ts + +```ts +// "/bullmq" entry: the BullMQ-on-Redis backend (producer) and the worker runner (consumer). +// bullmq and ioredis are optional peers; nothing else in the package imports them. +import { Job, Queue, UnrecoverableError, Worker, type ConnectionOptions, type JobsOptions } from "bullmq"; +import { Redis } from "ioredis"; +import { logError, runAttempt, type ErrorHandler } from "./define"; +import { isJobFailure, WorkerError } from "./errors"; +import { jobRegistry } from "./queue"; +import type { AnyJob, JobBackend, JobInfo, JobState, PreparedJob } from "./types"; + +export interface RedisOptions { + /** An ioredis connection you own (not closed by worker-kit). Give this or `redisUrl`. */ + connection?: Redis; + /** worker-kit opens (and closes) its own connection. */ + redisUrl?: string; + /** Redis key prefix: keys are `<prefix>:<queue>:…`. Default "mgpl". */ + prefix?: string; +} + +/** An ioredis connection as BullMQ needs it (`maxRetriesPerRequest: null` for blocking commands). */ +export function redisConnection(url: string): Redis { + return new Redis(url, { maxRetriesPerRequest: null }); +} + +/** `available` check for a connection: true only while it is "ready" (ioredis would otherwise queue commands while down). */ +export const redisReady = (connection: Redis) => () => connection.status === "ready"; + +function own(o: RedisOptions): { connection: Redis; owned: boolean } { + if (o.connection) return { connection: o.connection, owned: false }; + if (o.redisUrl) { + const connection = redisConnection(o.redisUrl); + // ioredis reconnects by itself; `available()` reports the outage. Without a listener every retry is printed. + connection.on("error", () => undefined); + return { connection, owned: true }; + } + throw new WorkerError("WORKER_UNAVAILABLE", "Give a Redis `connection` or a `redisUrl`"); +} + +const seconds = (ms: number) => Math.max(1, Math.ceil(ms / 1000)); + +/** BullMQ job options from a definition: the only place they are set. */ +export function bullJobOptions(job: PreparedJob): JobsOptions { + const { def } = job; + return { + jobId: job.id, + attempts: def.attempts, + backoff: { type: "exponential", delay: def.backoffMs }, + ...(job.delayMs > 0 ? { delay: job.delayMs } : {}), + removeOnComplete: def.keepResultsMs === 0 ? true : { age: seconds(def.keepResultsMs) }, + removeOnFail: def.keepFailedMs === 0 ? true : { age: seconds(def.keepFailedMs) }, + }; +} + +const STATES: Record<string, JobState> = { waiting: "waiting", "waiting-children": "waiting", prioritized: "waiting", active: "active", completed: "completed", failed: "failed", delayed: "delayed" }; + +async function toInfo(job: Job, queue: string): Promise<JobInfo> { + const state = STATES[await job.getState()] ?? "waiting"; + return { + id: job.id!, + name: job.name, + queue, + state, + attempts: job.attemptsMade, + error: state === "completed" ? null : job.failedReason || null, + result: state === "completed" ? ((job.returnvalue as unknown) ?? null) : null, + data: job.data as unknown, + createdAt: new Date(job.timestamp), + finishedAt: job.finishedOn && (state === "completed" || state === "failed") ? new Date(job.finishedOn) : null, + }; +} + +/** + * The producer side (API processes): adds jobs to Redis. Run `startWorker` somewhere to process them. + * `available()` is false while the Redis connection is not "ready" (use it with fallbackBackend). + */ +export function bullmqBackend(o: RedisOptions): JobBackend { + const { connection, owned } = own(o); + const prefix = o.prefix ?? "mgpl"; + const queues = new Map<string, Queue>(); + const queueFor = (name: string) => { + let q = queues.get(name); + if (!q) { + q = new Queue(name, { connection: connection as unknown as ConnectionOptions, prefix }); + queues.set(name, q); + } + return q; + }; + return { + mode: "bullmq", + async add(job) { + const added = await queueFor(job.def.queue).add(job.def.name, job.data, bullJobOptions(job)); + return toInfo(added, job.def.queue); + }, + async get(def: AnyJob, id: string) { + const job = await Job.fromId(queueFor(def.queue), id); + return job && job.name === def.name ? toInfo(job, def.queue) : null; + }, + available: redisReady(connection), + async close() { + await Promise.all([...queues.values()].map((q) => q.close())); + if (owned) await connection.quit().catch(() => undefined); + }, + }; +} + +export interface StartWorkerOptions extends RedisOptions { + /** The definitions to run. One BullMQ Worker per distinct queue. */ + jobs: readonly AnyJob[]; + /** Only these queues (default: every queue of `jobs`). */ + queues?: readonly string[]; + /** Jobs a worker runs at once, per queue. Default 4. */ + concurrency?: number; + /** Errors that are not a JobFailure (with the job context). Default: one line on stderr. */ + onError?: ErrorHandler; + /** Each failed attempt (after onError). Default: one warning line. */ + onFailed?: (info: { id: string | undefined; name: string; queue: string; attempt: number; error: string; final: boolean }) => void; +} + +export interface RunningWorker { + queues: string[]; + workers: Worker[]; + /** Finishes the jobs in hand, then stops. */ + close(): Promise<void>; +} + +/** The consumer side (`node dist/worker.js` in the api): runs the definitions' processors. */ +export function startWorker(o: StartWorkerOptions): RunningWorker { + const byName = jobRegistry(o.jobs); + const { connection, owned } = own(o); + const prefix = o.prefix ?? "mgpl"; + const onError = o.onError ?? logError; + const onFailed = + o.onFailed ?? ((f) => console.warn(`[worker-kit] ${f.queue}/${f.name} job ${f.id} failed (attempt ${f.attempt}${f.final ? ", final" : ""}): ${f.error}`)); + const names = [...new Set(o.jobs.map((j) => j.queue))].filter((q) => !o.queues || o.queues.includes(q)); + + const workers = names.map((queue) => { + const w = new Worker( + queue, + async (job: Job) => { + const def = byName.get(job.name); + if (!def || def.queue !== queue) throw new UnrecoverableError(`Unknown job "${job.name}" on queue "${queue}"`); + const ctx = { id: job.id ?? "", name: job.name, queue, attempt: job.attemptsMade + 1, attempts: job.opts.attempts ?? def.attempts }; + try { + return await runAttempt(def, job.data, ctx, onError); + } catch (e) { + const message = isJobFailure(e) ? e.message : "Job failed"; + // a JobFailure that retrying cannot fix fails the job at once + if (isJobFailure(e) && !e.retry) throw new UnrecoverableError(message); + throw new Error(message); + } + }, + { connection: connection as unknown as ConnectionOptions, prefix, concurrency: Math.max(1, o.concurrency ?? 4) }, + ); + w.on("failed", (job, err) => { + if (!job) return; + const final = err instanceof UnrecoverableError || err.name === "UnrecoverableError" || job.attemptsMade >= (job.opts.attempts ?? 1); + onFailed({ id: job.id, name: job.name, queue, attempt: job.attemptsMade, error: err.message, final }); + }); + // without a listener an "error" event (e.g. Redis dropped) would crash the process + w.on("error", (err) => console.error(`[worker-kit] ${queue} worker:`, err.message)); + return w; + }); + + return { + queues: names, + workers, + async close() { + await Promise.all(workers.map((w) => w.close())); + if (owned) await connection.quit().catch(() => undefined); + }, + }; +} +``` + + +--- + +## File: src/define.ts + +```ts +import { randomUUID } from "node:crypto"; +import { GENERIC_FAILURE, isJobFailure, JobFailure, WorkerError } from "./errors"; +import type { AnyJob, EnqueueOptions, JobContext, JobDefinition, PreparedJob } from "./types"; + +export interface DefineJobOptions<Data, Result> { + name: string; + queue?: string; + validate?: (data: Data) => void; + process: (data: Data, ctx: JobContext) => Promise<Result> | Result; + /** Default 3. */ + attempts?: number; + /** First retry delay; doubles each time. Default 2 s. */ + backoffMs?: number; + /** Default 1 hour. Use 0 for jobs whose data must not linger (e.g. a mail with a reset link). */ + keepResultsMs?: number; + /** Default max(1 day, keepResultsMs). */ + keepFailedMs?: number; + /** Default 512 KiB. */ + maxBytes?: number; + /** Default true. */ + refuseSecrets?: boolean; +} + +const NAME = /^[a-z0-9][a-z0-9._-]{0,63}$/; + +/** Keys that must not appear in job data: payloads carry ids, never tokens or secrets. */ +export const SECRET_KEY = /^(password|passwordhash|token|accesstoken|refreshtoken|resettoken|secret|clientsecret|apikey|api_key|authorization|cookie)$/i; + +export function defineJob<Data, Result>(o: DefineJobOptions<Data, Result>): JobDefinition<Data, Result> { + const queue = o.queue ?? o.name; + if (!NAME.test(o.name)) throw new WorkerError("WORKER_BAD_DATA", `Job name "${o.name}" must match ${NAME}`); + if (!NAME.test(queue)) throw new WorkerError("WORKER_BAD_DATA", `Queue name "${queue}" must match ${NAME}`); + const attempts = Math.max(1, Math.floor(o.attempts ?? 3)); + const keepResultsMs = Math.max(0, o.keepResultsMs ?? 3600_000); + return Object.freeze({ + name: o.name, + queue, + validate: o.validate, + process: o.process, + attempts, + backoffMs: Math.max(0, o.backoffMs ?? 2000), + keepResultsMs, + keepFailedMs: Math.max(0, o.keepFailedMs ?? Math.max(86_400_000, keepResultsMs)), + maxBytes: o.maxBytes ?? 512 * 1024, + refuseSecrets: o.refuseSecrets ?? true, + }); +} + +/** Delay before attempt `n + 1` after `n` failed attempts (exponential, the same as BullMQ's). */ +export const backoffDelay = (def: AnyJob, failedAttempts: number) => def.backoffMs * 2 ** (failedAttempts - 1); + +function findSecret(v: unknown, path: string, seen = 0): string | null { + if (seen > 64 || v === null || typeof v !== "object") return null; + if (Array.isArray(v)) { + for (let i = 0; i < v.length; i++) { + const hit = findSecret(v[i], `${path}[${i}]`, seen + 1); + if (hit) return hit; + } + return null; + } + for (const [k, child] of Object.entries(v)) { + if (SECRET_KEY.test(k)) return path ? `${path}.${k}` : k; + const hit = findSecret(child, path ? `${path}.${k}` : k, seen + 1); + if (hit) return hit; + } + return null; +} + +/** JSON round trip with the definition's checks, so every backend sees exactly what Redis would hold. */ +export function toJsonData(def: AnyJob, data: unknown): unknown { + let json: string | undefined; + try { + json = JSON.stringify(data); + } catch (e) { + throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is not JSON (${(e as Error).message})`); + } + if (json === undefined) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is not JSON`); + const bytes = Buffer.byteLength(json); + if (bytes > def.maxBytes) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is ${bytes} bytes, the limit is ${def.maxBytes}`); + const parsed: unknown = JSON.parse(json); + if (def.refuseSecrets) { + const key = findSecret(parsed, ""); + if (key) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data carries a secret-looking key "${key}"; pass an id and look the value up in the processor`); + } + return parsed; +} + +/** Validates and serialises a job for a backend. Runs in the caller, so bad data fails the request. */ +export function prepareJob(def: AnyJob, data: unknown, o: EnqueueOptions = {}): PreparedJob { + if (def.validate) { + try { + def.validate(data); + } catch (e) { + throw e instanceof WorkerError ? e : new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": ${(e as Error).message}`); + } + } + return { + id: randomUUID(), // unguessable; callers still check ownership on every read + def, + data: toJsonData(def, data), + delayMs: Math.max(0, Math.floor(o.delayMs ?? 0)), + }; +} + +export type ErrorHandler = (error: unknown, ctx: JobContext) => void; + +/** + * Runs one attempt. Every backend goes through here, so failures read the same everywhere: + * a JobFailure keeps its message and retry flag; anything else goes to `onError` (log it) and + * becomes a retryable "Job failed", so internals never reach the client. + */ +export async function runAttempt(def: AnyJob, data: unknown, ctx: JobContext, onError?: ErrorHandler): Promise<unknown> { + let result: unknown; + try { + result = await def.process(data, ctx); + } catch (e) { + if (isJobFailure(e)) throw e; + onError?.(e, ctx); + throw new JobFailure(GENERIC_FAILURE, true); + } + if (result === undefined) return null; + try { + return JSON.parse(JSON.stringify(result)) as unknown; + } catch (e) { + onError?.(e, ctx); + throw new JobFailure(GENERIC_FAILURE, false); + } +} + +/** The default onError: one line on stderr, without the job's data. */ +export const logError: ErrorHandler = (e, ctx) => { + console.error(`[worker-kit] ${ctx.queue}/${ctx.name} job ${ctx.id} attempt ${ctx.attempt}/${ctx.attempts} failed:`, e); +}; +``` + + +--- + +## File: src/errors.ts + +```ts +/** Error codes are a public contract: add, never rename. */ +export type WorkerErrorCode = + | "WORKER_UNKNOWN_JOB" // enqueue/get/run of a job name that is not defined + | "WORKER_BAD_DATA" // data is not JSON, too large, carries a secret key, or failed validate() + | "WORKER_DUPLICATE_JOB" // two definitions with the same name + | "WORKER_UNAVAILABLE" // no backend could take the job + | "WORKER_CLOSED"; // the queue was closed + +export class WorkerError extends Error { + constructor(readonly code: WorkerErrorCode, message: string) { + super(message); + this.name = "WorkerError"; + } +} + +/** + * Throw from a processor for a failure whose message is safe to show to the client. + * `retry: false` fails the job at once (retrying cannot help); `true` uses the remaining attempts. + * Any other error is logged through `onError` and shown as "Job failed". + */ +export class JobFailure extends Error { + constructor(message: string, readonly retry: boolean = true) { + super(message); + this.name = "JobFailure"; + } +} + +/** True for a JobFailure, also one made by another copy of this package. */ +export function isJobFailure(e: unknown): e is JobFailure { + return e instanceof JobFailure || (e instanceof Error && e.name === "JobFailure" && typeof (e as { retry?: unknown }).retry === "boolean"); +} + +/** The message shown for failures that are not a JobFailure. */ +export const GENERIC_FAILURE = "Job failed"; +``` + + +--- + +## File: src/fallback.ts + +```ts +// "The worker if it is there, otherwise standalone": jobs go to the primary backend (BullMQ on +// Redis) while it is available, and run on the fallback (usually in-process) when it is not. +import { WorkerError } from "./errors"; +import type { JobBackend, PreparedJob } from "./types"; + +export interface FallbackBackendOptions { + primary: JobBackend; + fallback: JobBackend; + /** When the primary is unavailable or `add` fails: "fallback" (default) or "throw" (WORKER_UNAVAILABLE). */ + whenFails?: "fallback" | "throw"; + /** Called each time a job went to the fallback (log it). */ + onFallback?: (reason: unknown, job: PreparedJob) => void; +} + +export function fallbackBackend(o: FallbackBackendOptions): JobBackend { + const { primary, fallback } = o; + const isAvailable = async (b: JobBackend) => { + try { + return (await (b.available?.() ?? true)) === true; + } catch { + return false; + } + }; + return { + mode: "fallback", + async add(job) { + let reason: unknown; + if (await isAvailable(primary)) { + try { + return await primary.add(job); + } catch (e) { + reason = e; + } + } else { + reason = new WorkerError("WORKER_UNAVAILABLE", `The ${primary.mode} backend is not available`); + } + if (o.whenFails === "throw") { + throw reason instanceof WorkerError ? reason : new WorkerError("WORKER_UNAVAILABLE", `Could not queue job "${job.def.name}": ${reason instanceof Error ? reason.message : String(reason)}`); + } + o.onFallback?.(reason, job); + return fallback.add(job); + }, + async get(def, id) { + // ids are UUIDs, so a job is on one side only; a primary that is down just has nothing to say + const fromPrimary = (await isAvailable(primary)) ? await primary.get(def, id).catch(() => null) : null; + return fromPrimary ?? fallback.get(def, id); + }, + async available() { + return (await isAvailable(primary)) || (await isAvailable(fallback)); + }, + async close() { + await Promise.all([primary.close(), fallback.close()]); + }, + }; +} +``` + + +--- + +## File: src/index.ts + +```ts +// @mgpl-diagnostics/worker-kit: background jobs. The main entry has no dependencies; BullMQ lives in "/bullmq". +export { backoffDelay, defineJob, logError, prepareJob, runAttempt, SECRET_KEY, toJsonData, type DefineJobOptions, type ErrorHandler } from "./define"; +export { GENERIC_FAILURE, isJobFailure, JobFailure, WorkerError, type WorkerErrorCode } from "./errors"; +export { fallbackBackend, type FallbackBackendOptions } from "./fallback"; +export { inlineBackend, type InlineBackendOptions } from "./inline"; +export { createJobQueue, jobRegistry, type Enqueuer, type JobQueue, type JobQueueOptions } from "./queue"; +export type { AnyJob, DataOf, EnqueueOptions, JobBackend, JobContext, JobDefinition, JobInfo, JobState, PreparedJob, ResultOf } from "./types"; +``` + + +--- + +## File: src/inline.ts + +```ts +// Jobs in this process, no Redis: development, tests, a single server, and the standalone side of +// fallbackBackend. Same definitions, same JSON data, same retries and same error messages as BullMQ. +import { backoffDelay, logError, runAttempt, type ErrorHandler } from "./define"; +import { GENERIC_FAILURE, isJobFailure, WorkerError } from "./errors"; +import type { AnyJob, JobBackend, JobInfo, PreparedJob } from "./types"; + +export interface InlineBackendOptions { + /** + * true (default): `add` runs the job at once and resolves when it has finished (the request waits), + * retrying immediately. false: `add` resolves at once and the job runs in the background of this + * process, with the definition's backoff between attempts. A delayed job always runs in the background. + */ + wait?: boolean; + /** Called with errors that are not a JobFailure. Default: one line on stderr. */ + onError?: ErrorHandler; +} + +interface Entry { + info: JobInfo; + def: AnyJob; +} + +export function inlineBackend(o: InlineBackendOptions = {}): JobBackend { + const wait = o.wait ?? true; + const onError = o.onError ?? logError; + const jobs = new Map<string, Entry>(); + const running = new Set<Promise<void>>(); + const timers = new Map<NodeJS.Timeout, () => void>(); + let closed = false; + + const prune = () => { + const now = Date.now(); + for (const [id, { info, def }] of jobs) { + const finished = info.finishedAt?.getTime(); + if (finished === undefined) continue; + const keep = info.state === "failed" ? def.keepFailedMs : def.keepResultsMs; + if (finished + keep <= now) jobs.delete(id); + } + }; + + const sleep = (ms: number) => + new Promise<void>((resolve) => { + const t = setTimeout(() => { + timers.delete(t); + resolve(); + }, ms); + timers.set(t, resolve); + }); + + /** Runs every attempt; never rejects. */ + async function run(entry: Entry, delayFirst: number, backoff: boolean): Promise<void> { + const { info, def } = entry; + if (delayFirst > 0) await sleep(delayFirst); + else if (backoff) await new Promise((r) => setImmediate(r)); // background: after `add` has returned + while (!closed) { + info.state = "active"; + info.attempts++; + try { + info.result = await runAttempt(def, info.data, { id: info.id, name: info.name, queue: info.queue, attempt: info.attempts, attempts: def.attempts }, onError); + info.state = "completed"; + info.error = null; + break; + } catch (e) { + info.error = isJobFailure(e) ? e.message : GENERIC_FAILURE; + if (!(isJobFailure(e) && e.retry) || info.attempts >= def.attempts) { + info.state = "failed"; + break; + } + info.state = "delayed"; + if (backoff) await sleep(backoffDelay(def, info.attempts)); + } + } + if (info.state === "completed" || info.state === "failed") info.finishedAt = new Date(); + } + + const track = (p: Promise<void>) => { + running.add(p); + void p.finally(() => running.delete(p)); + }; + + const snapshot = (info: JobInfo): JobInfo => ({ ...info }); + + return { + mode: "inline", + async add(job: PreparedJob) { + if (closed) throw new WorkerError("WORKER_CLOSED", "The job queue is closed"); + prune(); + const info: JobInfo = { + id: job.id, + name: job.def.name, + queue: job.def.queue, + state: job.delayMs > 0 ? "delayed" : "waiting", + attempts: 0, + error: null, + result: null, + data: job.data, + createdAt: new Date(), + finishedAt: null, + }; + const entry: Entry = { info, def: job.def }; + jobs.set(info.id, entry); + if (wait && job.delayMs === 0) { + await run(entry, 0, false); + const out = snapshot(info); + prune(); // keepResultsMs 0: gone for `get`, but the caller still gets the outcome + return out; + } + track(run(entry, job.delayMs, true)); + return snapshot(info); + }, + async get(def, id) { + prune(); + const e = jobs.get(id); + return e && e.info.name === def.name && e.info.queue === def.queue ? snapshot(e.info) : null; + }, + available: () => !closed, + async close() { + closed = true; + // pending delays and backoffs end now; jobs not yet finished are dropped (nothing persists in-process) + for (const [t, resolve] of timers) { + clearTimeout(t); + resolve(); + } + timers.clear(); + await Promise.allSettled([...running]); + jobs.clear(); + }, + }; +} +``` + + +--- + +## File: src/queue.ts + +```ts +// The producer API: one object the app's services use to start jobs and read them back, +// whatever backend sits underneath. +import { prepareJob } from "./define"; +import { WorkerError } from "./errors"; +import type { AnyJob, DataOf, EnqueueOptions, JobBackend, JobInfo, ResultOf } from "./types"; + +export interface JobQueueOptions { + backend: JobBackend; + /** Every job this queue may start. Names are unique. */ + jobs: readonly AnyJob[]; +} + +/** A one-job producer with mail-kit's `MailQueue` shape: `createMailer({ queue: jobs.enqueuer(mailJob) })`. */ +export interface Enqueuer<Data> { + /** Starts the job; resolves with its id. Rejects when it could not be queued or (in-process) failed. */ + enqueue(data: Data): Promise<string>; + available(): Promise<boolean>; +} + +export interface JobQueue { + readonly mode: string; + /** Validates the data here (bad data fails the caller), then hands the job to the backend. */ + enqueue<J extends AnyJob>(job: J, data: DataOf<J>, o?: EnqueueOptions): Promise<JobInfo<DataOf<J>, ResultOf<J>>>; + enqueue(job: string, data: unknown, o?: EnqueueOptions): Promise<JobInfo>; + /** The job with this id, only if it is a `job` job. Callers still check that the reader owns it. */ + get<J extends AnyJob>(job: J, id: string): Promise<JobInfo<DataOf<J>, ResultOf<J>> | null>; + get(job: string, id: string): Promise<JobInfo | null>; + enqueuer<J extends AnyJob>(job: J): Enqueuer<DataOf<J>>; + definition(name: string): AnyJob; + names(): string[]; + available(): Promise<boolean>; + close(): Promise<void>; +} + +/** Checks a set of definitions: unique names. Returns them by name. */ +export function jobRegistry(jobs: readonly AnyJob[]): Map<string, AnyJob> { + const byName = new Map<string, AnyJob>(); + for (const j of jobs) { + if (byName.has(j.name)) throw new WorkerError("WORKER_DUPLICATE_JOB", `Two jobs are named "${j.name}"`); + byName.set(j.name, j); + } + return byName; +} + +export function createJobQueue(o: JobQueueOptions): JobQueue { + const byName = jobRegistry(o.jobs); + let closed = false; + + const resolve = (job: AnyJob | string): AnyJob => { + const name = typeof job === "string" ? job : job.name; + const def = byName.get(name); + if (!def) throw new WorkerError("WORKER_UNKNOWN_JOB", `Unknown job "${name}" (known: ${[...byName.keys()].join(", ") || "none"})`); + if (typeof job !== "string" && job !== def) throw new WorkerError("WORKER_UNKNOWN_JOB", `Job "${name}" is not the definition this queue was created with`); + return def; + }; + + const enqueue = async (job: AnyJob | string, data: unknown, opts?: EnqueueOptions): Promise<JobInfo> => { + if (closed) throw new WorkerError("WORKER_CLOSED", "The job queue is closed"); + const def = resolve(job); + return o.backend.add(prepareJob(def, data, opts)); + }; + + const available = async () => { + if (closed) return false; + try { + return (await (o.backend.available?.() ?? true)) === true; + } catch { + return false; + } + }; + + const queue: JobQueue = { + mode: o.backend.mode, + enqueue: enqueue as JobQueue["enqueue"], + get: (async (job: AnyJob | string, id: string) => (typeof id === "string" && id ? o.backend.get(resolve(job), id) : null)) as JobQueue["get"], + enqueuer(job) { + const def = resolve(job); + return { + async enqueue(data) { + const info = await enqueue(def, data); + if (info.state === "failed") throw new WorkerError("WORKER_UNAVAILABLE", `Job "${def.name}" failed: ${info.error ?? "Job failed"}`); + return info.id; + }, + available, + }; + }, + definition: (name) => resolve(name), + names: () => [...byName.keys()], + available, + async close() { + if (closed) return; + closed = true; + await o.backend.close(); + }, + }; + return queue; +} +``` + + +--- + +## File: src/types.ts + +```ts +// The job model shared by every backend. A job is JSON data on a named queue; a definition says +// how to run it. Backends (in-process, BullMQ on Redis, fallback) only move and store jobs. + +/** A job's state, the same words on every backend. */ +export type JobState = "waiting" | "delayed" | "active" | "completed" | "failed"; + +/** What a processor gets besides the data. */ +export interface JobContext { + id: string; + name: string; + queue: string; + /** 1 on the first run, 2 on the first retry, … */ + attempt: number; + /** Total attempts allowed. */ + attempts: number; +} + +/** + * A kind of background work. Data and result cross JSON (Dates arrive as ISO strings), on every + * backend, so a job behaves the same in-process and on Redis. + */ +export interface JobDefinition<Data = unknown, Result = unknown> { + /** Job name, unique in a queue set, e.g. "password-reset". */ + readonly name: string; + /** Queue it runs on (Redis keys `<prefix>:<queue>:…`). Default: the name. */ + readonly queue: string; + /** Runs before enqueueing, in the caller's request: throw (WorkerError WORKER_BAD_DATA) to refuse. */ + readonly validate?: ((data: Data) => void) | undefined; + readonly process: (data: Data, ctx: JobContext) => Promise<Result> | Result; + /** Total attempts, with exponential backoff between them. */ + readonly attempts: number; + readonly backoffMs: number; + /** How long a finished job (and its result) stays readable. 0 = removed at once. */ + readonly keepResultsMs: number; + /** How long a failed job stays readable (at least a day by default, for inspection). */ + readonly keepFailedMs: number; + /** Largest JSON payload accepted, in bytes. */ + readonly maxBytes: number; + /** Refuse data with token / password / secret keys (see SECRET_KEY). */ + readonly refuseSecrets: boolean; +} + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export type AnyJob = JobDefinition<any, any>; +export type DataOf<J> = J extends JobDefinition<infer D, unknown> ? D : never; +export type ResultOf<J> = J extends JobDefinition<unknown, infer R> ? R : never; + +/** A job as callers see it. */ +export interface JobInfo<Data = unknown, Result = unknown> { + id: string; + name: string; + queue: string; + state: JobState; + /** Attempts made so far. */ + attempts: number; + /** Safe to show: a JobFailure's message, otherwise a generic "Job failed". */ + error: string | null; + result: Result | null; + data: Data; + createdAt: Date; + finishedAt: Date | null; +} + +export interface EnqueueOptions { + /** Run no sooner than this many ms from now. */ + delayMs?: number; +} + +/** A job ready for a backend: JSON-checked data plus the backend-independent options. */ +export interface PreparedJob { + id: string; + def: AnyJob; + data: unknown; + delayMs: number; +} + +/** + * Where jobs go. Implementations: `inlineBackend` (this process), `bullmqBackend` ("/bullmq", + * Redis) and `fallbackBackend` (a worker when available, else standalone). Write your own the same way. + */ +export interface JobBackend { + /** "inline", "bullmq", "fallback" or your own; shown in logs and health checks. */ + readonly mode: string; + add(job: PreparedJob): Promise<JobInfo>; + /** The job with this id on `def`'s queue, or null (unknown, expired, or another job's name). */ + get(def: AnyJob, id: string): Promise<JobInfo | null>; + /** Cheap readiness check (e.g. the Redis connection is "ready"). Default: assume yes. */ + available?(): boolean | Promise<boolean>; + close(): Promise<void>; +} +``` diff --git a/llms.txt b/llms.txt new file mode 100644 index 0000000..d93cd05 --- /dev/null +++ b/llms.txt @@ -0,0 +1,25 @@ +# @mgpl-diagnostics/worker-kit + +> Background jobs for the MGPL Diagnostics backend (Node.js ≥ 20). Typed job definitions and one queue API over BullMQ on Redis, in-process, or worker-when-available-else-standalone; plus the worker runner. The main entry has no runtime dependencies; bullmq and ioredis are optional peers used only by "/bullmq". + +Key facts for agents: +- `defineJob({ name, queue?, process(data, ctx), validate?, attempts = 3, backoffMs = 2000, keepResultsMs = 1h, keepFailedMs, maxBytes, refuseSecrets = true })`. +- `createJobQueue({ backend, jobs })` → `enqueue(job, data, { delayMs? })`, `get(job, id)`, `enqueuer(job)` (mail-kit MailQueue shape), `available()`, `close()`. +- Backends: `inlineBackend({ wait?, onError? })`, `fallbackBackend({ primary, fallback, whenFails?, onFallback? })`, and from "/bullmq" `bullmqBackend({ connection | redisUrl, prefix = "mgpl" })`. +- Worker: `startWorker({ jobs, connection | redisUrl, prefix, concurrency = 4, queues?, onError?, onFailed? })` → `{ queues, workers, close() }`. +- `JobFailure(message, retry = true)`: message shown to clients; `retry: false` fails at once. Other errors → "Job failed" + `onError`. +- `JobInfo`: id, name, queue, state (waiting | delayed | active | completed | failed), attempts, error, result, data, createdAt, finishedAt. +- Errors: `WorkerError.code` = WORKER_UNKNOWN_JOB, WORKER_BAD_DATA, WORKER_DUPLICATE_JOB, WORKER_UNAVAILABLE, WORKER_CLOSED. + +## Docs + +- [README](README.md): usage, backends, worker, mail-kit +- [AGENTS.md](AGENTS.md): map and rules + +## Source + +- [src/define.ts](src/define.ts): definitions, data checks, one attempt +- [src/queue.ts](src/queue.ts): the producer API +- [src/inline.ts](src/inline.ts): in-process backend +- [src/fallback.ts](src/fallback.ts): worker or standalone +- [src/bullmq.ts](src/bullmq.ts): BullMQ backend and worker diff --git a/package.json b/package.json new file mode 100644 index 0000000..8250b80 --- /dev/null +++ b/package.json @@ -0,0 +1,63 @@ +{ + "name": "@mgpl-diagnostics/worker-kit", + "version": "0.1.0", + "description": "MGPL Diagnostics background jobs: typed job definitions, one queue API over BullMQ on Redis or in-process, a worker runner, and fallback from the worker to standalone. No runtime dependencies; bullmq and ioredis are optional peers used only by the \"/bullmq\" entry.", + "type": "module", + "license": "SEE LICENSE IN LICENSE", + "repository": "github:mgpl-diagnostics/worker-kit", + "sideEffects": false, + "main": "./dist/index.js", + "types": "./dist/index.d.ts", + "exports": { + ".": { + "types": "./dist/index.d.ts", + "import": "./dist/index.js" + }, + "./bullmq": { + "types": "./dist/bullmq.d.ts", + "import": "./dist/bullmq.js" + }, + "./package.json": "./package.json" + }, + "files": [ + "dist", + "README.md", + "AGENTS.md", + "llms.txt", + "LICENSE", + "llms-full.txt" + ], + "publishConfig": { + "registry": "https://npm.pkg.github.com" + }, + "scripts": { + "build": "tsup", + "typecheck": "tsc --noEmit", + "test": "tsx scripts/test.ts", + "prepare": "tsup", + "llms": "node scripts/llms-full.mjs @mgpl-diagnostics/worker-kit README.md AGENTS.md llms.txt package.json src scripts" + }, + "peerDependencies": { + "bullmq": ">=5", + "ioredis": ">=5" + }, + "peerDependenciesMeta": { + "bullmq": { + "optional": true + }, + "ioredis": { + "optional": true + } + }, + "devDependencies": { + "@types/node": "^22.20.5", + "bullmq": "^5.81.5", + "ioredis": "^5.11.1", + "tsup": "^8.5.0", + "tsx": "^4.23.15", + "typescript": "^5.9.3" + }, + "engines": { + "node": ">=20" + } +} diff --git a/scripts/llms-full.mjs b/scripts/llms-full.mjs new file mode 100644 index 0000000..77ff18a --- /dev/null +++ b/scripts/llms-full.mjs @@ -0,0 +1,53 @@ +// Builds llms-full.txt: every doc and source file of the repo in one plain-text document for AI tools. +// node scripts/llms-full.mjs "<title>" <file-or-folder> [file-or-folder …] +// Folders are walked recursively. Output order: README, AGENTS, docs, then the rest alphabetically. +// Skipped: build output, dependencies, lockfiles, generated files, binaries. Files over 150 KB are +// listed (path + size) instead of inlined. CI checks the file is up to date (`npm run llms` + git diff). +import { readdirSync, readFileSync, statSync, writeFileSync } from "node:fs"; +import { extname, join, relative } from "node:path"; + +const [title, ...inputs] = process.argv.slice(2); +if (!title || !inputs.length) { + console.error('usage: node scripts/llms-full.mjs "<title>" <file-or-folder> …'); + process.exit(1); +} + +const TEXT = new Set([".ts", ".tsx", ".mts", ".cts", ".js", ".mjs", ".cjs", ".md", ".mdx", ".json", ".css", ".yml", ".yaml", ".txt", ".graphql", ".gql", ".sql", ".html"]); +const SKIP_DIRS = new Set(["node_modules", "dist", ".next", ".git", ".turbo", "coverage", "examples", ".pg-test"]); +const SKIP_FILES = /(^|\/)(package-lock\.json|pnpm-lock\.yaml|yarn\.lock|llms-full\.txt|[^/]*\.generated\.[^/]+|[^/]*\.log)$/; +const MAX = 150 * 1024; +const LANG = { ".ts": "ts", ".tsx": "tsx", ".mts": "ts", ".cts": "ts", ".js": "js", ".mjs": "js", ".cjs": "js", ".json": "json", ".css": "css", ".yml": "yaml", ".yaml": "yaml", ".graphql": "graphql", ".gql": "graphql", ".sql": "sql", ".html": "html" }; + +const posix = (p) => p.split("\\").join("/"); +const files = new Set(); +function walk(p) { + const st = statSync(p); + if (st.isDirectory()) { + for (const name of readdirSync(p)) if (!SKIP_DIRS.has(name)) walk(join(p, name)); + } else if (TEXT.has(extname(p).toLowerCase()) && !SKIP_FILES.test(posix(p))) { + files.add(posix(relative(".", p))); + } +} +for (const input of inputs) walk(input); + +const rank = (f) => (/^README\.md$/i.test(f) ? 0 : /^AGENTS\.md$/i.test(f) ? 1 : /^docs\//.test(f) ? 2 : 3); +const ordered = [...files].sort((a, b) => rank(a) - rank(b) || (a < b ? -1 : a > b ? 1 : 0)); + +const parts = [`# ${title} — full documentation and source\n\n> Generated by scripts/llms-full.mjs. Do not edit by hand. ${ordered.length} files.\n\n## Contents\n\n${ordered.map((f) => `- ${f}`).join("\n")}\n`]; +const omitted = []; +for (const f of ordered) { + // LF-normalised text, so Windows and Linux checkouts produce byte-identical output + const body = readFileSync(f, "utf8").replace(/\r\n/g, "\n").trim(); + const size = Buffer.byteLength(body); + if (size > MAX) { + omitted.push(`${f} (${(size / 1024).toFixed(0)} KB)`); + parts.push(`\n\n---\n\n## File: ${f}\n\n_Omitted: ${(size / 1024).toFixed(0)} KB. Read the file directly._\n`); + continue; + } + const lang = LANG[extname(f).toLowerCase()]; + const fence = body.includes("```") ? "````" : "```"; + parts.push(`\n\n---\n\n## File: ${f}\n\n${lang ? `${fence}${lang}\n${body}\n${fence}` : body}\n`); +} +const out = parts.join(""); +writeFileSync("llms-full.txt", out); +console.log(`llms-full.txt: ${ordered.length} files, ${(out.length / 1024).toFixed(0)} KB${omitted.length ? `; omitted (too large): ${omitted.join(", ")}` : ""}`); diff --git a/scripts/test.ts b/scripts/test.ts new file mode 100644 index 0000000..7b69222 --- /dev/null +++ b/scripts/test.ts @@ -0,0 +1,550 @@ +// worker-kit checks: definitions, data rules, the in-process backend, fallback, the producer API, +// and BullMQ on a real Redis, where the same scenario must give the same results as in-process. +// Redis: REDIS_URL, or a local `redis-server` started here; CI sets REQUIRE_REDIS=1 so it never skips. +import assert from "node:assert/strict"; +import { spawn, spawnSync, type ChildProcess } from "node:child_process"; +import { createServer } from "node:net"; +import { Queue } from "bullmq"; +import { Redis } from "ioredis"; +import { bullJobOptions, bullmqBackend, redisConnection, startWorker } from "../src/bullmq"; +import { + createJobQueue, + defineJob, + fallbackBackend, + inlineBackend, + isJobFailure, + JobFailure, + prepareJob, + WorkerError, + type AnyJob, + type JobBackend, + type JobContext, + type JobInfo, +} from "../src/index"; + +let passed = 0; +const failures: string[] = []; +async function check(name: string, fn: () => unknown): Promise<void> { + try { + await fn(); + passed++; + } catch (e) { + failures.push(`✗ ${name}\n ${(e as Error).stack?.split("\n").slice(0, 4).join("\n ")}`); + } +} +const code = async (fn: () => unknown, expected: string) => { + try { + await fn(); + } catch (e) { + assert.ok(e instanceof WorkerError, `expected WorkerError, got ${String(e)}`); + assert.equal(e.code, expected, e.message); + return e.message; + } + assert.fail(`expected ${expected}`); +}; +const sleep = (ms: number) => new Promise((r) => setTimeout(r, ms)); +async function waitFor<T>(fn: () => Promise<T | null | undefined | false>, ms = 10_000): Promise<T> { + const end = Date.now() + ms; + for (;;) { + const v = await fn(); + if (v) return v; + if (Date.now() > end) assert.fail("timed out"); + await sleep(20); + } +} +const quiet = () => undefined; + +// ---- the scenario every backend must run identically -------------------------------------------- + +interface SumData { + a: number; + b: number; + at: Date | string; +} +function scenario() { + const errors: string[] = []; + const calls: Record<string, number[]> = {}; + const seen = (ctx: JobContext) => (calls[ctx.name] ??= []).push(ctx.attempt); + const jobs = { + sum: defineJob<SumData, { sum: number; at: unknown; when: Date }>({ + name: "sum", + queue: "maths", + process: (d, ctx) => { + seen(ctx); + return { sum: d.a + d.b, at: d.at, when: new Date(0) }; + }, + }), + flaky: defineJob<{ n: number }, string>({ + name: "flaky", + queue: "maths", + attempts: 3, + backoffMs: 10, + process: (d, ctx) => { + seen(ctx); + if (ctx.attempt < 3) throw new JobFailure("Busy, try again", true); + return `ok ${d.n} on ${ctx.attempt}/${ctx.attempts}`; + }, + }), + fatal: defineJob<{ invoiceId: string }, never>({ + name: "fatal", + attempts: 5, + backoffMs: 10, + process: (d, ctx) => { + seen(ctx); + throw new JobFailure(`No invoice ${d.invoiceId}`, false); + }, + }), + crash: defineJob<{ id: string }, never>({ + name: "crash", + attempts: 2, + backoffMs: 10, + process: (_d, ctx) => { + seen(ctx); + throw new Error("connect ECONNREFUSED db.internal:5432 user=admin"); + }, + }), + }; + const onError = (e: unknown) => errors.push((e as Error).message); + return { jobs, list: Object.values(jobs) as AnyJob[], errors, calls, onError }; +} + +/** Runs the scenario on a backend; returns comparable results. */ +async function runScenario(backend: JobBackend, s: ReturnType<typeof scenario>) { + const q = createJobQueue({ backend, jobs: s.list }); + const started = [ + await q.enqueue(s.jobs.sum, { a: 2, b: 3, at: new Date("2026-10-14T10:00:00.000Z") }), + await q.enqueue(s.jobs.flaky, { n: 7 }), + await q.enqueue(s.jobs.fatal, { invoiceId: "INV-1" }), + await q.enqueue(s.jobs.crash, { id: "x" }), + ]; + const done: JobInfo[] = []; + for (const j of started) { + const def = q.definition(j.name); + done.push(await waitFor(async () => { + const info = await q.get(def, j.id); + return info && (info.state === "completed" || info.state === "failed") ? info : null; + })); + } + return { q, done: done.map(({ id: _id, createdAt, finishedAt, ...rest }) => ({ ...rest, timed: createdAt instanceof Date && finishedAt instanceof Date && finishedAt >= createdAt })) }; +} + +const expected = [ + { name: "sum", queue: "maths", state: "completed", attempts: 1, error: null, result: { sum: 5, at: "2026-10-14T10:00:00.000Z", when: "1970-01-01T00:00:00.000Z" }, data: { a: 2, b: 3, at: "2026-10-14T10:00:00.000Z" }, timed: true }, + { name: "flaky", queue: "maths", state: "completed", attempts: 3, error: null, result: "ok 7 on 3/3", data: { n: 7 }, timed: true }, + { name: "fatal", queue: "fatal", state: "failed", attempts: 1, error: "No invoice INV-1", result: null, data: { invoiceId: "INV-1" }, timed: true }, + { name: "crash", queue: "crash", state: "failed", attempts: 2, error: "Job failed", result: null, data: { id: "x" }, timed: true }, +]; + +// ---- definitions and data rules ----------------------------------------------------------------- + +await check("defineJob: defaults, queue = name, frozen", () => { + const j = defineJob({ name: "mail", process: () => 1 }); + assert.equal(j.queue, "mail"); + assert.equal(j.attempts, 3); + assert.equal(j.backoffMs, 2000); + assert.equal(j.keepResultsMs, 3600_000); + assert.equal(j.keepFailedMs, 86_400_000); + assert.equal(j.refuseSecrets, true); + assert.ok(Object.isFrozen(j)); + assert.equal(defineJob({ name: "x", keepResultsMs: 2 * 86_400_000, process: () => 1 }).keepFailedMs, 2 * 86_400_000); + assert.equal(defineJob({ name: "x", attempts: 0, process: () => 1 }).attempts, 1); +}); + +await check("defineJob: names are checked (they become Redis keys)", async () => { + await code(() => defineJob({ name: "Bad Name", process: () => 1 }), "WORKER_BAD_DATA"); + await code(() => defineJob({ name: "ok", queue: "a:b", process: () => 1 }), "WORKER_BAD_DATA"); + await code(() => defineJob({ name: "", process: () => 1 }), "WORKER_BAD_DATA"); +}); + +await check("data crosses JSON: Dates become ISO strings, undefined keys drop", () => { + const j = defineJob<Record<string, unknown>, null>({ name: "j", process: () => null }); + assert.deepEqual(prepareJob(j, { at: new Date(0), gone: undefined, n: 1 }).data, { at: "1970-01-01T00:00:00.000Z", n: 1 }); +}); + +await check("data that is not JSON, too large, or carries secrets is refused in the caller", async () => { + const j = defineJob<unknown, null>({ name: "j", maxBytes: 100, process: () => null }); + await code(() => prepareJob(j, { n: 1n }), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, undefined), "WORKER_BAD_DATA"); + const cyclic: Record<string, unknown> = {}; + cyclic.self = cyclic; + await code(() => prepareJob(j, cyclic), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, { s: "x".repeat(200) }), "WORKER_BAD_DATA"); + const m = await code(() => prepareJob(j, { user: { id: "u1", resetToken: "abc" } }), "WORKER_BAD_DATA"); + assert.match(m, /user\.resetToken/); + assert.doesNotMatch(m, /abc/); + await code(() => prepareJob(j, { list: [{ PASSWORD: "x" }] }), "WORKER_BAD_DATA"); + await code(() => prepareJob(j, { headers: { authorization: "Bearer x" } }), "WORKER_BAD_DATA"); + // ids and links are fine; opting out is explicit + prepareJob(j, { userId: "u1", tokenId: "t1", link: "https://x.example/r" }); + prepareJob(defineJob<unknown, null>({ name: "k", refuseSecrets: false, process: () => null }), { token: "x" }); +}); + +await check("validate() runs before enqueueing; its error becomes WORKER_BAD_DATA", async () => { + let ran = false; + const j = defineJob<{ email: string }, null>({ + name: "j", + validate: (d) => { + if (!d.email.includes("@")) throw new Error("email is not valid"); + }, + process: () => { + ran = true; + return null; + }, + }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [j] }); + assert.match(await code(() => q.enqueue(j, { email: "nope" }), "WORKER_BAD_DATA"), /email is not valid/); + assert.equal(ran, false); + assert.equal((await q.enqueue(j, { email: "a@example.com" })).state, "completed"); +}); + +await check("every job id is a fresh UUID", () => { + const j = defineJob({ name: "j", process: () => 1 }); + const ids = new Set(Array.from({ length: 50 }, () => prepareJob(j, {}).id)); + assert.equal(ids.size, 50); + for (const id of ids) assert.match(id, /^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/); +}); + +await check("JobFailure is recognised across package copies", () => { + const foreign = Object.assign(new Error("x"), { name: "JobFailure", retry: false }); + assert.ok(isJobFailure(foreign)); + assert.ok(isJobFailure(new JobFailure("y"))); + assert.equal(new JobFailure("y").retry, true); + assert.ok(!isJobFailure(new Error("z"))); +}); + +// ---- the producer API --------------------------------------------------------------------------- + +await check("createJobQueue: unknown jobs, duplicate names, foreign definitions", async () => { + const a = defineJob({ name: "a", process: () => 1 }); + await code(() => createJobQueue({ backend: inlineBackend(), jobs: [a, defineJob({ name: "a", process: () => 2 })] }), "WORKER_DUPLICATE_JOB"); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a] }); + assert.deepEqual(q.names(), ["a"]); + assert.equal(q.mode, "inline"); + await code(() => q.enqueue("b", {}), "WORKER_UNKNOWN_JOB"); + await code(() => q.enqueue(defineJob({ name: "a", process: () => 3 }), {}), "WORKER_UNKNOWN_JOB"); + assert.equal((await q.enqueue("a", {})).result, 1); + assert.equal(await q.get(a, ""), null); +}); + +await check("get only returns a job under its own definition", async () => { + const a = defineJob({ name: "a", queue: "shared", process: () => "A" }); + const b = defineJob({ name: "b", queue: "shared", process: () => "B" }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a, b] }); + const job = await q.enqueue(a, {}); + assert.equal((await q.get(a, job.id))?.result, "A"); + assert.equal(await q.get(b, job.id), null); + assert.equal(await q.get(a, "00000000-0000-4000-8000-000000000000"), null); +}); + +await check("close: enqueue refused afterwards, available() false", async () => { + const a = defineJob({ name: "a", process: () => 1 }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [a] }); + assert.equal(await q.available(), true); + await q.close(); + await q.close(); + assert.equal(await q.available(), false); + await code(() => q.enqueue(a, {}), "WORKER_CLOSED"); +}); + +// ---- in-process --------------------------------------------------------------------------------- + +await check("inline (waiting): the scenario", async () => { + const s = scenario(); + const { done } = await runScenario(inlineBackend({ onError: s.onError }), s); + assert.deepEqual(done, expected); + assert.deepEqual(s.calls, { sum: [1], flaky: [1, 2, 3], fatal: [1], crash: [1, 2] }); + assert.equal(s.errors.length, 2); +}); + +await check("inline (background): the scenario, with backoff between attempts", async () => { + const s = scenario(); + const t0 = Date.now(); + const { done } = await runScenario(inlineBackend({ wait: false, onError: s.onError }), s); + assert.deepEqual(done, expected); + assert.ok(Date.now() - t0 >= 30, "flaky waited 10 + 20 ms"); +}); + +await check("inline (background): add resolves before the job runs", async () => { + let release!: () => void; + const gate = new Promise<void>((r) => (release = r)); + const j = defineJob({ name: "slow", process: async () => (await gate, "done") }); + const q = createJobQueue({ backend: inlineBackend({ wait: false }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.state, "waiting"); + await sleep(5); + assert.equal((await q.get(j, info.id))?.state, "active"); + release(); + assert.equal((await waitFor(async () => ((await q.get(j, info.id))?.state === "completed" ? q.get(j, info.id) : null)))?.result, "done"); +}); + +await check("inline: errors that are not JobFailure are hidden and passed to onError", async () => { + const logged: unknown[] = []; + const j = defineJob({ name: "j", attempts: 1, process: () => Promise.reject(new Error("secret internals")) }); + const q = createJobQueue({ backend: inlineBackend({ onError: (e, ctx) => logged.push([(e as Error).message, ctx.name, ctx.attempt]) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.state, "failed"); + assert.equal(info.error, "Job failed"); + assert.deepEqual(logged, [["secret internals", "j", 1]]); +}); + +await check("inline: a result that is not JSON fails without retrying", async () => { + let runs = 0; + const j = defineJob({ name: "j", process: () => (runs++, { n: 1n }) }); + const info = await createJobQueue({ backend: inlineBackend({ onError: quiet }), jobs: [j] }).enqueue(j, {}); + assert.equal(info.state, "failed"); + assert.equal(runs, 1); +}); + +await check("inline: delayed jobs wait, then run", async () => { + const j = defineJob({ name: "later", process: () => "ran" }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [j] }); + const info = await q.enqueue(j, {}, { delayMs: 40 }); + assert.equal(info.state, "delayed"); + await sleep(10); + assert.equal((await q.get(j, info.id))?.state, "delayed"); + const done = await waitFor(async () => ((await q.get(j, info.id))?.state === "completed" ? q.get(j, info.id) : null)); + assert.equal(done?.result, "ran"); +}); + +await check("inline: results are kept for keepResultsMs, failures for keepFailedMs", async () => { + const gone = defineJob({ name: "gone", keepResultsMs: 0, process: () => "x" }); + const brief = defineJob({ name: "brief", keepResultsMs: 30, process: () => "y" }); + const bad = defineJob({ name: "bad", attempts: 1, keepFailedMs: 30, process: () => Promise.reject(new JobFailure("no", false)) }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [gone, brief, bad] }); + const g = await q.enqueue(gone, {}); + assert.equal(g.result, "x", "the caller still sees the outcome"); + assert.equal(await q.get(gone, g.id), null); + const b = await q.enqueue(brief, {}); + const f = await q.enqueue(bad, {}); + assert.equal((await q.get(brief, b.id))?.result, "y"); + assert.equal((await q.get(bad, f.id))?.error, "no"); + await sleep(40); + assert.equal(await q.get(brief, b.id), null); + assert.equal(await q.get(bad, f.id), null); +}); + +await check("inline: close finishes jobs in hand and drops pending delays", async () => { + let finished = false; + let delayedRan = false; + const slow = defineJob({ name: "slow", process: async () => (await sleep(30), (finished = true)) }); + const later = defineJob({ name: "later", process: () => (delayedRan = true) }); + const q = createJobQueue({ backend: inlineBackend({ wait: false }), jobs: [slow, later] }); + await q.enqueue(slow, {}); + await q.enqueue(later, {}, { delayMs: 60_000 }); + await sleep(5); + const t0 = Date.now(); + await q.close(); + assert.ok(finished); + assert.ok(!delayedRan); + assert.ok(Date.now() - t0 < 1000, "did not wait for the delay"); +}); + +// ---- fallback ----------------------------------------------------------------------------------- + +function fakeBackend(o: { available?: boolean; fails?: boolean }) { + const added: string[] = []; + const inner = inlineBackend(); + const b: JobBackend = { + mode: "fake", + async add(job) { + if (o.fails) throw new Error("redis went away"); + added.push(job.def.name); + return inner.add(job); + }, + get: (def, id) => inner.get(def, id), + available: () => o.available ?? true, + close: () => inner.close(), + }; + return { b, added }; +} + +await check("fallback: the primary when available", async () => { + const p = fakeBackend({}); + const fellBack: unknown[] = []; + const j = defineJob({ name: "j", process: () => 1 }); + const q = createJobQueue({ backend: fallbackBackend({ primary: p.b, fallback: inlineBackend(), onFallback: (r) => fellBack.push(r) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.deepEqual(p.added, ["j"]); + assert.equal(fellBack.length, 0); + assert.equal((await q.get(j, info.id))?.result, 1); + assert.equal(q.mode, "fallback"); +}); + +await check("fallback: standalone when the primary is unavailable or fails", async () => { + for (const o of [{ available: false }, { fails: true }]) { + const p = fakeBackend(o); + const reasons: string[] = []; + const j = defineJob({ name: "j", process: () => 2 }); + const q = createJobQueue({ backend: fallbackBackend({ primary: p.b, fallback: inlineBackend(), onFallback: (r) => reasons.push((r as Error).message) }), jobs: [j] }); + const info = await q.enqueue(j, {}); + assert.equal(info.result, 2); + assert.deepEqual(p.added, []); + assert.equal(reasons.length, 1); + assert.equal((await q.get(j, info.id))?.result, 2, "found on the fallback"); + assert.equal(await q.available(), true); + } +}); + +await check("fallback: whenFails 'throw' raises WORKER_UNAVAILABLE", async () => { + const j = defineJob({ name: "j", process: () => 1 }); + for (const o of [{ available: false }, { fails: true }]) { + const q = createJobQueue({ backend: fallbackBackend({ primary: fakeBackend(o).b, fallback: inlineBackend(), whenFails: "throw" }), jobs: [j] }); + await code(() => q.enqueue(j, {}), "WORKER_UNAVAILABLE"); + } +}); + +await check("enqueuer: mail-kit's MailQueue shape", async () => { + // mail-kit: interface MailQueue { enqueue(job): Promise<string | undefined>; available?(): boolean | Promise<boolean> } + interface MailQueueShape { + enqueue(job: { v: 1; template: string }): Promise<string | undefined>; + available?(): boolean | Promise<boolean>; + } + const sent: string[] = []; + const mail = defineJob<{ v: 1; template: string }, null>({ name: "mail", keepResultsMs: 0, process: (d) => (sent.push(d.template), null) }); + const broken = defineJob<{ v: 1; template: string }, null>({ name: "broken", attempts: 1, process: () => Promise.reject(new JobFailure("SMTP refused", false)) }); + const q = createJobQueue({ backend: inlineBackend(), jobs: [mail, broken] }); + const mq: MailQueueShape = q.enqueuer(mail); + const id = await mq.enqueue({ v: 1, template: "password-reset" }); + assert.equal(typeof id, "string"); + assert.deepEqual(sent, ["password-reset"]); + assert.equal(await mq.available?.(), true); + await assert.rejects(q.enqueuer(broken).enqueue({ v: 1, template: "x" }), /SMTP refused/); +}); + +// ---- BullMQ on Redis ---------------------------------------------------------------------------- + +async function freePort(): Promise<number> { + return new Promise((resolve) => { + const s = createServer().listen(0, "127.0.0.1", () => { + const port = (s.address() as { port: number }).port; + s.close(() => resolve(port)); + }); + }); +} + +let redisUrl = process.env.REDIS_URL; +let redisServer: ChildProcess | undefined; +if (!redisUrl && spawnSync("redis-server", ["--version"]).status === 0) { + const port = await freePort(); + redisServer = spawn("redis-server", ["--port", String(port), "--save", "", "--appendonly", "no"], { stdio: "ignore" }); + redisUrl = `redis://127.0.0.1:${port}`; + const probe = new Redis(redisUrl, { lazyConnect: true, maxRetriesPerRequest: 0, retryStrategy: () => 50 }); + probe.on("error", quiet); + await waitFor(async () => (await probe.connect().then(() => true, () => false)) && (await probe.ping()) === "PONG", 5000); + probe.disconnect(); +} + +if (!redisUrl) { + if (process.env.REQUIRE_REDIS) failures.push("✗ BullMQ checks: no Redis (set REDIS_URL or install redis-server)"); + else console.log("- BullMQ checks skipped: no REDIS_URL and no redis-server"); +} else { + const prefix = `wk-test-${process.pid}`; + const admin = new Redis(redisUrl); + + await check("bullmq: the scenario gives the same results as in-process", async () => { + const s = scenario(); + const worker = startWorker({ redisUrl, prefix, jobs: s.list, concurrency: 2, onError: s.onError, onFailed: quiet }); + assert.deepEqual(worker.queues.sort(), ["crash", "fatal", "maths"]); + const backend = bullmqBackend({ redisUrl, prefix }); + await waitFor(async () => backend.available?.()); + const { q, done } = await runScenario(backend, s); + assert.equal(q.mode, "bullmq"); + assert.deepEqual(done, expected); + assert.deepEqual(s.calls, { sum: [1], flaky: [1, 2, 3], fatal: [1], crash: [1, 2] }); + assert.equal(s.errors.length, 2); + await worker.close(); + await q.close(); + }); + + await check("bullmq: keys use the prefix and queue; job options come from the definition", async () => { + const j = defineJob({ name: "keys", queue: "kq", attempts: 4, backoffMs: 500, keepResultsMs: 60_000, keepFailedMs: 120_000, process: () => 1 }); + const prepared = prepareJob(j, { id: "a" }, { delayMs: 1000 }); + assert.deepEqual(bullJobOptions(prepared), { jobId: prepared.id, attempts: 4, backoff: { type: "exponential", delay: 500 }, delay: 1000, removeOnComplete: { age: 60 }, removeOnFail: { age: 120 } }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [j] }); + const info = await q.enqueue(j, { id: "a" }, { delayMs: 60_000 }); + assert.equal(info.state, "delayed"); + assert.equal((await q.get(j, info.id))?.state, "delayed"); + assert.ok((await admin.keys(`${prefix}:kq:*`)).length > 0); + await q.close(); + }); + + await check("bullmq: keepResultsMs 0 removes the job (and its data) once done", async () => { + const j = defineJob({ name: "mail", keepResultsMs: 0, process: () => "sent" }); + const worker = startWorker({ redisUrl, prefix, jobs: [j] }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [j] }); + const info = await q.enqueue(j, { link: "https://x.example/reset/abc" }); + await waitFor(async () => (await q.get(j, info.id)) === null); + assert.equal(await admin.exists(`${prefix}:mail:${info.id}`), 0); + await worker.close(); + await q.close(); + }); + + await check("bullmq: get under another definition returns null", async () => { + const a = defineJob({ name: "a", queue: "same", process: () => 1 }); + const b = defineJob({ name: "b", queue: "same", process: () => 2 }); + const q = createJobQueue({ backend: bullmqBackend({ redisUrl, prefix }), jobs: [a, b] }); + const info = await q.enqueue(a, {}, { delayMs: 60_000 }); + assert.equal((await q.get(a, info.id))?.name, "a"); + assert.equal(await q.get(b, info.id), null); + await q.close(); + }); + + await check("bullmq: a job name the worker does not know fails at once", async () => { + const known = defineJob({ name: "known", queue: "mixed", process: () => 1 }); + const failed: { name: string; final: boolean }[] = []; + const worker = startWorker({ redisUrl, prefix, jobs: [known], onFailed: (f) => failed.push(f) }); + const raw = new Queue("mixed", { connection: redisConnection(redisUrl), prefix }); + await raw.add("stranger", {}, { attempts: 3 }); + await waitFor(async () => failed.length > 0); + assert.deepEqual(failed.map((f) => [f.name, f.final]), [["stranger", true]]); + await raw.close(); + await worker.close(); + }); + + await check("bullmq: a shared connection is not closed by worker-kit", async () => { + const connection = redisConnection(redisUrl); + const j = defineJob({ name: "shared", process: () => "x" }); + const worker = startWorker({ connection, prefix, jobs: [j] }); + const q = createJobQueue({ backend: bullmqBackend({ connection, prefix }), jobs: [j] }); + const info = await q.enqueue(j, {}); + await waitFor(async () => (await q.get(j, info.id))?.state === "completed"); + await worker.close(); + await q.close(); + assert.equal(await connection.ping(), "PONG"); + await connection.quit(); + }); + + await check("bullmq + fallback: Redis down → jobs run standalone, without waiting", async () => { + const port = await freePort(); // nothing listens here + const down = new Redis(`redis://127.0.0.1:${port}`, { maxRetriesPerRequest: null, lazyConnect: true }); + down.on("error", quiet); + const ran: string[] = []; + const j = defineJob<{ to: string }, null>({ name: "mail", process: (d) => (ran.push(d.to), null) }); + const reasons: string[] = []; + const q = createJobQueue({ + backend: fallbackBackend({ primary: bullmqBackend({ connection: down, prefix }), fallback: inlineBackend(), onFallback: (r) => reasons.push((r as Error).message) }), + jobs: [j], + }); + const t0 = Date.now(); + const info = await q.enqueue(j, { to: "a@example.com" }); + assert.ok(Date.now() - t0 < 1000); + assert.equal(info.state, "completed"); + assert.deepEqual(ran, ["a@example.com"]); + assert.match(reasons[0] ?? "", /bullmq backend is not available/); + await q.close(); + down.disconnect(); + }); + + await check("bullmq: needs a connection or a URL", async () => { + await code(() => bullmqBackend({}), "WORKER_UNAVAILABLE"); + }); + + const leftovers = await admin.keys(`${prefix}:*`); + if (leftovers.length) await admin.del(...leftovers); + await admin.quit(); +} +redisServer?.kill(); + +for (const f of failures) console.log(f); +console.log(`\n${passed} passed, ${failures.length} failed`); +process.exit(failures.length ? 1 : 0); diff --git a/src/bullmq.ts b/src/bullmq.ts new file mode 100644 index 0000000..255c16d --- /dev/null +++ b/src/bullmq.ts @@ -0,0 +1,171 @@ +// "/bullmq" entry: the BullMQ-on-Redis backend (producer) and the worker runner (consumer). +// bullmq and ioredis are optional peers; nothing else in the package imports them. +import { Job, Queue, UnrecoverableError, Worker, type ConnectionOptions, type JobsOptions } from "bullmq"; +import { Redis } from "ioredis"; +import { logError, runAttempt, type ErrorHandler } from "./define"; +import { isJobFailure, WorkerError } from "./errors"; +import { jobRegistry } from "./queue"; +import type { AnyJob, JobBackend, JobInfo, JobState, PreparedJob } from "./types"; + +export interface RedisOptions { + /** An ioredis connection you own (not closed by worker-kit). Give this or `redisUrl`. */ + connection?: Redis; + /** worker-kit opens (and closes) its own connection. */ + redisUrl?: string; + /** Redis key prefix: keys are `<prefix>:<queue>:…`. Default "mgpl". */ + prefix?: string; +} + +/** An ioredis connection as BullMQ needs it (`maxRetriesPerRequest: null` for blocking commands). */ +export function redisConnection(url: string): Redis { + return new Redis(url, { maxRetriesPerRequest: null }); +} + +/** `available` check for a connection: true only while it is "ready" (ioredis would otherwise queue commands while down). */ +export const redisReady = (connection: Redis) => () => connection.status === "ready"; + +function own(o: RedisOptions): { connection: Redis; owned: boolean } { + if (o.connection) return { connection: o.connection, owned: false }; + if (o.redisUrl) { + const connection = redisConnection(o.redisUrl); + // ioredis reconnects by itself; `available()` reports the outage. Without a listener every retry is printed. + connection.on("error", () => undefined); + return { connection, owned: true }; + } + throw new WorkerError("WORKER_UNAVAILABLE", "Give a Redis `connection` or a `redisUrl`"); +} + +const seconds = (ms: number) => Math.max(1, Math.ceil(ms / 1000)); + +/** BullMQ job options from a definition: the only place they are set. */ +export function bullJobOptions(job: PreparedJob): JobsOptions { + const { def } = job; + return { + jobId: job.id, + attempts: def.attempts, + backoff: { type: "exponential", delay: def.backoffMs }, + ...(job.delayMs > 0 ? { delay: job.delayMs } : {}), + removeOnComplete: def.keepResultsMs === 0 ? true : { age: seconds(def.keepResultsMs) }, + removeOnFail: def.keepFailedMs === 0 ? true : { age: seconds(def.keepFailedMs) }, + }; +} + +const STATES: Record<string, JobState> = { waiting: "waiting", "waiting-children": "waiting", prioritized: "waiting", active: "active", completed: "completed", failed: "failed", delayed: "delayed" }; + +async function toInfo(job: Job, queue: string): Promise<JobInfo> { + const state = STATES[await job.getState()] ?? "waiting"; + return { + id: job.id!, + name: job.name, + queue, + state, + attempts: job.attemptsMade, + error: state === "completed" ? null : job.failedReason || null, + result: state === "completed" ? ((job.returnvalue as unknown) ?? null) : null, + data: job.data as unknown, + createdAt: new Date(job.timestamp), + finishedAt: job.finishedOn && (state === "completed" || state === "failed") ? new Date(job.finishedOn) : null, + }; +} + +/** + * The producer side (API processes): adds jobs to Redis. Run `startWorker` somewhere to process them. + * `available()` is false while the Redis connection is not "ready" (use it with fallbackBackend). + */ +export function bullmqBackend(o: RedisOptions): JobBackend { + const { connection, owned } = own(o); + const prefix = o.prefix ?? "mgpl"; + const queues = new Map<string, Queue>(); + const queueFor = (name: string) => { + let q = queues.get(name); + if (!q) { + q = new Queue(name, { connection: connection as unknown as ConnectionOptions, prefix }); + queues.set(name, q); + } + return q; + }; + return { + mode: "bullmq", + async add(job) { + const added = await queueFor(job.def.queue).add(job.def.name, job.data, bullJobOptions(job)); + return toInfo(added, job.def.queue); + }, + async get(def: AnyJob, id: string) { + const job = await Job.fromId(queueFor(def.queue), id); + return job && job.name === def.name ? toInfo(job, def.queue) : null; + }, + available: redisReady(connection), + async close() { + await Promise.all([...queues.values()].map((q) => q.close())); + if (owned) await connection.quit().catch(() => undefined); + }, + }; +} + +export interface StartWorkerOptions extends RedisOptions { + /** The definitions to run. One BullMQ Worker per distinct queue. */ + jobs: readonly AnyJob[]; + /** Only these queues (default: every queue of `jobs`). */ + queues?: readonly string[]; + /** Jobs a worker runs at once, per queue. Default 4. */ + concurrency?: number; + /** Errors that are not a JobFailure (with the job context). Default: one line on stderr. */ + onError?: ErrorHandler; + /** Each failed attempt (after onError). Default: one warning line. */ + onFailed?: (info: { id: string | undefined; name: string; queue: string; attempt: number; error: string; final: boolean }) => void; +} + +export interface RunningWorker { + queues: string[]; + workers: Worker[]; + /** Finishes the jobs in hand, then stops. */ + close(): Promise<void>; +} + +/** The consumer side (`node dist/worker.js` in the api): runs the definitions' processors. */ +export function startWorker(o: StartWorkerOptions): RunningWorker { + const byName = jobRegistry(o.jobs); + const { connection, owned } = own(o); + const prefix = o.prefix ?? "mgpl"; + const onError = o.onError ?? logError; + const onFailed = + o.onFailed ?? ((f) => console.warn(`[worker-kit] ${f.queue}/${f.name} job ${f.id} failed (attempt ${f.attempt}${f.final ? ", final" : ""}): ${f.error}`)); + const names = [...new Set(o.jobs.map((j) => j.queue))].filter((q) => !o.queues || o.queues.includes(q)); + + const workers = names.map((queue) => { + const w = new Worker( + queue, + async (job: Job) => { + const def = byName.get(job.name); + if (!def || def.queue !== queue) throw new UnrecoverableError(`Unknown job "${job.name}" on queue "${queue}"`); + const ctx = { id: job.id ?? "", name: job.name, queue, attempt: job.attemptsMade + 1, attempts: job.opts.attempts ?? def.attempts }; + try { + return await runAttempt(def, job.data, ctx, onError); + } catch (e) { + const message = isJobFailure(e) ? e.message : "Job failed"; + // a JobFailure that retrying cannot fix fails the job at once + if (isJobFailure(e) && !e.retry) throw new UnrecoverableError(message); + throw new Error(message); + } + }, + { connection: connection as unknown as ConnectionOptions, prefix, concurrency: Math.max(1, o.concurrency ?? 4) }, + ); + w.on("failed", (job, err) => { + if (!job) return; + const final = err instanceof UnrecoverableError || err.name === "UnrecoverableError" || job.attemptsMade >= (job.opts.attempts ?? 1); + onFailed({ id: job.id, name: job.name, queue, attempt: job.attemptsMade, error: err.message, final }); + }); + // without a listener an "error" event (e.g. Redis dropped) would crash the process + w.on("error", (err) => console.error(`[worker-kit] ${queue} worker:`, err.message)); + return w; + }); + + return { + queues: names, + workers, + async close() { + await Promise.all(workers.map((w) => w.close())); + if (owned) await connection.quit().catch(() => undefined); + }, + }; +} diff --git a/src/define.ts b/src/define.ts new file mode 100644 index 0000000..42e95b7 --- /dev/null +++ b/src/define.ts @@ -0,0 +1,133 @@ +import { randomUUID } from "node:crypto"; +import { GENERIC_FAILURE, isJobFailure, JobFailure, WorkerError } from "./errors"; +import type { AnyJob, EnqueueOptions, JobContext, JobDefinition, PreparedJob } from "./types"; + +export interface DefineJobOptions<Data, Result> { + name: string; + queue?: string; + validate?: (data: Data) => void; + process: (data: Data, ctx: JobContext) => Promise<Result> | Result; + /** Default 3. */ + attempts?: number; + /** First retry delay; doubles each time. Default 2 s. */ + backoffMs?: number; + /** Default 1 hour. Use 0 for jobs whose data must not linger (e.g. a mail with a reset link). */ + keepResultsMs?: number; + /** Default max(1 day, keepResultsMs). */ + keepFailedMs?: number; + /** Default 512 KiB. */ + maxBytes?: number; + /** Default true. */ + refuseSecrets?: boolean; +} + +const NAME = /^[a-z0-9][a-z0-9._-]{0,63}$/; + +/** Keys that must not appear in job data: payloads carry ids, never tokens or secrets. */ +export const SECRET_KEY = /^(password|passwordhash|token|accesstoken|refreshtoken|resettoken|secret|clientsecret|apikey|api_key|authorization|cookie)$/i; + +export function defineJob<Data, Result>(o: DefineJobOptions<Data, Result>): JobDefinition<Data, Result> { + const queue = o.queue ?? o.name; + if (!NAME.test(o.name)) throw new WorkerError("WORKER_BAD_DATA", `Job name "${o.name}" must match ${NAME}`); + if (!NAME.test(queue)) throw new WorkerError("WORKER_BAD_DATA", `Queue name "${queue}" must match ${NAME}`); + const attempts = Math.max(1, Math.floor(o.attempts ?? 3)); + const keepResultsMs = Math.max(0, o.keepResultsMs ?? 3600_000); + return Object.freeze({ + name: o.name, + queue, + validate: o.validate, + process: o.process, + attempts, + backoffMs: Math.max(0, o.backoffMs ?? 2000), + keepResultsMs, + keepFailedMs: Math.max(0, o.keepFailedMs ?? Math.max(86_400_000, keepResultsMs)), + maxBytes: o.maxBytes ?? 512 * 1024, + refuseSecrets: o.refuseSecrets ?? true, + }); +} + +/** Delay before attempt `n + 1` after `n` failed attempts (exponential, the same as BullMQ's). */ +export const backoffDelay = (def: AnyJob, failedAttempts: number) => def.backoffMs * 2 ** (failedAttempts - 1); + +function findSecret(v: unknown, path: string, seen = 0): string | null { + if (seen > 64 || v === null || typeof v !== "object") return null; + if (Array.isArray(v)) { + for (let i = 0; i < v.length; i++) { + const hit = findSecret(v[i], `${path}[${i}]`, seen + 1); + if (hit) return hit; + } + return null; + } + for (const [k, child] of Object.entries(v)) { + if (SECRET_KEY.test(k)) return path ? `${path}.${k}` : k; + const hit = findSecret(child, path ? `${path}.${k}` : k, seen + 1); + if (hit) return hit; + } + return null; +} + +/** JSON round trip with the definition's checks, so every backend sees exactly what Redis would hold. */ +export function toJsonData(def: AnyJob, data: unknown): unknown { + let json: string | undefined; + try { + json = JSON.stringify(data); + } catch (e) { + throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is not JSON (${(e as Error).message})`); + } + if (json === undefined) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is not JSON`); + const bytes = Buffer.byteLength(json); + if (bytes > def.maxBytes) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data is ${bytes} bytes, the limit is ${def.maxBytes}`); + const parsed: unknown = JSON.parse(json); + if (def.refuseSecrets) { + const key = findSecret(parsed, ""); + if (key) throw new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": data carries a secret-looking key "${key}"; pass an id and look the value up in the processor`); + } + return parsed; +} + +/** Validates and serialises a job for a backend. Runs in the caller, so bad data fails the request. */ +export function prepareJob(def: AnyJob, data: unknown, o: EnqueueOptions = {}): PreparedJob { + if (def.validate) { + try { + def.validate(data); + } catch (e) { + throw e instanceof WorkerError ? e : new WorkerError("WORKER_BAD_DATA", `Job "${def.name}": ${(e as Error).message}`); + } + } + return { + id: randomUUID(), // unguessable; callers still check ownership on every read + def, + data: toJsonData(def, data), + delayMs: Math.max(0, Math.floor(o.delayMs ?? 0)), + }; +} + +export type ErrorHandler = (error: unknown, ctx: JobContext) => void; + +/** + * Runs one attempt. Every backend goes through here, so failures read the same everywhere: + * a JobFailure keeps its message and retry flag; anything else goes to `onError` (log it) and + * becomes a retryable "Job failed", so internals never reach the client. + */ +export async function runAttempt(def: AnyJob, data: unknown, ctx: JobContext, onError?: ErrorHandler): Promise<unknown> { + let result: unknown; + try { + result = await def.process(data, ctx); + } catch (e) { + if (isJobFailure(e)) throw e; + onError?.(e, ctx); + throw new JobFailure(GENERIC_FAILURE, true); + } + if (result === undefined) return null; + try { + return JSON.parse(JSON.stringify(result)) as unknown; + } catch (e) { + onError?.(e, ctx); + throw new JobFailure(GENERIC_FAILURE, false); + } +} + +/** The default onError: one line on stderr, without the job's data. */ +export const logError: ErrorHandler = (e, ctx) => { + console.error(`[worker-kit] ${ctx.queue}/${ctx.name} job ${ctx.id} attempt ${ctx.attempt}/${ctx.attempts} failed:`, e); +}; diff --git a/src/errors.ts b/src/errors.ts new file mode 100644 index 0000000..23db8ed --- /dev/null +++ b/src/errors.ts @@ -0,0 +1,34 @@ +/** Error codes are a public contract: add, never rename. */ +export type WorkerErrorCode = + | "WORKER_UNKNOWN_JOB" // enqueue/get/run of a job name that is not defined + | "WORKER_BAD_DATA" // data is not JSON, too large, carries a secret key, or failed validate() + | "WORKER_DUPLICATE_JOB" // two definitions with the same name + | "WORKER_UNAVAILABLE" // no backend could take the job + | "WORKER_CLOSED"; // the queue was closed + +export class WorkerError extends Error { + constructor(readonly code: WorkerErrorCode, message: string) { + super(message); + this.name = "WorkerError"; + } +} + +/** + * Throw from a processor for a failure whose message is safe to show to the client. + * `retry: false` fails the job at once (retrying cannot help); `true` uses the remaining attempts. + * Any other error is logged through `onError` and shown as "Job failed". + */ +export class JobFailure extends Error { + constructor(message: string, readonly retry: boolean = true) { + super(message); + this.name = "JobFailure"; + } +} + +/** True for a JobFailure, also one made by another copy of this package. */ +export function isJobFailure(e: unknown): e is JobFailure { + return e instanceof JobFailure || (e instanceof Error && e.name === "JobFailure" && typeof (e as { retry?: unknown }).retry === "boolean"); +} + +/** The message shown for failures that are not a JobFailure. */ +export const GENERIC_FAILURE = "Job failed"; diff --git a/src/fallback.ts b/src/fallback.ts new file mode 100644 index 0000000..4975aa8 --- /dev/null +++ b/src/fallback.ts @@ -0,0 +1,55 @@ +// "The worker if it is there, otherwise standalone": jobs go to the primary backend (BullMQ on +// Redis) while it is available, and run on the fallback (usually in-process) when it is not. +import { WorkerError } from "./errors"; +import type { JobBackend, PreparedJob } from "./types"; + +export interface FallbackBackendOptions { + primary: JobBackend; + fallback: JobBackend; + /** When the primary is unavailable or `add` fails: "fallback" (default) or "throw" (WORKER_UNAVAILABLE). */ + whenFails?: "fallback" | "throw"; + /** Called each time a job went to the fallback (log it). */ + onFallback?: (reason: unknown, job: PreparedJob) => void; +} + +export function fallbackBackend(o: FallbackBackendOptions): JobBackend { + const { primary, fallback } = o; + const isAvailable = async (b: JobBackend) => { + try { + return (await (b.available?.() ?? true)) === true; + } catch { + return false; + } + }; + return { + mode: "fallback", + async add(job) { + let reason: unknown; + if (await isAvailable(primary)) { + try { + return await primary.add(job); + } catch (e) { + reason = e; + } + } else { + reason = new WorkerError("WORKER_UNAVAILABLE", `The ${primary.mode} backend is not available`); + } + if (o.whenFails === "throw") { + throw reason instanceof WorkerError ? reason : new WorkerError("WORKER_UNAVAILABLE", `Could not queue job "${job.def.name}": ${reason instanceof Error ? reason.message : String(reason)}`); + } + o.onFallback?.(reason, job); + return fallback.add(job); + }, + async get(def, id) { + // ids are UUIDs, so a job is on one side only; a primary that is down just has nothing to say + const fromPrimary = (await isAvailable(primary)) ? await primary.get(def, id).catch(() => null) : null; + return fromPrimary ?? fallback.get(def, id); + }, + async available() { + return (await isAvailable(primary)) || (await isAvailable(fallback)); + }, + async close() { + await Promise.all([primary.close(), fallback.close()]); + }, + }; +} diff --git a/src/index.ts b/src/index.ts new file mode 100644 index 0000000..f050576 --- /dev/null +++ b/src/index.ts @@ -0,0 +1,7 @@ +// @mgpl-diagnostics/worker-kit: background jobs. The main entry has no dependencies; BullMQ lives in "/bullmq". +export { backoffDelay, defineJob, logError, prepareJob, runAttempt, SECRET_KEY, toJsonData, type DefineJobOptions, type ErrorHandler } from "./define"; +export { GENERIC_FAILURE, isJobFailure, JobFailure, WorkerError, type WorkerErrorCode } from "./errors"; +export { fallbackBackend, type FallbackBackendOptions } from "./fallback"; +export { inlineBackend, type InlineBackendOptions } from "./inline"; +export { createJobQueue, jobRegistry, type Enqueuer, type JobQueue, type JobQueueOptions } from "./queue"; +export type { AnyJob, DataOf, EnqueueOptions, JobBackend, JobContext, JobDefinition, JobInfo, JobState, PreparedJob, ResultOf } from "./types"; diff --git a/src/inline.ts b/src/inline.ts new file mode 100644 index 0000000..977af3d --- /dev/null +++ b/src/inline.ts @@ -0,0 +1,129 @@ +// Jobs in this process, no Redis: development, tests, a single server, and the standalone side of +// fallbackBackend. Same definitions, same JSON data, same retries and same error messages as BullMQ. +import { backoffDelay, logError, runAttempt, type ErrorHandler } from "./define"; +import { GENERIC_FAILURE, isJobFailure, WorkerError } from "./errors"; +import type { AnyJob, JobBackend, JobInfo, PreparedJob } from "./types"; + +export interface InlineBackendOptions { + /** + * true (default): `add` runs the job at once and resolves when it has finished (the request waits), + * retrying immediately. false: `add` resolves at once and the job runs in the background of this + * process, with the definition's backoff between attempts. A delayed job always runs in the background. + */ + wait?: boolean; + /** Called with errors that are not a JobFailure. Default: one line on stderr. */ + onError?: ErrorHandler; +} + +interface Entry { + info: JobInfo; + def: AnyJob; +} + +export function inlineBackend(o: InlineBackendOptions = {}): JobBackend { + const wait = o.wait ?? true; + const onError = o.onError ?? logError; + const jobs = new Map<string, Entry>(); + const running = new Set<Promise<void>>(); + const timers = new Map<NodeJS.Timeout, () => void>(); + let closed = false; + + const prune = () => { + const now = Date.now(); + for (const [id, { info, def }] of jobs) { + const finished = info.finishedAt?.getTime(); + if (finished === undefined) continue; + const keep = info.state === "failed" ? def.keepFailedMs : def.keepResultsMs; + if (finished + keep <= now) jobs.delete(id); + } + }; + + const sleep = (ms: number) => + new Promise<void>((resolve) => { + const t = setTimeout(() => { + timers.delete(t); + resolve(); + }, ms); + timers.set(t, resolve); + }); + + /** Runs every attempt; never rejects. */ + async function run(entry: Entry, delayFirst: number, backoff: boolean): Promise<void> { + const { info, def } = entry; + if (delayFirst > 0) await sleep(delayFirst); + else if (backoff) await new Promise((r) => setImmediate(r)); // background: after `add` has returned + while (!closed) { + info.state = "active"; + info.attempts++; + try { + info.result = await runAttempt(def, info.data, { id: info.id, name: info.name, queue: info.queue, attempt: info.attempts, attempts: def.attempts }, onError); + info.state = "completed"; + info.error = null; + break; + } catch (e) { + info.error = isJobFailure(e) ? e.message : GENERIC_FAILURE; + if (!(isJobFailure(e) && e.retry) || info.attempts >= def.attempts) { + info.state = "failed"; + break; + } + info.state = "delayed"; + if (backoff) await sleep(backoffDelay(def, info.attempts)); + } + } + if (info.state === "completed" || info.state === "failed") info.finishedAt = new Date(); + } + + const track = (p: Promise<void>) => { + running.add(p); + void p.finally(() => running.delete(p)); + }; + + const snapshot = (info: JobInfo): JobInfo => ({ ...info }); + + return { + mode: "inline", + async add(job: PreparedJob) { + if (closed) throw new WorkerError("WORKER_CLOSED", "The job queue is closed"); + prune(); + const info: JobInfo = { + id: job.id, + name: job.def.name, + queue: job.def.queue, + state: job.delayMs > 0 ? "delayed" : "waiting", + attempts: 0, + error: null, + result: null, + data: job.data, + createdAt: new Date(), + finishedAt: null, + }; + const entry: Entry = { info, def: job.def }; + jobs.set(info.id, entry); + if (wait && job.delayMs === 0) { + await run(entry, 0, false); + const out = snapshot(info); + prune(); // keepResultsMs 0: gone for `get`, but the caller still gets the outcome + return out; + } + track(run(entry, job.delayMs, true)); + return snapshot(info); + }, + async get(def, id) { + prune(); + const e = jobs.get(id); + return e && e.info.name === def.name && e.info.queue === def.queue ? snapshot(e.info) : null; + }, + available: () => !closed, + async close() { + closed = true; + // pending delays and backoffs end now; jobs not yet finished are dropped (nothing persists in-process) + for (const [t, resolve] of timers) { + clearTimeout(t); + resolve(); + } + timers.clear(); + await Promise.allSettled([...running]); + jobs.clear(); + }, + }; +} diff --git a/src/queue.ts b/src/queue.ts new file mode 100644 index 0000000..97afee1 --- /dev/null +++ b/src/queue.ts @@ -0,0 +1,97 @@ +// The producer API: one object the app's services use to start jobs and read them back, +// whatever backend sits underneath. +import { prepareJob } from "./define"; +import { WorkerError } from "./errors"; +import type { AnyJob, DataOf, EnqueueOptions, JobBackend, JobInfo, ResultOf } from "./types"; + +export interface JobQueueOptions { + backend: JobBackend; + /** Every job this queue may start. Names are unique. */ + jobs: readonly AnyJob[]; +} + +/** A one-job producer with mail-kit's `MailQueue` shape: `createMailer({ queue: jobs.enqueuer(mailJob) })`. */ +export interface Enqueuer<Data> { + /** Starts the job; resolves with its id. Rejects when it could not be queued or (in-process) failed. */ + enqueue(data: Data): Promise<string>; + available(): Promise<boolean>; +} + +export interface JobQueue { + readonly mode: string; + /** Validates the data here (bad data fails the caller), then hands the job to the backend. */ + enqueue<J extends AnyJob>(job: J, data: DataOf<J>, o?: EnqueueOptions): Promise<JobInfo<DataOf<J>, ResultOf<J>>>; + enqueue(job: string, data: unknown, o?: EnqueueOptions): Promise<JobInfo>; + /** The job with this id, only if it is a `job` job. Callers still check that the reader owns it. */ + get<J extends AnyJob>(job: J, id: string): Promise<JobInfo<DataOf<J>, ResultOf<J>> | null>; + get(job: string, id: string): Promise<JobInfo | null>; + enqueuer<J extends AnyJob>(job: J): Enqueuer<DataOf<J>>; + definition(name: string): AnyJob; + names(): string[]; + available(): Promise<boolean>; + close(): Promise<void>; +} + +/** Checks a set of definitions: unique names. Returns them by name. */ +export function jobRegistry(jobs: readonly AnyJob[]): Map<string, AnyJob> { + const byName = new Map<string, AnyJob>(); + for (const j of jobs) { + if (byName.has(j.name)) throw new WorkerError("WORKER_DUPLICATE_JOB", `Two jobs are named "${j.name}"`); + byName.set(j.name, j); + } + return byName; +} + +export function createJobQueue(o: JobQueueOptions): JobQueue { + const byName = jobRegistry(o.jobs); + let closed = false; + + const resolve = (job: AnyJob | string): AnyJob => { + const name = typeof job === "string" ? job : job.name; + const def = byName.get(name); + if (!def) throw new WorkerError("WORKER_UNKNOWN_JOB", `Unknown job "${name}" (known: ${[...byName.keys()].join(", ") || "none"})`); + if (typeof job !== "string" && job !== def) throw new WorkerError("WORKER_UNKNOWN_JOB", `Job "${name}" is not the definition this queue was created with`); + return def; + }; + + const enqueue = async (job: AnyJob | string, data: unknown, opts?: EnqueueOptions): Promise<JobInfo> => { + if (closed) throw new WorkerError("WORKER_CLOSED", "The job queue is closed"); + const def = resolve(job); + return o.backend.add(prepareJob(def, data, opts)); + }; + + const available = async () => { + if (closed) return false; + try { + return (await (o.backend.available?.() ?? true)) === true; + } catch { + return false; + } + }; + + const queue: JobQueue = { + mode: o.backend.mode, + enqueue: enqueue as JobQueue["enqueue"], + get: (async (job: AnyJob | string, id: string) => (typeof id === "string" && id ? o.backend.get(resolve(job), id) : null)) as JobQueue["get"], + enqueuer(job) { + const def = resolve(job); + return { + async enqueue(data) { + const info = await enqueue(def, data); + if (info.state === "failed") throw new WorkerError("WORKER_UNAVAILABLE", `Job "${def.name}" failed: ${info.error ?? "Job failed"}`); + return info.id; + }, + available, + }; + }, + definition: (name) => resolve(name), + names: () => [...byName.keys()], + available, + async close() { + if (closed) return; + closed = true; + await o.backend.close(); + }, + }; + return queue; +} diff --git a/src/types.ts b/src/types.ts new file mode 100644 index 0000000..48b7f69 --- /dev/null +++ b/src/types.ts @@ -0,0 +1,90 @@ +// The job model shared by every backend. A job is JSON data on a named queue; a definition says +// how to run it. Backends (in-process, BullMQ on Redis, fallback) only move and store jobs. + +/** A job's state, the same words on every backend. */ +export type JobState = "waiting" | "delayed" | "active" | "completed" | "failed"; + +/** What a processor gets besides the data. */ +export interface JobContext { + id: string; + name: string; + queue: string; + /** 1 on the first run, 2 on the first retry, … */ + attempt: number; + /** Total attempts allowed. */ + attempts: number; +} + +/** + * A kind of background work. Data and result cross JSON (Dates arrive as ISO strings), on every + * backend, so a job behaves the same in-process and on Redis. + */ +export interface JobDefinition<Data = unknown, Result = unknown> { + /** Job name, unique in a queue set, e.g. "password-reset". */ + readonly name: string; + /** Queue it runs on (Redis keys `<prefix>:<queue>:…`). Default: the name. */ + readonly queue: string; + /** Runs before enqueueing, in the caller's request: throw (WorkerError WORKER_BAD_DATA) to refuse. */ + readonly validate?: ((data: Data) => void) | undefined; + readonly process: (data: Data, ctx: JobContext) => Promise<Result> | Result; + /** Total attempts, with exponential backoff between them. */ + readonly attempts: number; + readonly backoffMs: number; + /** How long a finished job (and its result) stays readable. 0 = removed at once. */ + readonly keepResultsMs: number; + /** How long a failed job stays readable (at least a day by default, for inspection). */ + readonly keepFailedMs: number; + /** Largest JSON payload accepted, in bytes. */ + readonly maxBytes: number; + /** Refuse data with token / password / secret keys (see SECRET_KEY). */ + readonly refuseSecrets: boolean; +} + +// eslint-disable-next-line @typescript-eslint/no-explicit-any +export type AnyJob = JobDefinition<any, any>; +export type DataOf<J> = J extends JobDefinition<infer D, unknown> ? D : never; +export type ResultOf<J> = J extends JobDefinition<unknown, infer R> ? R : never; + +/** A job as callers see it. */ +export interface JobInfo<Data = unknown, Result = unknown> { + id: string; + name: string; + queue: string; + state: JobState; + /** Attempts made so far. */ + attempts: number; + /** Safe to show: a JobFailure's message, otherwise a generic "Job failed". */ + error: string | null; + result: Result | null; + data: Data; + createdAt: Date; + finishedAt: Date | null; +} + +export interface EnqueueOptions { + /** Run no sooner than this many ms from now. */ + delayMs?: number; +} + +/** A job ready for a backend: JSON-checked data plus the backend-independent options. */ +export interface PreparedJob { + id: string; + def: AnyJob; + data: unknown; + delayMs: number; +} + +/** + * Where jobs go. Implementations: `inlineBackend` (this process), `bullmqBackend` ("/bullmq", + * Redis) and `fallbackBackend` (a worker when available, else standalone). Write your own the same way. + */ +export interface JobBackend { + /** "inline", "bullmq", "fallback" or your own; shown in logs and health checks. */ + readonly mode: string; + add(job: PreparedJob): Promise<JobInfo>; + /** The job with this id on `def`'s queue, or null (unknown, expired, or another job's name). */ + get(def: AnyJob, id: string): Promise<JobInfo | null>; + /** Cheap readiness check (e.g. the Redis connection is "ready"). Default: assume yes. */ + available?(): boolean | Promise<boolean>; + close(): Promise<void>; +} diff --git a/tsconfig.json b/tsconfig.json new file mode 100644 index 0000000..adfb128 --- /dev/null +++ b/tsconfig.json @@ -0,0 +1,15 @@ +{ + "compilerOptions": { + "target": "ES2022", + "lib": ["es2022"], + "module": "esnext", + "moduleResolution": "bundler", + "strict": true, + "noUncheckedIndexedAccess": true, + "skipLibCheck": true, + "isolatedModules": true, + "types": ["node"], + "noEmit": true + }, + "include": ["src", "scripts"] +} diff --git a/tsup.config.ts b/tsup.config.ts new file mode 100644 index 0000000..75b4502 --- /dev/null +++ b/tsup.config.ts @@ -0,0 +1,11 @@ +import { defineConfig } from "tsup"; + +export default defineConfig({ + entry: { index: "src/index.ts", bullmq: "src/bullmq.ts" }, + format: ["esm"], + dts: true, + target: "node20", + platform: "node", + clean: true, + external: [/^node:/, "bullmq", "ioredis"], +});