fix(selfhost): make retention pruning work on postgres

src/selfhost/pg-dialect.ts

@@ -1,9 +1,9 @@
 // SQLite → Postgres SQL dialect translation for the self-host Postgres backend (#977). gittensory's core and
 // drizzle-orm/d1 emit SQLite-dialect SQL; this translates the bounded set of SQLite-isms the codebase uses
-// (placeholders + a handful of scalar functions + INSERT OR REPLACE/IGNORE) so the SAME queries run on
-// Postgres. The timestamp columns are TEXT (ISO strings written by the app), so the datetime/CURRENT_TIMESTAMP
-// translations return TEXT in SQLite's format to preserve the existing text-comparison semantics. Validated
-// against a real Postgres (all 56 migrations + the runtime query paths).
+// (placeholders + a handful of scalar functions + INSERT OR REPLACE/IGNORE + the rowid pseudo-column) so
+// the SAME queries run on Postgres. The timestamp columns are TEXT (ISO strings written by the app), so the
+// datetime/CURRENT_TIMESTAMP translations return TEXT in SQLite's format to preserve the existing
+// text-comparison semantics. Validated against a real Postgres (all 56 migrations + the runtime query paths).
 
 // INSERT OR REPLACE needs an explicit conflict target on Postgres; map the (few) tables that use it to their PK.
 const REPLACE_CONFLICT_KEYS: Record<string, string[]> = {
@@ -14,14 +14,29 @@ const REPLACE_CONFLICT_KEYS: Record<string, string[]> = {
   orb_signals: ["instance_id", "repo_hash", "pr_hash"],
 };
 
-/** Replace `?` placeholders with `$1,$2,…`, skipping any `?` inside single-quoted string literals. */
+/** Replace `?` placeholders with `$1,$2,…`, skipping any `?` inside single-quoted string literals. A `?`
+ *  immediately followed by digits is SQLite's *numbered* placeholder (`?1`, `?2`, …, e.g. retention.ts's
+ *  `retentionWhere()` and repositories.ts's `claimRegateFanoutSlot()`) — its index is reused verbatim as
+ *  `$1`/`$2` rather than folded into the anonymous-placeholder counter below, otherwise `?1` corrupts to
+ *  `$1` + a literal trailing `1` (i.e. `$11`), which Postgres reads as bind parameter 11. Per SQLite's own
+ *  rule, a later anonymous `?` gets "one greater than the largest parameter number already assigned", so a
+ *  numbered placeholder also raises the anonymous counter's floor — otherwise a later `?` could collide with
+ *  an earlier `?N` (e.g. `?1` then `?` must yield `$1`, `$2`, not `$1`, `$1`). */
 export function toNumberedPlaceholders(sql: string): string {
   let out = "";
   let n = 0;
   let inString = false;
-  for (const ch of sql) {
+  for (let i = 0; i < sql.length; i++) {
+    const ch = sql[i] as string;
     if (ch === "'") inString = !inString;
     if (ch === "?" && !inString) {
+      const numbered = /^\d+/.exec(sql.slice(i + 1))?.[0];
+      if (numbered) {
+        n = Math.max(n, Number(numbered));
+        out += `$${numbered}`;
+        i += numbered.length;
+        continue;
+      }
       n += 1;
       out += `$${n}`;
     } else {
@@ -50,6 +65,18 @@ export function translateFunctions(sql: string): string {
   );
 }
 
+/** Translate SQLite's `rowid` pseudo-column to Postgres's `ctid` system column. Both give a stable,
+ *  per-row identifier for the lifetime of a single statement/snapshot — exactly how the codebase uses
+ *  it: `DELETE ... WHERE rowid IN (SELECT rowid FROM t WHERE ... LIMIT n)` for bounded batched pruning
+ *  (retention.ts) and `ORDER BY rowid` for insertion-order tie-breaking (orb/relay.ts, tests). `ctid` is
+ *  a *physical* row location that can change across `VACUUM FULL` / row rewrites, so this is only safe
+ *  for the codebase's existing usage — internal bookkeeping resolved within one statement — never for
+ *  durable application-facing row identity. Fixes the self-host Postgres dead-letter where the raw
+ *  `rowid` reached Postgres verbatim ("column \"rowid\" does not exist"). */
+export function translateRowid(sql: string): string {
+  return sql.replace(/\browid\b/gi, "ctid");
+}
+
 /** Translate INSERT OR REPLACE / INSERT OR IGNORE to Postgres ON CONFLICT. */
 export function translateInsertOr(sql: string): string {
   if (/^\s*INSERT\s+OR\s+IGNORE\s+INTO/i.test(sql)) {
@@ -87,7 +114,7 @@ export function stripConflictTargetQualifiers(sql: string): string {
 
 /** Translate a runtime query (SQLite → Postgres). */
 export function translateSql(sql: string): string {
-  return toNumberedPlaceholders(stripConflictTargetQualifiers(translateFunctions(translateInsertOr(sql))));
+  return toNumberedPlaceholders(stripConflictTargetQualifiers(translateRowid(translateFunctions(translateInsertOr(sql)))));
 }
 
 /** Migrations are applied as whole multi-statement files via exec(), so the statement-anchored

test/integration/selfhost-pg.test.ts

@@ -6,6 +6,8 @@ import { afterAll, beforeAll, describe, expect, it } from "vitest";
 import pg from "pg";
 import { runSelfHostMigrations } from "../../src/selfhost/migrate";
 import { createPgAdapter } from "../../src/selfhost/pg-adapter";
+import { pruneExpiredRecords } from "../../src/db/retention";
+import { processJob } from "../../src/queue/processors";
 
 const URL = process.env.PG_TEST_URL;
 const suite = URL ? describe : describe.skip;
@@ -55,4 +57,31 @@ suite("Postgres backend (#977) — real Postgres", () => {
     const still = await db.prepare("SELECT COUNT(*) AS n FROM system_flags WHERE key=?").bind("batch_probe").first<{ n: number }>();
     expect(still?.n).toBe(1); // the DELETE rolled back
   });
+
+  it("prunes rows past the retention window and processJob('prune-retention') does not dead-letter (regression for the live self-host incident: job _selfhost_jobs.id=61132 failed with 'column \"rowid\" does not exist')", async () => {
+    const db = createPgAdapter(pool);
+    const env = { DB: db } as unknown as Env;
+    const oldIso = new Date(Date.now() - 100 * 86_400_000).toISOString();
+    const recentIso = new Date(Date.now() - 1 * 86_400_000).toISOString();
+    for (const [id, createdAt] of [
+      ["pg-old-1", oldIso],
+      ["pg-old-2", oldIso],
+      ["pg-recent", recentIso],
+    ] as const) {
+      await db
+        .prepare("INSERT INTO ai_usage_events (id, feature, model, status, estimated_neurons, created_at) VALUES (?, 'f', 'm', 'ok', 1, ?)")
+        .bind(id, createdAt)
+        .run();
+    }
+
+    const results = await pruneExpiredRecords(env, { policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }] });
+    expect(results[0]?.deleted).toBe(2); // the two old rows, bounded-batch deleted via ctid (not rowid)
+    const remaining = await db.prepare("SELECT COUNT(*) AS n FROM ai_usage_events").first<{ n: number }>();
+    expect(remaining?.n).toBe(1);
+
+    // The exact live incident: the job queue's processJob("prune-retention") dispatch must not throw.
+    await expect(processJob(env, { type: "prune-retention", requestedBy: "schedule" })).resolves.toBeUndefined();
+    const audit = await db.prepare("SELECT outcome FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string }>();
+    expect(audit?.outcome).toBe("success");
+  });
 });

test/unit/selfhost-pg-dialect.test.ts

@@ -1,12 +1,27 @@
 import { describe, expect, it } from "vitest";
-import { stripConflictTargetQualifiers, toNumberedPlaceholders, translateDdl, translateFunctions, translateInsertOr, translateMigrationInserts, translateSql } from "../../src/selfhost/pg-dialect";
+import { stripConflictTargetQualifiers, toNumberedPlaceholders, translateDdl, translateFunctions, translateInsertOr, translateMigrationInserts, translateRowid, translateSql } from "../../src/selfhost/pg-dialect";
 
 describe("pg-dialect (#977 SQLite → Postgres)", () => {
   it("numbers placeholders, skipping `?` inside string literals", () => {
     expect(toNumberedPlaceholders("SELECT * FROM t WHERE a=? AND b=?")).toBe("SELECT * FROM t WHERE a=$1 AND b=$2");
     expect(toNumberedPlaceholders("SELECT '?' AS lit WHERE a=?")).toBe("SELECT '?' AS lit WHERE a=$1");
   });
 
+  it("REGRESSION: reuses a SQLite numbered placeholder's own index instead of corrupting it via the anonymous counter", () => {
+    // Before the fix: `?1` scanned as anonymous `?` (→ $1) followed by a literal `1`, corrupting to `$11`
+    // — a bind index Postgres has no value for. retentionWhere() (retention.ts) and claimRegateFanoutSlot()
+    // (repositories.ts) both use this numbered syntax.
+    expect(toNumberedPlaceholders("created_at < ?1")).toBe("created_at < $1");
+    expect(toNumberedPlaceholders("last_regate_fanout_at = ?1 WHERE id = 'singleton' AND (x IS NULL OR x < ?2)")).toBe(
+      "last_regate_fanout_at = $1 WHERE id = 'singleton' AND (x IS NULL OR x < $2)",
+    );
+    // A later anonymous `?` continues from the highest index already assigned (SQLite's own rule), so it
+    // must not collide with an earlier numbered placeholder.
+    expect(toNumberedPlaceholders("a=?1 AND b=?")).toBe("a=$1 AND b=$2");
+    // A literal `?1` inside a string is left untouched, same as a bare `?` literal.
+    expect(toNumberedPlaceholders("SELECT '?1' AS lit WHERE a=?")).toBe("SELECT '?1' AS lit WHERE a=$1");
+  });
+
   it("translates datetime/strftime/CURRENT_TIMESTAMP/json to Postgres (text-returning to match SQLite)", () => {
     expect(translateFunctions("x > datetime('now', ?)")).toContain("to_char(now() + (?)::interval");
     expect(translateFunctions("datetime('now')")).toContain("to_char(now(),");
@@ -80,4 +95,29 @@ describe("pg-dialect (#977 SQLite → Postgres)", () => {
     expect(out).toContain("values ($1, $2)"); // placeholders numbered
     expect(out).toContain("set \"status\" = $3");
   });
+
+  it("translates the rowid pseudo-column to Postgres's ctid system column", () => {
+    expect(translateRowid("SELECT rowid FROM t WHERE a = ?")).toBe("SELECT ctid FROM t WHERE a = ?");
+    expect(translateRowid("ORDER BY rowid DESC")).toBe("ORDER BY ctid DESC");
+    expect(translateRowid("ORDER BY ROWID ASC")).toBe("ORDER BY ctid ASC"); // case-insensitive
+    // Only the bare `rowid` token is rewritten — identifiers that merely contain it are left alone.
+    expect(translateRowid("SELECT row_id, my_rowid_col FROM t")).toBe("SELECT row_id, my_rowid_col FROM t");
+    expect(translateRowid("SELECT 1")).toBe("SELECT 1"); // no-op passthrough
+  });
+
+  it("REGRESSION (self-host Postgres prune-retention dead-letter): translateSql strips rowid from the exact batched-delete shape retention.ts emits", () => {
+    // The literal shape src/db/retention.ts's pruneExpiredRecords() builds for its bounded batched delete.
+    // Before the fix, this reached Postgres verbatim and failed with `column "rowid" does not exist`.
+    const deleteSql = 'DELETE FROM ai_usage_events WHERE rowid IN (SELECT rowid FROM ai_usage_events WHERE created_at < ?1 LIMIT 1000)';
+    const out = translateSql(deleteSql);
+    expect(out.toLowerCase()).not.toContain("rowid");
+    expect(out).toBe("DELETE FROM ai_usage_events WHERE ctid IN (SELECT ctid FROM ai_usage_events WHERE created_at < $1 LIMIT 1000)");
+  });
+
+  it("also fixes the rowid tie-break ORDER BY used by orb/relay.ts enrollment resolution", () => {
+    const sql = "SELECT relay_mode FROM orb_enrollments WHERE installation_id = ? ORDER BY enrolled_at DESC, rowid DESC";
+    const out = translateSql(sql);
+    expect(out.toLowerCase()).not.toContain("rowid");
+    expect(out).toContain("ORDER BY enrolled_at DESC, ctid DESC");
+  });
 });

test/unit/selfhost-pg-retention.test.ts

@@ -0,0 +1,123 @@
+// Unit tests for data-retention pruning (src/db/retention.ts) against the Postgres backend (#977). Mocks
+// pg.Pool so no real DB is needed — real-Postgres integration coverage lives in test/integration/selfhost-pg.test.ts.
+// retention.ts itself is unchanged: it still emits SQLite-dialect SQL (rowid, `?1`-style numbered
+// placeholders); src/selfhost/pg-dialect.ts's translateSql() is what makes it Postgres-safe, same as every
+// other query path on this backend. These tests exercise that translation end-to-end through the real
+// pruneExpiredRecords()/processJob() call path, not just the dialect translator in isolation.
+import { describe, expect, it, vi } from "vitest";
+import type { Pool } from "pg";
+import { createPgAdapter } from "../../src/selfhost/pg-adapter";
+import { pruneExpiredRecords } from "../../src/db/retention";
+import { processJob, runRetentionPrune } from "../../src/queue/processors";
+
+interface MockPgPool {
+  pool: Pool;
+  calls: string[];
+  remaining: Record<string, number>;
+}
+
+/** A minimal fake Postgres that also acts as a regression guard: if untranslated SQLite SQL (the `rowid`
+ *  pseudo-column) ever reaches it again, it throws the exact error a real Postgres raised in the live
+ *  self-host incident (dead-lettered job `_selfhost_jobs.id = 61132`, `prune-retention`, 5 attempts). */
+function makeRetentionPgPool(remaining: Record<string, number> = {}): MockPgPool {
+  const calls: string[] = [];
+  const fn = vi.fn().mockImplementation(async (sql: unknown) => {
+    const q = String(sql);
+    calls.push(q);
+    if (/\browid\b/i.test(q)) throw new Error('column "rowid" does not exist');
+
+    const countMatch = /^SELECT count\(\*\) AS n FROM (\w+) WHERE/i.exec(q);
+    if (countMatch) {
+      const table = countMatch[1] as string;
+      return { rows: [{ n: remaining[table] ?? 0 }], rowCount: 1 };
+    }
+
+    const deleteMatch = /^DELETE FROM (\w+) WHERE ctid IN \(SELECT ctid FROM \1 WHERE .*? LIMIT (\d+)\)$/i.exec(q);
+    if (deleteMatch) {
+      const table = deleteMatch[1] as string;
+      const limit = Number(deleteMatch[2]);
+      const have = remaining[table] ?? 0;
+      const changes = Math.min(have, limit);
+      remaining[table] = have - changes;
+      return { rows: [], rowCount: changes };
+    }
+
+    if (/^insert into "?audit_events"?/i.test(q)) return { rows: [], rowCount: 1 };
+
+    return { rows: [], rowCount: 0 };
+  });
+  return { pool: { query: fn } as unknown as Pool, calls, remaining };
+}
+
+function makeEnv(remaining: Record<string, number> = {}): { env: Env; mock: MockPgPool } {
+  const mock = makeRetentionPgPool(remaining);
+  return { env: { DB: createPgAdapter(mock.pool) } as unknown as Env, mock };
+}
+
+describe("pruneExpiredRecords on the Postgres backend (#977)", () => {
+  it("dry-run counts eligible rows without issuing any delete", async () => {
+    const { env, mock } = makeEnv({ ai_usage_events: 5 });
+    const results = await pruneExpiredRecords(env, {
+      dryRun: true,
+      policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }],
+    });
+    expect(results[0]?.deleted).toBe(5);
+    expect(mock.calls.some((q) => /^DELETE/i.test(q))).toBe(false);
+    expect(mock.remaining.ai_usage_events).toBe(5); // untouched
+  });
+
+  it("deletes across multiple bounded batches and stops at the per-table cap, same as the SQLite path", async () => {
+    const { env, mock } = makeEnv({ ai_usage_events: 5 });
+    const results = await pruneExpiredRecords(env, {
+      batchSize: 2,
+      maxPerTable: 4,
+      policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }],
+    });
+    expect(results[0]?.deleted).toBe(4); // 2 + 2, then cap reached
+    expect(mock.remaining.ai_usage_events).toBe(1); // one row left for the next run
+    const deletes = mock.calls.filter((q) => /^DELETE/i.test(q));
+    expect(deletes).toHaveLength(2);
+  });
+
+  it("keeps the audit_events durable-event exclusion in the translated Postgres delete", async () => {
+    const { env, mock } = makeEnv({ audit_events: 1 });
+    await pruneExpiredRecords(env, { policy: [{ table: "audit_events", column: "created_at", days: 90 }] });
+    const [deleteSql] = mock.calls.filter((q) => /^DELETE/i.test(q));
+    expect(deleteSql).toContain("event_type NOT IN ('github_app.pr_public_surface_published')");
+    expect(deleteSql?.toLowerCase()).not.toContain("rowid");
+  });
+
+  it("translates the numbered `?1` cutoff placeholder to Postgres's $1, not a corrupted $11", async () => {
+    const { env, mock } = makeEnv({ ai_usage_events: 0 });
+    await pruneExpiredRecords(env, {
+      dryRun: true,
+      policy: [{ table: "ai_usage_events", column: "created_at", days: 90 }],
+    });
+    const [countSql] = mock.calls;
+    expect(countSql).toContain("created_at < $1");
+    expect(countSql).not.toMatch(/\$1\d/); // not $11, $12, ...
+  });
+});
+
+describe("runRetentionPrune + processJob on the Postgres backend (#977)", () => {
+  it("audits a dry-run without deleting", async () => {
+    const { env, mock } = makeEnv({ ai_usage_events: 2 });
+    await runRetentionPrune(env, "test", true);
+    const [insertSql] = mock.calls.filter((q) => /^insert into "?audit_events"?/i.test(q));
+    expect(insertSql).toBeDefined();
+    expect(mock.remaining.ai_usage_events).toBe(2); // nothing deleted
+  });
+
+  it("REGRESSION (self-host dead-letter, job id 61132): processJob prune-retention no longer throws the rowid column error on Postgres", async () => {
+    const { env } = makeEnv(); // every table reports 0 eligible rows — exercises the full default policy
+    await expect(processJob(env, { type: "prune-retention", requestedBy: "schedule" })).resolves.toBeUndefined();
+  });
+
+  it("processJob prune-retention deletes eligible rows and records a success audit event on Postgres", async () => {
+    const { env, mock } = makeEnv({ ai_usage_events: 3 });
+    await processJob(env, { type: "prune-retention", requestedBy: "schedule" });
+    expect(mock.remaining.ai_usage_events).toBe(0);
+    const [insertSql] = mock.calls.filter((q) => /^insert into "?audit_events"?/i.test(q));
+    expect(insertSql).toBeDefined();
+  });
+});