import { unlink } from "node:fs/promises";
import { join, resolve } from "node:path";
import { db, sql } from "@websensor/db";
import { config, log } from "./config";
import { m } from "./metrics";

/**
 * Storage lifecycle (spec §58). Evidence that matters is never touched:
 *   - snapshots referenced by an event (old or new side) keep raw body + canonical forever;
 *   - snapshots referenced by a *change* keep raw + canonical for `keepChangedRawDays`;
 *   - "baseline / unchanged" snapshots (no change, no event) lose their RAW body after `keepRawDays`
 *     but keep their canonical representation and every hash, so history/compare still work.
 * Blobs are content-addressed and shared: a raw blob is deleted only when no other snapshot row
 * still references its key. Runs incrementally (bounded batches).
 */
export async function pruneRawSnapshots(opts: { keepRawDays?: number; keepChangedRawDays?: number; batch?: number } = {}): Promise<{ examined: number; pruned: number; freedKeys: number }> {
  const keepRawDays = opts.keepRawDays ?? config.retention.rawDays;
  const keepChangedRawDays = opts.keepChangedRawDays ?? config.retention.changedRawDays;
  const batch = opts.batch ?? 2000;
  const rows = await db.execute<{ id: string; storage_key: string }>(sql`
    select s.id, s.storage_key from snapshots s
    where s.storage_key is not null
      and s.captured_at < now() - make_interval(days => ${keepRawDays})
      and not exists (select 1 from events e where e.new_snapshot_id = s.id or e.old_snapshot_id = s.id)
      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}))
    order by s.captured_at asc limit ${batch}`);
  let pruned = 0;
  let freed = 0;
  const root = resolve(process.env.BLOB_STORE_DIR ?? "./data/blobs");
  for (const r of rows.rows) {
    // detach the raw body from this snapshot (canonical_storage_key stays)
    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}`);
    pruned++;
    // delete the file only if no other row references the same content-addressed key
    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}`);
    if (still.rows.every((x) => Number(x.n) === 0)) {
      await unlink(join(root, r.storage_key + ".zst")).catch(() => undefined);
      freed++;
      m.prunedBlobs.inc();
    }
  }
  if (pruned) log.info({ examined: rows.rows.length, pruned, freed }, "raw snapshot bodies pruned");
  return { examined: rows.rows.length, pruned, freedKeys: freed };
}

/** Old raw changes that never became events: keep metadata, drop nothing (they are small). Old runs are pruned in scheduler.ts. */
export async function pruneNotifications(): Promise<void> {
  await db.execute(sql`delete from notifications where created_at < now() - interval '90 days'`);
}