release: WRNexusJS 0.3.0

packages/queue/package.json

@@ -1,6 +1,6 @@
 {
   "name": "@wrnexus/queue",
-  "version": "0.2.79",
+  "version": "0.3.0",
   "private": true,
   "type": "module",
   "main": "src/index.ts",

packages/queue/src/index.ts

@@ -20,6 +20,9 @@ export interface Job<T = unknown> {
   runAt: number;
   /** If set, re-enqueue this job this many ms after each successful run. */
   repeat?: number;
+  priority: number;
+  idempotencyKey?: string;
+  createdAt: number;
 }
 
 export type JobHandler<T = unknown> = (job: Job<T>) => void | Promise<void>;
@@ -31,6 +34,10 @@ export interface AddOptions {
   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;
 }
 
 export interface QueueOptions {
@@ -42,6 +49,8 @@ export interface QueueOptions {
   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;
   /** Clock injection (tests). Default Date.now. */
   now?: () => number;
 }
@@ -54,18 +63,66 @@ export interface Queue {
   start(): void;
   stop(): void;
   size(): number;
+  get(id: string): Job | undefined;
+  list(name?: string): Job[];
+  cancel(id: string): boolean;
+}
+
+export interface JobDefinition<I> {
+  name: string;
+  options?: Omit<AddOptions, "idempotencyKey">;
+  run: JobHandler<I>;
+}
+
+export function defineJob<I>(definition: JobDefinition<I>): JobDefinition<I> {
+  return definition;
+}
+
+export interface WorkflowStep<I, O> {
+  name: string;
+  run(input: I): O | Promise<O>;
+}
+
+export function defineWorkflow<T>(name: string, steps: Array<WorkflowStep<any, any>>) {
+  return {
+    name,
+    steps,
+    async run(input: T): Promise<unknown> {
+      let value: unknown = input;
+      for (const step of steps) value = await step.run(value);
+      return value;
+    },
+  };
+}
+
+export function cronToInterval(cron: string): number {
+  const aliases: Record<string, number> = {
+    "@hourly": 60 * 60 * 1000,
+    "@daily": 24 * 60 * 60 * 1000,
+    "@weekly": 7 * 24 * 60 * 60 * 1000,
+  };
+  if (aliases[cron]) return aliases[cron];
+  const everyMinutes = /^\*\/(\d+)\s+\*\s+\*\s+\*\s+\*$/.exec(cron.trim());
+  if (everyMinutes) return Number(everyMinutes[1]) * 60 * 1000;
+  throw new Error(`WRN-CRON-UNSUPPORTED: '${cron}'. Use @hourly, @daily, @weekly, or */N * * * *.`);
 }
 
 export function createQueue(options: QueueOptions = {}): Queue {
   const defaultMax = options.maxAttempts ?? 3;
   const backoffMs = options.backoffMs ?? 1000;
   const pollMs = options.pollMs ?? 250;
+  const concurrency = options.concurrency ?? Number.POSITIVE_INFINITY;
   if (!Number.isInteger(defaultMax) || defaultMax < 1)
     throw new RangeError("queue maxAttempts must be a positive integer");
   if (!Number.isFinite(backoffMs) || backoffMs < 0)
     throw new RangeError("queue backoffMs must be a non-negative number");
   if (!Number.isFinite(pollMs) || pollMs < 1)
     throw new RangeError("queue pollMs must be at least 1ms");
+  if (!(
+    concurrency === Number.POSITIVE_INFINITY ||
+    (Number.isInteger(concurrency) && concurrency > 0)
+  ))
+    throw new RangeError("queue concurrency must be a positive integer");
   const now = options.now ?? Date.now;
 
   const jobs: Job[] = [];
@@ -100,7 +157,10 @@ export function createQueue(options: QueueOptions = {}): Queue {
     draining = true;
     try {
       const cutoff = at ?? now();
-      const due = jobs.filter((j) => j.runAt <= cutoff && handlers.has(j.name));
+      const due = jobs
+        .filter((j) => j.runAt <= cutoff && handlers.has(j.name))
+        .sort((a, b) => b.priority - a.priority || a.runAt - b.runAt || a.createdAt - b.createdAt)
+        .slice(0, concurrency);
       await Promise.all(due.map(runJob));
       return due.length;
     } finally {
@@ -120,14 +180,24 @@ export function createQueue(options: QueueOptions = {}): Queue {
         throw new RangeError("job delayMs must be a non-negative number");
       if (opts.repeat !== undefined && (!Number.isFinite(opts.repeat) || opts.repeat <= 0))
         throw new RangeError("job repeat must be a positive number");
+      if (opts.priority !== undefined && !Number.isFinite(opts.priority))
+        throw new RangeError("job priority must be a finite number");
+      if (opts.idempotencyKey) {
+        const existing = jobs.find((job) => job.idempotencyKey === opts.idempotencyKey);
+        if (existing) return existing as Job<typeof data>;
+      }
+      const createdAt = now();
       const job: Job = {
         id: `job_${++seq}`,
         name,
         data,
         attempts: 0,
         maxAttempts: opts.maxAttempts ?? defaultMax,
-        runAt: now() + (opts.delayMs ?? 0),
+        runAt: createdAt + (opts.delayMs ?? 0),
         repeat: opts.repeat,
+        priority: opts.priority ?? 0,
+        idempotencyKey: opts.idempotencyKey,
+        createdAt,
       };
       jobs.push(job);
       return job as Job<typeof data>;
@@ -145,5 +215,13 @@ export function createQueue(options: QueueOptions = {}): Queue {
       timer = null;
     },
     size: () => jobs.length,
+    get: (id) => jobs.find((job) => job.id === id),
+    list: (name) => jobs.filter((job) => !name || job.name === name).map((job) => ({ ...job })),
+    cancel(id) {
+      const index = jobs.findIndex((job) => job.id === id);
+      if (index < 0) return false;
+      jobs.splice(index, 1);
+      return true;
+    },
   };
 }

packages/queue/test/queue.test.ts

@@ -1,5 +1,5 @@
 import { test, expect } from "bun:test";
-import { createQueue } from "../src/index.ts";
+import { createQueue, cronToInterval, defineWorkflow } from "../src/index.ts";
 
 test("processes a job", async () => {
   const queue = createQueue();
@@ -104,3 +104,37 @@ test("drains independent due jobs concurrently", async () => {
   releases.forEach((release) => release());
   expect(await draining).toBe(2);
 });
+
+test("prioritizes due jobs and respects concurrency", async () => {
+  const queue = createQueue({ concurrency: 1 });
+  const order: string[] = [];
+  queue.process<string>("work", (job) => {
+    order.push(job.data);
+  });
+  await queue.add("work", "low", { priority: 1 });
+  await queue.add("work", "high", { priority: 10 });
+  expect(await queue.drain()).toBe(1);
+  expect(order).toEqual(["high"]);
+  expect(queue.size()).toBe(1);
+});
+
+test("deduplicates, lists and cancels queued work", async () => {
+  const queue = createQueue();
+  const first = await queue.add("sync", { id: 1 }, { idempotencyKey: "customer:1" });
+  const second = await queue.add("sync", { id: 2 }, { idempotencyKey: "customer:1" });
+  expect(second.id).toBe(first.id);
+  expect(queue.list("sync")).toHaveLength(1);
+  expect(queue.get(first.id)?.data).toEqual({ id: 1 });
+  expect(queue.cancel(first.id)).toBe(true);
+  expect(queue.cancel(first.id)).toBe(false);
+});
+
+test("runs typed workflows and parses supported cron expressions", async () => {
+  const workflow = defineWorkflow<number>("double-and-label", [
+    { name: "double", run: (value: number) => value * 2 },
+    { name: "label", run: (value: number) => `value:${value}` },
+  ]);
+  expect(await workflow.run(5)).toBe("value:10");
+  expect(cronToInterval("@hourly")).toBe(3_600_000);
+  expect(cronToInterval("*/5 * * * *")).toBe(300_000);
+});