import { env } from "cloudflare:workers";
import { repoDb } from "../scoped-db";
import { pullBaseRepoIds } from "../pull-scope";
import { notFoundError } from "../errors";
import { loadRepoById } from "../repos";
import { canBridge, canBridgeBase, isPullHeadRepo } from "./bridge-policy";
import { reachableViaFetch } from "./commit-graph";
import { listRefs } from "./refs";
import type { ResolvedRevision } from "./refs";
import { withStore } from "./store";
import type { GitStore } from "./store";
import type { CommitSummary, RepoRow } from "~/lib/types";

const MAX_COMMIT_READS = 3000;
const MAX_LOG_PAGES = 10;
const LOG_PAGE = 1000;
const MAX_TIPS = 50;
const MAX_PULL_CANDIDATES = 5;
/** Commits are only reachable from descendants, which are (nearly always) committed later. Allow skew. */
const CLOCK_SKEW_MS = 24 * 60 * 60 * 1000;
const REACHABLE_TTL_SECONDS = 60 * 60 * 24;
const UNREACHABLE_TTL_SECONDS = 5 * 60;

type ReachRepo = Pick<RepoRow, "id" | "artifacts_name" | "default_branch" | "artifacts_remote">;

/** First-parent history of the default branch, paged server-side by Artifacts: cheap for most links. */
async function onDefaultBranch(store: GitStore, branch: string, target: CommitSummary): Promise<boolean> {
  for (let page = 0; page < MAX_LOG_PAGES; page++) {
    const commits = await store.log(branch, LOG_PAGE, page * LOG_PAGE);
    if (commits.some((commit) => commit.sha === target.sha)) return true;
    const oldest = commits.at(-1);
    if (commits.length < LOG_PAGE || !oldest || oldest.committedAt < target.committedAt - CLOCK_SKEW_MS) return false;
  }
  return false;
}

/** Date-pruned walk from every visible tip: stops descending once commits are older than the target. */
async function fromTips(store: GitStore, tips: string[], target: CommitSummary): Promise<boolean> {
  const cutoff = target.committedAt - CLOCK_SKEW_MS;
  const seen = new Set<string>();
  const queue = [...tips];
  let reads = 0;
  for (let index = 0; index < queue.length; index++) {
    const sha = queue[index];
    if (sha === target.sha) return true;
    if (seen.has(sha)) continue;
    seen.add(sha);
    if (++reads > MAX_COMMIT_READS) return false;
    const commit = await store.commit(sha);
    if (!commit || commit.committedAt < cutoff) continue;
    queue.push(...commit.parents);
  }
  return false;
}

type PullCandidate = { repo_id: string; head_repo_id: string };

/**
 * Whether `sha` is a recorded commit of an OPEN pull request that may currently flow into `repoId`: for a
 * base repo the fork head must still be allowed to bridge into it; for a head repo the base must be
 * public (only public base tips are ever bridged into forks). Closed pull requests and partners whose
 * visibility changed no longer vouch for swept objects, even while Artifacts still holds them.
 */
async function knownPullCommit(repo: ReachRepo, sha: string): Promise<boolean> {
  // Pull requests based here live in this repo's object; those headed here from a fork live in its parent's.
  const lists = await Promise.all(
    (await pullBaseRepoIds(repo.id)).map((baseId) =>
      repoDb(baseId).all<PullCandidate>(
        `SELECT p.repo_id, p.head_repo_id FROM pull_requests p JOIN issues i ON i.id = p.issue_id
         WHERE (p.repo_id = ?1 OR p.head_repo_id = ?1) AND i.state = 'open'
           AND (p.head_sha = ?2 OR p.base_sha = ?2 OR p.merge_base_sha = ?2 OR p.merge_commit_sha = ?2)
         LIMIT ?3`,
        repo.id,
        sha,
        MAX_PULL_CANDIDATES,
      ),
    ),
  );
  const rows = lists.flat().slice(0, MAX_PULL_CANDIDATES);
  for (const row of rows) {
    if (row.repo_id === row.head_repo_id) return true;
    const partnerId = row.repo_id === repo.id ? row.head_repo_id : row.repo_id;
    const [self, partner] = await Promise.all([loadRepoById(repo.id), loadRepoById(partnerId)]);
    if (!self || !partner) continue;
    if (row.repo_id === repo.id) {
      if (isPullHeadRepo(self, partner) && canBridge(partner, self)) return true;
    } else if (canBridgeBase(partner)) {
      return true;
    }
  }
  return false;
}

function generationKey(repoId: string): string {
  return `reach:gen:${repoId}`;
}

/**
 * Invalidates every cached reachability answer for a repository. Cached positives can outlive the hidden
 * ref that justified them (visibility change, pull request close, force push cleanup), so the cache key
 * carries a per-repo generation that callers bump instead of enumerating keys.
 */
export async function forgetReachability(repoId: string): Promise<void> {
  await env.KV.put(generationKey(repoId), String(Date.now()));
}

/**
 * Whether `sha` is an ancestor of (or equal to) one of `tips`, bounded and failing closed. Used where a
 * caller-supplied SHA must belong to a specific line of history (a pull request's head or base) rather
 * than to anything the repository happens to hold.
 */
export async function isReachableFrom(store: GitStore, tips: string[], sha: string): Promise<boolean> {
  if (tips.includes(sha)) return true;
  const target = await store.commit(sha);
  if (!target) return false;
  const fetched = await reachableViaFetch(store.artifactsName, tips, sha);
  if (fetched !== null) return fetched;
  return fromTips(store, tips, target);
}

/**
 * Whether a commit SHA may be browsed: it must be reachable from a visible branch or tag or an open pull
 * request's head (or be a pull request's recorded head/base/merge commit), never only from hidden refs or
 * dangling after a force push.
 * Results are cached; walks are bounded and fail closed.
 */
export async function isReachableCommit(repo: ReachRepo, sha: string): Promise<boolean> {
  const refs = await listRefs(repo);
  // listRefs keeps each annotated tag's peeled `^{}` commit, so tags walk from commits, not tag objects.
  const tips = [...refs.branches, ...refs.tags].map((ref) => ref.sha);
  if (tips.includes(sha)) return true;
  const generation = (await env.KV.get(generationKey(repo.id))) ?? "0";
  const key = `reach:v3:${repo.id}:${generation}:${sha}`;
  const cached = await env.KV.get(key);
  if (cached !== null) return cached === "1";
  let reachable = await knownPullCommit(repo, sha);
  if (!reachable) {
    reachable = await withStore(repo.artifacts_name, async (store) => {
      const target = await store.commit(sha);
      if (!target) return false;
      if (await onDefaultBranch(store, repo.default_branch, target)) return true;
      const pullHeads = await repoDb(repo.id).all<{ head_sha: string }>(
        `SELECT p.head_sha FROM pull_requests p JOIN issues i ON i.id = p.issue_id
         WHERE p.repo_id = ? AND i.state = 'open' ORDER BY i.updated_at DESC LIMIT ?`,
        repo.id,
        MAX_TIPS,
      );
      const walkTips = [...new Set([...tips, ...pullHeads.map((pull) => pull.head_sha)])].slice(0, MAX_TIPS * 2);
      const fetched = await reachableViaFetch(repo.artifacts_name, walkTips, target.sha);
      if (fetched !== null) return fetched;
      return fromTips(store, walkTips, target);
    });
  }
  await env.KV.put(key, reachable ? "1" : "0", {
    expirationTtl: reachable ? REACHABLE_TTL_SECONDS : UNREACHABLE_TTL_SECONDS,
  });
  return reachable;
}

/** Throws not-found for SHA revisions that are not reachable from visible refs; branches and tags pass. */
export async function assertVisibleRevision(repo: ReachRepo, revision: ResolvedRevision): Promise<void> {
  if (revision.kind !== "commit") return;
  if (!(await isReachableCommit(repo, revision.sha))) throw notFoundError("Commit not found");
}