import { env, waitUntil } from "cloudflare:workers";
import { dropPullRefsForIssue } from "./git/pull-bridge";
import { repoDb, stmt } from "./scoped-db";
import { locate } from "./scoped-db/locators";
import { pullBaseRepoIds } from "./pull-scope";
import { randomId } from "./crypto";
import { adjustOpenCounter } from "./issue-counters";
import { hasPermission, loadRepoById, repoPermission } from "./repos";
import { loadProtections, protectionFor, bypassesProtection } from "./protection";
import { listCommitStatuses, requiredCheckResults } from "./statuses";
import { authenticatedRemote, mintRepoToken, revokeRepoToken, revokeRepoTokens } from "./git/artifacts";
import { ZERO_SHA, withStore } from "./git/store";
import { isAncestor } from "./git/history";
import { triggerPush } from "./events/push-trigger";
import { inWorkerMergeability, landInWorker, loadPullRepos } from "./pull-merge";
import type { Landed } from "./pull-merge";
import { RebaseConflictError, rebaseInWorker } from "./pull-rebase";
import { actorViewer, afterMerged, codeOwnerBlockers, kickMergeAutomation } from "./merge-automation";
import { ensurePullHeadAccessible } from "./pull-guard";
import type { MergeMethod, Permission, RepoRow } from "~/lib/types";
import { issuesChanged } from "./repo-state";

const MERGE_TOKEN_TTL_SECONDS = 1800;
const MAX_EXTERNAL_MERGE_CHECKS = 10;
const EXIT_CONFLICT = 2;
const EXIT_HEAD_MOVED = 3;
const EXIT_BASE_MOVED = 6;

export type PullRow = {
  issue_id: string;
  repo_id: string;
  number: number;
  title: string;
  body: string;
  state: "open" | "closed";
  author_id: string;
  locked: number;
  head_repo_id: string;
  head_ref: string;
  head_sha: string;
  base_ref: string;
  base_sha: string;
  merge_base_sha: string | null;
  draft: number;
  mergeable: "unknown" | "clean" | "conflict";
  merged_at: number | null;
  merged_by: string | null;
  merge_commit_sha: string | null;
  merge_method: string | null;
  stacked_on_issue_id?: string | null;
  restack_upstream_sha?: string | null;
};

export const PULL_SELECT = `SELECT p.*, i.number, i.title, i.body, i.state, i.author_id, i.locked, i.repo_id
  FROM pull_requests p JOIN issues i ON i.id = p.issue_id`;

/** Like `loadIssue`, hidden pull requests are left out unless a site-admin read path asks for them. */
export async function loadPull(
  repoId: string,
  number: number,
  { includeHidden = false }: { includeHidden?: boolean } = {},
): Promise<PullRow | null> {
  return repoDb(repoId).first<PullRow>(
    `${PULL_SELECT} WHERE i.repo_id = ? AND i.number = ? AND i.is_pull = 1${includeHidden ? "" : " AND i.hidden_at IS NULL"}`,
    repoId,
    number,
  );
}

export async function loadPullByIssueId(repoId: string, issueId: string): Promise<PullRow | null> {
  return repoDb(repoId).first<PullRow>(`${PULL_SELECT} WHERE p.issue_id = ?`, issueId);
}

/** For callers that only hold an issue id (queue messages, stored references): finds its repository first. */
export async function locatePullByIssueId(issueId: string): Promise<PullRow | null> {
  const repoId = await locate("issue", issueId);
  return repoId ? loadPullByIssueId(repoId, issueId) : null;
}

/** An issue (`id`) or a pull request (`issue_id`), with the repository whose object holds its timeline. */
export type IssueEventTarget = { repo_id: string } & ({ id: string } | { issue_id: string });

export async function addIssueEvent(
  target: IssueEventTarget,
  actorId: string | null,
  kind: string,
  data: Record<string, unknown>,
) {
  await repoDb(target.repo_id).run(
    "INSERT INTO issue_events (id, issue_id, actor_id, kind, data, created_at) VALUES (?, ?, ?, ?, ?, ?)",
    randomId(),
    "issue_id" in target ? target.issue_id : target.id,
    actorId,
    kind,
    JSON.stringify(data),
    Date.now(),
  );
}

/** Open pull requests from any base repository whose head is `headRepoId`'s `headRef`. */
export async function openPullsWithHead(headRepoId: string, headRef: string): Promise<PullRow[]> {
  const baseIds = await pullBaseRepoIds(headRepoId);
  const lists = await Promise.all(
    baseIds.map((baseId) =>
      repoDb(baseId).all<PullRow>(
        `${PULL_SELECT} WHERE p.head_repo_id = ? AND p.head_ref = ? AND i.state = 'open'`,
        headRepoId,
        headRef,
      ),
    ),
  );
  return lists.flat();
}

export type SynchronizedPull = { pull: PullRow; before: string };

/** Applies a pushed branch update to every open pull request that uses it as head or base. */
export async function syncPullsForPush(
  repo: RepoRow,
  branch: string,
  after: string,
  pusherId: string | null,
): Promise<{ synchronized: SynchronizedPull[]; mergedExternally: PullRow[]; baseUpdated: PullRow[] }> {
  const now = Date.now();
  const heads = await openPullsWithHead(repo.id, branch);
  const synchronized: SynchronizedPull[] = [];
  const protections = await loadProtections(repo.id);
  for (const pull of heads) {
    if (after === ZERO_SHA) {
      await addIssueEvent(pull, pusherId, "head_ref_deleted", { ref: branch });
      continue;
    }
    if (pull.head_sha === after) continue;
    // A fork pull request whose author lost access is closed here instead of relaying the new head.
    if (!(await ensurePullHeadAccessible(pull))) continue;
    const scoped = repoDb(pull.repo_id);
    await scoped.batch([
      stmt("UPDATE pull_requests SET head_sha = ?, mergeable = 'unknown' WHERE issue_id = ?", after, pull.issue_id),
      stmt("UPDATE issues SET updated_at = ? WHERE id = ?", now, pull.issue_id),
    ]);
    await addIssueEvent(pull, pusherId, "head_pushed", { before: pull.head_sha, after });
    const baseRepoProtections = pull.repo_id === repo.id ? protections : await loadProtections(pull.repo_id);
    if (protectionFor(baseRepoProtections, pull.base_ref)?.dismissStaleReviews) {
      await scoped.run(
        "UPDATE reviews SET state = 'dismissed' WHERE issue_id = ? AND state = 'approved' AND commit_sha IS NOT ?",
        pull.issue_id,
        after,
      );
    }
    synchronized.push({ pull: { ...pull, head_sha: after }, before: pull.head_sha });
  }

  const bases = await repoDb(repo.id).all<PullRow>(
    `${PULL_SELECT} WHERE i.repo_id = ? AND p.base_ref = ? AND i.state = 'open'`,
    repo.id,
    branch,
  );
  const mergedExternally: PullRow[] = [];
  const baseUpdated: PullRow[] = [];
  if (after !== ZERO_SHA) {
    await withStore(repo.artifacts_name, async (store) => {
      for (const [index, pull] of bases.entries()) {
        await repoDb(repo.id).run(
          `UPDATE pull_requests SET base_sha = ?, mergeable = 'unknown',
             restack_upstream_sha = COALESCE(restack_upstream_sha, base_sha) WHERE issue_id = ?`,
          after,
          pull.issue_id,
        );
        const merged =
          pull.head_repo_id === repo.id &&
          index < MAX_EXTERNAL_MERGE_CHECKS &&
          (pull.head_sha === after || (await isAncestor(store, pull.head_sha, after)));
        if (merged) {
          await markMerged(pull, pusherId, after, "push");
          mergedExternally.push(pull);
        } else {
          baseUpdated.push({ ...pull, base_sha: after });
        }
      }
    });
  }
  return { synchronized, mergedExternally, baseUpdated };
}

export async function markMerged(
  pull: PullRow,
  actorId: string | null,
  mergeSha: string,
  method: string,
): Promise<void> {
  const now = Date.now();
  const [, closed] = await repoDb(pull.repo_id).batch([
    stmt(
      `UPDATE pull_requests SET merged_at = ?, merged_by = ?, merge_commit_sha = ?, merge_method = ?, mergeable = 'clean'
       WHERE issue_id = ?`,
      now,
      actorId,
      mergeSha,
      method,
      pull.issue_id,
    ),
    stmt(
      "UPDATE issues SET state = 'closed', closed_at = ?, closed_by = ?, updated_at = ? WHERE id = ? AND state = 'open'",
      now,
      actorId,
      now,
      pull.issue_id,
    ),
  ]);
  if (closed && closed.meta.changes > 0) {
    await adjustOpenCounter({ id: pull.issue_id, repo_id: pull.repo_id, is_pull: 1 }, -1);
  }
  issuesChanged(pull.repo_id, [pull.issue_id]);
  await addIssueEvent(pull, actorId, "merged", { sha: mergeSha, method });
  await afterMerged(pull, actorId, mergeSha, method === "squash" || method === "rebase" ? method : "merge");
  waitUntil(dropPullRefsForIssue(pull.repo_id, pull.issue_id));
}

function gitTaskRunner(repoId: string) {
  return env.GIT_RUNNER.get(env.GIT_RUNNER.idFromName(`git:${repoId}`));
}

/** Runs `task` with authenticated base/head remotes whose dedicated tokens are revoked afterwards. */
async function withPullRemotes<T>(
  pull: PullRow,
  scope: "read" | "write",
  task: (remotes: { baseUrl: string; headUrl: string; baseRepo: RepoRow }) => Promise<T>,
): Promise<T> {
  const [baseRepo, headRepo] = await Promise.all([loadRepoById(pull.repo_id), loadRepoById(pull.head_repo_id)]);
  if (!baseRepo || !headRepo) throw new Error("Pull request repositories are missing");
  const baseToken = await mintRepoToken(baseRepo.artifacts_name, scope, MERGE_TOKEN_TTL_SECONDS);
  const headToken =
    headRepo.id === baseRepo.id
      ? baseToken
      : await mintRepoToken(headRepo.artifacts_name, "read", MERGE_TOKEN_TTL_SECONDS).catch(async (error: unknown) => {
          await revokeRepoToken(baseToken);
          throw error;
        });
  try {
    return await task({
      baseRepo,
      baseUrl: authenticatedRemote(baseRepo.artifacts_remote, baseToken.secret),
      headUrl: authenticatedRemote(headRepo.artifacts_remote, headToken.secret),
    });
  } finally {
    waitUntil(revokeRepoTokens([baseToken, headToken]));
  }
}

const FETCH_SCRIPT = `set -uo pipefail
work="$(mktemp -d)"
trap 'rm -rf "$work"' EXIT
cd "$work"
git init -q .
git config user.name "$MERGER_NAME"
git config user.email "$MERGER_EMAIL"
git fetch -q "$BASE_URL" "+refs/heads/$BASE_REF:refs/gitfor/base" || { echo "failed to fetch base" >&2; exit 4; }
git fetch -q "$HEAD_URL" "+refs/heads/$HEAD_REF:refs/gitfor/head" || { echo "failed to fetch head" >&2; exit 4; }
if [ "$(git rev-parse refs/gitfor/head)" != "$EXPECTED_HEAD" ]; then echo "head branch changed" >&2; exit 3; fi
echo "BEFORE $(git rev-parse refs/gitfor/base)"
`;

const MERGEABILITY_SCRIPT = `${FETCH_SCRIPT}
if git merge-tree --write-tree --name-only --no-messages refs/gitfor/base refs/gitfor/head > "$work/out"; then
  echo "CLEAN"
else
  echo "CONFLICT"
  tail -n +2 "$work/out"
fi
`;

const MERGE_SCRIPT = `${FETCH_SCRIPT}
git checkout -q -B work refs/gitfor/base
case "$MERGE_METHOD" in
  merge)
    git merge -q --no-ff -m "$MERGE_MESSAGE" refs/gitfor/head || { git merge --abort; echo "merge conflict" >&2; exit 2; }
    ;;
  squash)
    git merge -q --squash refs/gitfor/head || { echo "merge conflict" >&2; exit 2; }
    GIT_AUTHOR_NAME="$AUTHOR_NAME" GIT_AUTHOR_EMAIL="$AUTHOR_EMAIL" git commit -q --allow-empty -m "$MERGE_MESSAGE" || exit 5
    ;;
  rebase)
    git checkout -q -B rebasing refs/gitfor/head
    git rebase -q work || { git rebase --abort; echo "rebase conflict" >&2; exit 2; }
    git checkout -q work
    git merge -q --ff-only rebasing || exit 5
    ;;
esac
git push -q "$BASE_URL" "HEAD:refs/heads/$BASE_REF" || { echo "push rejected; base branch moved" >&2; exit 6; }
echo "AFTER $(git rev-parse HEAD)"
`;

async function containerMergeability(pull: PullRow): Promise<"clean" | "conflict" | "unknown"> {
  const result = await withPullRemotes(pull, "read", (remotes) =>
    gitTaskRunner(pull.repo_id).runGitTask(MERGEABILITY_SCRIPT, {
      MERGER_NAME: "gitfor",
      MERGER_EMAIL: "gitfor@localhost",
      BASE_URL: remotes.baseUrl,
      HEAD_URL: remotes.headUrl,
      BASE_REF: pull.base_ref,
      HEAD_REF: pull.head_ref,
      EXPECTED_HEAD: pull.head_sha,
    }),
  );
  return result.stdout.includes("CLEAN") ? "clean" : result.stdout.includes("CONFLICT") ? "conflict" : "unknown";
}

/** Decides mergeability with the in-Worker tree merge, using the git container only when the Worker cannot. */
export async function checkMergeability(pull: PullRow): Promise<"clean" | "conflict" | "unknown"> {
  if (!(await ensurePullHeadAccessible(pull))) return "unknown";
  const state = (await inWorkerMergeability(pull)) ?? (await containerMergeability(pull));
  if (state !== "unknown") {
    const updated = await repoDb(pull.repo_id).run(
      "UPDATE pull_requests SET mergeable = ? WHERE issue_id = ? AND head_sha = ?",
      state,
      pull.issue_id,
      pull.head_sha,
    );
    if (state === "clean" && updated.meta.changes > 0) await kickMergeAutomation(pull);
  }
  return state;
}

const MAX_GATE_REVIEWERS = 50;

/** Reviewers who currently hold write access; reviews from removed or downgraded users no longer count. */
async function reviewersWithWrite(repoId: string, userIds: string[]): Promise<Set<string>> {
  const repo = await loadRepoById(repoId);
  if (!repo) return new Set();
  const checked = await Promise.all(
    userIds.slice(0, MAX_GATE_REVIEWERS).map(async (id) => {
      const viewer = await actorViewer(id);
      return viewer && hasPermission(await repoPermission(repo, viewer), "write") ? id : null;
    }),
  );
  return new Set(checked.filter((id): id is string => id !== null));
}

export type MergeGate = { allowed: boolean; blockers: string[]; bypassed: string[] };

/** Evaluates branch protection requirements (approvals, checks, change requests) for merging. */
export async function mergeGate(
  pull: PullRow,
  permission: Permission,
  options: { skipChecks?: boolean } = {},
): Promise<MergeGate> {
  const blockers: string[] = [];
  if (pull.state !== "open") blockers.push("Pull request is not open");
  if (pull.draft === 1) blockers.push("Pull request is still a draft");
  if (pull.mergeable === "conflict") blockers.push("Merge conflicts must be resolved");
  const protection = protectionFor(await loadProtections(pull.repo_id), pull.base_ref);
  const protectionBlockers: string[] = [];
  if (protection) {
    const reviews = await repoDb(pull.repo_id).all<{ author_id: string; state: string; commit_sha: string | null }>(
      `SELECT author_id, state, commit_sha FROM reviews WHERE issue_id = ? AND state != 'commented' ORDER BY created_at`,
      pull.issue_id,
    );
    const latest = new Map<string, string>();
    for (const review of reviews) {
      if (review.author_id === pull.author_id) continue;
      const stale =
        protection.dismissStaleReviews && review.state === "approved" && review.commit_sha !== pull.head_sha;
      latest.set(review.author_id, stale ? "dismissed" : review.state);
    }
    if (latest.size > MAX_GATE_REVIEWERS) protectionBlockers.push("Too many reviewers to evaluate reviews");
    const writers = await reviewersWithWrite(pull.repo_id, [...latest.keys()]);
    for (const id of [...latest.keys()]) if (!writers.has(id)) latest.delete(id);
    const approvals = [...latest.values()].filter((state) => state === "approved").length;
    if (approvals < protection.requiredApprovals) {
      protectionBlockers.push(`${protection.requiredApprovals - approvals} more approving review(s) required`);
    }
    if ([...latest.values()].includes("changes_requested")) protectionBlockers.push("Changes were requested");
    if (protection.requireCodeOwnerReview) {
      const approvers = new Set([...latest].flatMap(([id, state]) => (state === "approved" ? [id] : [])));
      protectionBlockers.push(...(await codeOwnerBlockers(pull, approvers)));
    }
    if (protection.requiredChecks.length > 0 && !options.skipChecks) {
      const statuses = await listCommitStatuses(pull.repo_id, pull.head_sha);
      const repo = await loadRepoById(pull.repo_id);
      const results = repo
        ? await requiredCheckResults(repo, pull.base_sha, statuses, protection.requiredChecks)
        : protection.requiredChecks.map(() => "missing" as const);
      protection.requiredChecks.forEach((check, index) => {
        if (results[index] === "untrusted") {
          protectionBlockers.push(`Required check "${check}" did not run from the base branch's workflow definition`);
        } else if (results[index] === "stale") {
          protectionBlockers.push(
            `Required check "${check}" ran from an outdated workflow definition; re-run it against the current base branch`,
          );
        } else if (results[index] !== "success") {
          protectionBlockers.push(`Required check "${check}" has not passed`);
        }
      });
    }
  }
  const bypass = protection !== null && bypassesProtection(protection, permission);
  if (!bypass) blockers.push(...protectionBlockers);
  const canWrite = permission === "write" || permission === "admin";
  return { allowed: blockers.length === 0 && canWrite, blockers, bypassed: bypass ? protectionBlockers : [] };
}

export class MergeError extends Error {}

export async function performMerge(input: {
  pull: PullRow;
  method: MergeMethod;
  message: string;
  merger: { id: string; name: string; email: string; timezone?: string };
  author: { name: string; email: string };
}): Promise<string> {
  let landed: Landed | null;
  try {
    landed =
      input.method === "rebase"
        ? await rebaseInWorker({ pull: input.pull, merger: input.merger })
        : await landInWorker({ ...input, method: input.method, merger: input.merger, author: input.author });
  } catch (error) {
    if (!(error instanceof RebaseConflictError)) throw error;
    throw new MergeError("These commits cannot be rebased cleanly; use a merge commit or squash instead");
  }
  if (landed) {
    await markMerged(input.pull, input.merger.id, landed.after, input.method);
    await triggerPush({
      source: "internal",
      repoId: input.pull.repo_id,
      artifactsName: (await loadPullRepos(input.pull)).baseRepo.artifacts_name,
      ref: `refs/heads/${input.pull.base_ref}`,
      before: landed.before,
      after: landed.after,
      pusherId: input.merger.id,
    });
    return landed.after;
  }
  const { result, baseRepo } = await withPullRemotes(input.pull, "write", async (remotes) => ({
    baseRepo: remotes.baseRepo,
    result: await gitTaskRunner(input.pull.repo_id).runGitTask(MERGE_SCRIPT, {
      MERGER_NAME: input.merger.name,
      MERGER_EMAIL: input.merger.email,
      AUTHOR_NAME: input.author.name,
      AUTHOR_EMAIL: input.author.email,
      BASE_URL: remotes.baseUrl,
      HEAD_URL: remotes.headUrl,
      BASE_REF: input.pull.base_ref,
      HEAD_REF: input.pull.head_ref,
      EXPECTED_HEAD: input.pull.head_sha,
      MERGE_METHOD: input.method,
      MERGE_MESSAGE: input.message,
    }),
  }));
  const before = /BEFORE ([0-9a-f]{40})/.exec(result.stdout)?.[1];
  const after = /AFTER ([0-9a-f]{40})/.exec(result.stdout)?.[1];
  if (result.code !== 0 || !before || !after) {
    if (result.code === EXIT_CONFLICT) {
      await repoDb(input.pull.repo_id).run(
        "UPDATE pull_requests SET mergeable = 'conflict' WHERE issue_id = ?",
        input.pull.issue_id,
      );
      throw new MergeError("This branch has conflicts that must be resolved");
    }
    if (result.code === EXIT_HEAD_MOVED) throw new MergeError("The head branch changed; reload and try again");
    if (result.code === EXIT_BASE_MOVED) throw new MergeError("The base branch moved during the merge; try again");
    throw new MergeError(`Merge failed: ${result.stderr.trim() || `exit ${result.code}`}`);
  }
  await markMerged(input.pull, input.merger.id, after, input.method);
  await triggerPush({
    source: "internal",
    repoId: baseRepo.id,
    artifactsName: baseRepo.artifacts_name,
    ref: `refs/heads/${input.pull.base_ref}`,
    before,
    after,
    pusherId: input.merger.id,
  });
  return after;
}