TypeScript 55.4%
Python 43.2%
SQL 1.2%
1import { unlink } from "node:fs/promises";2import { join, resolve } from "node:path";3import { db, sql } from "@websensor/db";4import { config, log } from "./config";5import { m } from "./metrics";67/**8 * Storage lifecycle (spec §58). Evidence that matters is never touched:9 * - snapshots referenced by an event (old or new side) keep raw body + canonical forever;10 * - snapshots referenced by a *change* keep raw + canonical for `keepChangedRawDays`;11 * - "baseline / unchanged" snapshots (no change, no event) lose their RAW body after `keepRawDays`12 * but keep their canonical representation and every hash, so history/compare still work.13 * Blobs are content-addressed and shared: a raw blob is deleted only when no other snapshot row14 * still references its key. Runs incrementally (bounded batches).15 */16export async function pruneRawSnapshots(opts: { keepRawDays?: number; keepChangedRawDays?: number; batch?: number } = {}): Promise<{ examined: number; pruned: number; freedKeys: number }> {17 const keepRawDays = opts.keepRawDays ?? config.retention.rawDays;18 const keepChangedRawDays = opts.keepChangedRawDays ?? config.retention.changedRawDays;19 const batch = opts.batch ?? 2000;20 const rows = await db.execute<{ id: string; storage_key: string }>(sql`21 select s.id, s.storage_key from snapshots s22 where s.storage_key is not null23 and s.captured_at < now() - make_interval(days => ${keepRawDays})24 and not exists (select 1 from events e where e.new_snapshot_id = s.id or e.old_snapshot_id = s.id)25 and not exists (select 1 from changes c where (c.new_snapshot_id = s.id or c.old_snapshot_id = s.id) and c.detected_at >= now() - make_interval(days => ${keepChangedRawDays}))26 order by s.captured_at asc limit ${batch}`);27 let pruned = 0;28 let freed = 0;29 const root = resolve(process.env.BLOB_STORE_DIR ?? "./data/blobs");30 for (const r of rows.rows) {31 // detach the raw body from this snapshot (canonical_storage_key stays)32 await db.execute(sql`update snapshots set storage_key = null, extra = coalesce(extra, '{}'::jsonb) || jsonb_build_object('raw_pruned_at', now()) where id = ${r.id}`);33 pruned++;34 // delete the file only if no other row references the same content-addressed key35 const still = await db.execute<{ n: string }>(sql`select count(*)::text as n from snapshots where storage_key = ${r.storage_key} or canonical_storage_key = ${r.storage_key} union all select count(*)::text from changes where diff_storage_key = ${r.storage_key}`);36 if (still.rows.every((x) => Number(x.n) === 0)) {37 await unlink(join(root, r.storage_key + ".zst")).catch(() => undefined);38 freed++;39 m.prunedBlobs.inc();40 }41 }42 if (pruned) log.info({ examined: rows.rows.length, pruned, freed }, "raw snapshot bodies pruned");43 return { examined: rows.rows.length, pruned, freedKeys: freed };44}4546/** Old raw changes that never became events: keep metadata, drop nothing (they are small). Old runs are pruned in scheduler.ts. */47export async function pruneNotifications(): Promise<void> {48 await db.execute(sql`delete from notifications where created_at < now() - interval '90 days'`);49}50