453 lines
33 KiB
Plaintext
453 lines
33 KiB
Plaintext
page wrnexusqueue {
|
|
seo {
|
|
title = "@wrnexus/queue"
|
|
description = "Background jobs with delay, concurrency, retry, and repetition."
|
|
}
|
|
|
|
view {
|
|
<div class="docs-shell">
|
|
<a href="#main" class="skip-link">Skip to content</a>
|
|
<header class="topbar">
|
|
<a class="brand" href="/"><span>W</span> WRNexusJS</a>
|
|
<nav aria-label="Primary"><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a></nav>
|
|
<div class="topbar-actions"><a class="preview-pill" href="/access">Private preview · v0.8.8</a><button data-wire-theme-toggle class="theme-button" aria-label="Toggle color theme" title="Toggle color theme">◐</button></div>
|
|
</header>
|
|
<div class="mobile-doc-nav"><details><summary>Browse documentation</summary><nav><a href="/getting-started">Get started</a><a href="/packages">Packages</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/language">Language</a><a href="/architecture">Architecture</a><a href="/tutorial">Tutorial</a><a href="/guides/project-structure">Guides</a><a href="/examples">Examples</a><a href="/search">Search</a></nav></details></div>
|
|
<main id="main" class="portal-main docs-layout docs-layout--navigation"><aside class="docs-section-menu" aria-label="Documentation sections"><nav>
|
|
<section><strong>Get started</strong><a href="/getting-started">Quick start</a><a href="/tutorial">Tutorial</a><a href="/guides/full-stack-auth-demo">Auth dashboard demo</a><a href="/guides/configuration-and-profiles">Configuration</a></section>
|
|
<section><strong>Build</strong><a href="/guides/pages-and-components">Pages and components</a><a href="/guides/routing">Routing</a><a href="/guides/server-data">Server data</a><a href="/guides/forms-and-validation">Forms and validation</a><a href="/guides/database">Database</a></section>
|
|
<section><strong>Auth and security</strong><a href="/guides/authentication">Authentication</a><a href="/guides/authorization">Authorization</a><a href="/guides/security">Application security</a><a href="/security">Security policy</a></section>
|
|
<section><strong>Reference</strong><a href="/packages">Packages overview</a><a href="/packages/cli">CLI</a><a href="/language">.wrn language</a><a href="https://component.wrnexusjs.dev/">Components</a><a href="/examples">Examples</a></section>
|
|
<section class="package-menu"><strong>Packages</strong><a href="/packages/ai">@wrnexus/ai</a><a href="/packages/auth">@wrnexus/auth</a><a href="/packages/authz">@wrnexus/authz</a><a href="/packages/benchmark">@wrnexus/benchmark</a><a href="/packages/cache">@wrnexus/cache</a><a href="/packages/captcha">@wrnexus/captcha</a><a href="/packages/cli">@wrnexus/cli</a><a href="/packages/compiler">@wrnexus/compiler</a><a href="/packages/content">@wrnexus/content</a><a href="/packages/core">@wrnexus/core</a><a href="/packages/csr">@wrnexus/csr</a><a href="/packages/db">@wrnexus/db</a><a href="/packages/dev-server">@wrnexus/dev-server</a><a href="/packages/dev-toolbar">@wrnexus/dev-toolbar</a><a href="/packages/encryption">@wrnexus/encryption</a><a href="/packages/graphql">@wrnexus/graphql</a><a href="/packages/helpers">@wrnexus/helpers</a><a href="/packages/i18n">@wrnexus/i18n</a><a href="/packages/identity">@wrnexus/identity</a><a href="/packages/image">@wrnexus/image</a><a href="/packages/jwt">@wrnexus/jwt</a><a href="/packages/language-server">@wrnexus/language-server</a><a href="/packages/mcp">@wrnexus/mcp</a><a href="/packages/mobile">@wrnexus/mobile</a><a href="/packages/native">@wrnexus/native</a><a href="/packages/oauth">@wrnexus/oauth</a><a href="/packages/observability">@wrnexus/observability</a><a href="/packages/playground">@wrnexus/playground</a><a href="/packages/plugin">@wrnexus/plugin</a><a href="/packages/pubsub">@wrnexus/pubsub</a><a href="/packages/pwa">@wrnexus/pwa</a><a href="/packages/queue">@wrnexus/queue</a><a href="/packages/reactive">@wrnexus/reactive</a><a href="/packages/realtime">@wrnexus/realtime</a><a href="/packages/router">@wrnexus/router</a><a href="/packages/security">@wrnexus/security</a><a href="/packages/ssr">@wrnexus/ssr</a><a href="/packages/store">@wrnexus/store</a><a href="/packages/styles">@wrnexus/styles</a><a href="/packages/syntax">@wrnexus/syntax</a><a href="/packages/test">@wrnexus/test</a><a href="/packages/tracking">@wrnexus/tracking</a><a href="/packages/typecheck">@wrnexus/typecheck</a><a href="/packages/ui">@wrnexus/ui</a><a href="/packages/uploader">@wrnexus/uploader</a><a href="/packages/validation">@wrnexus/validation</a></section>
|
|
</nav></aside><div class="docs-reading-column">
|
|
<article class="documentation prose standalone package-document"><nav class="breadcrumbs" aria-label="Breadcrumb"><a href="/">Home</a><span>/</span><a href="/packages">Packages</a><span>/</span><span aria-current="page">@wrnexus/queue</span></nav><section class="doc-intro"><span class="eyebrow">Data · Package reference</span><h1>@wrnexus/queue</h1><p>Background jobs with delay, concurrency, retry, and repetition.</p><div class="doc-meta"><span>v0.8.8</span><span>Private registry</span><span>Data</span></div><section id="access" class="access-callout"><h2>Install the package</h2><p>After WorkRoot approves private registry access, install the release-aligned package:</p><pre><code>bun add @wrnexus/queue@0.8.8</code><button type="button" class="copy-button" aria-label="Copy installation command">Copy</button></pre><p><a href="/access">Request preview access</a>. Never put registry tokens in source control.</p></section></section><section id="guide"><blockquote>A background job queue with delays, retries + exponential backoff, recurring jobs, and concurrent workers.</blockquote>
|
|
<p>Part of the <strong>WRNexusJS</strong> framework — an SSR-first, Bun-native full-stack web framework.</p>
|
|
<h3 id="overview">Overview</h3>
|
|
<p><code>@wrnexus/queue</code> is a server-side in-process job queue. You register named workers, enqueue jobs (optionally delayed or recurring), and let the queue poll and run them on a timer — with per-job retry limits and doubling backoff between attempts. The default store lives in memory; the design allows a pluggable driver to back it with Redis/SQL for durability across restarts. Reach for it when you need to defer work (emails, webhooks, cleanup) off the request path without a heavyweight external broker. Tests can drive it deterministically via <code>drain()</code>.</p>
|
|
<pre data-language="bash"><code>bun add @wrnexus/queue</code></pre>
|
|
<blockquote>Private package — the machine must be authenticated to the <code>wrnexus</code> npm org</blockquote>
|
|
<blockquote>(a read token in <code>~/.npmrc</code>). Requires <strong>Bun</strong> (Node is not supported).</blockquote>
|
|
<h3 id="api">API</h3>
|
|
<p>The package exports a single factory plus its supporting types.</p>
|
|
<h4 id="createqueue-options-queue"><code>createQueue(options?): Queue</code></h4>
|
|
<p>Creates a new queue instance.</p>
|
|
<pre data-language="ts"><code>function createQueue(options?: QueueOptions): Queue;</code></pre>
|
|
<h4 id="queueoptions"><code>QueueOptions</code></h4>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Option</th><th>Type</th><th>Default</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>maxAttempts</code></td><td><code>number</code></td><td><code>3</code></td><td>Default max attempts per job before it is dead-lettered.</td></tr><tr><td><code>backoffMs</code></td><td><code>number</code></td><td><code>1000</code></td><td>Base retry backoff in ms; doubles per attempt.</td></tr><tr><td><code>pollMs</code></td><td><code>number</code></td><td><code>250</code></td><td>Poll interval used once <code>start()</code> is called (ms).</td></tr><tr><td><code>onFailed</code></td><td><code>(job: Job, error: unknown) => void</code></td><td>—</td><td>Called when a job exhausts its attempts.</td></tr><tr><td><code>concurrency</code></td><td><code>number</code></td><td>unlimited</td><td>Maximum jobs claimed by one <code>drain()</code> call.</td></tr><tr><td><code>capacity</code></td><td><code>number</code></td><td>unlimited</td><td>Maximum queued plus active jobs before adds reject.</td></tr><tr><td><code>now</code></td><td><code>() => number</code></td><td><code>Date.now</code></td><td>Clock injection for deterministic tests.</td></tr></tbody></table></div>
|
|
<h4 id="queue"><code>Queue</code></h4>
|
|
<p>The object returned by <code>createQueue</code>.</p>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Method</th><th>Signature</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>add</code></td><td><code>add<T>(name, data: T, options?: AddOptions): Promise<Job<T>></code></td><td>Enqueue a job under a worker name. Returns the created job.</td></tr><tr><td><code>process</code></td><td><code>process<T>(name, handler: JobHandler<T>): void</code></td><td>Register the worker that runs jobs of the given name.</td></tr><tr><td><code>drain</code></td><td><code>drain(now?: number): Promise<number></code></td><td>Run every job whose <code>runAt ≤ now</code>, once. Returns how many ran.</td></tr><tr><td><code>start</code></td><td><code>start(): void</code></td><td>Begin polling every <code>pollMs</code>. No-op if already started.</td></tr><tr><td><code>stop</code></td><td><code>stop(): void</code></td><td>Stop the poll timer.</td></tr><tr><td><code>shutdown</code></td><td><code>shutdown({ force? }): Promise<void></code></td><td>Stop accepting jobs and await active work; force aborts it.</td></tr><tr><td><code>size</code></td><td><code>size(): number</code></td><td>Number of jobs currently queued.</td></tr><tr><td><code>get/list</code></td><td><code>get(id)</code> / <code>list(name?)</code></td><td>Inspect defensive copies of pending jobs.</td></tr><tr><td><code>cancel</code></td><td><code>cancel(id): boolean</code></td><td>Remove queued work or abort an active handler.</td></tr><tr><td><code>failed</code></td><td><code>failed(): Job[]</code></td><td>Inspect exhausted jobs in the dead-letter collection.</td></tr><tr><td><code>retry</code></td><td><code>retry(id): Promise<boolean></code></td><td>Reset and requeue a dead-lettered job.</td></tr></tbody></table></div>
|
|
<h4 id="addoptions"><code>AddOptions</code></h4>
|
|
<div class="table-wrap"><table>
|
|
<thead><tr><th>Option</th><th>Type</th><th>Description</th></tr></thead>
|
|
<tbody><tr><td><code>delayMs</code></td><td><code>number</code></td><td>Delay before the job becomes runnable (ms).</td></tr><tr><td><code>maxAttempts</code></td><td><code>number</code></td><td>Max attempts before dead-lettering. Defaults to the queue's <code>maxAttempts</code>.</td></tr><tr><td><code>repeat</code></td><td><code>number</code></td><td>Re-enqueue this job this many ms after each successful run (recurring).</td></tr><tr><td><code>priority</code></td><td><code>number</code></td><td>Higher values are selected first among due jobs.</td></tr><tr><td><code>idempotencyKey</code></td><td><code>string</code></td><td>Return the matching pending job instead of enqueueing a duplicate.</td></tr></tbody></table></div>
|
|
<h4 id="jobhandler-t"><code>JobHandler<T></code></h4>
|
|
<pre data-language="ts"><code>type JobHandler<T = unknown> = (
|
|
job: Job<T>,
|
|
context: { signal: AbortSignal },
|
|
) => void | Promise<void>;</code></pre>
|
|
<h4 id="job-t"><code>Job<T></code></h4>
|
|
<pre data-language="ts"><code>interface Job<T = unknown> {
|
|
id: string; // e.g. "job_1"
|
|
name: string;
|
|
data: T;
|
|
attempts: number;
|
|
maxAttempts: number;
|
|
runAt: number; // epoch ms; job runs when now ≥ runAt
|
|
repeat?: number; // if set, re-enqueue this many ms after each success
|
|
}</code></pre>
|
|
<h3 id="usage">Usage</h3>
|
|
<p>Register workers, enqueue jobs, then start the poller:</p>
|
|
<pre data-language="ts"><code>import { createQueue } from "@wrnexus/queue";
|
|
|
|
const queue = createQueue({ maxAttempts: 3, backoffMs: 1000 });
|
|
|
|
// Register a worker for the "email" job name.
|
|
queue.process<{ to: string }>("email", async (job) => {
|
|
await send(job.data.to);
|
|
});
|
|
|
|
// Enqueue a delayed job with up to 3 attempts.
|
|
await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });
|
|
|
|
queue.start(); // begin polling; queue.stop() to halt</code></pre>
|
|
<p>Use <code>context.signal</code> in network/database calls so forced shutdown and active cancellation finish promptly. For process termination, prefer <code>await queue.shutdown()</code>; use <code>{ force: true }</code> only after your grace period.</p>
|
|
<h4 id="durable-queue">Durable queue</h4>
|
|
<p><code>createDurableQueue({ store })</code> retains jobs until their handler succeeds and supports atomic leases when a driver implements <code>QueueStore.claim</code>. It exposes the same cancellation/shutdown behavior plus <code>list</code>, <code>failed</code>, and <code>retry</code>. The included <code>memoryQueueStore()</code> is useful for tests; production Redis/SQL drivers should make <code>claim()</code> atomic to prevent two workers executing one job.</p>
|
|
<h4 id="recurring-jobs">Recurring jobs</h4>
|
|
<p>Pass <code>repeat</code> to re-enqueue a job a fixed interval after each successful run:</p>
|
|
<pre data-language="ts"><code>queue.process("heartbeat", async () => ping());
|
|
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute</code></pre>
|
|
<h4 id="handling-permanent-failures">Handling permanent failures</h4>
|
|
<p>When a job's <code>attempts</code> reaches <code>maxAttempts</code>, it is dropped and <code>onFailed</code> fires instead of retrying:</p>
|
|
<pre data-language="ts"><code>const queue = createQueue({
|
|
onFailed: (job, error) => {
|
|
console.error(`job ${job.id} (${job.name}) gave up`, error);
|
|
},
|
|
});</code></pre>
|
|
<h4 id="deterministic-testing">Deterministic testing</h4>
|
|
<p>Instead of <code>start()</code>, inject a clock and drive the queue with <code>drain()</code>:</p>
|
|
<pre data-language="ts"><code>let clock = 0;
|
|
const queue = createQueue({ now: () => clock });
|
|
|
|
queue.process("task", async () => {
|
|
/* ... */
|
|
});
|
|
await queue.add("task", {}, { delayMs: 5000 });
|
|
|
|
clock = 5000;
|
|
const ran = await queue.drain(); // => 1</code></pre>
|
|
<h4 id="durable-workflows-and-approvals">Durable workflows and approvals</h4>
|
|
<p><code>createWorkflowEngine(store)</code> executes dependency-ordered steps and persists every transition, result, progress update, failure, cancellation, and approval record. Approval steps pause safely and can resume after a process restart because the snapshot lives in the supplied <code>WorkflowStore</code>.</p>
|
|
<pre data-language="ts"><code>const workflow = defineDurableWorkflow({
|
|
name: "publish-report",
|
|
steps: [
|
|
{ name: "build", run: buildReport },
|
|
{ name: "approve", dependsOn: ["build"], approval: true, run: (report) => report },
|
|
{ name: "publish", dependsOn: ["approve"], run: publishReport },
|
|
],
|
|
});
|
|
|
|
const run = await engine.start(workflow, input);
|
|
await engine.approve(workflow, run.id, "approve", currentUser.id);</code></pre>
|
|
<p>Use <code>memoryWorkflowStore()</code> for tests. Production stores implement the small <code>get</code>, <code>put</code>, and <code>list</code> contract using the same transactional database or durable service as the application.</p>
|
|
<h3 id="retry-backoff-behavior">Retry & backoff behavior</h3>
|
|
<ul>
|
|
<li>On a thrown handler error, the job is retried while <code>attempts < maxAttempts</code>.</li>
|
|
<li>The next <code>runAt</code> is set to <code>now + backoffMs * 2^(attempts - 1)</code> (exponential</li>
|
|
<p>backoff): with <code>backoffMs: 1000</code> the delays are 1s, 2s, 4s, …</p>
|
|
<li>A job whose worker name has no registered handler stays queued until one is</li>
|
|
<p>registered (it is not counted as runnable by <code>drain</code>).</p>
|
|
<li><code>drain</code> is re-entrant-safe: overlapping calls are skipped while one is running.</li>
|
|
</ul>
|
|
<h3 id="requirements-notes">Requirements / Notes</h3>
|
|
<ul>
|
|
<li><strong>Bun-only</strong> runtime (Node is not supported), consistent with the rest of the</li>
|
|
<p>WRNexusJS framework. The queue itself relies only on standard timers (<code>setInterval</code>/<code>clearInterval</code>) and has no runtime dependencies.</p>
|
|
<li>The default store is in-process, so queued jobs do not survive a restart; a</li>
|
|
<p>pluggable driver is intended for backing it with Redis/SQL for durability.</p>
|
|
<li>Works alongside <code>@wrnexus/core</code> for offloading work from the request path.</li>
|
|
</ul></section><section id="api" class="api"><h2>Complete TypeScript API</h2><p>Generated from the exact installed package declarations.</p><pre data-language="typescript"><code>import { ExecutionContext, Context } from '@wrnexus/core';
|
|
import { SubjectContext } from '@wrnexus/rpc';
|
|
|
|
/**
|
|
* Persistence contract for the durable queue.
|
|
*
|
|
* Distributed drivers should implement `claim()` atomically and exclude leased
|
|
* jobs from `due()` until their lease expires. Calling `put()` must replace the
|
|
* stored record and release any previous lease for that job.
|
|
*/
|
|
interface QueueStore {
|
|
put(job: Job): Promise<void>;
|
|
get(id: string): Promise<Job | null>;
|
|
remove(id: string): Promise<void>;
|
|
due(now: number, limit: number): Promise<Job[]>;
|
|
list(name?: string): Promise<Job[]>;
|
|
claim?(id: string, worker: string, leaseUntil: number): Promise<boolean>;
|
|
}
|
|
declare function memoryQueueStore(): QueueStore;
|
|
interface DurableQueueOptions {
|
|
store?: QueueStore;
|
|
workerId?: string;
|
|
maxAttempts?: number;
|
|
concurrency?: number;
|
|
capacity?: number;
|
|
leaseMs?: number;
|
|
backoff?: (attempt: number) => number;
|
|
now?: () => number;
|
|
onDeadLetter?: (job: Job, error: unknown) => void | Promise<void>;
|
|
context?: (job: Job, signal: AbortSignal) => ExecutionContext;
|
|
}
|
|
interface DurableQueue {
|
|
add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
|
|
process<T>(name: string, handler: JobHandler<T>): void;
|
|
drain(): Promise<number>;
|
|
cancel(id: string): Promise<boolean>;
|
|
shutdown(options?: {
|
|
force?: boolean;
|
|
}): Promise<void>;
|
|
list(name?: string): Promise<Job[]>;
|
|
failed(): Job[];
|
|
retry(id: string): Promise<boolean>;
|
|
}
|
|
declare function createDurableQueue(options?: DurableQueueOptions): DurableQueue;
|
|
|
|
interface RedisQueueClient {
|
|
get(key: string): Promise<string | null>;
|
|
set(key: string, value: string, options?: {
|
|
NX?: boolean;
|
|
PX?: number;
|
|
}): Promise<unknown>;
|
|
del(...keys: string[]): Promise<unknown>;
|
|
zadd(key: string, score: number, member: string): Promise<unknown>;
|
|
zrem(key: string, member: string): Promise<unknown>;
|
|
zrangebyscore(key: string, min: number, max: number, options?: {
|
|
limit: [number, number];
|
|
}): Promise<string[]>;
|
|
smembers(key: string): Promise<string[]>;
|
|
sadd(key: string, member: string): Promise<unknown>;
|
|
srem(key: string, member: string): Promise<unknown>;
|
|
}
|
|
/** Redis-backed queue store using only the common client command surface. */
|
|
declare function redisQueueStore(client: RedisQueueClient, prefix?: string): QueueStore;
|
|
interface SqlQueueClient {
|
|
query<T = Record<string, unknown>>(sql: string, parameters?: unknown[]): Promise<{
|
|
rows: T[];
|
|
}>;
|
|
}
|
|
/** PostgreSQL store with atomic SKIP LOCKED leasing and JSON payloads. */
|
|
declare function postgresQueueStore(db: SqlQueueClient, table?: string): QueueStore;
|
|
declare const POSTGRES_QUEUE_SCHEMA = "CREATE TABLE IF NOT EXISTS wrnexus_jobs (\n id text PRIMARY KEY, name text NOT NULL, payload jsonb NOT NULL, run_at bigint NOT NULL,\n priority integer NOT NULL DEFAULT 0, lease_owner text, lease_until bigint\n); CREATE INDEX IF NOT EXISTS wrnexus_jobs_due ON wrnexus_jobs (run_at, priority DESC);";
|
|
|
|
interface ScheduledJob<T = unknown> {
|
|
name: string;
|
|
data: T;
|
|
everyMs: number;
|
|
options?: AddOptions;
|
|
}
|
|
interface QueueScheduler {
|
|
start(): void;
|
|
stop(): void;
|
|
tick(now?: number): Promise<number>;
|
|
snapshot(): {
|
|
running: boolean;
|
|
schedules: number;
|
|
nextRuns: Record<string, number>;
|
|
};
|
|
}
|
|
/** Restart-safe scheduler when used with a durable queue and stable idempotency buckets. */
|
|
declare function createQueueScheduler(queue: DurableQueue, schedules: ScheduledJob[], options?: {
|
|
pollMs?: number;
|
|
now?: () => number;
|
|
}): QueueScheduler;
|
|
declare function addBatch<T>(queue: DurableQueue, name: string, values: T[], options?: AddOptions): Promise<Job<T>[]>;
|
|
interface QueueDashboardSnapshot {
|
|
generatedAt: number;
|
|
pending: number;
|
|
failed: number;
|
|
byName: Record<string, number>;
|
|
oldestRunAt?: number;
|
|
}
|
|
declare function queueDashboardSnapshot(queue: DurableQueue): Promise<QueueDashboardSnapshot>;
|
|
declare function renderQueueDashboard(snapshot: QueueDashboardSnapshot): string;
|
|
/** Long-running scheduler/worker loop suitable for a dedicated process or container. */
|
|
declare function runQueueDaemon(queue: DurableQueue, scheduler: QueueScheduler, options?: {
|
|
signal?: AbortSignal;
|
|
pollMs?: number;
|
|
onError?: (error: unknown) => void;
|
|
}): Promise<void>;
|
|
|
|
type WorkflowStatus = "pending" | "running" | "waiting-approval" | "completed" | "failed" | "cancelled";
|
|
interface WorkflowStep$1<I = unknown, O = unknown> {
|
|
name: string;
|
|
dependsOn?: string[];
|
|
approval?: boolean;
|
|
run(input: I, context: WorkflowRunContext): O | Promise<O>;
|
|
}
|
|
interface WorkflowRunContext {
|
|
workflowId: string;
|
|
step: string;
|
|
results: Readonly<Record<string, unknown>>;
|
|
signal: AbortSignal;
|
|
progress(value: number, message?: string): void;
|
|
}
|
|
interface WorkflowSnapshot {
|
|
id: string;
|
|
name: string;
|
|
status: WorkflowStatus;
|
|
input: unknown;
|
|
results: Record<string, unknown>;
|
|
completed: string[];
|
|
waitingFor?: string;
|
|
progress: number;
|
|
message?: string;
|
|
error?: string;
|
|
updatedAt: number;
|
|
}
|
|
interface WorkflowStore {
|
|
get(id: string): Promise<WorkflowSnapshot | null>;
|
|
put(snapshot: WorkflowSnapshot): Promise<void>;
|
|
list(): Promise<WorkflowSnapshot[]>;
|
|
}
|
|
declare function memoryWorkflowStore(): WorkflowStore;
|
|
interface WorkflowDefinition<I = unknown> {
|
|
name: string;
|
|
steps: WorkflowStep$1<any, any>[];
|
|
/** Compile-time input marker; definitions do not store runtime input values. */
|
|
readonly __input?: I;
|
|
}
|
|
interface WorkflowEngine {
|
|
start<I>(definition: WorkflowDefinition<I>, input: I, id?: string): Promise<WorkflowSnapshot>;
|
|
resume<I>(definition: WorkflowDefinition<I>, id: string): Promise<WorkflowSnapshot>;
|
|
approve<I>(definition: WorkflowDefinition<I>, id: string, step: string, actor: string): Promise<WorkflowSnapshot>;
|
|
cancel(id: string): Promise<boolean>;
|
|
get(id: string): Promise<WorkflowSnapshot | null>;
|
|
list(): Promise<WorkflowSnapshot[]>;
|
|
}
|
|
declare function createWorkflowEngine(store?: WorkflowStore): WorkflowEngine;
|
|
declare function defineDurableWorkflow<I>(definition: WorkflowDefinition<I>): WorkflowDefinition<I>;
|
|
|
|
interface SubjectEnvelope<T> {
|
|
payload: T;
|
|
identity?: string;
|
|
}
|
|
interface SubjectJob<T> extends Omit<Job<SubjectEnvelope<T>>, "data"> {
|
|
data: T;
|
|
subject?: SubjectContext;
|
|
}
|
|
interface SubjectQueue {
|
|
add<T>(ctx: Context, name: string, data: T, options?: AddOptions): Promise<Job<SubjectEnvelope<T>>>;
|
|
process<T>(name: string, handler: (job: SubjectJob<T>, context: {
|
|
signal: AbortSignal;
|
|
}) => void | Promise<void>): void;
|
|
}
|
|
/** Queue adapter that persists a signed end-user context alongside job data. */
|
|
/** Works with both the in-memory Queue and createDurableQueue(). */
|
|
declare function subjectQueue(queue: Queue | DurableQueue): SubjectQueue;
|
|
|
|
/**
|
|
* @wrnexus/queue — a background job queue with delays, retries + backoff, and
|
|
* concurrent workers. The default store is in-process; a pluggable driver lets
|
|
* you back it with Redis/SQL for durability across restarts.
|
|
*
|
|
* const queue = createQueue();
|
|
* queue.process("email", async (job) => { await send(job.data); });
|
|
* await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });
|
|
* queue.start(); // begin polling; queue.stop() to halt
|
|
*
|
|
* Tests can drive it deterministically with `await queue.drain(now)`.
|
|
*/
|
|
interface Job<T = unknown> {
|
|
id: string;
|
|
name: string;
|
|
data: T;
|
|
attempts: number;
|
|
maxAttempts: number;
|
|
runAt: number;
|
|
/** If set, re-enqueue this job this many ms after each successful run. */
|
|
repeat?: number;
|
|
priority: number;
|
|
idempotencyKey?: string;
|
|
createdAt: number;
|
|
}
|
|
interface JobContext {
|
|
/** Aborted when an active job is cancelled or the queue is force-stopped. */
|
|
signal: AbortSignal;
|
|
/** The same trusted context shape used by HTTP, actions, realtime and webhooks. */
|
|
execution: ExecutionContext;
|
|
}
|
|
type JobHandler<T = unknown> = (job: Job<T>, context: JobContext) => void | Promise<void>;
|
|
interface AddOptions {
|
|
/** Delay before the job becomes runnable (ms). */
|
|
delayMs?: number;
|
|
/** Max attempts before it's dead-lettered. Default from queue options. */
|
|
maxAttempts?: number;
|
|
/** Re-enqueue this job this many ms after each successful run (recurring). */
|
|
repeat?: number;
|
|
/** Higher-priority jobs run first when multiple jobs are due. */
|
|
priority?: number;
|
|
/** Prevent duplicate queued work with the same stable key. */
|
|
idempotencyKey?: string;
|
|
}
|
|
interface QueueOptions {
|
|
/** Default max attempts per job. Default 3. */
|
|
maxAttempts?: number;
|
|
/** Base retry backoff (ms); doubles per attempt. Default 1000. */
|
|
backoffMs?: number;
|
|
/** Poll interval when started (ms). Default 250. */
|
|
pollMs?: number;
|
|
/** Called when a job exhausts its attempts. */
|
|
onFailed?: (job: Job, error: unknown) => void;
|
|
/** Maximum jobs executed in one drain. Default: unlimited. */
|
|
concurrency?: number;
|
|
/** Maximum queued + active jobs. Adds reject once this limit is reached. */
|
|
capacity?: number;
|
|
/** Clock injection (tests). Default Date.now. */
|
|
now?: () => number;
|
|
context?: (job: Job, signal: AbortSignal) => ExecutionContext;
|
|
}
|
|
interface Queue {
|
|
add<T>(name: string, data: T, options?: AddOptions): Promise<Job<T>>;
|
|
process<T>(name: string, handler: JobHandler<T>): void;
|
|
/** Run every job whose runAt ≤ now, once. Returns how many ran. */
|
|
drain(now?: number): Promise<number>;
|
|
start(): void;
|
|
stop(): void;
|
|
/** Stop accepting work and wait for active handlers (or abort them). */
|
|
shutdown(options?: {
|
|
force?: boolean;
|
|
}): Promise<void>;
|
|
size(): number;
|
|
get(id: string): Job | undefined;
|
|
list(name?: string): Job[];
|
|
cancel(id: string): boolean;
|
|
failed(): Job[];
|
|
retry(id: string): Promise<boolean>;
|
|
}
|
|
interface JobDefinition<I> {
|
|
name: string;
|
|
options?: Omit<AddOptions, "idempotencyKey">;
|
|
run: JobHandler<I>;
|
|
}
|
|
declare function defineJob<I>(definition: JobDefinition<I>): JobDefinition<I>;
|
|
interface WorkflowStep<I, O> {
|
|
name: string;
|
|
run(input: I): O | Promise<O>;
|
|
}
|
|
declare function defineWorkflow<T>(name: string, steps: Array<WorkflowStep<any, any>>): {
|
|
name: string;
|
|
steps: WorkflowStep<any, any>[];
|
|
run(input: T): Promise<unknown>;
|
|
};
|
|
declare function cronToInterval(cron: string): number;
|
|
declare function createQueue(options?: QueueOptions): Queue;
|
|
|
|
export { type AddOptions, type DurableQueue, type DurableQueueOptions, type Job, type JobContext, type JobDefinition, type JobHandler, POSTGRES_QUEUE_SCHEMA, type Queue, type QueueDashboardSnapshot, type QueueOptions, type QueueScheduler, type QueueStore, type RedisQueueClient, type ScheduledJob, type SqlQueueClient, type SubjectJob, type SubjectQueue, type WorkflowDefinition, type WorkflowEngine, type WorkflowRunContext, type WorkflowSnapshot, type WorkflowStatus, type WorkflowStep, type WorkflowStore, addBatch, createDurableQueue, createQueue, createQueueScheduler, createWorkflowEngine, cronToInterval, defineDurableWorkflow, defineJob, defineWorkflow, memoryQueueStore, memoryWorkflowStore, postgresQueueStore, queueDashboardSnapshot, redisQueueStore, renderQueueDashboard, runQueueDaemon, subjectQueue };
|
|
</code></pre></section><section id="examples" class="examples"><h2>Examples</h2><p>Copy-ready examples from the installed package documentation.</p><div class="example-grid"><article class="example-card"><h3>Register workers, enqueue jobs, then start the poller</h3><pre data-language="ts"><code>import { createQueue } from "@wrnexus/queue";
|
|
|
|
const queue = createQueue({ maxAttempts: 3, backoffMs: 1000 });
|
|
|
|
// Register a worker for the "email" job name.
|
|
queue.process<{ to: string }>("email", async (job) => {
|
|
await send(job.data.to);
|
|
});
|
|
|
|
// Enqueue a delayed job with up to 3 attempts.
|
|
await queue.add("email", { to: "a@b.com" }, { delayMs: 5000, maxAttempts: 3 });
|
|
|
|
queue.start(); // begin polling; queue.stop() to halt</code></pre></article><article class="example-card"><h3>Recurring jobs</h3><pre data-language="ts"><code>queue.process("heartbeat", async () => ping());
|
|
await queue.add("heartbeat", {}, { repeat: 60_000 }); // runs ~every minute</code></pre></article><article class="example-card"><h3>Handling permanent failures</h3><pre data-language="ts"><code>const queue = createQueue({
|
|
onFailed: (job, error) => {
|
|
console.error(`job ${job.id} (${job.name}) gave up`, error);
|
|
},
|
|
});</code></pre></article><article class="example-card"><h3>Deterministic testing</h3><pre data-language="ts"><code>let clock = 0;
|
|
const queue = createQueue({ now: () => clock });
|
|
|
|
queue.process("task", async () => {
|
|
/* ... */
|
|
});
|
|
await queue.add("task", {}, { delayMs: 5000 });
|
|
|
|
clock = 5000;
|
|
const ran = await queue.drain(); // => 1</code></pre></article><article class="example-card"><h3>Durable workflows and approvals</h3><pre data-language="ts"><code>const workflow = defineDurableWorkflow({
|
|
name: "publish-report",
|
|
steps: [
|
|
{ name: "build", run: buildReport },
|
|
{ name: "approve", dependsOn: ["build"], approval: true, run: (report) => report },
|
|
{ name: "publish", dependsOn: ["approve"], run: publishReport },
|
|
],
|
|
});
|
|
|
|
const run = await engine.start(workflow, input);
|
|
await engine.approve(workflow, run.id, "approve", currentUser.id);</code></pre></article></div></section></article></div>
|
|
<aside class="on-this-page"><h2>On this page</h2><nav><a class="toc-level-2" href="#guide">Guide</a><a class="toc-level-3" href="#overview">Overview</a><a class="toc-level-3" href="#api">API</a><a class="toc-level-4" href="#createqueue-options-queue">createQueue(options?): Queue</a><a class="toc-level-4" href="#queueoptions">QueueOptions</a><a class="toc-level-4" href="#queue">Queue</a><a class="toc-level-4" href="#addoptions">AddOptions</a><a class="toc-level-4" href="#jobhandler-t">JobHandler<T></a><a class="toc-level-4" href="#job-t">Job<T></a><a class="toc-level-3" href="#usage">Usage</a><a class="toc-level-4" href="#durable-queue">Durable queue</a><a class="toc-level-4" href="#recurring-jobs">Recurring jobs</a><a class="toc-level-4" href="#handling-permanent-failures">Handling permanent failures</a><a class="toc-level-4" href="#deterministic-testing">Deterministic testing</a><a class="toc-level-4" href="#durable-workflows-and-approvals">Durable workflows and approvals</a><a class="toc-level-3" href="#retry-backoff-behavior">Retry & backoff behavior</a><a class="toc-level-3" href="#requirements-notes">Requirements / Notes</a><a class="toc-level-2" href="#api">Complete API</a><a class="toc-level-2" href="#examples">Examples</a></nav></aside>
|
|
</main>
|
|
<footer><div class="footer-brand"><span class="footer-mark" aria-hidden="true">W</span><p><strong>WRNexusJS 0.8.8</strong><span>Complete API documentation generated from installed package declarations.</span></p></div><nav aria-label="Footer"><a href="/packages">All packages</a><a href="/getting-started">Get started</a><a href="/security">Security</a><a href="/support">Support</a><a href="/llms.txt">AI guide</a></nav><p class="footer-meta">Private Developer Preview · Bun-native</p></footer>
|
|
</div>
|
|
}
|
|
}
|