import { test, expect } from "bun:test"; import { createQueue, cronToInterval, defineWorkflow } from "../src/index.ts"; test("processes a job", async () => { const queue = createQueue(); const done: string[] = []; queue.process<{ to: string }>("email", (job) => { done.push(job.data.to); }); await queue.add("email", { to: "a@b.com" }); expect(queue.size()).toBe(1); await queue.drain(); expect(done).toEqual(["a@b.com"]); expect(queue.size()).toBe(0); }); test("delayed jobs only run once due", async () => { let clock = 1000; const queue = createQueue({ now: () => clock }); const ran: number[] = []; queue.process("x", () => { ran.push(clock); }); await queue.add("x", 1, { delayMs: 500 }); await queue.drain(1200); // not yet due (runAt = 1500) expect(ran.length).toBe(0); clock = 1600; await queue.drain(); expect(ran.length).toBe(1); }); test("retries with backoff, then dead-letters", async () => { let clock = 0; const failed: unknown[] = []; const queue = createQueue({ maxAttempts: 3, backoffMs: 100, now: () => clock, onFailed: (job) => failed.push(job.id), }); let calls = 0; queue.process("flaky", () => { calls++; throw new Error("boom"); }); await queue.add("flaky", {}); await queue.drain(); // attempt 1 → reschedule at 100 expect(calls).toBe(1); clock = 100; await queue.drain(); // attempt 2 → reschedule at 0+200 clock = 300; await queue.drain(); // attempt 3 → dead-letter expect(calls).toBe(3); expect(failed.length).toBe(1); expect(queue.size()).toBe(0); }); test("repeat re-enqueues a recurring job after each success", async () => { let clock = 0; const queue = createQueue({ now: () => clock }); let runs = 0; queue.process("tick", () => { runs++; }); await queue.add("tick", {}, { repeat: 1000 }); await queue.drain(); // run 1 → re-enqueued at 1000 expect(runs).toBe(1); await queue.drain(); // not due yet expect(runs).toBe(1); clock = 1000; await queue.drain(); // run 2 → re-enqueued at 2000 expect(runs).toBe(2); expect(queue.size()).toBe(1); // always one pending }); test("jobs without a worker stay queued", async () => { const queue = createQueue(); await queue.add("later", {}); await queue.drain(); expect(queue.size()).toBe(1); // no handler yet queue.process("later", () => {}); await queue.drain(); expect(queue.size()).toBe(0); }); test("rejects invalid queue and job timing options", async () => { expect(() => createQueue({ pollMs: 0 })).toThrow("pollMs"); expect(() => createQueue({ maxAttempts: 0 })).toThrow("maxAttempts"); const queue = createQueue(); await expect(queue.add("", {})).rejects.toThrow("name"); await expect(queue.add("job", {}, { delayMs: -1 })).rejects.toThrow("delayMs"); await expect(queue.add("job", {}, { repeat: 0 })).rejects.toThrow("repeat"); }); test("drains independent due jobs concurrently", async () => { const queue = createQueue(); const releases: Array<() => void> = []; queue.process("work", () => new Promise((resolve) => releases.push(resolve))); await queue.add("work", 1); await queue.add("work", 2); const draining = queue.drain(); await Promise.resolve(); expect(releases).toHaveLength(2); 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("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("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); });