WRNexusJS/packages/queue/test/queue.test.ts

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<void>((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<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);
});