From 4a1f89885a660834fe1c6d932758ce0ea05ec1ff Mon Sep 17 00:00:00 2001 From: JSONbored <49853598+JSONbored@users.noreply.github.com> Date: Sat, 13 Jun 2026 16:33:15 -0700 Subject: [PATCH] feat(db): scheduled data-retention pruning for log/snapshot tables Adds a conservative, auto-pruning retention job so high-volume append-only and superseded-snapshot tables don't grow unbounded (precautionary D1 hygiene; the DB isn't at its size limit yet). - src/db/retention.ts: RETENTION_POLICY (webhook_events 30d, audit_events 90d, ai_usage_events 90d, product_usage_events 180d, github_rate_limit_observations 30d, signal_snapshots/score_previews/repo_snapshots 90d) + pruneExpiredRecords, which deletes rows older than each window in bounded rowid batches (cap per table per run). Table/column names come only from the hardcoded policy and are identifier-validated; the cutoff is bound. Current-state/reference tables (repositories, settings, PRs, issues, contributors, repository_ai_keys, etc.) are intentionally NOT pruned. - Wired as a daily (03:00 UTC) prune-retention queue job via the cron; processJob runs it and audits the outcome (retention.prune). - GET /v1/internal/retention/preview: read-only dry-run that reports the rows the next prune would delete, per table (deletes nothing). Tests cover dry-run vs real delete, the batch loop + per-table cap, the identifier guard, the audit, processJob, the preview route, and that protected tables are excluded. Coverage holds above the 97% gate. --- src/api/routes.ts | 8 +++ src/db/retention.ts | 80 ++++++++++++++++++++++++++ src/index.ts | 4 ++ src/queue/processors.ts | 20 +++++++ src/types.ts | 5 ++ test/unit/retention.test.ts | 112 ++++++++++++++++++++++++++++++++++++ 6 files changed, 229 insertions(+) create mode 100644 src/db/retention.ts create mode 100644 test/unit/retention.test.ts diff --git a/src/api/routes.ts b/src/api/routes.ts index 7be7401054..9e688fd969 100644 --- a/src/api/routes.ts +++ b/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); diff --git a/src/db/retention.ts b/src/db/retention.ts new file mode 100644 index 0000000000..0ad30dbe9d --- /dev/null +++ b/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 { + 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; +} diff --git a/src/index.ts b/src/index.ts index 8298f9cca0..9d2692fc2e 100644 --- a/src/index.ts +++ b/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" }); diff --git a/src/queue/processors.ts b/src/queue/processors.ts index 6276eb8ee8..f5ae238d9d 100644 --- a/src/queue/processors.ts +++ b/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 { + 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 { switch (message.type) { case "refresh-registry": @@ -248,6 +265,9 @@ export async function processJob(env: Env, message: JobMessage): Promise { 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; diff --git a/src/types.ts b/src/types.ts index 54a77be066..8bcd0c6a3a 100644 --- a/src/types.ts +++ b/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"; diff --git a/test/unit/retention.test.ts b/test/unit/retention.test.ts new file mode 100644 index 0000000000..cbcf485a31 --- /dev/null +++ b/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 + }); +});