import { env } from "cloudflare:workers";
import { all, batch, chunks, placeholders, run, stmt } from "../db";
import { repoDb } from "../scoped-db";
import { siteScopesFrom } from "./store";
import { manifestKey, objectKey, SITE_LIMITS } from "./config";
import type { SiteManifest } from "./config";
import { notifyUsers } from "../notify/deliver";
import { publicUrl } from "../events/emit";

/**
 * Mark-and-sweep for the content-addressed Sites objects under `pages/objects/`. A run marks every
 * hash a retained manifest references (active, superseded within the history limit, previews and commit previews),
 * then deletes unmarked objects uploaded more than GRACE_MS before the run started. Deploys refresh any object they
 * reuse that is older than REFRESH_MS (collect.ts), so an object a deploy in flight relies on is never old enough
 * to be swept. State lives in KV, so each cron tick does a bounded step and the next one resumes.
 */
export const GC_GRACE_MS = 7 * 24 * 60 * 60 * 1000;
const RUN_INTERVAL_MS = 24 * 60 * 60 * 1000;
const RETRY_MS = 60 * 60 * 1000;
const LOCK = "pages-gc";
const LOCK_MS = 4 * 60 * 1000;
/** A retained deployment whose manifest has been gone this long is an orphan, skipped instead of blocking GC. */
const ORPHAN_MS = 7 * 24 * 60 * 60 * 1000;
const STATE_KEY = "pages:gc:state";
const PREFIX = "pages/objects/";
const MARK_PAGE = 50;
/** Bounds R2 work per step (one list, a head per candidate, one delete) to stay far below the subrequest limit. */
const LIST_PAGE = 100;
const HEAD_BATCH = 50;
const STEPS_PER_TICK = 5;
const INSERT_ROWS = 80;

type GcState =
  | { phase: "idle"; nextAt: number }
  | { phase: "clear" | "mark"; startedAt: number; cursor: string }
  | { phase: "sweep"; startedAt: number; cursor: string | null; deleted: number };

const RETAINED_SQL = `SELECT id, status, created_at FROM (
    SELECT id, status, created_at FROM (
      SELECT id, status, created_at, ROW_NUMBER() OVER (PARTITION BY site_id ORDER BY created_at DESC) AS n
      FROM site_deployments WHERE status IN ('active', 'superseded', 'queued', 'building')
    ) WHERE status != 'superseded' OR n <= ?1
    UNION SELECT d.id, 'active' AS status, d.created_at FROM site_pr_aliases a
      JOIN site_deployments d ON d.id = a.deployment_id
  ) WHERE id > ?2 ORDER BY id LIMIT ?3`;

async function loadState(): Promise<GcState> {
  return (await env.KV.get<GcState>(STATE_KEY, "json")) ?? { phase: "idle", nextAt: 0 };
}

async function saveState(state: GcState): Promise<void> {
  await env.KV.put(STATE_KEY, JSON.stringify(state));
}

/** The hashes a deployment's manifest references, or null when the manifest can't be read. */
async function manifestHashes(id: string): Promise<string[] | null> {
  const manifest = await env.KV.get<SiteManifest>(manifestKey(id), "json");
  if (!manifest) return null;
  return [...Object.values(manifest.files), ...Object.values(manifest.worker ?? {})].map(([sha]) => sha);
}

async function mark(hashes: string[]): Promise<void> {
  const unique = [...new Set(hashes)];
  const statements = chunks(unique, INSERT_ROWS).map((group) =>
    stmt(`INSERT OR IGNORE INTO r2_gc_marks (hash) VALUES ${group.map(() => "(?)").join(", ")}`, ...group),
  );
  for (const group of chunks(statements, 50)) await batch(group);
}

/** One bounded step; returns the state to continue from. */
async function step(state: GcState, now: number): Promise<GcState> {
  if (state.phase === "idle") {
    if (now < state.nextAt) return state;
    return { phase: "clear", startedAt: now, cursor: "" };
  }
  if (state.phase === "clear") {
    const removed = await run("DELETE FROM r2_gc_marks WHERE rowid IN (SELECT rowid FROM r2_gc_marks LIMIT 5000)");
    return removed.meta.changes > 0 ? state : { phase: "mark", startedAt: state.startedAt, cursor: "" };
  }
  if (state.phase === "mark") {
    // The cursor is `<scope>\n<deployment id>`: scopes are walked in order, each one page at a time.
    const [cursorScope = "", cursorId = ""] = state.cursor.split("\n");
    const [scope] = await siteScopesFrom(cursorScope, 1);
    if (scope === undefined) return { phase: "sweep", startedAt: state.startedAt, cursor: null, deleted: 0 };
    const rows = await repoDb(scope).all<{ id: string; status: string; created_at: number }>(
      RETAINED_SQL,
      SITE_LIMITS.historyLimit,
      scope === cursorScope ? cursorId : "",
      MARK_PAGE,
    );
    const found = await Promise.all(rows.map((row) => manifestHashes(row.id)));
    // Queued and building deploys have no manifest yet (their objects stay fresh); any other gap aborts the run.
    const missing = rows.filter(
      (row, index) => found[index] === null && row.status !== "queued" && row.status !== "building",
    );
    const orphans = missing.filter((row) => row.created_at < now - ORPHAN_MS);
    if (orphans.length > 0) {
      console.warn("Hosting object GC skipping orphaned deployments with no manifest", orphans.slice(0, 5));
    }
    const unexplained = missing.filter((row) => row.created_at >= now - ORPHAN_MS);
    if (unexplained.length > 0) {
      console.error("Hosting object GC aborted: unreadable retained manifests", unexplained.slice(0, 5));
      await notifyAbort(now, unexplained.length);
      return { phase: "idle", nextAt: now + RETRY_MS };
    }
    await mark(found.flatMap((hashes) => hashes ?? []));
    if (rows.length < MARK_PAGE) return { ...state, cursor: `${scope}\u0000\n` };
    return { ...state, cursor: `${scope}\n${rows[rows.length - 1].id}` };
  }
  if (state.phase !== "sweep") return state;
  const listed = await env.BUCKET.list({ prefix: PREFIX, limit: LIST_PAGE, cursor: state.cursor ?? undefined });
  const cutoff = state.startedAt - GC_GRACE_MS;
  const old = listed.objects.filter((object) => object.uploaded.getTime() < cutoff);
  const doomed: string[] = [];
  for (const group of chunks(old.map((object) => object.key.slice(PREFIX.length)))) {
    const marked = await all<{ hash: string }>(
      `SELECT hash FROM r2_gc_marks WHERE hash IN (${placeholders(group.length)})`,
      ...group,
    );
    const keep = new Set(marked.map((row) => row.hash));
    doomed.push(...group.filter((hash) => !keep.has(hash)));
  }
  // A deploy may have re-put an object since the listing; only delete what is still old right now.
  const stale: string[] = [];
  for (const group of chunks(doomed, HEAD_BATCH)) {
    const heads = await Promise.all(group.map((hash) => env.BUCKET.head(objectKey(hash))));
    stale.push(...group.filter((_, index) => (heads[index]?.uploaded.getTime() ?? cutoff) < cutoff));
  }
  if (stale.length > 0) await env.BUCKET.delete(stale.map(objectKey));
  const deleted = state.deleted + stale.length;
  if (listed.truncated) return { ...state, cursor: listed.cursor, deleted };
  if (deleted > 0) console.warn("Hosting object GC deleted unreferenced objects", { deleted });
  return { phase: "idle", nextAt: state.startedAt + RUN_INTERVAL_MS };
}

async function acquire(holder: string, now: number): Promise<boolean> {
  await run("INSERT OR IGNORE INTO cron_locks (name, holder, expires_at) VALUES (?, NULL, 0)", LOCK);
  const taken = await run(
    "UPDATE cron_locks SET holder = ?, expires_at = ? WHERE name = ? AND (holder IS NULL OR expires_at < ?)",
    holder,
    now + LOCK_MS,
    LOCK,
    now,
  );
  return taken.meta.changes > 0;
}

async function notifyAbort(now: number, count: number): Promise<void> {
  const admins = await all<{ account_id: string }>("SELECT account_id FROM users WHERE is_admin = 1 LIMIT 50");
  await notifyUsers({
    category: "quota",
    repoId: null,
    accountId: null,
    access: "account",
    recipients: admins.map((row) => row.account_id),
    title: `Hosting object GC paused: ${count} retained deployments have no readable manifest`,
    details: ["Nothing was deleted. The run retries hourly; see the worker logs for the deployment ids."],
    url: publicUrl("/admin"),
    dedupeKey: `pages-gc-abort:${new Date(now).toISOString().slice(0, 10)}`,
  }).catch((error: unknown) => console.error("GC abort notification failed", error));
}

/** Cron: advances the collector by a few bounded steps under a D1 lease, so overlapping ticks never interleave. */
export async function collectHostingObjects(now = Date.now()): Promise<void> {
  const holder = crypto.randomUUID();
  if (!(await acquire(holder, now))) return;
  try {
    await advance(now);
  } finally {
    await run("UPDATE cron_locks SET holder = NULL WHERE name = ? AND holder = ?", LOCK, holder);
  }
}

async function advance(now: number): Promise<void> {
  let state = await loadState();
  for (let index = 0; index < STEPS_PER_TICK; index += 1) {
    const next = await step(state, now);
    if (next === state) break;
    state = next;
    await saveState(state);
    if (state.phase === "idle") break;
  }
}