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

import { test, expect } from "bun:test";
import {
  createDurableQueue,
  createQueue,
  cronToInterval,
  defineWorkflow,
  memoryQueueStore,
  defineWorker,
  runWorker,
} from "../src/index.ts";

test("durable queues batch enqueue and worker lifecycle are idempotent", async () => {
  const queue = createDurableQueue();
  const seen: number[] = [];
  const worker = runWorker(
    queue,
    defineWorker<number>({ name: "number", handler: (job) => void seen.push(job.data) }),
  );
  await worker.start();
  await worker.start();
  const jobs = await queue.addBatch("number", [{ data: 1 }, { data: 2 }]);
  expect(jobs).toHaveLength(2);
  await queue.drain();
  expect(seen).toEqual([1, 2]);
  await worker.stop();
  expect(worker.running).toBe(false);
});

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("enforces capacity and can retry dead-lettered work", async () => {
  const queue = createQueue({ capacity: 1, maxAttempts: 1 });
  const job = await queue.add("work", {});
  await expect(queue.add("work", {})).rejects.toThrow("WRN-QUEUE-CAPACITY");
  queue.process("work", () => {
    throw new Error("nope");
  });
  await queue.drain();
  expect(queue.failed().map(({ id }) => id)).toEqual([job.id]);
  expect(await queue.retry(job.id)).toBe(true);
  expect(queue.failed()).toHaveLength(0);
  expect(queue.size()).toBe(1);
});

test("cancels active work with an AbortSignal and force-shuts down", async () => {
  const queue = createQueue();
  let aborted = false;
  queue.process(
    "work",
    (_job, { signal }) =>
      new Promise<void>((resolve) => {
        signal.addEventListener("abort", () => {
          aborted = true;
          resolve();
        });
      }),
  );
  const job = await queue.add("work", {});
  const draining = queue.drain();
  await Promise.resolve();
  expect(queue.cancel(job.id)).toBe(true);
  await draining;
  expect(aborted).toBe(true);
  expect(queue.size()).toBe(0);
  await queue.shutdown({ force: true });
  await expect(queue.add("work", {})).rejects.toThrow("WRN-QUEUE-CLOSED");
});

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

test("durable queue persists retries and dead-letters exhausted jobs", async () => {
  let clock = 0;
  const store = memoryQueueStore();
  const dead: string[] = [];
  const queue = createDurableQueue({
    store,
    maxAttempts: 2,
    now: () => clock,
    backoff: () => 100,
    onDeadLetter: (job) => void dead.push(job.id),
  });
  let calls = 0;
  queue.process("flaky", () => {
    calls++;
    throw new Error("boom");
  });

  const job = await queue.add("flaky", { id: 1 });
  expect(await queue.drain()).toBe(1);
  expect((await store.get(job.id))?.attempts).toBe(1);
  clock = 100;
  expect(await queue.drain()).toBe(1);
  expect(await store.get(job.id)).toBeNull();
  expect(calls).toBe(2);
  expect(dead).toEqual([job.id]);
  expect(queue.failed()).toHaveLength(1);

  expect(await queue.retry(job.id)).toBe(true);
  expect((await store.get(job.id))?.attempts).toBe(0);
});

test("durable queue does not remove a job before its handler succeeds", async () => {
  const store = memoryQueueStore();
  const queue = createDurableQueue({ store });
  let visibleDuringHandler = false;
  queue.process("work", async (job) => {
    visibleDuringHandler = (await store.get(job.id)) !== null;
  });
  const job = await queue.add("work", {});

  await queue.drain();

  expect(visibleDuringHandler).toBe(true);
  expect(await store.get(job.id)).toBeNull();
});

test("durable queue supports capacity, cancellation, and shutdown", async () => {
  const store = memoryQueueStore();
  const queue = createDurableQueue({ store, capacity: 1 });
  const job = await queue.add("work", {});
  await expect(queue.add("work", {})).rejects.toThrow("WRN-QUEUE-CAPACITY");
  expect((await queue.list()).map(({ id }) => id)).toEqual([job.id]);
  expect(await queue.cancel(job.id)).toBe(true);
  expect(await queue.cancel(job.id)).toBe(false);
  await queue.shutdown();
  await expect(queue.add("work", {})).rejects.toThrow("WRN-QUEUE-CLOSED");
});

test("durable queue has no implicit capacity scan or 10k ceiling", async () => {
  const base = memoryQueueStore();
  let sizeCalls = 0;
  const store = {
    ...base,
    async size() {
      sizeCalls++;
      return 10_001;
    },
  };
  const queue = createDurableQueue({ store });
  await expect(queue.add("work", {})).resolves.toMatchObject({ name: "work" });
  expect(sizeCalls).toBe(0);
});

test("force shutdown aborts an active durable job", async () => {
  const store = memoryQueueStore();
  const queue = createDurableQueue({ store });
  let aborted = false;
  queue.process(
    "work",
    (_job, { signal }) =>
      new Promise<void>((resolve) => {
        signal.addEventListener("abort", () => {
          aborted = true;
          resolve();
        });
      }),
  );
  const job = await queue.add("work", {});
  const draining = queue.drain();
  await Promise.resolve();
  await queue.shutdown({ force: true });
  await draining;
  expect(aborted).toBe(true);
  expect(await store.get(job.id)).toBeNull();
});