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

import { expect, test } from "bun:test";
import { createDurableQueue, defineQueue, memoryQueueStore } from "../src/index.ts";

test("defineQueue creates typed producers and registers workers", async () => {
  const seen: number[] = [];
  const completed: number[] = [];
  const email = defineQueue({
    name: "email",
    options: { store: memoryQueueStore() },
    jobs: {
      send: {
        maxAttempts: 4,
        idempotency: (data: { messageId: number }) => `message:${data.messageId}`,
        validate: (data: unknown): data is { messageId: number } =>
          typeof data === "object" &&
          data !== null &&
          Number.isInteger((data as { messageId?: unknown }).messageId),
        run: async ({ messageId }: { messageId: number }) => void seen.push(messageId),
        success: async ({ messageId }: { messageId: number }) => void completed.push(messageId),
      },
    },
  });

  const first = await email.send.add({ messageId: 7 });
  const duplicate = await email.jobs.send.add({ messageId: 7 });
  expect(duplicate.id).toBe(first.id);
  expect(await email.send.status(first.id)).toBe("queued");
  expect(await email.drain()).toBe(1);
  expect(seen).toEqual([7]);
  expect(completed).toEqual([7]);
  expect(await email.send.status(first.id)).toBe("completed");
});

test("durable queues expose lifecycle, health and events", async () => {
  const events: string[] = [];
  const queue = createDurableQueue({
    pollMs: 10,
    onEvent: (event) => void events.push(event.type),
  });
  queue.process("work", async () => {});
  queue.start();
  expect(queue.isRunning()).toBe(true);
  await queue.add("work", {});
  await queue.drain();
  const health = await queue.health();
  expect(health.pending).toBe(0);
  expect(events).toContain("job.added");
  expect(events).toContain("job.completed");
  queue.stop();
  expect(queue.isRunning()).toBe(false);
  await queue.shutdown();
});

test("defineQueue validates payloads before persistence", async () => {
  const queue = defineQueue({
    name: "safe",
    jobs: {
      number: {
        validate: (data: unknown): data is number => typeof data === "number",
        run: async (_data: number) => {},
      },
    },
  });
  await expect(queue.number.add("no" as never)).rejects.toThrow("WRN-QUEUE-PAYLOAD");
});

// `shutdown()` used to close a queue permanently: it set `accepting = false`
// and `start()` refused for ever after. That made a queue single-use, which
// breaks any process that boots more than one app -- a test suite closing one
// harness and opening the next, a hot reload, a multi-tenant host. The failure
// was remote from its cause: the SECOND app to boot threw WRN-QUEUE-CLOSED out
// of the dev server, so a perfectly good app failed because an unrelated one
// had shut down earlier in the same process.
test("a queue that was shut down can be started again", async () => {
  const { createDurableQueue, memoryQueueStore } = await import("../src/index.ts");
  const queue = createDurableQueue({ pollMs: 5, store: memoryQueueStore() });

  queue.start();
  await queue.shutdown();
  expect(queue.health ? (await queue.health()).accepting : false).toBe(false);

  // The explicit intent to run again reopens it.
  queue.start();
  expect((await queue.health()).accepting).toBe(true);

  // And it genuinely works, rather than merely reporting that it does.
  const done: string[] = [];
  queue.process("work", async (job: { data: { id: string } }) => void done.push(job.data.id));
  await queue.add("work", { id: "after-restart" });
  await queue.drain();
  expect(done).toEqual(["after-restart"]);

  await queue.shutdown();
});