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

import { afterEach, expect, test } from "bun:test";
import { createDurableQueue, createQueue, subjectQueue } from "../src/index.ts";

const secret = process.env.WRNEXUS_RPC_SECRET;
const app = process.env.WRNEXUS_APP_NAME;
afterEach(() => {
  if (secret === undefined) delete process.env.WRNEXUS_RPC_SECRET;
  else process.env.WRNEXUS_RPC_SECRET = secret;
  if (app === undefined) delete process.env.WRNEXUS_APP_NAME;
  else process.env.WRNEXUS_APP_NAME = app;
});

test("subjectQueue supplies a verified subject to the worker", async () => {
  process.env.WRNEXUS_RPC_SECRET = "test-rpc-secret-at-least-32-chars-long";
  process.env.WRNEXUS_APP_NAME = "orders";
  const queue = createQueue();
  const subjectAware = subjectQueue(queue);
  let seen: string | undefined;
  subjectAware.process<{ orderId: string }>("email", (job) => {
    expect(job.data.orderId).toBe("o1");
    seen = job.subject?.subjectId;
  });
  await subjectAware.add({ user: { id: "u1" }, locals: {} } as never, "email", { orderId: "o1" });
  await queue.drain();
  expect(seen).toBe("u1");
});

test("subjectQueue preserves identity when a durable queue persists the job", async () => {
  process.env.WRNEXUS_RPC_SECRET = "test-rpc-secret-at-least-32-chars-long";
  process.env.WRNEXUS_APP_NAME = "orders";
  const queue = createDurableQueue();
  const subjectAware = subjectQueue(queue);
  let seen: string | undefined;
  subjectAware.process<{ orderId: string }>("email", (job) => {
    seen = job.subject?.subjectId;
  });
  await subjectAware.add({ user: { id: "u1" }, locals: {} } as never, "email", { orderId: "o1" });
  await queue.drain();
  expect(seen).toBe("u1");
});