import { env } from "cloudflare:workers";
import { all, batch, first, run, stmt } from "../../db";
import { toHex } from "../../crypto";
import { enforceRateLimit } from "../../limits";
import { recordUsage } from "../../metering/record";
import { parseTreeObject } from "../../git/snapshot";
import { isSafeTreePath } from "../../git/tree-path";
import { tarHeader, tarPadding, TAR_BLOCK } from "../../tar";
import type { PackObject, ParsedPack } from "../../git/pack";
import {
  ActionSourceError,
  fetchCommitObjects,
  GithubUnavailableError,
  isReachableFromRefs,
  resolveGithubRef,
} from "./github";

/**
 * Action archives are content-addressed by commit SHA and shared by every account: public code is identical
 * for everyone, so it is downloaded from GitHub once and kept in R2. The tarball's sha256 is recorded and
 * checked by runners, so neither GitHub nor a tampered R2 object can change what a run executes.
 */
const MAX_FILES = 20_000;
const MAX_TREE_VISITS = 20_000;
export const MAX_ARCHIVE_BYTES = 64 * 1024 * 1024;
const MAX_MANIFESTS = 100;
const REF_CACHE_SECONDS = 300;
/** Fallback when GitHub refuses: the last commit a tag resolved to, for up to a week. */
const LAST_REF_SECONDS = 7 * 24 * 60 * 60;
const NEGATIVE_PIN_CACHE_MS = 60 * 60 * 1000;
/** A pin that was reachable is re-checked weekly, so a SHA force-pushed off every ref stops being trusted. */
const POSITIVE_PIN_CACHE_MS = 7 * 24 * 60 * 60 * 1000;
const TOUCH_INTERVAL_MS = 24 * 60 * 60 * 1000;
const IDLE_MS = 30 * 24 * 60 * 60 * 1000;
const PRUNE_BATCH = 100;
const MANIFEST_NAMES = ["action.yml", "action.yaml"];
const FILE_MODES: Record<string, number> = { "100644": 0o644, "100755": 0o755 };
const TREE_LINE = /^tree ([0-9a-f]{40,64})$/m;
const COMMITTER_TIME = /^committer .* (\d+) [+-]\d{4}$/m;
const decoder = new TextDecoder();

export type ArchiveRow = {
  key: string;
  owner: string;
  repo: string;
  sha: string;
  r2_key: string;
  size: number;
  sha256: string;
  created_at: number;
  last_used_at: number;
};

export function archiveKey(owner: string, repo: string, sha: string): string {
  return `${owner.toLowerCase()}/${repo.toLowerCase()}@${sha}`;
}

type BuiltArchive = { tgz: Uint8Array<ArrayBuffer>; manifests: Array<{ path: string; source: string }> };

/**
 * Builds a deterministic .tgz of the commit's tree from verified objects only: every object is looked up by
 * the SHA its parent names, and parsePack computed each SHA from content, so the tree is exactly the commit.
 * Symlinks and submodules are left out (a symlink could point extraction outside the action directory).
 */
export async function buildArchive(parsed: ParsedPack, sha: string): Promise<BuiltArchive> {
  const objects = new Map<string, PackObject>(parsed.objects.map((object) => [object.sha, object]));
  const commit = objects.get(sha);
  if (!commit || commit.type !== "commit") throw new ActionSourceError(`GitHub did not return commit ${sha}`);
  const header = decoder.decode(commit.content.subarray(0, Math.min(commit.content.byteLength, 64 * 1024)));
  const rootSha = TREE_LINE.exec(header)?.[1];
  if (!rootSha) throw new ActionSourceError(`commit ${sha} has no tree`);
  const mtime = Number(COMMITTER_TIME.exec(header)?.[1] ?? "0");
  const chunks: BlobPart[] = [];
  const manifests = new Map<string, string>();
  let files = 0;
  let visits = 0;
  let bytes = TAR_BLOCK * 2;
  // Trees and blobs are shared by SHA, so a repeated subtree is parsed once and its blobs are never copied; the
  // visit and byte caps bound the tarball a hostile DAG of reused trees would otherwise expand into.
  const parsedTrees = new Map<string, ReturnType<typeof parseTreeObject>>();
  const entriesOf = (treeSha: string) => {
    if (++visits > MAX_TREE_VISITS)
      throw new ActionSourceError(`the action tree has more than ${MAX_TREE_VISITS} directory entries`);
    const cached = parsedTrees.get(treeSha);
    if (cached) return cached;
    const tree = objects.get(treeSha);
    if (!tree || tree.type !== "tree") throw new ActionSourceError(`tree ${treeSha} is missing from the download`);
    const entries = parseTreeObject(tree.content);
    parsedTrees.set(treeSha, entries);
    return entries;
  };
  const walk = (treeSha: string, prefix: string) => {
    for (const entry of entriesOf(treeSha)) {
      const path = prefix ? `${prefix}/${entry.name}` : entry.name;
      if (!isSafeTreePath(path)) continue;
      if (entry.type === "tree") {
        walk(entry.hash, path);
        continue;
      }
      const mode = FILE_MODES[entry.mode];
      if (mode === undefined) continue;
      const blob = objects.get(entry.hash);
      if (!blob || blob.type !== "blob") throw new ActionSourceError(`file ${path} is missing from the download`);
      if (++files > MAX_FILES) throw new ActionSourceError(`the action has more than ${MAX_FILES} files`);
      const fileHeader = tarHeader(path, blob.content.byteLength, mode, "0", "", mtime);
      if (!fileHeader) continue;
      const padding = tarPadding(blob.content.byteLength);
      bytes += fileHeader.byteLength + blob.content.byteLength + padding;
      if (bytes > MAX_ARCHIVE_BYTES) {
        throw new ActionSourceError(`the action is larger than ${MAX_ARCHIVE_BYTES / 1024 / 1024} MB`);
      }
      chunks.push(fileHeader, asBlobPart(blob.content));
      if (padding > 0) chunks.push(new Uint8Array(padding));
      const known = manifests.get(prefix);
      if (
        MANIFEST_NAMES.includes(entry.name) &&
        (known ? entry.name === "action.yml" : manifests.size < MAX_MANIFESTS)
      ) {
        manifests.set(prefix, decoder.decode(blob.content));
      }
    }
  };
  walk(rootSha, "");
  chunks.push(new Uint8Array(TAR_BLOCK * 2));
  const tar = new Blob(chunks).stream().pipeThrough(new CompressionStream("gzip"));
  return {
    tgz: new Uint8Array(await new Response(tar).arrayBuffer()),
    manifests: [...manifests].map(([path, source]) => ({ path, source })),
  };
}

function asBlobPart(bytes: Uint8Array): Uint8Array<ArrayBuffer> {
  return bytes.buffer instanceof ArrayBuffer
    ? new Uint8Array(bytes.buffer, bytes.byteOffset, bytes.byteLength)
    : new Uint8Array(bytes);
}

async function sha256Hex(bytes: Uint8Array<ArrayBuffer>): Promise<string> {
  return toHex(await crypto.subtle.digest("SHA-256", bytes));
}

export async function touch(row: ArchiveRow): Promise<void> {
  if (Date.now() - row.last_used_at < TOUCH_INTERVAL_MS) return;
  await run("UPDATE action_archives SET last_used_at = ? WHERE key = ?", Date.now(), row.key);
}

/** The stored archive for owner/repo@sha, downloading and storing it first when no run has used it yet. */
export async function ensureArchive(
  owner: string,
  repo: string,
  sha: string,
  billing: { accountId: string; repoId: string; runId: string },
): Promise<ArchiveRow> {
  const key = archiveKey(owner, repo, sha);
  const existing = await first<ArchiveRow>("SELECT * FROM action_archives WHERE key = ?", key);
  if (existing) {
    await touch(existing);
    return existing;
  }
  await enforceRateLimit("HEAVY_LIMITER", `action-fetch:${billing.accountId}`);
  const built = await buildArchive(await fetchCommitObjects(owner, repo, sha), sha);
  const digest = await sha256Hex(built.tgz);
  const r2Key = `actions/github.com/${key.replace("@", "/")}-${digest.slice(0, 16)}.tgz`;
  await env.BUCKET.put(r2Key, built.tgz, { httpMetadata: { contentType: "application/gzip" } });
  const now = Date.now();
  let inserted: D1Result;
  try {
    [inserted] = await batch([
      stmt(
        `INSERT OR IGNORE INTO action_archives (key, owner, repo, sha, r2_key, size, sha256, created_at, last_used_at)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)`,
        key,
        owner.toLowerCase(),
        repo.toLowerCase(),
        sha,
        r2Key,
        built.tgz.byteLength,
        digest,
        now,
        now,
      ),
      ...built.manifests.map((manifest) =>
        stmt(
          "INSERT OR IGNORE INTO action_manifests (archive_key, path, source) VALUES (?, ?, ?)",
          key,
          manifest.path,
          manifest.source,
        ),
      ),
    ]);
  } catch (error) {
    // Racing writers produce the same deterministic key; only drop the object if no row ended up using it.
    const owner = await first<{ r2_key: string }>("SELECT r2_key FROM action_archives WHERE key = ?", key);
    if (owner?.r2_key !== r2Key) await env.BUCKET.delete(r2Key);
    throw error;
  }
  const row = await first<ArchiveRow>("SELECT * FROM action_archives WHERE key = ?", key);
  if (!row) throw new Error(`action archive ${key} vanished after insert`);
  if (inserted.meta.changes === 0 && row.r2_key !== r2Key) {
    await env.BUCKET.delete(r2Key);
    return row;
  }
  await recordUsage({
    accountId: billing.accountId,
    repoId: billing.repoId,
    meter: "ops.action_fetches",
    quantity: 1,
    attributes: { action: key, run: billing.runId, bytes: built.tgz.byteLength },
    idempotencyKey: `action-fetch:${key}`,
  }).catch((error: unknown) => console.warn("action fetch metering failed", key, error));
  return row;
}

/** action.yml (or action.yaml) of `path` inside a stored archive. */
export async function archiveManifest(key: string, path: string): Promise<string | null> {
  const row = await first<{ source: string }>(
    "SELECT source FROM action_manifests WHERE archive_key = ? AND path = ?",
    key,
    path,
  );
  return row?.source ?? null;
}

/** A tag or branch resolved to its commit, cached briefly so a burst of runs lists refs once. */
export async function resolveActionRef(owner: string, repo: string, ref: string): Promise<string> {
  const cacheKey = `action-ref:${owner.toLowerCase()}/${repo.toLowerCase()}@${ref}`;
  const cached = await env.KV.get(cacheKey);
  if (cached) return cached;
  let sha: string;
  try {
    sha = await resolveGithubRef(owner, repo, ref);
  } catch (error) {
    // While GitHub is rate-limiting or down, the last known commit for this tag beats failing the run.
    const last = error instanceof GithubUnavailableError ? await env.KV.get(`last-${cacheKey}`) : null;
    if (last) return last;
    throw error;
  }
  await Promise.all([
    env.KV.put(cacheKey, sha, { expirationTtl: REF_CACHE_SECONDS }),
    env.KV.put(`last-${cacheKey}`, sha, { expirationTtl: LAST_REF_SECONDS }),
  ]);
  return sha;
}

/** Refuses a pinned SHA that no branch or tag of owner/repo reaches (see isReachableFromRefs). */
export async function verifyPin(owner: string, repo: string, sha: string): Promise<void> {
  const key = archiveKey(owner, repo, sha);
  const cached = await first<{ reachable: number; checked_at: number }>(
    "SELECT reachable, checked_at FROM action_pin_checks WHERE key = ?",
    key,
  );
  let reachable = cached?.reachable === 1;
  const age = cached ? Date.now() - cached.checked_at : Infinity;
  if (age > (reachable ? POSITIVE_PIN_CACHE_MS : NEGATIVE_PIN_CACHE_MS)) {
    reachable = await isReachableFromRefs(owner, repo, sha);
    await run(
      `INSERT INTO action_pin_checks (key, reachable, checked_at) VALUES (?, ?, ?)
       ON CONFLICT (key) DO UPDATE SET reachable = excluded.reachable, checked_at = excluded.checked_at`,
      key,
      reachable ? 1 : 0,
      Date.now(),
    );
  }
  if (!reachable) {
    throw new ActionSourceError(
      `${sha} is not on any branch or tag of github.com/${owner}/${repo}; it may come from a fork, so it was refused`,
    );
  }
}

/**
 * Forgets an archive row whose R2 object is gone, so the next ensureArchive downloads it again. Conditional on
 * the row still naming that object: another runner may already have rebuilt it under a new row.
 */
export async function forgetBrokenArchive(row: ArchiveRow): Promise<void> {
  const deleted = await first<{ key: string }>(
    "DELETE FROM action_archives WHERE key = ? AND r2_key = ? AND created_at = ? RETURNING key",
    row.key,
    row.r2_key,
    row.created_at,
  );
  if (deleted) await run("DELETE FROM action_manifests WHERE archive_key = ?", row.key);
}

/** Streams a stored archive to a runner. */
export async function readArchive(key: string): Promise<{ row: ArchiveRow; body: ReadableStream } | null> {
  const row = await first<ArchiveRow>("SELECT * FROM action_archives WHERE key = ?", key);
  if (!row) return null;
  const object = await env.BUCKET.get(row.r2_key);
  if (!object) return null;
  return { row, body: object.body };
}

/** Cron: drops archives no run has used for 30 days; the next run re-downloads and re-verifies them. */
export async function pruneIdleActionArchives(): Promise<void> {
  const cutoff = Date.now() - IDLE_MS;
  const idle = await all<{ key: string }>(
    "SELECT key FROM action_archives WHERE last_used_at < ? LIMIT ?",
    cutoff,
    PRUNE_BATCH,
  );
  for (const { key } of idle) {
    // Conditional on still being idle, so an archive a run just touched survives; R2 goes only for deleted rows.
    const deleted = await first<{ r2_key: string }>(
      "DELETE FROM action_archives WHERE key = ? AND last_used_at < ? RETURNING r2_key",
      key,
      cutoff,
    );
    if (!deleted) continue;
    await run("DELETE FROM action_manifests WHERE archive_key = ?", key);
    // Narrows (doesn't close) the race with a run rebuilding the same deterministic object; if it loses, the
    // serve path finds the object missing and rebuilds it (forgetBrokenArchive).
    const reused = await first<{ key: string }>("SELECT key FROM action_archives WHERE r2_key = ?", deleted.r2_key);
    if (!reused) await env.BUCKET.delete(deleted.r2_key);
  }
}