Implement Phase 3 PostgreSQL queue API

__tests__/utils/specHelper.ts

@@ -120,10 +120,26 @@ export async function dropSchema(): Promise<void> {
 }
 
 /**
- * @throws Always — dequeue lands in Phase 3.
+ * Fetch and remove one ready job from the default test queue.
+ *
+ * @returns The node-resque encoded payload, or `null` when the queue is empty.
  */
-export async function popFromQueue(): Promise<never> {
-  throw new Error("not implemented");
+export async function popFromQueue(): Promise<string | null> {
+  const connection = new Connection(cleanConnectionDetails());
+  await connection.connect();
+  try {
+    const jobs = await connection.boss.fetch<{
+      class: string;
+      queue: string;
+      args: unknown[];
+    }>(queue, { batchSize: 1 });
+    const job = jobs[0];
+    if (!job) return null;
+    await connection.boss.deleteJob(queue, job.id);
+    return JSON.stringify(job.data);
+  } finally {
+    await connection.end();
+  }
 }
 
 const specHelper = {

docs/plans/02-connection-and-schema.md

@@ -36,6 +36,7 @@ export interface ConnectionOptions {
 
 export interface QueueOptions {
   connection?: ConnectionOptions;
+  queue?: string | string[]; // node-resque constructor compatibility
 }
 
 export interface WorkerOptions extends QueueOptions {
@@ -231,3 +232,4 @@ Do not defer these to Phase 8.
 - 2026-08-26: pg-boss is a named ESM export (`import { PgBoss } from "pg-boss"`), not a default export. Migrator instances use `migrate: true` + `supervise: false` + `schedule: false`.
 - 2026-08-26: Version bumped `0.0.1` → `0.1.0` (first user-facing API: `Connection`).
 - 2026-08-26: Node ESM (`"type": "module"`) requires relative import specifiers with `.js` extensions in emitted `dist/` (e.g. `from "./core/connection.js"`). Without them, `node scripts/assert-node-package.mjs` fails with `ERR_MODULE_NOT_FOUND` even though `tsc` and Bun tests pass.
+- 2026-08-29: Phase 3 restored node-resque's optional `QueueOptions.queue` field. Queue methods still take an explicit queue name, but retaining the constructor field lets existing typed call sites migrate without an excess-property error.

docs/plans/03-queue.md

@@ -1,6 +1,6 @@
 # Phase 3 — Queue
 
-**Status:** not-started  
+**Status:** done  
 **Depends on:** Phase 2
 
 ## Goal
@@ -116,12 +116,12 @@ Skip only tests that poke Redis keys directly if any remain inside queue.ts (the
 ## Acceptance criteria
 
 - All Queue methods exist with JSDoc copied/adapted from node-resque
-- `__tests__/core/queue.ts` port is green on CI (worker-status tests that start a Worker wait for Phase 4 — split those into a `describe` marked pending **or** implement after Phase 4; prefer implementing worker methods against empty tables so idle tests pass, and mark `active workingOn` pending)
+- `__tests__/core/queue.test.ts` is green. Worker-status methods are tested with seeded `pgrq_workers` and active pg-boss jobs; Phase 4 will additionally exercise them through a live Worker.
 
 Recommended split:
 
-- Phase 3: enqueue, delayed, delete, failed (inject failed rows via SQL/`fail`), locks, stats, leader (null), idle workers
-- Phase 4: active `workingOn`, `forceCleanWorker` with a live worker
+- Phase 3: enqueue, delayed, delete, failed (inject failed rows via SQL/`fail`), locks, stats, leader, and worker-table behavior
+- Phase 4: repeat active `workingOn` / cleanup behavior end-to-end through a live Worker
 
 ## Next phase needs
 
@@ -130,3 +130,16 @@ Recommended split:
 ## Lessons learned
 
 - 2026-08-26: Bun requires `*.test.ts` filenames for discovery; Phase 3 ports should use `__tests__/core/queue.test.ts` (not bare `queue.ts`) while keeping node-resque describe/test titles.
+- 2026-08-29: pg-boss v12 queues are explicit configuration rows/partitions, so Queue lazily calls `getQueue`/`createQueue` before `send`. Queue defaults are `retryLimit: 0` and `deleteAfterSeconds: 0` (pg-boss defines `0` as never auto-delete); scheduler retention remains authoritative.
+- 2026-08-29: Delayed duplicate identity is reserved in `pgrq_locks` under a private `timestamps:{payload}:delayed:{second}` key until the scheduled second passes. This makes concurrent duplicate checks atomic without modifying pg-boss's partitioned `job` schema; `locks()` intentionally exposes only plugin `lock:*` and `workerslock:*` rows.
+- 2026-08-29: Upstream queue tests schedule at Unix millisecond `10000`, but pg-boss correctly treats 1970 timestamps as immediately runnable. The PostgreSQL port uses future rounded timestamps while retaining the same test titles and timestamp-unit assertions.
+- 2026-08-29: Phase 3 initially completed with 36 Queue tests passing against PostgreSQL. Follow-up review added direct metadata coverage for all Queue worker-status methods; Phase 4 retains responsibility for end-to-end live-Worker coverage.
+- 2026-08-29: `pgrq_stats` stores numeric counters, but `Queue.stats()` stringifies them to retain node-resque's Redis `MGET` response shape (`{ processed: "2", failed: "1" }`).
+- 2026-08-29: Bugbot: an in-process `knownQueues` cache survived `delQueue` on another `Queue` instance, so later `send` skipped `createQueue`. `ensureQueue` now always checks pg-boss and retries once on `Queue does not exist`.
+- 2026-08-29: Bugbot: `delQueue` only skipped `delete_queue` when `active` rows remained, so a concurrent `created` insert could be dropped. It now locks the pg-boss queue row, deletes non-active jobs, and drops the queue only when no rows remain.
+- 2026-08-29: Bugbot: `forceCleanWorker` inserted a second `failed` row and left the original job `active`. It now updates the in-flight job to `failed` (by id when recorded, otherwise by matching `data`) and only inserts if no active row exists.
+- 2026-08-29: Bugbot: the data-only fallback could fail every identical `active` payload. The update now selects a single matching row (`LIMIT 1 … FOR UPDATE`).
+- 2026-08-29: Bugbot: `delayedAt` omitted `start_after > now()`, so a timestamp whose second had arrived still listed jobs that `length`/`queued` already treated as ready. It now uses the same delayed filter as `timestamps` / `scheduledAt` / `delDelayed`.
+- 2026-08-29: Bugbot: delayed duplicate lock keys embedded the encoded JSON and overflowed the `pgrq_locks` btree for large payloads. Keys now use `sha256(encoded)` plus the timestamp second.
+- 2026-08-29: Bugbot: `delQueue` rebuilt those keys from jsonb-loaded args, whose object key order can differ from `JSON.stringify` at enqueue time. The hash now canonicalizes nested object keys so delete and re-enqueue agree.
+- 2026-08-29: Coverage audit found untested `del(count)`, `delByFunction(start, stop)`, expired-lock cleanup, concurrent delayed enqueue, queue-row serialization, `cleanOldWorkers`, `retryStuckJobs`, active `workingOn`, and unknown-worker errors. These now have focused PostgreSQL tests rather than being deferred wholesale to Phase 4.

docs/plans/08-conformance-tests.md

@@ -17,8 +17,8 @@ Source of truth: [actionhero/node-resque](https://github.com/actionhero/node-res
 | --- | --- |
 | Phase 1 | `specHelper` skeleton, smoke `SELECT 1`, `test.yaml` (lint / build / Postgres / complete) |
 | Phase 2 | `__tests__/core/connection.test.ts` + `connectionError.test.ts` (+ illegal schema, BYO pool, migrate, locks, leader) |
-| Phase 3 | `__tests__/core/queue.test.ts` (minus live-worker slices deferred to 4) |
-| Phase 4 | `__tests__/core/worker.test.ts`, remaining queue worker-status, multi-process + priority extras |
+| Phase 3 | `__tests__/core/queue.test.ts` (worker-table behavior seeded directly) |
+| Phase 4 | `__tests__/core/worker.test.ts`, live-Worker queue-status integration, multi-process + priority extras |
 | Phase 5 | `__tests__/core/scheduler.test.ts`, automigrate + sweeper extras |
 | Phase 6 | `__tests__/plugins/*.test.ts` |
 | Phase 7 | `__tests__/core/multiWorker.test.ts` |
@@ -151,6 +151,7 @@ If an assertion cannot be identical, add a row (may already have rows from earli
 | --- | --- | --- | --- |
 | keys built with a custom namespace | `connection.key("thing") === "customNamespace:thing"` | `connection.schema === customSchema` and `pgrq_locks` exists in that schema | Keys are not Redis-prefixed; schema replaces namespace |
 | removes the redis event listeners when end | `redis.listenerCount("error"|"end")` | `pool`/`boss` `listenerCount("error")` with BYO pool | No Redis `end` event; we forward `error` only |
+| queue delayed-job tests using timestamp `10000` | Redis keeps the 1970 timestamp in a delayed list until Scheduler transfers it | Use a future rounded timestamp and assert the same seconds/ms conversions | pg-boss `startAfter` is eligibility time, so a past timestamp is immediately ready by design |
 
 PRs that add rows must explain. "Postgres is different" is not enough if the Queue API can still match.
 
@@ -176,3 +177,5 @@ Docs site can describe a real API. Phase 10 can trust tests that have been runni
 - 2026-08-26: Phase 1 corrected the runner to `node:test` on a Bun + Node matrix. Isolation uses `--max-concurrency=1` / `--test-concurrency=1`, not `bun test --concurrency=1`.
 - 2026-08-26: Phase 1 reverted the suite to `bun:test`. Node is covered by importing `dist/` (`test:node-package`), not by running this matrix on `node --test`.
 - 2026-08-26: Phase 2 — Bun requires `.test.ts` (or `.spec` / `_test_` / `_spec_`) in the filename. Matrix paths are `__tests__/core/<name>.test.ts` while describe/test titles stay node-resque-identical. Later phases must not copy bare `connection.ts`-style names or CI will skip them.
+- 2026-08-29: Phase 3 preserves upstream Queue test titles but replaces hard-coded 1970 delayed timestamps with future rounded values. This is a required semantic adaptation because pg-boss uses `startAfter` directly rather than waiting for a scheduler to move a Redis-list item.
+- 2026-08-29: Phase 3's Queue suite now covers active/old worker metadata, force-clean, and retry-stuck behavior through seeded pg-boss and `pgrq_workers` rows. Phase 4 still repeats these paths with a live Worker but no Queue titles remain skipped.

package.json

@@ -1,6 +1,6 @@
 {
   "name": "pgboss-queue",
-  "version": "0.1.0",
+  "version": "0.2.0",
   "description": "A PostgreSQL-backed background job queue with the node-resque runtime model",
   "type": "module",
   "main": "./dist/index.js",

src/core/connection.ts

@@ -59,6 +59,8 @@ export interface ConnectionOptions {
  */
 export interface QueueOptions {
   connection?: ConnectionOptions;
+  /** Optional default queue retained for node-resque constructor compatibility. */
+  queue?: string | string[];
 }
 
 /**

src/core/plugin.ts

@@ -0,0 +1,69 @@
+import type { JobDefinition, PluginHost } from "../types/job.js";
+import type { Queue } from "./queue.js";
+
+/**
+ * Base class for node-resque-compatible job plugins.
+ *
+ * Subclasses may implement any enqueue/perform hook. Returning `false` from a
+ * before hook prevents the operation.
+ */
+export abstract class Plugin {
+  /** Plugin name used to look up `pluginOptions`. */
+  readonly name: string;
+  /** Queue or Worker instance invoking this plugin. */
+  readonly worker: PluginHost;
+  /** Queue instance associated with the host, when available. */
+  readonly queueObject: Queue | undefined;
+  /** Queue name for this operation. */
+  readonly queue: string;
+  /** Registered job name. */
+  readonly func: string;
+  /** Registered job definition. */
+  readonly job: JobDefinition;
+  /** Arguments encoded for the job. */
+  readonly args: unknown[];
+  /** Options selected by plugin name. */
+  readonly options: Record<string, unknown>;
+
+  /**
+   * @param worker - Queue or Worker invoking the hook.
+   * @param func - Registered job name.
+   * @param queue - Queue name for this operation.
+   * @param job - Registered job definition.
+   * @param args - Job arguments.
+   * @param options - Plugin-specific options.
+   */
+  constructor(
+    worker: PluginHost,
+    func: string,
+    queue: string,
+    job: JobDefinition,
+    args: unknown[],
+    options: Record<string, unknown>,
+  ) {
+    this.name = this.constructor.name || "Node Resque Plugin";
+    this.worker = worker;
+    this.func = func;
+    this.queue = queue;
+    this.job = job;
+    this.args = args;
+    this.options = options;
+
+    const host = worker as PluginHost & { queueObject?: Queue };
+    this.queueObject =
+      host.queueObject ?? (isQueue(worker) ? worker : undefined);
+  }
+
+  /** Run before enqueue. Return `false` to suppress enqueueing. */
+  beforeEnqueue?(): boolean | Promise<boolean>;
+  /** Run after enqueue. */
+  afterEnqueue?(): boolean | Promise<boolean>;
+  /** Run before performing. Return `false` to suppress execution. */
+  beforePerform?(): boolean | Promise<boolean>;
+  /** Run after performing. */
+  afterPerform?(): boolean | Promise<boolean>;
+}
+
+function isQueue(host: PluginHost): host is Queue {
+  return "enqueue" in host;
+}

src/core/pluginRunner.ts

@@ -0,0 +1,122 @@
+import {
+  type JobDefinition,
+  type Jobs,
+  jobDefinition,
+  type PluginConstructor,
+  type PluginHost,
+} from "../types/job.js";
+import type { Plugin } from "./plugin.js";
+
+/** Hook names understood by the plugin runner. */
+export type PluginHook =
+  | "beforeEnqueue"
+  | "afterEnqueue"
+  | "beforePerform"
+  | "afterPerform";
+
+/**
+ * Run all plugins for one job in declaration order.
+ *
+ * @param self - Queue or Worker invoking the plugins.
+ * @param type - Hook to invoke.
+ * @param func - Registered job name.
+ * @param queue - Queue name.
+ * @param job - Job declaration, including function-form jobs.
+ * @param args - Job arguments.
+ * @returns `false` when a plugin vetoes the operation, otherwise `true`.
+ */
+export async function runPlugins(
+  self: PluginHost,
+  type: PluginHook,
+  func: string,
+  queue: string,
+  job: Jobs[string] | undefined,
+  args: unknown[],
+): Promise<boolean> {
+  const definition = jobDefinition(job);
+  if (!definition?.plugins?.length) return true;
+
+  for (const reference of definition.plugins) {
+    const result = await runPlugin(
+      self,
+      reference,
+      type,
+      func,
+      queue,
+      definition,
+      args,
+    );
+    if (result === false) return false;
+  }
+
+  return true;
+}
+
+/**
+ * Construct and run one plugin hook.
+ *
+ * @param self - Queue or Worker invoking the plugin.
+ * @param reference - Plugin constructor or built-in plugin name.
+ * @param type - Hook to invoke.
+ * @param func - Registered job name.
+ * @param queue - Queue name.
+ * @param job - Normalized job definition.
+ * @param args - Job arguments.
+ * @returns Hook result, defaulting to `true` when the hook is absent.
+ * @throws If a named plugin module does not export the requested plugin.
+ */
+export async function runPlugin(
+  self: PluginHost,
+  reference: string | PluginConstructor,
+  type: PluginHook,
+  func: string,
+  queue: string,
+  job: JobDefinition,
+  args: unknown[],
+): Promise<boolean> {
+  const Constructor =
+    typeof reference === "string"
+      ? await loadNamedPlugin(reference)
+      : reference;
+
+  const name = Constructor.name || "Node Resque Plugin";
+  const options = job.pluginOptions?.[name] ?? {};
+  const plugin = new Constructor(self, func, queue, job, args, options);
+  const hook = plugin[type];
+  if (typeof hook !== "function") return true;
+
+  return (await hook.call(plugin)) !== false;
+}
+
+async function loadNamedPlugin(name: string): Promise<PluginConstructor> {
+  const module: unknown = await import(`../plugins/${name}.js`);
+  if (!isModuleRecord(module)) {
+    throw new Error(`Plugin module "${name}" is invalid`);
+  }
+
+  const Constructor = module[name];
+  if (typeof Constructor !== "function") {
+    throw new Error(`Plugin "${name}" is not exported`);
+  }
+
+  return Constructor as PluginConstructor;
+}
+
+function isModuleRecord(value: unknown): value is Record<string, unknown> {
+  return typeof value === "object" && value !== null;
+}
+
+/** @deprecated Use {@link runPlugins}. Kept for node-resque source compatibility. */
+export const RunPlugins = runPlugins;
+
+/** @deprecated Use {@link runPlugin}. Kept for node-resque source compatibility. */
+export const RunPlugin = runPlugin;
+
+// Assert the indexed hook surface remains compatible with Plugin.
+const _pluginTypeCheck: PluginHook[] = [
+  "beforeEnqueue",
+  "afterEnqueue",
+  "beforePerform",
+  "afterPerform",
+];
+void (_pluginTypeCheck satisfies Array<keyof Plugin>);