import { env } from "cloudflare:workers";
import { chunks, first, placeholders } from "../db";
import { repoDb, stmt } from "../scoped-db";
import { randomId } from "../crypto";
import { CAPS, enforceRateLimit, numberVar } from "../limits";
import { HttpError } from "../errors";
import type { JobTokenAccess } from "./job-token";
import { meterCacheFreed, meterCacheStorage } from "./usage-hooks";

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;
export const IDLE_EVICTION_MS = 7 * 24 * 60 * 60 * 1000;
export const ABANDONED_RESERVATION_MS = 60 * 60 * 1000;
const VERSION = /^[0-9a-f]{1,64}$/;
const SHA256 = /^[0-9a-f]{64}$/;
const SCOPE_HEX_LENGTH = 32;
export const CONTENT_SHA256_HEADER = "x-gitfor-content-sha256";

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;
};

/** `readRefs`: refs a restore may read from; untrusted (fork) runs only see caches their own ref saved (CI-A10). */
type CacheScope = { repoId: string; ref: string; defaultRef: string; readRefs: 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 run = await repoDb(access.repoId).first<{
    ref: string;
    untrusted: number;
    source_repo_id: string;
    repo_id: string;
  }>(
    `SELECT w.ref, w.untrusted, w.source_repo_id, w.repo_id FROM workflow_jobs j
     JOIN workflow_runs w ON w.id = j.run_id WHERE j.id = ?`,
    access.jobId,
  );
  const repo = run
    ? await first<{ default_branch: string }>("SELECT default_branch FROM repos WHERE id = ?", run.repo_id)
    : null;
  if (!run || !repo) return null;
  const row = { ...run, default_branch: repo.default_branch };
  const defaultRef = `refs/heads/${row.default_branch}`;
  const fork = row.untrusted === 1 || row.source_repo_id !== access.repoId;
  return { repoId: access.repoId, ref: row.ref, defaultRef, readRefs: fork ? [row.ref] : [row.ref, defaultRef] };
}

/**
 * Like GitHub, a trusted run can restore caches saved from its own ref or the default branch, but only saves to
 * its own ref, so pull requests can't poison caches used by the default branch. Unlike GitHub, fork (untrusted)
 * runs are also denied reads of default-branch caches: those are written by trusted runs and often contain files
 * such as `~/.npmrc` that a fork must never see (CI-A10).
 */
async function findEntry(scope: CacheScope, key: string, version: string, restore: string[]) {
  const refs = placeholders(scope.readRefs.length);
  const order = "ORDER BY (ref = ?) DESC, created_at DESC LIMIT 1";
  const db = repoDb(scope.repoId);
  const exact = await db.first<CacheRow>(
    `SELECT * FROM action_caches WHERE repo_id = ? AND key = ? AND version = ? AND ref IN (${refs}) AND size > 0 ${order}`,
    scope.repoId,
    key,
    version,
    ...scope.readRefs,
    scope.ref,
  );
  if (exact) return { row: exact, hit: "exact" as const };
  for (const prefix of restore) {
    const row = await db.first<CacheRow>(
      `SELECT * FROM action_caches WHERE repo_id = ? AND substr(key, 1, length(?)) = ? AND version = ? AND ref IN (${refs})
         AND size > 0 ${order}`,
      scope.repoId,
      prefix,
      prefix,
      version,
      ...scope.readRefs,
      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 repoDb(scope.repoId).run("DELETE FROM action_caches WHERE id = ?", found.row.id);
    return json({ error: "cache miss" }, 404);
  }
  await repoDb(scope.repoId).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-gitfor-cache-hit": found.hit,
      "x-gitfor-cache-key": encodeURIComponent(found.row.key),
    },
  });
}

/** Deletes entries, and their R2 objects once no other entry shares the same content-addressed object. */
async function deleteEntries(repoId: string, rows: Array<Pick<CacheRow, "id" | "r2_key">>): Promise<void> {
  if (rows.length === 0) return;
  const db = repoDb(repoId);
  const sized: Array<{ repo_id: string; size: number }> = [];
  for (const group of chunks(rows.map((row) => row.id))) {
    sized.push(
      ...(await db.all<{ repo_id: string; size: number }>(
        `SELECT repo_id, size FROM action_caches WHERE size > 0 AND id IN (${placeholders(group.length)})`,
        ...group,
      )),
    );
  }
  await db.batch(rows.map((row) => stmt("DELETE FROM action_caches WHERE id = ?", row.id)));
  for (const row of sized) await meterCacheFreed(row.repo_id, row.size);
  const keys = [...new Set(rows.map((row) => row.r2_key))];
  const orphaned: string[] = [];
  for (const key of keys) {
    const shared = await db.first<{ ok: number }>("SELECT 1 AS ok FROM action_caches WHERE r2_key = ? LIMIT 1", key);
    if (!shared) orphaned.push(key);
  }
  if (orphaned.length > 0) await env.BUCKET.delete(orphaned);
}

async function sha256Hex(text: string): Promise<string> {
  const digest = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(text));
  return [...new Uint8Array(digest)].map((byte) => byte.toString(16).padStart(2, "0")).join("");
}

/**
 * Objects live at `cache/<repo>/<scope>/<sha256(content)>`, where the scope is derived from the ref the entry was
 * saved from. A run only ever writes its own ref's scope, so a fork pull request can't place content in the
 * default branch's scope, and content addressing means a verified object can't be swapped for other bytes.
 */
async function objectKey(scope: CacheScope, contentSha256: string): Promise<string> {
  const scopeHash = (await sha256Hex(scope.ref)).slice(0, SCOPE_HEX_LENGTH);
  return `cache/${scope.repoId}/${scopeHash}/${contentSha256}`;
}

/**
 * Streams the body to R2 once, verifying it twice: R2 rejects the put atomically when the bytes don't match the
 * declared SHA-256 (so no unverified object is ever visible), and a DigestStream fed chunk by chunk on the way
 * (with backpressure, so nothing is buffered) confirms it.
 */
async function putVerified(key: string, body: ReadableStream, length: number, contentSha256: string): Promise<void> {
  // Workers expose DigestStream on `crypto`, not as a global.
  const digest = new (crypto as unknown as { DigestStream: typeof DigestStream }).DigestStream("SHA-256");
  const digestWriter = digest.getWriter();
  const hashing = new TransformStream<Uint8Array, Uint8Array>({
    async transform(chunk, controller) {
      await digestWriter.write(chunk);
      controller.enqueue(chunk);
    },
    async flush() {
      await digestWriter.close();
    },
  });
  let stored: R2Object | null;
  try {
    stored = await env.BUCKET.put(key, body.pipeThrough(hashing).pipeThrough(new FixedLengthStream(length)), {
      sha256: contentSha256,
    });
  } catch {
    throw new HttpError(400, "cache upload does not match its declared SHA-256 or content-length");
  }
  const actual = [...new Uint8Array(await digest.digest)].map((byte) => byte.toString(16).padStart(2, "0")).join("");
  if (!stored || stored.size !== length || actual !== contentSha256) {
    throw new HttpError(400, "cache upload does not match its declared SHA-256 or content-length");
  }
}

/** 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 db = repoDb(repoId);
  const usage = await db.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 db.all<{ id: string; r2_key: string; size: number }>(
    "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,
  );
  const evict: Array<{ id: string; r2_key: string }> = [];
  for (const row of candidates) {
    if (bytes + incoming <= quota && count < CAPS.cacheEntriesPerRepo) break;
    evict.push(row);
    bytes -= row.size;
    count--;
  }
  await deleteEntries(repoId, evict);
}

/**
 * Re-checks the quota after eviction counting every row, including this and other in-flight reservations (their
 * negative sizes), so concurrent saves or a too-small eviction batch can't push the repository over (CI-P7).
 */
export async function assertCacheFits(repoId: string): Promise<void> {
  const usage = await repoDb(repoId).first<{ bytes: number | null; count: number }>(
    "SELECT SUM(ABS(size)) AS bytes, COUNT(*) AS count FROM action_caches WHERE repo_id = ?",
    repoId,
  );
  if ((usage?.bytes ?? 0) > cacheQuotaBytes() || (usage?.count ?? 0) > CAPS.cacheEntriesPerRepo) {
    throw new HttpError(413, "cache quota exceeded for this repository");
  }
}

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 contentSha256 = (request.headers.get(CONTENT_SHA256_HEADER) ?? "").toLowerCase();
  if (!SHA256.test(contentSha256)) return json({ error: `${CONTENT_SHA256_HEADER} header required` }, 400);
  const id = randomId();
  const now = Date.now();
  const r2Key = await objectKey(scope, contentSha256);
  const reserved = await repoDb(scope.repoId).run(
    `INSERT OR IGNORE INTO action_caches (id, repo_id, ref, key, version, size, r2_key, created_at, last_used_at,
       content_sha256) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
    id,
    scope.repoId,
    scope.ref,
    key,
    version,
    -length,
    r2Key,
    now,
    now,
    contentSha256,
  );
  if (reserved.meta.changes !== 1) return json({ error: "cache entry already exists" }, 409);
  try {
    await makeRoom(scope.repoId, length, id);
    await assertCacheFits(scope.repoId);
    // Always write, even if the object exists: overwriting identical verified bytes is idempotent and avoids
    // racing a concurrent delete of a shared object.
    await putVerified(r2Key, request.body, length, contentSha256);
    await repoDb(scope.repoId).run("UPDATE action_caches SET size = ? WHERE id = ?", length, id);
  } catch (error) {
    await deleteEntries(scope.repoId, [{ id, r2_key: r2Key }]);
    if (error instanceof HttpError) return json({ error: error.message }, error.status);
    throw error;
  }
  await meterCacheStorage(scope.repoId, length);
  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 repoDb(repoId).first<CacheRow>(
    "SELECT * FROM action_caches WHERE id = ? AND repo_id = ?",
    id,
    repoId,
  );
  if (row) await deleteEntries(repoId, [row]);
}

/**
 * Drops entries unused for a week (GitHub's policy) and abandoned upload reservations (negative size). Run by the
 * RepoState alarm when the oldest entry comes due.
 */
export async function evictIdleCachesForRepo(repoId: string, now: number): Promise<boolean> {
  const rows = await repoDb(repoId).all<{ id: string; r2_key: string }>(
    `SELECT id, r2_key FROM action_caches WHERE last_used_at < ? OR (size < 0 AND created_at < ?) LIMIT ?`,
    now - IDLE_EVICTION_MS,
    now - ABANDONED_RESERVATION_MS,
    EVICTION_BATCH,
  );
  await deleteEntries(repoId, rows);
  return rows.length === EVICTION_BATCH;
}