`apps/workers/src/queue-resilience.itest.ts`

79 lines
import { afterAll, beforeAll, describe, expect, it } from "vitest";
import {
  getQueue,
  enqueue,
  stopQueue,
  computeQueueHealth,
  getQueueHealthState,
  recordQueueHealthy,
} from "@homesonspec/shared";

/**
 * Integration: exercises the TK-10240 pg-boss resilience facade against a REAL
 * (test) Postgres — the "test vs the live crawler" coverage the mocked unit
 * suite in packages/shared/queue.test.ts cannot provide. Proves against real
 * pg-boss + Postgres:
 *   (1) enqueue() writes a real, fetchable job through Postgres,
 *   (2) getQueue() attaches an 'error' listener so a boss error is NON-FATAL
 *       (the ride-out that replaced process.exit(1)) and flips the health probe,
 *       and a canary success recovers it,
 *   (3) stopQueue() tears down cleanly and getQueue() yields a fresh instance
 *       with the ride-out handler re-attached.
 *
 * Uses fetch()/complete() rather than a persistent work() subscription so no
 * poller outlives the test (a lingering subscription throws "Workers are
 * disabled" after stopQueue). pg-boss creates its own `pgboss` schema on
 * start(); safe because the integration setup pins DATABASE_URL to a throwaway
 * *_test database.
 */

const JOB = "resilience-itest-job";

beforeAll(() => {
  // Guard: never touch a non-test database.
  expect(process.env.DATABASE_URL).toMatch(/homesonspec_test/);
});

afterAll(async () => {
  await stopQueue();
});

describe("pg-boss resilience facade (real Postgres)", () => {
  it("enqueue writes a real, fetchable job through Postgres", async () => {
    const boss = await getQueue();
    await boss.createQueue(JOB);
    const id = await enqueue(JOB, { n: 42 });
    expect(id).toBeTruthy();

    const jobs = await boss.fetch<{ n: number }>(JOB);
    expect(jobs[0]?.data).toEqual({ n: 42 });
    if (jobs[0]) await boss.complete(JOB, jobs[0].id);
  }, 20_000);

  it("a boss 'error' is non-fatal (ride-out) and flips health; a canary success recovers it", async () => {
    const boss = await getQueue();
    // The whole point of TK-10240: getQueue() registers an 'error' listener so a
    // transient pg-boss/pg-pool error is recorded + survived instead of bubbling
    // to an uncaught 'error' event (which would crash the process). Node only
    // throws for an 'error' emit with ZERO listeners — so this emit NOT throwing
    // is itself the proof the ride-out handler is wired.
    expect(boss.listenerCount("error")).toBeGreaterThanOrEqual(1);
    boss.emit("error", new Error("synthetic transient blip"));

    expect(computeQueueHealth(getQueueHealthState(), Date.now()).healthy).toBe(false);

    // A real canary roundtrip (what startQueueHealthCanary does) proves recovery.
    await boss.getQueueSize(JOB);
    recordQueueHealthy();
    expect(computeQueueHealth(getQueueHealthState(), Date.now()).healthy).toBe(true);
  }, 20_000);

  it("stopQueue tears down cleanly; getQueue yields a fresh instance with the handler re-attached", async () => {
    const before = await getQueue();
    await stopQueue();
    const after = await getQueue();
    expect(after).not.toBe(before);
    expect(after.listenerCount("error")).toBeGreaterThanOrEqual(1);
  }, 20_000);
});