test: install the pg-boss schema the HTTP suite reads

test/integration/http-transport.integration.test.ts

@@ -8,6 +8,7 @@ import { createReadOnlyDb } from '../../src/db/pool.js';
 import type { ReadOnlyDb } from '../../src/db/readonly.js';
 import { collectSnapshot, diagnose } from '../../src/diagnose/engine.js';
 import { PgBossAdapter } from '../../src/pgboss/adapter.js';
+import { installPgBossSchema } from './pgboss-fixture.js';
 
 /**
  * The HTTP transport against a real Postgres, through a stand-in for the
@@ -28,6 +29,13 @@ import { PgBossAdapter } from '../../src/pgboss/adapter.js';
  */
 const DATABASE_URL = process.env['QUEUE_DOCTOR_TEST_DATABASE_URL'];
 const TOKEN = 'x'.repeat(64);
+/**
+ * The suite installs the schema it reads. Pointing the adapter at a default
+ * `pgboss` this file never created passed only where an earlier run had left
+ * one behind — green on a long-lived developer database, red on CI's fresh one
+ * from the day it landed.
+ */
+const SCHEMA = 'qd_it_http';
 
 const describeIfConfigured = DATABASE_URL === undefined ? describe.skip : describe;
 
@@ -80,6 +88,8 @@ describeIfConfigured('HTTP SQL transport (integration)', () => {
   const seen: Array<{ query: string; params: unknown }> = [];
 
   beforeAll(async () => {
+    await installPgBossSchema({ alias: 'pgboss12', schema: SCHEMA, databaseUrl: DATABASE_URL! });
+
     const { Pool } = (await import('pg')).default;
     const pool = new Pool({ connectionString: DATABASE_URL, max: 4 });
 
@@ -157,15 +167,15 @@ describeIfConfigured('HTTP SQL transport (integration)', () => {
         QUEUE_DOCTOR_LOG_LEVEL: 'silent',
       }),
     );
-  }, 60_000);
+  }, 180_000);
 
   afterAll(async () => {
     await db.close();
     await new Promise<void>((resolve) => server.close(() => resolve()));
   });
 
   it('runs a full diagnose through the endpoint', async () => {
-    const adapter = new PgBossAdapter(db, 'pgboss');
+    const adapter = new PgBossAdapter(db, SCHEMA);
 
     const snapshot = await collectSnapshot(adapter, {
       windowSeconds: 86_400,
@@ -174,6 +184,9 @@ describeIfConfigured('HTTP SQL transport (integration)', () => {
     const { findings } = diagnose(snapshot);
 
     expect(snapshot.probe.backend).toBe('pg-boss');
+    // A schema the transport could not read would still satisfy the
+    // assertions below, so state plainly that it read a real one.
+    expect(snapshot.probe.supported).toBe(true);
     expect(Array.isArray(findings)).toBe(true);
     // The point of the suite: every query the engine needs, over HTTP, none
     // refused by the endpoint's guards.

test/integration/pgboss-fixture.ts

@@ -0,0 +1,87 @@
+/* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-explicit-any -- pg-boss is loaded through a computed specifier so several major versions can be installed side by side; TypeScript cannot type a dynamic import. */
+
+/**
+ * Shared setup for the suites that need a real pg-boss schema.
+ *
+ * Every integration suite must build the schema it reads. A suite that points
+ * an adapter at a schema it did not install passes only where some earlier run
+ * happened to leave one behind — which is exactly how the HTTP-transport suite
+ * came to be green on a developer's long-lived database and red on CI's fresh
+ * one from the day it landed.
+ */
+
+interface BossLike {
+  start(): Promise<unknown>;
+  stop(options?: { graceful?: boolean }): Promise<unknown>;
+  on(event: string, listener: (err: unknown) => void): void;
+  createQueue?(name: string): Promise<unknown>;
+  schedule?(name: string, cron: string, data?: unknown, options?: unknown): Promise<unknown>;
+  send?(name: string, data?: unknown): Promise<unknown>;
+}
+
+/** pg-boss moved from a default export to a named one; accept either. */
+function resolveConstructor(mod: any): new (options: Record<string, unknown>) => BossLike {
+  const candidate = typeof mod === 'function' ? mod : (mod.PgBoss ?? mod.default);
+  if (typeof candidate !== 'function') {
+    throw new Error('Could not find a pg-boss constructor on the imported module');
+  }
+  return candidate as new (options: Record<string, unknown>) => BossLike;
+}
+
+export interface InstallOptions {
+  /** npm alias the pg-boss version is installed under (e.g. `pgboss12`). */
+  readonly alias: string;
+  /** Dedicated schema, so every version can coexist in one database. */
+  readonly schema: string;
+  readonly databaseUrl: string;
+}
+
+/**
+ * Install pg-boss into its own schema and seed one queue, job and schedule.
+ *
+ * The schema is dropped first. CI gets a fresh Postgres per run, but a
+ * developer's container is long-lived, so without this the fixtures stack up
+ * and a second run reads a database the first one shaped.
+ *
+ * Returns the pg-boss version that was installed, which the version suite
+ * quotes when a schema it has not been verified against turns up.
+ */
+export async function installPgBossSchema(options: InstallOptions): Promise<string> {
+  const { Pool } = (await import('pg')).default;
+  const setup = new Pool({ connectionString: options.databaseUrl });
+  try {
+    await setup.query(`drop schema if exists ${options.schema} cascade`);
+  } finally {
+    await setup.end();
+  }
+
+  // A computed specifier keeps TypeScript from trying to resolve the alias.
+  const specifier: string = options.alias;
+  const mod = await import(specifier);
+  const PgBoss = resolveConstructor(mod);
+  const pkg = await import(`${specifier}/package.json`, { with: { type: 'json' } });
+  const version: string = pkg.default.version;
+
+  const boss = new PgBoss({
+    connectionString: options.databaseUrl,
+    schema: options.schema,
+    // Migrations only; nothing here needs a running scheduler or supervisor.
+    schedule: false,
+    supervise: false,
+  });
+  boss.on('error', () => {
+    /* migration-time noise is not the subject of these tests */
+  });
+  await boss.start();
+  if (typeof boss.createQueue === 'function') {
+    await boss.createQueue('it/alpha');
+  }
+  if (typeof boss.send === 'function') {
+    await boss.send('it/alpha', { hello: 'world' });
+  }
+  if (typeof boss.schedule === 'function') {
+    await boss.schedule('it/alpha', '0 3 * * *', {}, { tz: 'UTC' });
+  }
+  await boss.stop({ graceful: false });
+  return version;
+}

test/integration/pgboss-versions.integration.test.ts

@@ -1,4 +1,3 @@
-/* eslint-disable @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-explicit-any -- pg-boss is loaded through a computed specifier so several major versions can be installed side by side; TypeScript cannot type a dynamic import. */
 import { afterAll, beforeAll, describe, expect, it } from 'vitest';
 
 import { PgBossAdapter } from '../../src/pgboss/adapter.js';
@@ -7,6 +6,7 @@ import { createReadOnlyDb } from '../../src/db/pool.js';
 import type { ReadOnlyDb } from '../../src/db/readonly.js';
 import { DEFAULT_THRESHOLDS } from '../../src/diagnose/types.js';
 import { collectSnapshot, diagnose } from '../../src/diagnose/engine.js';
+import { installPgBossSchema } from './pgboss-fixture.js';
 
 /**
  * The claim these tests exist to keep honest.
@@ -23,15 +23,6 @@ import { collectSnapshot, diagnose } from '../../src/diagnose/engine.js';
  */
 const DATABASE_URL = process.env['QUEUE_DOCTOR_TEST_DATABASE_URL'];
 
-interface BossLike {
-  start(): Promise<unknown>;
-  stop(options?: { graceful?: boolean }): Promise<unknown>;
-  on(event: string, listener: (err: unknown) => void): void;
-  createQueue?(name: string): Promise<unknown>;
-  schedule?(name: string, cron: string, data?: unknown, options?: unknown): Promise<unknown>;
-  send?(name: string, data?: unknown): Promise<unknown>;
-}
-
 interface VersionUnderTest {
   /** npm alias the version is installed under. */
   readonly alias: string;
@@ -48,47 +39,6 @@ const VERSIONS: readonly VersionUnderTest[] = [
   { alias: 'pgboss12', schema: 'qd_it_v12', expectedDialect: 'pgboss-v11plus', expectedSupported: true },
 ];
 
-/** pg-boss moved from a default export to a named one; accept either. */
-function resolveConstructor(mod: any): new (options: Record<string, unknown>) => BossLike {
-  const candidate = typeof mod === 'function' ? mod : (mod.PgBoss ?? mod.default);
-  if (typeof candidate !== 'function') {
-    throw new Error('Could not find a pg-boss constructor on the imported module');
-  }
-  return candidate as new (options: Record<string, unknown>) => BossLike;
-}
-
-async function installSchema(alias: string, schema: string): Promise<string> {
-  // A computed specifier keeps TypeScript from trying to resolve the alias.
-  const specifier: string = alias;
-  const mod = await import(specifier);
-  const PgBoss = resolveConstructor(mod);
-  const pkg = await import(`${specifier}/package.json`, { with: { type: 'json' } });
-  const version: string = pkg.default.version;
-
-  const boss = new PgBoss({
-    connectionString: DATABASE_URL,
-    schema,
-    // Migrations only; nothing here needs a running scheduler or supervisor.
-    schedule: false,
-    supervise: false,
-  });
-  boss.on('error', () => {
-    /* migration-time noise is not the subject of these tests */
-  });
-  await boss.start();
-  if (typeof boss.createQueue === 'function') {
-    await boss.createQueue('it/alpha');
-  }
-  if (typeof boss.send === 'function') {
-    await boss.send('it/alpha', { hello: 'world' });
-  }
-  if (typeof boss.schedule === 'function') {
-    await boss.schedule('it/alpha', '0 3 * * *', {}, { tz: 'UTC' });
-  }
-  await boss.stop({ graceful: false });
-  return version;
-}
-
 const describeIfConfigured = DATABASE_URL === undefined ? describe.skip : describe;
 
 describeIfConfigured('pg-boss versions (integration)', () => {
@@ -97,7 +47,14 @@ describeIfConfigured('pg-boss versions (integration)', () => {
 
   beforeAll(async () => {
     for (const version of VERSIONS) {
-      installed.set(version.alias, await installSchema(version.alias, version.schema));
+      installed.set(
+        version.alias,
+        await installPgBossSchema({
+          alias: version.alias,
+          schema: version.schema,
+          databaseUrl: DATABASE_URL!,
+        }),
+      );
     }
   }, 180_000);