import { repoDb } from "./scoped-db";
import { randomId } from "./crypto";
import { loadRepoById, repoPermission } from "./repos";
import { addIssueEvent, loadPullByIssueId, markMerged, mergeGate, MergeError, performMerge } from "./pulls";
import type { PullRow } from "./pulls";
import { loadProtections, protectionFor } from "./protection";
import { listCommitStatuses, requiredCheckResults } from "./statuses";
import { buildSpeculativeMerge, fastForward, inWorkerMergeability } from "./pull-merge";
import { queueMergeFix } from "./mergefix/service";
import { branchSha, listRefs } from "./git/refs";
import { deleteRef } from "./git/writer";
import { triggerPush } from "./events/push-trigger";
import { scheduleWorkflows } from "./ci/scheduler";
import { assertBelowCap } from "./limits";
import { actorViewer, announceMerged, defaultMergeMessage, kickBranch, mergeIdentity } from "./merge-automation";
import type { MergeMethod, RepoRow, Viewer } from "~/lib/types";

const MAX_QUEUE_LENGTH = 50;
const MAX_ENTRIES_PER_PASS = 5;
const CHECK_TIMEOUT_MS = 2 * 60 * 60 * 1000;
/** Safety re-poll while waiting on CI; status updates normally wake the queue first. */
export const QUEUE_POLL_MS = 10 * 60 * 1000;
/** Backoff when a pass could not make progress (e.g. the base moved during landing). */
const RETRY_BACKOFF_MS = 15_000;

export type QueueMethod = "merge" | "squash";

type QueueEntry = {
  id: string;
  issue_id: string;
  method: QueueMethod;
  state: "queued" | "testing";
  speculative_sha: string | null;
  base_sha: string | null;
  head_sha: string | null;
  enqueued_by: string;
  updated_at: number;
};

export function queueRef(baseRef: string, number: number): string {
  return `gitfor-merge-queue/${baseRef}/pr-${number}`;
}

export async function enqueuePull(pull: PullRow, method: QueueMethod, viewer: Viewer): Promise<void> {
  const active = await repoDb(pull.repo_id).first<{ count: number; mine: number }>(
    `SELECT COUNT(*) AS count, SUM(issue_id = ?) AS mine FROM merge_queue_entries
     WHERE repo_id = ? AND base_ref = ? AND state IN ('queued', 'testing')`,
    pull.issue_id,
    pull.repo_id,
    pull.base_ref,
  );
  if ((active?.mine ?? 0) > 0) throw new MergeError("This pull request is already in the merge queue");
  assertBelowCap(active?.count ?? 0, MAX_QUEUE_LENGTH, "pull requests in a merge queue");
  const now = Date.now();
  await insertEntry(pull, method, viewer, now);
  await addIssueEvent(pull, viewer.id, "added_to_merge_queue", {});
  await kickBranch(pull.repo_id, pull.base_ref);
}

async function insertEntry(pull: PullRow, method: QueueMethod, viewer: Viewer, now: number): Promise<void> {
  try {
    await repoDb(pull.repo_id).run(
      `INSERT INTO merge_queue_entries (id, repo_id, base_ref, issue_id, method, state, enqueued_by, created_at, updated_at)
     VALUES (?, ?, ?, ?, ?, 'queued', ?, ?, ?)`,
      randomId(),
      pull.repo_id,
      pull.base_ref,
      pull.issue_id,
      method,
      viewer.id,
      now,
      now,
    );
  } catch (error) {
    if (error instanceof Error && /UNIQUE/i.test(error.message)) {
      throw new MergeError("This pull request is already in the merge queue");
    }
    throw error;
  }
}

export async function dequeuePull(pull: PullRow, actorId: string | null, reason: string | null): Promise<void> {
  const db = repoDb(pull.repo_id);
  const entry = await db.first<QueueEntry>(
    "SELECT * FROM merge_queue_entries WHERE issue_id = ? AND state IN ('queued', 'testing')",
    pull.issue_id,
  );
  if (!entry) return;
  await db.run(
    "UPDATE merge_queue_entries SET state = ?, error = ?, updated_at = ? WHERE id = ?",
    reason ? "failed" : "removed",
    reason,
    Date.now(),
    entry.id,
  );
  await addIssueEvent(pull, actorId, "removed_from_merge_queue", reason ? { reason } : {});
  await cleanupScratch(pull, entry);
  await kickBranch(pull.repo_id, pull.base_ref);
}

async function queuePosition(pull: PullRow): Promise<{ position: number; state: string } | null> {
  const db = repoDb(pull.repo_id);
  const entry = await db.first<{ state: string; created_at: number }>(
    "SELECT state, created_at FROM merge_queue_entries WHERE issue_id = ? AND state IN ('queued', 'testing')",
    pull.issue_id,
  );
  if (!entry) return null;
  const ahead = await db.first<{ count: number }>(
    `SELECT COUNT(*) AS count FROM merge_queue_entries
     WHERE repo_id = ? AND base_ref = ? AND state IN ('queued', 'testing') AND created_at < ?`,
    pull.repo_id,
    pull.base_ref,
    entry.created_at,
  );
  return { position: (ahead?.count ?? 0) + 1, state: entry.state };
}

/** Auto-merge and merge-queue state shown in the merge box. */
export async function pullAutomation(pull: PullRow) {
  const [autoMerge, queue] = await Promise.all([
    repoDb(pull.repo_id).first<{ method: MergeMethod; enabled_by: string }>(
      "SELECT method, enabled_by FROM pull_auto_merges WHERE issue_id = ?",
      pull.issue_id,
    ),
    queuePosition(pull),
  ]);
  return { autoMerge: autoMerge && { method: autoMerge.method, enabledBy: autoMerge.enabled_by }, queue };
}

async function cleanupScratch(pull: PullRow, entry: Pick<QueueEntry, "speculative_sha">) {
  if (!entry.speculative_sha) return;
  const repo = await loadRepoById(pull.repo_id);
  if (!repo) return;
  const ref = queueRef(pull.base_ref, pull.number);
  const sha = branchSha(await listRefs(repo), ref);
  if (sha) await deleteRef(repo, `refs/heads/${ref}`, sha).catch((error) => console.error("Queue ref cleanup", error));
}

async function fail(pull: PullRow, reason: string) {
  await dequeuePull(pull, null, reason);
}

/**
 * Re-reads the entry and the pull right before landing: a dequeue, or a head push that happened after this
 * entry was picked up, must stop the merge.
 */
async function stillLandable(entry: QueueEntry, pull: PullRow): Promise<boolean> {
  const row = await repoDb(pull.repo_id).first<{ state: string; head_sha: string }>(
    `SELECT e.state, p.head_sha FROM merge_queue_entries e JOIN pull_requests p ON p.issue_id = e.issue_id
     WHERE e.id = ?`,
    entry.id,
  );
  return row !== null && (row.state === "queued" || row.state === "testing") && row.head_sha === pull.head_sha;
}

async function land(repo: RepoRow, pull: PullRow, entry: QueueEntry, actor: Viewer, sha: string, before: string) {
  await markMerged(pull, actor.id, sha, entry.method);
  await triggerPush({
    source: "internal",
    repoId: repo.id,
    artifactsName: repo.artifacts_name,
    ref: `refs/heads/${pull.base_ref}`,
    before,
    after: sha,
    pusherId: actor.id,
  });
  await cleanupScratch(pull, entry);
  await announceMerged(repo, pull, actor, sha);
}

async function startEntry(repo: RepoRow, pull: PullRow, entry: QueueEntry, actor: Viewer): Promise<number | null> {
  const gate = await mergeGate(pull, await repoPermission(repo, actor), { skipChecks: true });
  if (!gate.allowed) {
    await fail(pull, gate.blockers.join("; ") || "Merging is blocked");
    return 0;
  }
  const baseSha = branchSha(await listRefs(repo), pull.base_ref);
  if (!baseSha) {
    await fail(pull, "The base branch no longer exists");
    return 0;
  }
  const headRepo = pull.head_repo_id === repo.id ? repo : await loadRepoById(pull.head_repo_id);
  if (!headRepo) {
    await fail(pull, "The head repository was deleted");
    return 0;
  }
  const identity = await mergeIdentity(pull, actor);
  const message = defaultMergeMessage(pull, headRepo, entry.method);
  const speculative = await buildSpeculativeMerge({
    pull,
    baseSha,
    ref: queueRef(pull.base_ref, pull.number),
    method: entry.method,
    message,
    ...identity,
  });
  const required = protectionFor(await loadProtections(repo.id), pull.base_ref)?.requiredChecks ?? [];
  if (!speculative) {
    if ((await inWorkerMergeability(pull)) === "conflict" && (await queueMergeFix(repo, pull, actor, entry.method))) {
      await fail(
        pull,
        "This pull request has conflicts; an AI fix was started and it will rejoin the queue once CI passes on the fix",
      );
      return 0;
    }
    const fullGate = await mergeGate(pull, await repoPermission(repo, actor));
    if (!fullGate.allowed) {
      await fail(pull, fullGate.blockers.join("; ") || "Merging is blocked");
      return 0;
    }
    if (!(await stillLandable(entry, pull))) return 0;
    const sha = await performMerge({ pull, method: entry.method, message, ...identity });
    await announceMerged(repo, pull, actor, sha);
    return 0;
  }
  if (required.length === 0) {
    if (!(await stillLandable(entry, pull))) return 0;
    if (await fastForward(repo, pull.base_ref, baseSha, speculative)) {
      await land(repo, pull, entry, actor, speculative, baseSha);
      return 0;
    }
    return RETRY_BACKOFF_MS;
  }
  await repoDb(repo.id).run(
    "UPDATE merge_queue_entries SET state = 'testing', speculative_sha = ?, base_sha = ?, head_sha = ?, updated_at = ? WHERE id = ?",
    speculative,
    baseSha,
    pull.head_sha,
    Date.now(),
    entry.id,
  );
  await scheduleWorkflows(repo, {
    event: "pull_request",
    action: "synchronize",
    ref: `refs/heads/${queueRef(pull.base_ref, pull.number)}`,
    sha: speculative,
    actorId: actor.id,
    actorName: actor.name,
    title: `Merge queue: ${pull.title}`,
    prNumber: pull.number,
    headRef: pull.head_ref,
    baseRef: pull.base_ref,
    changedPaths: null,
    inputs: {},
    workflowPath: null,
    sourceRepo: repo,
    workflowSource: { repo, sha: baseSha },
    untrusted: pull.head_repo_id !== repo.id,
  });
  return QUEUE_POLL_MS;
}

async function checkEntry(repo: RepoRow, pull: PullRow, entry: QueueEntry, actor: Viewer): Promise<number | null> {
  const speculative = entry.speculative_sha;
  const baseSha = entry.base_sha;
  if (!speculative || !baseSha || entry.head_sha !== pull.head_sha) {
    await repoDb(repo.id).run(
      "UPDATE merge_queue_entries SET state = 'queued', speculative_sha = NULL, base_sha = NULL, updated_at = ? WHERE id = ?",
      Date.now(),
      entry.id,
    );
    return 0;
  }
  const required = protectionFor(await loadProtections(repo.id), pull.base_ref)?.requiredChecks ?? [];
  const statuses = await listCommitStatuses(repo.id, speculative);
  const results = await requiredCheckResults(repo, baseSha, statuses, required);
  const failed = required.findIndex((_, index) => results[index] === "failure" || results[index] === "error");
  if (failed >= 0) {
    await fail(pull, `Required check "${required[failed]}" failed on the merge group`);
    return 0;
  }
  const untrusted = required.findIndex((_, index) => results[index] === "untrusted" || results[index] === "stale");
  if (untrusted >= 0) {
    await fail(pull, `Required check "${required[untrusted]}" did not run from the base branch's workflow definition`);
    return 0;
  }
  if (results.some((state) => state !== "success")) {
    if (Date.now() - entry.updated_at > CHECK_TIMEOUT_MS) {
      await fail(pull, "Timed out waiting for required checks");
      return 0;
    }
    return QUEUE_POLL_MS;
  }
  const gate = await mergeGate(pull, await repoPermission(repo, actor), { skipChecks: true });
  if (!gate.allowed) {
    await fail(pull, gate.blockers.join("; ") || "Merging is blocked");
    return 0;
  }
  if (!(await stillLandable(entry, pull))) return 0;
  if (await fastForward(repo, pull.base_ref, baseSha, speculative)) {
    await land(repo, pull, entry, actor, speculative, baseSha);
  } else {
    await repoDb(repo.id).run(
      "UPDATE merge_queue_entries SET state = 'queued', speculative_sha = NULL, base_sha = NULL, updated_at = ? WHERE id = ?",
      Date.now(),
      entry.id,
    );
  }
  return 0;
}

/**
 * Advances the head of the branch's merge queue one step. Returns the delay before the next pass,
 * or null when the queue is empty. Runs only inside the branch's Durable Object, so it is serialized.
 */
export async function processQueue(repoId: string, baseRef: string): Promise<number | null> {
  const repo = await loadRepoById(repoId);
  if (!repo) return null;
  for (let pass = 0; pass < MAX_ENTRIES_PER_PASS; pass++) {
    const [entry] = await repoDb(repoId).all<QueueEntry>(
      `SELECT * FROM merge_queue_entries WHERE repo_id = ? AND base_ref = ? AND state IN ('queued', 'testing')
       ORDER BY created_at LIMIT 1`,
      repoId,
      baseRef,
    );
    if (!entry) return null;
    const pull = await loadPullByIssueId(repoId, entry.issue_id);
    const actor = await actorViewer(entry.enqueued_by);
    if (!pull || pull.state !== "open" || !actor) {
      await repoDb(repoId).run(
        "UPDATE merge_queue_entries SET state = 'removed', updated_at = ? WHERE id = ?",
        Date.now(),
        entry.id,
      );
      continue;
    }
    try {
      const delay =
        entry.state === "queued"
          ? await startEntry(repo, pull, entry, actor)
          : await checkEntry(repo, pull, entry, actor);
      if (delay !== 0) return delay;
    } catch (error) {
      if (!(error instanceof MergeError)) console.error("Merge queue entry failed", entry.id, error);
      await fail(pull, error instanceof MergeError ? error.message : "The merge queue hit an internal error");
    }
  }
  return RETRY_BACKOFF_MS;
}