Replace pg-boss with owned PostgreSQL queue store

.env.example

@@ -1 +1 @@
-DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test
+DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test

.github/workflows/test.yaml

@@ -33,7 +33,7 @@ jobs:
         image: postgres:16
         env:
           POSTGRES_PASSWORD: postgres
-          POSTGRES_DB: pgboss_queue_test
+          POSTGRES_DB: pgqueue_test
         options: >-
           --health-cmd pg_isready
           --health-interval 10s
@@ -47,7 +47,7 @@ jobs:
       - run: bun install --frozen-lockfile
       - run: bun run test
         env:
-          DATABASE_URL: postgres://postgres:postgres@localhost:5432/pgboss_queue_test
+          DATABASE_URL: postgres://postgres:postgres@localhost:5432/pgqueue_test
 
   node-package:
     runs-on: ubuntu-latest

CLAUDE.md

@@ -4,7 +4,7 @@ Guidance for Claude Code and other agents working in this repository.
 
 ## Project
 
-`pgboss-queue` is a TypeScript background-job library. The public API is the node-resque trio — `Queue`, `Worker`, `Scheduler` — plus `MultiWorker` and `Plugins`. Storage is PostgreSQL via [pg-boss](https://github.com/timgit/pg-boss), not Redis.
+`pg-queue` is a TypeScript background-job library. The public API is the node-resque trio — `Queue`, `Worker`, `Scheduler` — plus `MultiWorker` and `Plugins`. Storage is PostgreSQL through our versioned `pgrq_*` schema, not Redis.
 
 This is an Actionhero project (sibling of [node-resque](https://github.com/actionhero/node-resque) and [keryx](https://github.com/actionhero/keryx)). It exists so Keryx and other apps can keep the resque worker/scheduler pattern without a Redis dependency for jobs.
 
@@ -60,17 +60,17 @@ bun docs:dev                # VitePress (Phase 9)
 Local Postgres: set `DATABASE_URL` (see `.env.example`). CI starts Postgres as a workflow service; there is no `docker-compose.yml`.
 
 ```bash
-# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgboss_queue_test
+# DATABASE_URL=postgres://postgres:postgres@127.0.0.1:5432/pgqueue_test
 ```
 
 Tests create and tear down the configured schema per file (see `specHelper`). Never point tests at a production database.
 
 ## Architecture (do not violate)
 
-1. **node-resque is the runtime model.** Classes, events, plugin hooks, worker names (`hostname:pid[+id]`), queue priority (array order), leader-elected scheduler. Port behavior; do not "simplify" it into pg-boss's `work()` helper.
-2. **pg-boss is the job store.** Jobs live in pg-boss's `job` table. Dequeue with `fetch` + `SKIP LOCKED` (or equivalent SQL), not `LPOP`. Delayed jobs use `startAfter`. Do not reimplement a job table next to pg-boss.
-3. **We own metadata pg-boss does not.** Workers, heartbeats, leader lock, plugin locks, and processed/failed counters live in *our* tables in the same schema (or a documented adjacent schema). See Phase 2.
-4. **The elected scheduler is the only migrator and sweeper.** `automigrate` and completed-job deletion run on the leader, never on every worker. Workers start pg-boss with `migrate: false` and `supervise: false`.
+1. **node-resque is the runtime model.** Classes, events, plugin hooks, worker names (`hostname:pid[+id]`), queue priority (array order), leader-elected scheduler. Port behavior rather than replacing it with a generic handler API.
+2. **We own the job store.** Jobs live in `pgrq_jobs`; queue names live in `pgrq_queues`. Dequeue atomically with `FOR UPDATE SKIP LOCKED`, not `LPOP`. Delayed jobs use `start_after`.
+3. **We own metadata.** Workers, heartbeats, leader lock, plugin locks, and processed/failed counters live in `pgrq_*` tables in the same schema. See Phase 2.
+4. **The elected scheduler is the only migrator and sweeper.** `automigrate` and completed-job deletion run on the leader, never on every worker.
 5. **Keryx PR #519 is research, not a copy target.** Steal: connection strings, schema isolation, SQL introspection on `job`, `short` policy for singleton pending jobs, `deleteAfterSeconds` thinking. Do not steal: dropping worker heartbeats, dropping plugins, replacing `Worker` with a single `boss.work` handler, removing the scheduler.
 
 ## Public API
@@ -111,8 +111,8 @@ type ConnectionOptions = {
   user?: string;
   password?: string;
   ssl?: boolean | object;
-  pool?: import("pg").Pool;           // bring-your-own (maps to pg-boss `db`)
-  schema?: string;                    // default "pgboss_queue" (was Redis namespace)
+  pool?: import("pg").Pool;           // bring-your-own pool
+  schema?: string;                    // default "pgqueue" (was Redis namespace)
 };
 
 type SchedulerOptions = ConnectionOptions & {
@@ -137,7 +137,7 @@ Do **not** accept `pkg: "ioredis"`, `redis: Redis`, or `database: number`. Those
 - **Biome** for format/lint (keryx-style), not Prettier.
 - **Tests use `bun:test`**, not Jest. Port node-resque tests faithfully: same `describe` / `test` names, same assertions, Postgres `specHelper` instead of Redis. Node must still be able to import the compiled package (`node scripts/assert-node-package.mjs`); do not run the Bun suite on Node.
 - **Every behavior change ships with tests.** A PR with no test changes is a red flag unless it is docs-only.
-- **Do not add dependencies** unless a phase plan names them. Expected runtime deps: `pg-boss`, `pg`. Dev: `typescript`, `@types/pg`, `biome`, `bun` types.
+- **Do not add dependencies** unless a phase plan names them. Expected runtime dependency: `pg`. Dev: `typescript`, `@types/pg`, `biome`, `bun` types.
 
 ## Testing rules
 
@@ -183,4 +183,3 @@ Follow keryx: bump `version` in `package.json` on every user-facing PR (patch fo
 - node-resque README + `src/core/*` + `__tests__/*` — source of truth for behavior
 - node-resque examples (`example.ts`, `multiWorker.ts`, `scheduledJobs.ts`, `retry.ts`, `stuckWorker.ts`, `cluster.ts`)
 - [keryx#519](https://github.com/actionhero/keryx/pull/519) — `PgBossBackend.ts`, `TaskBackend.ts`, `config/tasks.ts`
-- pg-boss docs — constructor options (`connectionString`, `schema`, `migrate`, `supervise`, `schedule`), `send` / `fetch` / `complete` / `fail`, `startAfter`, `deleteAfterSeconds`

README.md

@@ -1,11 +1,11 @@
-# pgboss-queue
+# pg-queue
 
 **Background jobs in Node.js, backed by Postgres.**
 
-`pgboss-queue` is a queue-based job system with the same Worker / Scheduler / Queue API as [node-resque](https://github.com/actionhero/node-resque): priority queues, delayed jobs, plugins, locking, failed-job management, and a leader-elected scheduler. Storage is PostgreSQL via [pg-boss](https://github.com/timgit/pg-boss) (`SELECT … FOR UPDATE SKIP LOCKED`), not Redis.
+`pg-queue` is a queue-based job system with the same Worker / Scheduler / Queue API as [node-resque](https://github.com/actionhero/node-resque): priority queues, delayed jobs, plugins, locking, failed-job management, and a leader-elected scheduler. Storage is PostgreSQL (`SELECT … FOR UPDATE SKIP LOCKED`), not Redis.
 
 ```ts
-import { Queue, Worker, Scheduler, Plugins } from "pgboss-queue";
+import { Queue, Worker, Scheduler, Plugins } from "pg-queue";
 
 const connection = {
   connectionString: process.env.DATABASE_URL,
@@ -55,7 +55,7 @@ Pass a Postgres URL (not a Redis URL):
 ```ts
 const connection = {
   connectionString: "postgres://user:pass@host:5432/dbname",
-  schema: "pgboss_queue", // optional; default "pgboss_queue"
+  schema: "pgqueue", // optional; default "pgqueue"
 };
 ```
 
@@ -77,7 +77,7 @@ const pool = new pg.Pool({ connectionString: process.env.DATABASE_URL });
 const connection = { pool };
 ```
 
-`schema` isolates this library's tables (pg-boss's `job` table plus worker/lock/leader metadata) inside one Postgres database. It must be a legal SQL identifier (`letters`, `numbers`, `_`).
+`schema` isolates this library's `pgrq_*` tables inside one Postgres database. It defaults to `pgqueue` and must be a legal SQL identifier (`letters`, `numbers`, `_`) that does not begin with PostgreSQL's reserved `pg_` prefix.
 
 Coming from node-resque: replace `{ host, port, password, database: 0 }` / `{ redis }` / `{ namespace: "resque" }` with `{ connectionString }` or `{ pool }` and `{ schema }`.
 
@@ -142,7 +142,7 @@ Worker names must follow `hostname:pid` or `hostname:pid+unique_id` if you run m
 
 | Option | Default | Meaning |
 | --- | --- | --- |
-| `automigrate` | `true` | Leader applies pg-boss schema migrations and metadata tables. Workers never migrate. |
+| `automigrate` | `true` | Leader applies the bundled versioned SQL migrations. Workers never migrate. |
 | `completeJobRetentionMs` | `24 * 60 * 60 * 1000` | Leader deletes **completed** (and cancelled) jobs older than this. Failed jobs are kept until you retry or remove them. `false` disables the sweeper. `0` deletes completed jobs as soon as the leader sees them. |
 | `stuckWorkerTimeout` | 1 hour | If a worker has not pinged within this window, fail its in-flight job and remove it. Set `false` to disable. |
 | `leaderLockTimeout` | 180 seconds | Leader lock TTL; refreshed while the leader is alive. |
@@ -237,7 +237,7 @@ schedule.scheduleJob("0 * * * * *", async () => {
 Jobs may list plugins that extend `Plugin`. Hooks: `beforeEnqueue`, `afterEnqueue`, `beforePerform`, `afterPerform`. `before*` hooks return `true` to continue or `false` to skip.
 
 ```ts
-import { Plugin } from "pgboss-queue";
+import { Plugin } from "pg-queue";
 
 class MyPlugin extends Plugin {
   async beforeEnqueue() {
@@ -274,7 +274,7 @@ Inspect or delete plugin locks with `queue.locks()` and `queue.delLock(key)`.
 `MultiWorker` wraps `Worker` and scales the number of in-process workers from `minTaskProcessors` to `maxTaskProcessors` based on event-loop delay (more workers for I/O-bound jobs, fewer when the loop is blocked).
 
 ```ts
-import { MultiWorker } from "pgboss-queue";
+import { MultiWorker } from "pg-queue";
 
 const multiWorker = new MultiWorker(
   {
@@ -306,9 +306,9 @@ Raise your Postgres pool `max` when using a large `maxTaskProcessors`. Events ma
 - PostgreSQL 13+ (`SKIP LOCKED`)
 
 ```bash
-npm install pgboss-queue
+npm install pg-queue
 # or
-bun add pgboss-queue
+bun add pg-queue
 ```
 
 ## License

__tests__/core/connection.test.ts

@@ -16,11 +16,17 @@ describe("connection", () => {
     await specHelper.disconnect();
   });
 
+  test("uses PostgreSQL-safe project defaults", () => {
+    const connection = new Connection();
+    expect(connection.schema).toBe("pgqueue");
+    expect(connection.options.application_name).toBe("pg-queue");
+  });
+
   test("should start with no redis keys in the namespace", async () => {
     // Adapt: after cleanup, no job rows and no pgrq_* rows
     const pool = await specHelper.connect();
     const jobCount = await pool.query<{ count: string }>(
-      `SELECT count(*)::text AS count FROM ${specHelper.schema}.job`,
+      `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_jobs`,
     );
     const lockCount = await pool.query<{ count: string }>(
       `SELECT count(*)::text AS count FROM ${specHelper.schema}.pgrq_locks`,
@@ -71,7 +77,7 @@ describe("connection", () => {
     });
 
     test("keys built with a custom namespace are correct", async () => {
-      // Adapt: `schema` option sets pg-boss schema; migrate sees that schema
+      // Adapt: `schema` selects the isolated pg-queue schema
       const customSchema = "custom_namespace_test";
       const custom = new Connection({
         connectionString: process.env.DATABASE_URL,
@@ -137,20 +143,16 @@ describe("connection", () => {
     // Skip: empty schema illegal; we reject
   });
 
-  test("removes the redis event listeners when end", async () => {
-    // Adapt: pool / boss error listeners removed on end()
+  test("removes the postgres event listener when end", async () => {
     const pool = new Pool({ connectionString: process.env.DATABASE_URL });
     const connection = new Connection({
       pool,
       schema: specHelper.schema,
     });
     await connection.connect();
     expect(pool.listenerCount("error")).toBe(1);
-    expect(connection.boss.listenerCount("error")).toBe(1);
-    const boss = connection.boss;
     await connection.end();
     expect(pool.listenerCount("error")).toBe(0);
-    expect(boss.listenerCount("error")).toBe(0);
     await pool.end();
   });
 
@@ -165,6 +167,21 @@ describe("connection", () => {
     await connection.end();
   });
 
+  test("connect is idempotent and supports reconnect after end", async () => {
+    const connection = new Connection(specHelper.cleanConnectionDetails());
+    await Promise.all([connection.connect(), connection.connect()]);
+    expect(connection.connected).toBe(true);
+    expect(connection.pool.listenerCount("error")).toBe(1);
+
+    await connection.end();
+    expect(connection.connected).toBe(false);
+    expect(() => connection.pool).toThrow("Connection is not connected");
+
+    await connection.connect();
+    expect(connection.connected).toBe(true);
+    await connection.end();
+  });
+
   test("connect with discrete host/port/user/password/database", async () => {
     const databaseUrl = process.env.DATABASE_URL;
     expect(databaseUrl).toBeDefined();
@@ -197,13 +214,16 @@ describe("connection", () => {
   });
 
   test("reject illegal schema", () => {
-    expect(() => new Connection({ schema: "pgboss-queue" })).toThrow(
+    expect(() => new Connection({ schema: "pg-queue" })).toThrow(
       /Invalid schema/,
     );
     expect(() => new Connection({ schema: "public; drop" })).toThrow(
       /Invalid schema/,
     );
     expect(() => new Connection({ schema: "" })).toThrow(/Invalid schema/);
+    expect(() => new Connection({ schema: "pg_queue" })).toThrow(
+      /reserves the "pg_" prefix/,
+    );
   });
 
   test("reject Redis options", () => {
@@ -230,7 +250,7 @@ describe("connection", () => {
     ).toThrow(/database/);
   });
 
-  test("migrate() creates pg-boss job table and pgrq_* tables", async () => {
+  test("migrate() creates all versioned pg-queue tables", async () => {
     const freshSchema = "pgrq_migrate_once";
     const pool = await specHelper.connect();
     await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
@@ -249,17 +269,35 @@ describe("connection", () => {
        ORDER BY table_name`,
       [
         freshSchema,
-        ["job", "pgrq_leader", "pgrq_locks", "pgrq_stats", "pgrq_workers"],
+        [
+          "pgrq_jobs",
+          "pgrq_leader",
+          "pgrq_locks",
+          "pgrq_migrations",
+          "pgrq_queues",
+          "pgrq_stats",
+          "pgrq_workers",
+        ],
       ],
     );
     expect(tables.rows.map((row) => row.table_name)).toEqual([
-      "job",
+      "pgrq_jobs",
       "pgrq_leader",
       "pgrq_locks",
+      "pgrq_migrations",
+      "pgrq_queues",
       "pgrq_stats",
       "pgrq_workers",
     ]);
 
+    const migrations = await connection.query<{
+      version: number;
+      name: string;
+    }>(
+      `SELECT version, name FROM ${freshSchema}.pgrq_migrations ORDER BY version`,
+    );
+    expect(migrations.rows).toEqual([{ version: 1, name: "initial" }]);
+
     await connection.end();
     await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
   });
@@ -272,6 +310,92 @@ describe("connection", () => {
     await connection.end();
   });
 
+  test("concurrent migrate() calls serialize safely", async () => {
+    const freshSchema = "pgrq_migrate_concurrent";
+    const pool = await specHelper.connect();
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+    const a = new Connection({
+      connectionString: process.env.DATABASE_URL,
+      schema: freshSchema,
+    });
+    const b = new Connection({
+      connectionString: process.env.DATABASE_URL,
+      schema: freshSchema,
+    });
+
+    await Promise.all([a.migrate(), b.migrate()]);
+    const result = await a.query<{ count: string }>(
+      `SELECT count(*)::text AS count FROM ${freshSchema}.pgrq_migrations`,
+    );
+    expect(Number(result.rows[0]?.count)).toBe(1);
+
+    await Promise.all([a.end(), b.end()]);
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+  });
+
+  test("migrate() can establish its own connection", async () => {
+    const freshSchema = "pgrq_migrate_connect";
+    const pool = await specHelper.connect();
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+    const connection = new Connection({
+      connectionString: process.env.DATABASE_URL,
+      schema: freshSchema,
+    });
+
+    await connection.migrate();
+    expect(connection.connected).toBe(true);
+    expect(
+      (
+        await connection.query<{ version: number }>(
+          `SELECT version FROM ${freshSchema}.pgrq_migrations`,
+        )
+      ).rows,
+    ).toEqual([{ version: 1 }]);
+
+    await connection.end();
+    await pool.query(`DROP SCHEMA IF EXISTS ${freshSchema} CASCADE`);
+  });
+
+  test("fetchJob atomically claims ready jobs and skips delayed jobs", async () => {
+    await specHelper.cleanup();
+    const connection = new Connection(specHelper.cleanConnectionDetails());
+    await connection.connect();
+    await connection.query(
+      `INSERT INTO ${specHelper.schema}.pgrq_queues (name) VALUES ('claims')`,
+    );
+    await connection.query(
+      `INSERT INTO ${specHelper.schema}.pgrq_jobs
+         (name, data, priority, start_after)
+       VALUES
+         ('claims', '{"value":"low"}', 0, now()),
+         ('claims', '{"value":"high"}', 10, now()),
+         ('claims', '{"value":"later"}', 100, now() + interval '1 hour')`,
+    );
+
+    const [first, second] = await Promise.all([
+      connection.fetchJob<{ value: string }>("claims"),
+      connection.fetchJob<{ value: string }>("claims"),
+    ]);
+    expect(new Set([first?.id, second?.id]).size).toBe(2);
+    expect([first?.data.value, second?.data.value].sort()).toEqual([
+      "high",
+      "low",
+    ]);
+    expect(await connection.fetchJob("claims")).toBeNull();
+
+    const states = await connection.query<{ state: string; count: string }>(
+      `SELECT state, count(*)::text AS count
+       FROM ${specHelper.schema}.pgrq_jobs
+       GROUP BY state
+       ORDER BY state`,
+    );
+    expect(states.rows).toEqual([
+      { state: "active", count: "2" },
+      { state: "created", count: "1" },
+    ]);
+    await connection.end();
+  });
+
   test("tryLeader: only one of two connections wins; after expiry the other wins", async () => {
     await specHelper.cleanup();
     const a = new Connection(specHelper.cleanConnectionDetails());