import { jobTokenAccess, jobTokenRepoPermission, markJobTokenPush } from "../ci/job-token";
import { writeRefusal } from "~/lib/email-verification";
import { enforceRateLimit, requestIp } from "../limits";
import { HttpError } from "../errors";
import { hasPermission, loadRepo, repoReadOnlyReason, repoTakenDown } from "../repos";
import { scopedRepoPermission, tokenCredential } from "../token-scope";
import {
  bypassesProtection,
  directUpdateViolation,
  forcePushViolation,
  loadProtections,
  protectionFor,
  tagProtectionFor,
  tagUpdateViolation,
} from "../protection";
import { findForcePushes } from "./force-push";
import type { BranchProtection } from "../protection";
import { triggerPush } from "../events/push-trigger";
import { isHiddenRef } from "./hidden-refs";
import {
  filterAdvertisement,
  filterLsRefs,
  inspectUploadRequest,
  MalformedUploadRequestError,
} from "./advertise-filter";
import { scanPushBody } from "../security/push-protection";
import { basicAuthHeader, forgetRepoToken, repoTokenSecret } from "./artifacts";
import { appliedCommands } from "./receive-report";
import { concatBytes, pktLine, readUntilFlush, reportStatusResponse } from "./pktline";
import { invalidateRefs } from "./refs";
import { ZERO_SHA } from "./store";
import { scratchGitPermission, scratchPushPolicy } from "../sites/scratch-git";
import { isScratchRepo } from "../sites/config";
import type { Permission, RepoRow, Viewer } from "~/lib/types";
import { randomId } from "../crypto";
import { countingBody, haveDetector, meterGitOp } from "../metering/git";
import { planBlockReason } from "../metering/quota";
import { captureServerEvent } from "../analytics";

const GIT_PATH = /^\/([^/]+)\/([^/]+?)(?:\.git)?\/(info\/refs|git-upload-pack|git-receive-pack)$/;
const MAX_COMMAND_BYTES = 1024 * 1024;
const MAX_PUSH_COMMANDS = 1000;
const MAX_TRIGGERED_REFS = 100;
const FORWARDED_REQUEST_HEADERS = ["content-type", "accept", "content-encoding", "git-protocol", "user-agent"];
const FORWARDED_RESPONSE_HEADERS = ["content-type", "cache-control", "expires", "pragma"];
const HTTP_UNAUTHORIZED = 401;
const HTTP_FORBIDDEN = 403;
const SHA_PATTERN = /^[0-9a-f]{40}$/;
/** git's `probe_rpc` for large pushes (http.postBuffer exceeded): a lone flush packet, sent before the real body. */
const PROBE_BODY = "0000";
const decoder = new TextDecoder();

type GitService = "git-upload-pack" | "git-receive-pack";
type GitRoute = { owner: string; repo: string; action: "info/refs" | GitService };
type Command = { oldSha: string; newSha: string; ref: string };

export function matchGitRoute(url: URL): GitRoute | null {
  const match = GIT_PATH.exec(url.pathname);
  if (!match) return null;
  const [, owner, repo, action] = match;
  if (action !== "info/refs" && action !== "git-upload-pack" && action !== "git-receive-pack") return null;
  return { owner, repo, action };
}

function challenge(message: string): Response {
  return new Response(`${message}\n`, {
    status: 401,
    headers: { "www-authenticate": 'Basic realm="gitfor"', "content-type": "text/plain" },
  });
}

/** A v0 upload-pack `ERR` packet, which git prints as "remote error: …" instead of a confusing protocol failure. */
function uploadPackError(message: string): Response {
  return new Response(pktLine(`ERR ${message}\n`), {
    headers: { "content-type": "application/x-git-upload-pack-result", "cache-control": "no-cache" },
  });
}

function textResponse(status: number, message: string): Response {
  return new Response(`${message}\n`, { status, headers: { "content-type": "text/plain" } });
}

function serviceFor(route: GitRoute, url: URL): GitService | null {
  if (route.action !== "info/refs") return route.action;
  const service = url.searchParams.get("service");
  return service === "git-upload-pack" || service === "git-receive-pack" ? service : null;
}

function parseCommands(lines: string[]): { commands: Command[]; capabilities: string[] } {
  const commands: Command[] = [];
  let capabilities: string[] = [];
  for (const [index, line] of lines.entries()) {
    const [commandPart, caps] = line.split("\0");
    if (index === 0 && caps) capabilities = caps.trim().split(" ");
    const [oldSha, newSha, ref] = commandPart.trim().split(" ");
    if (oldSha && newSha && ref && SHA_PATTERN.test(oldSha) && SHA_PATTERN.test(newSha)) {
      commands.push({ oldSha, newSha, ref });
    }
  }
  return { commands, capabilities };
}

/**
 * The rule decisions for a push, without I/O: immediate violations, updates that must be fast-forwards,
 * and whether an unverifiable fast-forward check may pass. The last is only true for admins when no
 * matching rule binds them (`enforce_admins`): a rule that explicitly covers admins must fail closed.
 */
export function protectionVerdicts(
  protections: BranchProtection[],
  permission: Permission,
  commands: Command[],
): { violations: Map<string, string>; needsFastForward: Command[]; allowUnverifiable: boolean } {
  const violations = new Map<string, string>();
  const needsFastForward: Command[] = [];
  let allowUnverifiable = permission === "admin";
  for (const command of commands) {
    const isTag = command.ref.startsWith("refs/tags/");
    if (!isTag && !command.ref.startsWith("refs/heads/")) continue;
    const protection = isTag
      ? tagProtectionFor(protections, command.ref)
      : protectionFor(protections, command.ref.slice("refs/heads/".length));
    const isDeletion = command.newSha === ZERO_SHA;
    const isUpdate = command.oldSha !== ZERO_SHA && !isDeletion;
    const violation = isTag
      ? tagUpdateViolation(protection, permission, isDeletion ? "delete" : isUpdate ? "move" : "create")
      : directUpdateViolation(protection, permission, isDeletion);
    if (violation) violations.set(command.ref, violation);
    else if (!isTag && isUpdate && forcePushViolation(protection, permission)) {
      needsFastForward.push(command);
      if (protection && !bypassesProtection(protection, permission)) allowUnverifiable = false;
    }
  }
  return { violations, needsFastForward, allowUnverifiable };
}

/** Branch-protection and hidden-ref checks for a receive-pack, shared with mirrors that write into a repo. */
export async function checkProtections(
  repo: RepoRow,
  permission: Permission,
  commands: Command[],
  body: ReadableStream<Uint8Array>,
  gzip: boolean,
): Promise<{ violations: Map<string, string>; body: ReadableStream<Uint8Array> }> {
  const hidden = new Map<string, string>();
  for (const command of commands) {
    if (isHiddenRef(command.ref)) hidden.set(command.ref, "refs/pull and refs/gitfor are managed by gitfor");
  }
  if (hidden.size > 0) return { violations: hidden, body };
  const protections = await loadProtections(repo.id);
  if (protections.length === 0) return { violations: hidden, body };
  const { violations, needsFastForward, allowUnverifiable } = protectionVerdicts(protections, permission, commands);
  if (violations.size > 0 || needsFastForward.length === 0) return { violations, body };
  const checked = await findForcePushes(repo, needsFastForward, body, gzip, allowUnverifiable);
  for (const [ref, reason] of checked.rejected) violations.set(ref, reason);
  return { violations, body: checked.body };
}

function rejectPush(
  commands: Command[],
  violations: Map<string, string>,
  capabilities: string[],
  message = "gitfor: push rejected by branch protection",
): Response {
  if (!capabilities.includes("report-status")) {
    return textResponse(403, [message, ...violations.values()].join("\n"));
  }
  const useSideband = capabilities.includes("side-band-64k") || capabilities.includes("side-band");
  const results = commands.map((command) => ({
    ref: command.ref,
    error: violations.get(command.ref) ?? "atomic push failed: another ref was rejected",
  }));
  return new Response(reportStatusResponse(results, useSideband, message), {
    headers: { "content-type": "application/x-git-receive-pack-result", "cache-control": "no-cache" },
  });
}

function observeReceivePack(
  response: Response,
  onComplete: (report: Uint8Array) => Promise<void>,
  ctx: ExecutionContext,
): Response {
  const chunks: Uint8Array[] = [];
  let finish: (() => void) | undefined;
  const done = new Promise<void>((resolve) => {
    finish = resolve;
  });
  const transform = new TransformStream<Uint8Array, Uint8Array>({
    transform(chunk, controller) {
      chunks.push(chunk);
      controller.enqueue(chunk);
    },
    flush() {
      finish?.();
    },
  });
  ctx.waitUntil(done.then(() => onComplete(concatBytes(chunks))));
  return new Response(response.body?.pipeThrough(transform) ?? null, {
    status: response.status,
    headers: copyHeaders(response.headers, FORWARDED_RESPONSE_HEADERS),
  });
}

function copyHeaders(source: Headers, names: string[]): Headers {
  const headers = new Headers();
  for (const name of names) {
    const value = source.get(name);
    if (value) headers.set(name, value);
  }
  return headers;
}

/** Exported for tests. */
export async function authorize(
  request: Request,
  repo: RepoRow | null,
  needed: Permission,
): Promise<{ viewer: Viewer | null; permission: Permission; jobToken?: boolean } | Response> {
  // Scratch push tokens and CI job tokens carry no viewer, so a hidden or taken-down repository refuses them outright.
  const hidden = repo !== null && (Boolean(repo.hidden_at) || repoTakenDown(repo));
  const scratch = hidden ? null : await scratchGitPermission(request, repo);
  if (scratch === "write") return { viewer: null, permission: scratch };
  if (scratch === "none") return challenge("Invalid or expired scratch push token");
  if (needed !== "read" && repo && isScratchRepo(repo)) {
    return challenge("Scratch repositories accept pushes only with their scratch push token");
  }
  const jobAccess = hidden ? null : await jobTokenAccess(request);
  if (jobAccess) {
    const permission = repo ? jobTokenRepoPermission(jobAccess, repo.id) : "none";
    if (hasPermission(permission, needed)) return { viewer: null, permission, jobToken: true };
    return permission === "none" ? textResponse(404, "Repository not found") : textResponse(403, "Permission denied");
  }
  const credential = (await tokenCredential(request)) ?? { viewer: null, grant: null };
  const { viewer } = credential;
  const authenticated = viewer !== null || credential.grant !== null;
  if (!repo) return authenticated ? textResponse(404, "Repository not found") : challenge("Authentication required");
  const permission = await scopedRepoPermission(repo, credential, "contents");
  if (hasPermission(permission, needed)) return { viewer, permission };
  if (!authenticated) return challenge("Authentication required");
  return permission === "none" ? textResponse(404, "Repository not found") : textResponse(403, "Permission denied");
}

async function rateLimited(key: string): Promise<Response | null> {
  try {
    await enforceRateLimit("GIT_LIMITER", key);
    return null;
  } catch (error) {
    if (error instanceof HttpError) return textResponse(error.status, error.message);
    throw error;
  }
}

export async function handleGitRequest(request: Request, route: GitRoute, ctx: ExecutionContext): Promise<Response> {
  const url = new URL(request.url);
  const service = serviceFor(route, url);
  if (!service) return textResponse(400, "Unsupported git service");
  const isInfoRefs = route.action === "info/refs";
  if (isInfoRefs ? request.method !== "GET" : request.method !== "POST") {
    return textResponse(405, "Method not allowed");
  }

  // GIT-P6: anonymous requests pay the per-IP limit before any lookups, and failed credentials pay it too, so
  // probing and guessing are throttled without charging shared CI/NAT IPs for valid signed-in traffic.
  const ipKey = `git:ip:${requestIp(request)}`;
  const credentialed = request.headers.has("authorization");
  if (!credentialed) {
    const limited = await rateLimited(ipKey);
    if (limited) return limited;
  }
  const repo = await loadRepo(route.owner, route.repo);
  const needed: Permission = service === "git-receive-pack" ? "write" : "read";
  const auth = await authorize(request, repo, needed);
  if (auth instanceof Response) return (credentialed && (await rateLimited(ipKey))) || auth;
  const limited = auth.viewer
    ? await rateLimited(`git:user:${auth.viewer.id}`)
    : credentialed
      ? await rateLimited(ipKey)
      : null;
  if (limited) return limited;
  if (!repo) return textResponse(404, "Repository not found");
  if (repo.status !== "ready") return textResponse(503, `Repository is ${repo.status}`);
  const readOnly = service === "git-receive-pack" ? (repoReadOnlyReason(repo) ?? writeRefusal(auth.viewer)) : null;
  if (readOnly) return textResponse(403, readOnly);

  let body: ReadableStream<Uint8Array> | null = request.body;
  let commands: Command[] = [];
  let probe = false;
  if (
    service === "git-receive-pack" &&
    !isInfoRefs &&
    request.headers.get("content-length") === String(PROBE_BODY.length)
  ) {
    const text = await request.text();
    if (text !== PROBE_BODY) return textResponse(400, "Malformed receive-pack request");
    probe = true;
    body = new Blob([text]).stream();
  }
  if (service === "git-receive-pack" && !isInfoRefs && request.body && !probe) {
    const [inspect, forward] = request.body.tee();
    body = forward;
    const encoding = (request.headers.get("content-encoding") ?? "").trim().toLowerCase();
    if (encoding !== "" && encoding !== "identity" && encoding !== "gzip") {
      await Promise.all([inspect.cancel(), forward.cancel()]);
      return textResponse(415, "Unsupported content-encoding");
    }
    const encoded = encoding === "gzip";
    let lines: string[];
    try {
      lines = await readUntilFlush(
        encoded ? inspect.pipeThrough(new DecompressionStream("gzip")) : inspect,
        MAX_COMMAND_BYTES,
      );
    } catch {
      await forward.cancel();
      return textResponse(400, "Malformed receive-pack request");
    }
    const parsed = parseCommands(lines);
    commands = parsed.commands;
    const commandLines = lines.filter((line) => !line.startsWith("shallow "));
    if (commands.length === 0 || commands.length !== commandLines.length) {
      await forward.cancel();
      return textResponse(400, "Malformed receive-pack request");
    }
    if (!parsed.capabilities.some((cap) => cap === "report-status" || cap === "report-status-v2")) {
      await forward.cancel();
      return textResponse(400, "Pushes must request report-status");
    }
    if (commands.length > MAX_PUSH_COMMANDS) {
      await forward.cancel();
      return textResponse(413, `Pushes may update at most ${MAX_PUSH_COMMANDS} refs at once`);
    }
    if (commands.some((command) => command.newSha !== ZERO_SHA)) {
      const blocked = await planBlockReason(repo.owner_id, "storage.artifacts");
      if (blocked) {
        await forward.cancel();
        const declined = new Map(commands.map((command) => [command.ref, "repository storage limit reached"]));
        return rejectPush(commands, declined, parsed.capabilities, `gitfor: ${blocked}; deleting refs still works`);
      }
    }
    const scratch = await scratchPushPolicy(
      repo,
      commands.map((command) => command.ref),
      forward,
    );
    if ("error" in scratch) {
      const declined = new Map(commands.map((command) => [command.ref, scratch.error]));
      return rejectPush(commands, declined, parsed.capabilities, `gitfor: ${scratch.error}`);
    }
    const checked = await checkProtections(repo, auth.permission, commands, scratch.body, encoded);
    if (checked.violations.size > 0) {
      await checked.body.cancel().catch(() => undefined);
      return rejectPush(commands, checked.violations, parsed.capabilities);
    }
    const expectsPack = commands.some((command) => command.newSha !== ZERO_SHA);
    const isAdmin = auth.permission === "admin";
    const scan = await scanPushBody(repo, auth.viewer?.id ?? null, checked.body, encoded, expectsPack, isAdmin);
    if (scan.rejection) {
      const declined = new Map(commands.map((command) => [command.ref, "push declined due to secrets"]));
      return rejectPush(commands, declined, parsed.capabilities, scan.rejection);
    }
    body = scan.body;
  }
  let uploadCommand: string | null = null;
  if (service === "git-upload-pack" && !isInfoRefs && request.body) {
    const encoding = (request.headers.get("content-encoding") ?? "").trim().toLowerCase();
    if (encoding !== "" && encoding !== "identity" && encoding !== "gzip") {
      return textResponse(415, "Unsupported content-encoding");
    }
    const [inspect, forward] = request.body.tee();
    body = forward;
    let hidden: boolean;
    try {
      let v0Deepen: boolean | undefined;
      ({ hidden, command: uploadCommand, v0Deepen } = await inspectUploadRequest(repo, inspect, encoding === "gzip"));
      if (v0Deepen) {
        await forward.cancel();
        return uploadPackError("shallow clones need git protocol v2 (git 2.18+; remove protocol.version=0)");
      }
    } catch (error) {
      await forward.cancel();
      if (error instanceof MalformedUploadRequestError) return textResponse(400, error.message);
      throw error;
    }
    if (hidden) {
      await forward.cancel();
      return textResponse(403, "Requested objects are not available");
    }
  }

  const secret = await repoTokenSecret(repo, service === "git-receive-pack" ? "write" : "read");
  const upstreamUrl = isInfoRefs
    ? `${repo.artifacts_remote}/info/refs?service=${service}`
    : `${repo.artifacts_remote}/${service}`;
  const headers = copyHeaders(request.headers, FORWARDED_REQUEST_HEADERS);
  headers.set("authorization", basicAuthHeader(secret));
  const meterOp = (op: "push" | "fetch" | "clone", bytes: number) =>
    ctx.waitUntil(
      meterGitOp({ accountId: repo.owner_id, repoId: repo.id, op, bytes, requestId: randomId() }).catch(
        (error: unknown) => console.warn("git metering failed", error),
      ),
    );
  let pushBytes: Promise<number> = Promise.resolve(0);
  const haves = haveDetector();
  // Git only gzips large negotiation bodies, which means `have` lines: treat those as fetches.
  const gzipRequest = (request.headers.get("content-encoding") ?? "").trim().toLowerCase() === "gzip";
  if (!isInfoRefs && service === "git-receive-pack") {
    pushBytes = new Promise((resolve) => {
      body = countingBody(body, resolve);
    });
  } else if (!isInfoRefs) {
    body = countingBody(
      body,
      () => undefined,
      (chunk) => (gzipRequest ? undefined : haves.push(chunk)),
    );
  }
  const upstream = await fetch(upstreamUrl, { method: request.method, headers, body: isInfoRefs ? null : body });
  if (upstream.status === HTTP_UNAUTHORIZED || upstream.status === HTTP_FORBIDDEN) {
    // A cached token Artifacts no longer accepts; the client's retry gets a freshly minted one.
    ctx.waitUntil(forgetRepoToken(repo.artifacts_name, service === "git-receive-pack" ? "write" : "read"));
  }

  if (uploadCommand === "ls-refs" && upstream.ok) {
    const filtered = filterLsRefs(new Uint8Array(await upstream.arrayBuffer()), repo.default_branch);
    return new Response(filtered, {
      status: upstream.status,
      headers: copyHeaders(upstream.headers, FORWARDED_RESPONSE_HEADERS),
    });
  }

  if (isInfoRefs && upstream.ok) {
    const filtered = filterAdvertisement(new Uint8Array(await upstream.arrayBuffer()), repo.default_branch);
    return new Response(filtered, {
      status: upstream.status,
      headers: copyHeaders(upstream.headers, FORWARDED_RESPONSE_HEADERS),
    });
  }

  if (service === "git-upload-pack" && !isInfoRefs && upstream.ok) {
    return new Response(
      countingBody(upstream.body, (bytes) => meterOp(gzipRequest || haves.seen ? "fetch" : "clone", bytes)),
      { status: upstream.status, headers: copyHeaders(upstream.headers, FORWARDED_RESPONSE_HEADERS) },
    );
  }
  if (service === "git-receive-pack" && !isInfoRefs && upstream.ok && commands.length > 0) {
    ctx.waitUntil(pushBytes.then((bytes) => meterOp("push", bytes)));
  }

  if (service !== "git-receive-pack" || isInfoRefs || !upstream.ok || commands.length === 0) {
    return new Response(upstream.body, {
      status: upstream.status,
      headers: copyHeaders(upstream.headers, FORWARDED_RESPONSE_HEADERS),
    });
  }

  const pusherId = auth.viewer?.id ?? null;
  const jobToken = auth.jobToken === true;
  return observeReceivePack(
    upstream,
    async (report) => {
      await invalidateRefs(repo.id);
      const applied = appliedCommands(report, commands);
      // Only commits a job token actually landed are exempt from triggering workflows; marking before the push was
      // validated would let a rejected push suppress CI for those SHAs when a human pushes them later (CI-A5).
      if (jobToken) {
        await markJobTokenPush(
          repo.id,
          applied.filter((command) => command.newSha !== ZERO_SHA).map((command) => command.newSha),
        );
      }
      if (applied.length > 0) {
        captureServerEvent(
          "push",
          pusherId,
          {
            refs: applied.length,
            deletes: applied.filter((command) => command.newSha === ZERO_SHA).length,
            job_token: jobToken,
            private_repo: repo.visibility !== "public",
          },
          { ownerAccountId: repo.owner_id },
        );
      }
      for (const command of applied.slice(0, MAX_TRIGGERED_REFS)) {
        await triggerPush({
          source: "proxy",
          repoId: repo.id,
          artifactsName: repo.artifacts_name,
          ref: command.ref,
          before: command.oldSha,
          after: command.newSha,
          pusherId,
        });
      }
      if (applied.length === 0 && !decoder.decode(report).includes("ng ")) {
        console.warn("receive-pack report had no ok lines", decoder.decode(report).slice(0, 200));
      }
    },
    ctx,
  );
}