feat(db): scheduled data-retention pruning for log/snapshot tables

src/api/routes.ts

@@ -93,6 +93,7 @@ import {
   upsertRepositoryAiKey,
   deleteRepositoryAiKey,
 } from "../db/repositories";
+import { pruneExpiredRecords, RETENTION_POLICY } from "../db/retention";
 import {
   backfillOpenPullRequestDetails,
   backfillRegisteredRepositories,
@@ -2480,6 +2481,13 @@ export function createApp() {
     return c.json(await getRepositoryAiKeyStatus(c.env, fullName));
   });
 
+  // Read-only retention preview: counts the rows the daily prune cron would delete, per table. Does NOT
+  // delete anything (dry-run); the actual prune runs on the schedule via the prune-retention job.
+  app.get("/v1/internal/retention/preview", async (c) => {
+    const results = await pruneExpiredRecords(c.env, { dryRun: true });
+    return c.json({ policy: RETENTION_POLICY, eligible: results, totalEligible: results.reduce((sum, r) => sum + r.deleted, 0) });
+  });
+
   app.post("/v1/internal/repos/:owner/:repo/ai-key", async (c) => {
     const body = await c.req.json().catch(() => null);
     const parsed = repositoryAiKeySchema.safeParse(body);

src/db/retention.ts

@@ -0,0 +1,80 @@
+import { nowIso } from "../utils/json";
+
+/**
+ * Data-retention policy for the high-volume, append-only / log / superseded-snapshot tables. These hold
+ * pure history (logs, usage metrics, ephemeral observations) or snapshots where only the latest matters,
+ * so rows older than the window can be safely deleted. Current-state and reference tables (repositories,
+ * repository_settings, pull_requests, issues, contributors, registry/scoring snapshots, repository_ai_keys,
+ * focus manifests, etc.) are intentionally EXCLUDED — they are not append-only logs.
+ *
+ * `column` is the row's primary timestamp (ISO-8601). Windows are deliberately conservative.
+ */
+export type RetentionRule = { table: string; column: string; days: number };
+
+export const RETENTION_POLICY: readonly RetentionRule[] = [
+  { table: "webhook_events", column: "received_at", days: 30 },
+  { table: "audit_events", column: "created_at", days: 90 },
+  { table: "ai_usage_events", column: "created_at", days: 90 },
+  { table: "product_usage_events", column: "occurred_at", days: 180 },
+  { table: "github_rate_limit_observations", column: "observed_at", days: 30 },
+  { table: "signal_snapshots", column: "generated_at", days: 90 },
+  { table: "score_previews", column: "generated_at", days: 90 },
+  { table: "repo_snapshots", column: "fetched_at", days: 90 },
+];
+
+export type PruneResult = { table: string; column: string; cutoff: string; deleted: number };
+
+const SAFE_IDENTIFIER = /^[a-z_]+$/;
+const BATCH_SIZE = 1000;
+// Bound work per table per run so a first prune of a large backlog cannot blow the D1 statement budget;
+// the daily cron drains any remainder over subsequent runs.
+const MAX_DELETED_PER_TABLE = 50_000;
+const MS_PER_DAY = 86_400_000;
+
+function cutoffIso(days: number, nowMs: number): string {
+  return new Date(nowMs - days * MS_PER_DAY).toISOString();
+}
+
+/**
+ * Delete (or, in dry-run, count) rows older than each table's retention window. Returns per-table results.
+ * Table/column names come only from the hardcoded {@link RETENTION_POLICY} (never user input) and are
+ * identifier-validated defensively; the cutoff is bound as a parameter. Deletes run in bounded batches.
+ */
+export async function pruneExpiredRecords(
+  env: Env,
+  options: { dryRun?: boolean; nowMs?: number; policy?: readonly RetentionRule[]; batchSize?: number; maxPerTable?: number } = {},
+): Promise<PruneResult[]> {
+  const dryRun = options.dryRun ?? false;
+  const nowMs = options.nowMs ?? Date.parse(nowIso());
+  const policy = options.policy ?? RETENTION_POLICY;
+  const batchSize = options.batchSize ?? BATCH_SIZE;
+  const maxPerTable = options.maxPerTable ?? MAX_DELETED_PER_TABLE;
+  const results: PruneResult[] = [];
+
+  for (const rule of policy) {
+    if (!SAFE_IDENTIFIER.test(rule.table) || !SAFE_IDENTIFIER.test(rule.column)) {
+      throw new Error(`Unsafe retention identifier: ${rule.table}.${rule.column}`);
+    }
+    const cutoff = cutoffIso(rule.days, nowMs);
+
+    if (dryRun) {
+      const row = await env.DB.prepare(`SELECT count(*) AS n FROM ${rule.table} WHERE ${rule.column} < ?1`).bind(cutoff).first<{ n: number }>();
+      results.push({ table: rule.table, column: rule.column, cutoff, deleted: Number(row?.n ?? 0) });
+      continue;
+    }
+
+    let deleted = 0;
+    // Batched delete by rowid so each statement is bounded; loop until a short batch or the per-run cap.
+    for (;;) {
+      const result = await env.DB.prepare(`DELETE FROM ${rule.table} WHERE rowid IN (SELECT rowid FROM ${rule.table} WHERE ${rule.column} < ?1 LIMIT ${batchSize})`)
+        .bind(cutoff)
+        .run();
+      const changes = Number(result.meta?.changes ?? 0);
+      deleted += changes;
+      if (changes < batchSize || deleted >= maxPerTable) break;
+    }
+    results.push({ table: rule.table, column: rule.column, cutoff, deleted });
+  }
+
+  return results;
+}

src/index.ts

@@ -53,6 +53,10 @@ async function enqueueScheduledJobs(env: Env, controller: ScheduledController):
   if (isHourly && scheduledAt.getUTCDay() === 1 && hour === 12) {
     jobs.push({ type: "generate-weekly-value-report", requestedBy: "schedule", variant: "operator", days: 7 });
   }
+  // Prune expired log/snapshot rows once a day (03:00 UTC) per the conservative RETENTION_POLICY.
+  if (isHourly && hour === 3) {
+    jobs.push({ type: "prune-retention", requestedBy: "schedule" });
+  }
   if (isFullSyncWindow) {
     jobs.push({ type: "generate-signal-snapshots", requestedBy: "schedule" });
     jobs.push({ type: "build-burden-forecasts", requestedBy: "schedule" });

src/queue/processors.ts

@@ -53,6 +53,7 @@ import {
   upsertPullRequestFromGitHub,
   upsertRepositoryFromGitHub,
 } from "../db/repositories";
+import { pruneExpiredRecords } from "../db/retention";
 import {
   backfillOpenPullRequestDetails,
   backfillRegisteredRepositories,
@@ -143,6 +144,22 @@ const OFFICIAL_MINER_DETECTION_UNAVAILABLE_TTL_MS = 60 * 1000;
 const PR_PUBLIC_SURFACE_ACTIONS = new Set(["opened", "reopened", "synchronize", "ready_for_review", "edited"]);
 const PR_GATE_CLOSED_ACTIONS = new Set(["closed"]);
 
+/**
+ * Run (or dry-run) the data-retention prune across the configured log/snapshot tables and audit the
+ * outcome. The per-table windows live in RETENTION_POLICY; only append-only/superseded tables are pruned.
+ */
+export async function runRetentionPrune(env: Env, requestedBy: string, dryRun: boolean): Promise<void> {
+  const results = await pruneExpiredRecords(env, { dryRun });
+  const totalDeleted = results.reduce((sum, result) => sum + result.deleted, 0);
+  await recordAuditEvent(env, {
+    eventType: "retention.prune",
+    actor: requestedBy,
+    outcome: dryRun ? "completed" : "success",
+    detail: dryRun ? `dry-run: ${totalDeleted} row(s) eligible` : `pruned ${totalDeleted} row(s)`,
+    metadata: { dryRun, totalDeleted, perTable: Object.fromEntries(results.map((r) => [r.table, r.deleted])) },
+  });
+}
+
 export async function processJob(env: Env, message: JobMessage): Promise<void> {
   switch (message.type) {
     case "refresh-registry":
@@ -248,6 +265,9 @@ export async function processJob(env: Env, message: JobMessage): Promise<void> {
     case "rollup-product-usage":
       await rollupProductUsageDaily(env, { ...(message.day ? { day: message.day } : {}), ...(message.days === undefined ? {} : { days: message.days }) });
       return;
+    case "prune-retention":
+      await runRetentionPrune(env, message.requestedBy, message.dryRun ?? false);
+      return;
     case "generate-weekly-value-report":
       await generateWeeklyValueReport(env, { variant: message.variant ?? "operator", ...(message.days === undefined ? {} : { days: message.days }) });
       return;

src/types.ts

@@ -99,6 +99,11 @@ export type JobMessage =
       day?: string;
       days?: number;
     }
+  | {
+      type: "prune-retention";
+      requestedBy: "schedule" | "api" | "test";
+      dryRun?: boolean;
+    }
   | {
       type: "generate-weekly-value-report";
       requestedBy: "schedule" | "api" | "test";

test/unit/retention.test.ts

@@ -0,0 +1,112 @@
+import { eq } from "drizzle-orm";
+import { describe, expect, it } from "vitest";
+import { createApp } from "../../src/api/routes";
+import { getDb } from "../../src/db/client";
+import { pruneExpiredRecords, RETENTION_POLICY } from "../../src/db/retention";
+import { aiUsageEvents, webhookEvents } from "../../src/db/schema";
+import { processJob, runRetentionPrune } from "../../src/queue/processors";
+import { createTestEnv } from "../helpers/d1";
+
+const NOW = Date.parse("2026-06-13T00:00:00.000Z");
+const daysAgo = (n: number) => new Date(NOW - n * 86_400_000).toISOString();
+
+async function seed(env: Env) {
+  const db = getDb(env.DB);
+  // webhook_events window = 30d; two old + one recent.
+  await db.insert(webhookEvents).values([
+    { deliveryId: "wh-old-1", eventName: "push", payloadHash: "h", status: "processed", receivedAt: daysAgo(40) },
+    { deliveryId: "wh-old-2", eventName: "push", payloadHash: "h", status: "processed", receivedAt: daysAgo(35) },
+    { deliveryId: "wh-recent", eventName: "push", payloadHash: "h", status: "processed", receivedAt: daysAgo(1) },
+  ]);
+  // ai_usage_events window = 90d; one old + one recent.
+  await db.insert(aiUsageEvents).values([
+    { id: "ai-old", feature: "f", model: "m", status: "ok", estimatedNeurons: 1, createdAt: daysAgo(100) },
+    { id: "ai-recent", feature: "f", model: "m", status: "ok", estimatedNeurons: 1, createdAt: daysAgo(2) },
+  ]);
+}
+
+const countWebhook = async (env: Env) => (await env.DB.prepare("SELECT count(*) AS n FROM webhook_events").first<{ n: number }>())?.n ?? 0;
+
+describe("pruneExpiredRecords", () => {
+  it("dry-run reports eligible rows per table without deleting anything", async () => {
+    const env = createTestEnv();
+    await seed(env);
+    const results = await pruneExpiredRecords(env, { dryRun: true, nowMs: NOW });
+    const wh = results.find((r) => r.table === "webhook_events");
+    const ai = results.find((r) => r.table === "ai_usage_events");
+    expect(wh?.deleted).toBe(2);
+    expect(ai?.deleted).toBe(1);
+    expect(await countWebhook(env)).toBe(3); // nothing actually deleted
+  });
+
+  it("deletes rows older than the window and keeps recent ones", async () => {
+    const env = createTestEnv();
+    await seed(env);
+    const results = await pruneExpiredRecords(env, { nowMs: NOW });
+    expect(results.find((r) => r.table === "webhook_events")?.deleted).toBe(2);
+    expect(results.find((r) => r.table === "ai_usage_events")?.deleted).toBe(1);
+    expect(await countWebhook(env)).toBe(1);
+    const remaining = await env.DB.prepare("SELECT delivery_id FROM webhook_events").first<{ delivery_id: string }>();
+    expect(remaining?.delivery_id).toBe("wh-recent");
+  });
+
+  it("deletes across multiple batches and stops at the per-table cap", async () => {
+    const env = createTestEnv();
+    const db = getDb(env.DB);
+    await db.insert(webhookEvents).values(
+      Array.from({ length: 5 }, (_, i) => ({ deliveryId: `wh-${i}`, eventName: "push", payloadHash: "h", status: "processed", receivedAt: daysAgo(40) })),
+    );
+    // batchSize 2 forces multiple iterations; maxPerTable 4 forces the cap break before all 5 are gone.
+    const results = await pruneExpiredRecords(env, { nowMs: NOW, batchSize: 2, maxPerTable: 4, policy: [{ table: "webhook_events", column: "received_at", days: 30 }] });
+    expect(results[0]?.deleted).toBe(4); // 2 + 2, then cap reached
+    expect(await countWebhook(env)).toBe(1); // one old row left for the next run
+  });
+
+  it("rejects an unsafe table/column identifier (defensive guard)", async () => {
+    const env = createTestEnv();
+    await expect(pruneExpiredRecords(env, { policy: [{ table: "webhook_events; DROP TABLE x", column: "received_at", days: 1 }] })).rejects.toThrow("Unsafe retention identifier");
+  });
+
+  it("the policy only targets append-only/log/snapshot tables (no current-state tables)", () => {
+    const tables = RETENTION_POLICY.map((r) => r.table);
+    for (const protectedTable of ["repositories", "repository_settings", "pull_requests", "issues", "repository_ai_keys", "contributors"]) {
+      expect(tables).not.toContain(protectedTable);
+    }
+  });
+});
+
+describe("runRetentionPrune + processJob", () => {
+  it("audits a dry-run without deleting", async () => {
+    const env = createTestEnv();
+    await seed(env);
+    await runRetentionPrune(env, "test", true);
+    expect(await countWebhook(env)).toBe(3);
+    const audit = await env.DB.prepare("SELECT outcome, detail FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string; detail: string }>();
+    expect(audit?.outcome).toBe("completed");
+    expect(audit?.detail).toMatch(/dry-run/);
+  });
+
+  it("processJob prune-retention deletes and audits", async () => {
+    const env = createTestEnv();
+    await seed(env);
+    await processJob(env, { type: "prune-retention", requestedBy: "schedule" });
+    expect(await countWebhook(env)).toBe(1);
+    const audit = await env.DB.prepare("SELECT outcome FROM audit_events WHERE event_type = ?").bind("retention.prune").first<{ outcome: string }>();
+    expect(audit?.outcome).toBe("success");
+  });
+});
+
+describe("retention preview route", () => {
+  it("GET /v1/internal/retention/preview returns eligible counts and deletes nothing", async () => {
+    const app = createApp();
+    const env = createTestEnv();
+    await seed(env);
+    const res = await app.request("/v1/internal/retention/preview", { headers: { authorization: `Bearer ${env.INTERNAL_JOB_TOKEN}` } }, env);
+    expect(res.status).toBe(200);
+    const body = (await res.json()) as { totalEligible: number; eligible: Array<{ table: string; deleted: number }> };
+    expect(body.totalEligible).toBeGreaterThanOrEqual(3);
+    // The 2 rows aged 35/40d are eligible regardless of when the suite runs (route uses real `now`).
+    expect(body.eligible.find((r) => r.table === "webhook_events")?.deleted).toBeGreaterThanOrEqual(2);
+    expect(await countWebhook(env)).toBe(3); // preview is read-only
+  });
+});