import { advertisement, artifactsSource, relayPack } from "./relay";
import { isHiddenRef, pullBaseRef, pullHeadRef } from "./hidden-refs";
import { canBridge, canBridgeBase, pullHeadAccessible } from "./bridge-policy";
import { branchSha, listRefs } from "./refs";
import { pushToArtifacts } from "./writer";
import { withStore, ZERO_SHA } from "./store";
import { forgetReachability } from "./reachable";
import { enforceRateLimit, CAPS } from "../limits";
import { run } from "../db";
import { repoDb } from "../scoped-db";
import { pullBaseRepoIds } from "../pull-scope";
import { loadRepoById } from "../repos";
import { dropPullAliasesForIssue } from "../sites/pr-aliases";
import {
  bridgeQuotaAvailable,
  recordBridgedBytes,
  releaseBridgedBytes,
  releaseBridgedBytesForRepo,
} from "./bridge-quota";
import type { BridgeCharge } from "./bridge-quota";
import type { RepoRow } from "~/lib/types";

const MAX_BRIDGE_BYTES = 64 * 1024 * 1024;
const DROP_ATTEMPTS = 3;

async function hasCommit(repo: RepoRow, sha: string): Promise<boolean> {
  return withStore(repo.artifacts_name, async (store) => (await store.commit(sha)) !== null);
}

/**
 * Ensures `sha` and its history exist in `to`, pinned by the hidden `ref` so Artifacts keeps them.
 * Objects are relayed Artifacts-to-Artifacts, never from user-supplied URLs; the hidden ref is filtered
 * out of push processing, so no CI, webhooks or indexing run for it. Returns false when refused.
 */
/**
 * Re-reads both repos right before writing, so a bridge that started before a visibility change cannot
 * pin objects under the old visibility after the change's cleanup ran.
 */
async function stillAllowed(from: RepoRow, to: RepoRow, extra: (from: RepoRow) => boolean): Promise<boolean> {
  const [freshFrom, freshTo] = await Promise.all([loadRepoById(from.id), loadRepoById(to.id)]);
  return Boolean(
    freshFrom &&
    freshTo &&
    freshFrom.visibility === from.visibility &&
    freshTo.visibility === to.visibility &&
    canBridge(freshFrom, freshTo) &&
    extra(freshFrom),
  );
}

async function bridgeCommit(
  from: RepoRow,
  to: RepoRow,
  sha: string,
  ref: string,
  charge: BridgeCharge,
  extra: (from: RepoRow) => boolean = () => true,
): Promise<boolean> {
  if (from.id === to.id) return true;
  if (!canBridge(from, to) || !extra(from)) return false;
  // Objects outlive the hidden ref until GC, so the fetch want-check must apply to refs/pull bridges too.
  if (isHiddenRef(ref)) await run("UPDATE repos SET bridged_into = 1 WHERE id = ? AND bridged_into = 0", to.id);
  const targetRefs = await advertisement(await artifactsSource(to), true);
  const current = targetRefs.find((candidate) => candidate.name === ref)?.sha ?? ZERO_SHA;
  if (current === sha) return true;
  if (await hasCommit(to, sha)) {
    if (!(await stillAllowed(from, to, extra))) return false;
    await pushToArtifacts(to, [{ ref, oldSha: current, newSha: sha }], []);
    return afterBridge(from, to);
  }
  if (!(await bridgeQuotaAvailable(charge))) {
    console.warn("bridge refused: head owner's bridged bytes are over the cap", charge);
    return false;
  }
  await enforceRateLimit("HEAVY_LIMITER", `bridge:from:${from.id}`);
  await enforceRateLimit("HEAVY_LIMITER", `bridge:${to.id}`);
  if (!(await stillAllowed(from, to, extra))) return false;
  const report = await relayPack({
    source: await artifactsSource(from),
    target: to,
    wants: [sha],
    haves: [...new Set(targetRefs.map((candidate) => candidate.sha))],
    commands: [{ name: ref, sha, oldSha: current }],
    maxBytes: MAX_BRIDGE_BYTES,
  });
  if (report.accepted > 0) await recordBridgedBytes(charge, report.bytes);
  return report.accepted > 0 && (await afterBridge(from, to));
}

/** If visibility changed while the relay ran, the new ref may predate the cleanup: queue another one. */
async function afterBridge(from: RepoRow, to: RepoRow): Promise<boolean> {
  const [freshFrom, freshTo] = await Promise.all([loadRepoById(from.id), loadRepoById(to.id)]);
  if (freshFrom?.visibility === from.visibility && freshTo?.visibility === to.visibility) return true;
  for (const repo of [from, to]) {
    await requestHiddenRefCleanup(repo.id);
    await dropHiddenRefsForRepo(repo.id);
  }
  return false;
}

type BridgePull = { number: number; head_sha: string; repo_id: string; base_ref: string };
type BridgeHeadPull = BridgePull & { author_id: string };

/**
 * Makes a fork pull request's head commit readable from the base repository. Refused unless the author
 * can still read both sides and the head is a fork of the base (`pullHeadAccessible`).
 */
export async function bridgePullHead(pull: BridgeHeadPull, baseRepo: RepoRow, headRepo: RepoRow): Promise<boolean> {
  try {
    if (!(await pullHeadAccessible(pull, baseRepo, headRepo))) return false;
    return await bridgeCommit(
      headRepo,
      baseRepo,
      pull.head_sha,
      pullHeadRef(pull.number),
      { headOwnerId: headRepo.owner_id, baseRepoId: baseRepo.id, number: pull.number },
      () => headRepo.fork_of_id === baseRepo.id && headRepo.detached !== 1,
    );
  } catch (error) {
    console.warn("pull head bridge failed", pull.number, error);
    return false;
  }
}

/**
 * Makes the current tip of the base branch readable from the fork, for "update branch". Refused unless
 * every head-repo reader may see it (see `canBridgeBase`) and `baseSha` is the live base tip.
 */
export async function bridgePullBase(
  pull: BridgePull,
  baseRepo: RepoRow,
  headRepo: RepoRow,
  baseSha: string,
): Promise<boolean> {
  try {
    if (!canBridgeBase(baseRepo)) return false;
    if (branchSha(await listRefs(baseRepo), pull.base_ref) !== baseSha) return false;
    return await bridgeCommit(
      baseRepo,
      headRepo,
      baseSha,
      pullBaseRef(baseRepo.id, pull.number),
      { headOwnerId: headRepo.owner_id, baseRepoId: baseRepo.id, number: pull.number },
      canBridgeBase,
    );
  } catch (error) {
    console.warn("pull base bridge failed", pull.number, error);
    return false;
  }
}

async function dropRefs(repo: RepoRow, matches: (ref: string) => boolean): Promise<void> {
  const refs = await advertisement(await artifactsSource(repo), true);
  const doomed = refs.filter((candidate) => matches(candidate.name));
  if (doomed.length === 0) return;
  await pushToArtifacts(
    repo,
    doomed.map((candidate) => ({ ref: candidate.name, oldSha: candidate.sha, newSha: ZERO_SHA })),
    [],
  );
  await forgetReachability(repo.id);
}

async function withRetries(label: string, task: () => Promise<void>): Promise<boolean> {
  for (let attempt = 1; attempt <= DROP_ATTEMPTS; attempt++) {
    try {
      await task();
      return true;
    } catch (error) {
      if (attempt === DROP_ATTEMPTS) console.warn(label, error);
    }
  }
  return false;
}

/** Removes the hidden refs of a closed fork pull request so its unmerged objects can be collected. */
async function dropPullRefs(pull: BridgePull, baseRepo: RepoRow, headRepo: RepoRow): Promise<boolean> {
  if (baseRepo.id === headRepo.id) return true;
  const headRef = pullHeadRef(pull.number);
  const baseRef = pullBaseRef(baseRepo.id, pull.number);
  const results = await Promise.all([
    withRetries("drop pull ref", () => dropRefs(baseRepo, (ref) => ref === headRef)),
    withRetries("drop pull ref", () => dropRefs(headRepo, (ref) => ref === baseRef)),
  ]);
  return results.every(Boolean);
}

/** Drops hidden refs and site preview aliases when a pull request closes; runs after the response, never failing the close. */
export async function dropPullRefsForIssue(repoId: string, issueId: string): Promise<void> {
  await dropPullAliasesForIssue(issueId).catch((error: unknown) =>
    console.warn("Pull preview alias cleanup failed", issueId, error),
  );
  await dropPullRefsChecked(repoId, issueId);
}

/** True when every hidden ref of the pull request is gone (or it never had any). */
async function dropPullRefsChecked(repoId: string, issueId: string): Promise<boolean> {
  const row = await repoDb(repoId).first<BridgePull & { head_repo_id: string }>(
    `SELECT i.number, p.head_sha, p.repo_id, p.head_repo_id, p.base_ref FROM pull_requests p
     JOIN issues i ON i.id = p.issue_id WHERE p.issue_id = ?`,
    issueId,
  );
  if (!row || row.repo_id === row.head_repo_id) return true;
  const [baseRepo, headRepo] = await Promise.all([loadRepoById(row.repo_id), loadRepoById(row.head_repo_id)]);
  if (!baseRepo || !headRepo) return true;
  const dropped = await dropPullRefs(row, baseRepo, headRepo);
  if (dropped) await releaseBridgedBytes(repoId, issueId);
  return dropped;
}

const SWEEP_PULLS_PER_RUN = 25;

/** SQL over `pull_requests p JOIN issues i`: closed fork pull requests whose refs were not swept since closing. */
export const UNSWEPT_CLOSED_PULLS = `i.state = 'closed' AND p.repo_id != p.head_repo_id
  AND (p.refs_swept_at IS NULL OR p.refs_swept_at < COALESCE(i.closed_at, 0))`;

/**
 * The close-time drop is best effort, so the repository's alarm (repo-state/housekeeping.ts) retries closed fork
 * pull requests whose refs were not swept since they last closed. Reopened pull requests re-bridge lazily.
 */
export async function sweepClosedPullRefsForRepo(repoId: string): Promise<void> {
  const db = repoDb(repoId);
  const pulls = await db.all<{ issue_id: string }>(
    `SELECT p.issue_id FROM pull_requests p JOIN issues i ON i.id = p.issue_id
     WHERE ${UNSWEPT_CLOSED_PULLS} ORDER BY i.closed_at LIMIT ?`,
    SWEEP_PULLS_PER_RUN,
  );
  for (const pull of pulls) {
    if (await dropPullRefsChecked(repoId, pull.issue_id)) {
      await db.run("UPDATE pull_requests SET refs_swept_at = ? WHERE issue_id = ?", Date.now(), pull.issue_id);
    }
  }
}

type CleanupProgress = { ok: boolean; cursor: string | null };

/** Partner repositories of `repoId`'s fork pull requests after `cursor`, in id order, one page at a time. */
async function partnerPage(repoId: string, cursor: string | null): Promise<string[]> {
  const heads = await repoDb(repoId).all<{ partner: string }>(
    `SELECT DISTINCT head_repo_id AS partner FROM pull_requests
     WHERE head_repo_id != ?1 AND head_repo_id > ?2 ORDER BY partner LIMIT ?3`,
    repoId,
    cursor ?? "",
    CAPS.hiddenRefPartnersPerAttempt,
  );
  // As a fork, the only other base its pull requests can target is its parent.
  const parents = (await pullBaseRepoIds(repoId)).filter((id) => id !== repoId && id > (cursor ?? ""));
  const bases = await Promise.all(
    parents.map(async (parentId) =>
      (await repoDb(parentId).first("SELECT 1 FROM pull_requests WHERE head_repo_id = ? LIMIT 1", repoId))
        ? [parentId]
        : [],
    ),
  );
  return [...new Set([...heads.map((row) => row.partner), ...bases.flat()])]
    .sort()
    .slice(0, CAPS.hiddenRefPartnersPerAttempt);
}

/** The hidden refs `partnerId` holds for pull requests shared with `repoId`. */
async function partnerRefs(repoId: string, partnerId: string): Promise<Set<string>> {
  const select = `SELECT i.number, p.repo_id FROM pull_requests p JOIN issues i ON i.id = p.issue_id
     WHERE p.repo_id = ?1 AND p.head_repo_id = ?2`;
  const pulls = (
    await Promise.all([
      repoDb(repoId).all<{ number: number; repo_id: string }>(select, repoId, partnerId),
      repoDb(partnerId).all<{ number: number; repo_id: string }>(select, partnerId, repoId),
    ])
  ).flat();
  return new Set(
    pulls.map((pull) => (pull.repo_id === repoId ? pullBaseRef(repoId, pull.number) : pullHeadRef(pull.number))),
  );
}

/**
 * After a visibility change, objects bridged under the old visibility may now be readable by the wrong
 * people. Drops every hidden ref in the repository and every hidden ref other repos hold for its pull
 * requests; open pull requests re-bridge lazily under the new visibility. Partners are processed in
 * bounded pages (`cursor` persists across alarm runs), so a repo with thousands of fork pull requests
 * cannot exhaust one invocation. Returns the next cursor, or null once every partner was handled.
 */
async function dropHiddenRefsNow(repoId: string, cursor: string | null): Promise<CleanupProgress> {
  const repo = await loadRepoById(repoId);
  if (!repo) return { ok: true, cursor: null };
  if (cursor === null && !(await withRetries("drop hidden refs", () => dropRefs(repo, isHiddenRef)))) {
    return { ok: false, cursor: null };
  }
  let ok = true;
  const partners = await partnerPage(repoId, cursor);
  for (const partnerId of partners) {
    const partner = await loadRepoById(partnerId);
    if (!partner) continue;
    const refs = await partnerRefs(repoId, partnerId);
    ok = (await withRetries("drop hidden refs", () => dropRefs(partner, (ref) => refs.has(ref)))) && ok;
  }
  const exhausted = partners.length < CAPS.hiddenRefPartnersPerAttempt;
  return { ok, cursor: ok && !exhausted ? (partners.at(-1) ?? null) : null };
}

/** Records the cleanup durably first, so a failed or interrupted drop is retried by the RepoState alarm. */
export async function requestHiddenRefCleanup(repoId: string): Promise<void> {
  await repoDb(repoId).run(
    `INSERT INTO hidden_ref_cleanups (repo_id, attempts, requested_at, next_attempt_at, cursor) VALUES (?, 0, ?, 0, NULL)
     ON CONFLICT (repo_id) DO UPDATE SET requested_at = excluded.requested_at, attempts = 0, next_attempt_at = 0, cursor = NULL`,
    repoId,
    Date.now(),
  );
}

export const MAX_CLEANUP_ATTEMPTS = 12;
const CLEANUP_BACKOFF_MS = 5 * 60 * 1000;
const MAX_BACKOFF_MS = 24 * 60 * 60 * 1000;

/**
 * Runs one bounded page of a repository's hidden-ref cleanup. A successful but incomplete page stores its
 * cursor and is picked up by the next alarm without counting as a failed attempt; a failed page backs
 * off exponentially.
 */
export async function dropHiddenRefsForRepo(repoId: string): Promise<void> {
  const db = repoDb(repoId);
  const row = await db.first<{ attempts: number; cursor: string | null }>(
    "SELECT attempts, cursor FROM hidden_ref_cleanups WHERE repo_id = ?",
    repoId,
  );
  const progress = await dropHiddenRefsNow(repoId, row?.cursor ?? null);
  if (progress.ok && progress.cursor === null) {
    await releaseBridgedBytesForRepo(repoId);
    await db.run("DELETE FROM hidden_ref_cleanups WHERE repo_id = ?", repoId);
    return;
  }
  if (progress.ok) {
    await db.run(
      "UPDATE hidden_ref_cleanups SET cursor = ?, next_attempt_at = ? WHERE repo_id = ?",
      progress.cursor,
      Date.now(),
      repoId,
    );
    return;
  }
  const attempts = (row?.attempts ?? 0) + 1;
  const delay = Math.min(CLEANUP_BACKOFF_MS * 2 ** attempts, MAX_BACKOFF_MS);
  await db.run(
    "UPDATE hidden_ref_cleanups SET attempts = ?, next_attempt_at = ? WHERE repo_id = ?",
    attempts,
    Date.now() + delay,
    repoId,
  );
  if (attempts >= MAX_CLEANUP_ATTEMPTS) {
    console.error("hidden ref cleanup gave up; drop refs/gitfor and refs/pull manually", { repoId, attempts });
  }
}