import { env } from "cloudflare:workers";
import { batch, first, run, stmt } from "../db";
import { randomId } from "../crypto";
import { CAPS, enforceRateLimit, numberVar } from "../limits";
import { HttpError } from "../errors";
import type { JobTokenAccess } from "./job-token";

const DEFAULT_CACHE_BYTES_PER_REPO = 2 * 1024 * 1024 * 1024;
const MAX_ENTRY_BYTES = 1024 * 1024 * 1024;
const MAX_KEY_LENGTH = 512;
const MAX_RESTORE_KEYS = 10;
const EVICTION_BATCH = 100;
const IDLE_EVICTION_MS = 7 * 24 * 60 * 60 * 1000;
const VERSION = /^[0-9a-f]{1,64}$/;

type CacheRow = {
  id: string;
  repo_id: string;
  ref: string;
  key: string;
  version: string;
  size: number;
  r2_key: string;
  created_at: number;
  last_used_at: number;
};

type CacheScope = { repoId: string; ref: string; defaultRef: string };

export function cacheQuotaBytes(): number {
  return numberVar(env.CACHE_BYTES_PER_REPO, DEFAULT_CACHE_BYTES_PER_REPO);
}

function json(data: unknown, status: number): Response {
  return Response.json(data, { status });
}

async function scopeFor(access: JobTokenAccess): Promise<CacheScope | null> {
  const row = await first<{ ref: string; default_branch: string }>(
    `SELECT w.ref, r.default_branch FROM workflow_jobs j JOIN workflow_runs w ON w.id = j.run_id
     JOIN repos r ON r.id = w.repo_id WHERE j.id = ?`,
    access.jobId,
  );
  return row ? { repoId: access.repoId, ref: row.ref, defaultRef: `refs/heads/${row.default_branch}` } : null;
}

/**
 * Like GitHub, a run can restore caches saved from its own ref or the default branch, but only saves to its
 * own ref, so pull requests (including forks) can't poison caches used by the default branch.
 */
async function findEntry(scope: CacheScope, key: string, version: string, restore: string[]) {
  const order = "ORDER BY (ref = ?) DESC, created_at DESC LIMIT 1";
  const exact = await first<CacheRow>(
    `SELECT * FROM action_caches WHERE repo_id = ? AND key = ? AND version = ? AND ref IN (?, ?) AND size > 0 ${order}`,
    scope.repoId,
    key,
    version,
    scope.ref,
    scope.defaultRef,
    scope.ref,
  );
  if (exact) return { row: exact, hit: "exact" as const };
  for (const prefix of restore) {
    const row = await first<CacheRow>(
      `SELECT * FROM action_caches WHERE repo_id = ? AND substr(key, 1, length(?)) = ? AND version = ? AND ref IN (?, ?)
         AND size > 0 ${order}`,
      scope.repoId,
      prefix,
      prefix,
      version,
      scope.ref,
      scope.defaultRef,
      scope.ref,
    );
    if (row) return { row, hit: "partial" as const };
  }
  return null;
}

async function restore(scope: CacheScope, url: URL): Promise<Response> {
  const key = url.searchParams.get("key") ?? "";
  const version = url.searchParams.get("version") ?? "";
  const restoreKeys = url.searchParams.getAll("restore").slice(0, MAX_RESTORE_KEYS);
  const found = await findEntry(scope, key, version, restoreKeys);
  if (!found) return json({ error: "cache miss" }, 404);
  const object = await env.BUCKET.get(found.row.r2_key);
  if (!object) {
    await run("DELETE FROM action_caches WHERE id = ?", found.row.id);
    return json({ error: "cache miss" }, 404);
  }
  await run("UPDATE action_caches SET last_used_at = ? WHERE id = ?", Date.now(), found.row.id);
  return new Response(object.body, {
    headers: {
      "content-type": "application/gzip",
      "content-length": String(object.size),
      "x-stig-cache-hit": found.hit,
      "x-stig-cache-key": encodeURIComponent(found.row.key),
    },
  });
}

async function deleteEntries(rows: Array<Pick<CacheRow, "id" | "r2_key">>): Promise<void> {
  if (rows.length === 0) return;
  await env.BUCKET.delete(rows.map((row) => row.r2_key));
  await batch(rows.map((row) => stmt("DELETE FROM action_caches WHERE id = ?", row.id)));
}

/** Evicts least-recently-used entries until `incoming` more bytes fit in the repository's cache quota. */
async function makeRoom(repoId: string, incoming: number, keepId: string): Promise<void> {
  const quota = cacheQuotaBytes();
  const usage = await first<{ bytes: number | null; count: number }>(
    "SELECT SUM(ABS(size)) AS bytes, COUNT(*) AS count FROM action_caches WHERE repo_id = ? AND id != ?",
    repoId,
    keepId,
  );
  let bytes = usage?.bytes ?? 0;
  let count = usage?.count ?? 0;
  if (bytes + incoming <= quota && count < CAPS.cacheEntriesPerRepo) return;
  const candidates = await stmt(
    "SELECT id, r2_key, ABS(size) AS size FROM action_caches WHERE repo_id = ? AND id != ? AND size > 0 ORDER BY last_used_at LIMIT ?",
    repoId,
    keepId,
    EVICTION_BATCH,
  ).all<{ id: string; r2_key: string; size: number }>();
  const evict: Array<{ id: string; r2_key: string }> = [];
  for (const row of candidates.results) {
    if (bytes + incoming <= quota && count < CAPS.cacheEntriesPerRepo) break;
    evict.push(row);
    bytes -= row.size;
    count--;
  }
  await deleteEntries(evict);
}

async function save(scope: CacheScope, request: Request, url: URL): Promise<Response> {
  const key = url.searchParams.get("key") ?? "";
  const version = url.searchParams.get("version") ?? "";
  if (!key || key.length > MAX_KEY_LENGTH || key.includes(",")) return json({ error: "invalid cache key" }, 400);
  if (!VERSION.test(version)) return json({ error: "invalid cache version" }, 400);
  const length = Number(request.headers.get("content-length") ?? "NaN");
  if (!request.body || !Number.isFinite(length) || length <= 0) return json({ error: "content-length required" }, 411);
  if (length > Math.min(MAX_ENTRY_BYTES, cacheQuotaBytes())) return json({ error: "cache entry too large" }, 413);
  const id = randomId();
  const now = Date.now();
  const reserved = await run(
    `INSERT OR IGNORE INTO action_caches (id, repo_id, ref, key, version, size, r2_key, created_at, last_used_at)
     VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    id,
    scope.repoId,
    scope.ref,
    key,
    version,
    -length,
    `cache/${scope.repoId}/${id}.tgz`,
    now,
    now,
  );
  if (reserved.meta.changes !== 1) return json({ error: "cache entry already exists" }, 409);
  try {
    await makeRoom(scope.repoId, length, id);
    const stored = await env.BUCKET.put(`cache/${scope.repoId}/${id}.tgz`, request.body);
    if (!stored || stored.size !== length) throw new Error("cache upload size does not match content-length");
    await run("UPDATE action_caches SET size = ? WHERE id = ?", length, id);
  } catch (error) {
    await deleteEntries([{ id, r2_key: `cache/${scope.repoId}/${id}.tgz` }]);
    throw error;
  }
  return json({ ok: true }, 201);
}

export async function handleCacheApi(request: Request, url: URL, access: JobTokenAccess): Promise<Response> {
  try {
    await enforceRateLimit("HEAVY_LIMITER", `cache:${access.jobId}`);
  } catch (error) {
    if (error instanceof HttpError) return json({ error: error.message }, error.status);
    throw error;
  }
  const scope = await scopeFor(access);
  if (!scope) return json({ error: "job not found" }, 404);
  if (request.method === "GET") return restore(scope, url);
  if (request.method === "PUT") return save(scope, request, url);
  return json({ error: "method not allowed" }, 405);
}

export async function deleteCacheEntry(repoId: string, id: string): Promise<void> {
  const row = await first<CacheRow>("SELECT * FROM action_caches WHERE id = ? AND repo_id = ?", id, repoId);
  if (row) await deleteEntries([row]);
}

/** Cron: drops entries unused for a week (GitHub's policy) and abandoned upload reservations (negative size). */
export async function evictIdleCaches(): Promise<void> {
  const now = Date.now();
  const rows = await stmt(
    `SELECT id, r2_key FROM action_caches WHERE last_used_at < ? OR (size < 0 AND created_at < ?) LIMIT ?`,
    now - IDLE_EVICTION_MS,
    now - 60 * 60 * 1000,
    EVICTION_BATCH,
  ).all<{ id: string; r2_key: string }>();
  await deleteEntries(rows.results);
}