/**
 * The dispatcher's caller side of settlement: the periodic sweeps that watch
 * what happened to work already merged or already ended.
 *
 * Named `settle-pass` against the top-level `settlement.ts` because they are
 * two different jobs. `settlement.ts` settles *one* run — it is called at the
 * end of a launch and knows a worker's outcome. Everything here is a sweep with
 * a cursor and a batch size: the merged base's CI, the configured base's health,
 * retained worktrees past their keep window, and the historical-infra backfill
 * that reclassifies rows settled before the classifier knew about them.
 *
 * They belong together because they share one shape and one cost model — bounded
 * GitHub reads per tick against a persisted cursor — and because none of them
 * may ever block a dispatch: a sweep that fell behind must cost the next tick a
 * batch, never a run.
 */
import { infraLogSignature, infraSignatureVersion } from "../failure-class.ts";
import { errText, log, safeEscalate } from "../log.ts";
import { GhPrMissingError } from "../tracker/github.ts";
import type { BaseHealth, RunRecord, SettlementFlag, WorkflowRun } from "../types.ts";
import { cleanupRetainedWorktree, mirrorPathFor, type RetainedWorktreeCleanup } from "../worktree.ts";
import { githubRepo, type Deps, type RetainedCleanupCursor } from "./deps.ts";

export const BASE_CHECK_BATCH = 20;
export const BASE_CHECK_WINDOW_MS = 24 * 60 * 60 * 1_000;
export const BASE_STATUS_WINDOW_MS = 7 * 24 * 60 * 60 * 1_000;

export const SUCCESSFUL_WORKFLOW_CONCLUSIONS = new Set(["success", "neutral", "skipped"]);

export const FAILING_WORKFLOW_CONCLUSIONS = new Set([
  "failure",
  "cancelled",
  "timed_out",
  "action_required",
  "startup_failure",
  "stale",
]);

export function appendSettlementFlag(run: RunRecord, flag: SettlementFlag): SettlementFlag[] {
  const flags = run.settlementFlags ?? [];
  return flags.some((existing) => existing.kind === flag.kind && existing.detail === flag.detail)
    ? flags
    : [...flags, flag];
}

/**
 * Observe Actions on exact merge commits for up to one day. A running workflow
 * stays quiet and pending; a failure becomes durable evidence on the merged row
 * and pages exactly once because the row leaves `pending` before delivery.
 */
export async function watchMergedBase(d: Pick<Deps, "project" | "tracker" | "store" | "escalate">): Promise<void> {
  const now = Date.now();
  for (const run of d.store.runsNeedingBaseCheck(d.project.name, BASE_CHECK_BATCH)) {
    if (
      run.endedAt === undefined ||
      now - run.endedAt > BASE_CHECK_WINDOW_MS ||
      run.mergeSha === undefined ||
      run.baseRef === undefined
    ) {
      d.store.updateRun(run.id, { baseCheck: "unknown", baseCheckAt: now });
      log(`#${run.issue} base check unknown: merge identity is absent or older than 24h`);
      continue;
    }

    const repo = d.project.routing.repos[run.repo];
    const repoIdentity = repo === undefined ? undefined : githubRepo(repo.cloneUrl);
    if (repoIdentity === undefined) {
      d.store.updateRun(run.id, { baseCheck: "unknown", baseCheckAt: now });
      log(`#${run.issue} base check unknown: routed repository ${run.repo} has no GitHub identity`);
      continue;
    }

    let workflows;
    try {
      workflows = await d.tracker.workflowRunsAt(repoIdentity, run.mergeSha, {
        event: "push",
        branch: run.baseRef,
      });
    } catch (err) {
      log(`#${run.issue} base check unavailable (${errText(err)}) — retrying next tick`);
      continue;
    }
    if (workflows === undefined) {
      log(`#${run.issue} base check unavailable for ${run.mergeSha} — retrying next tick`);
      continue;
    }
    if (workflows.length === 0) {
      log(
        `#${run.issue} base check: no push-triggered run yet for ${run.mergeSha} — retrying next tick`,
      );
      continue;
    }
    if (workflows.some((workflow) => workflow.status !== "completed")) continue;

    const failed = workflows.find(
      (workflow) =>
        workflow.conclusion !== undefined &&
        FAILING_WORKFLOW_CONCLUSIONS.has(workflow.conclusion),
    );
    if (failed === undefined) {
      const unknown = workflows.find(
        (workflow) =>
          workflow.conclusion === undefined ||
          !SUCCESSFUL_WORKFLOW_CONCLUSIONS.has(workflow.conclusion),
      );
      if (unknown !== undefined) {
        log(
          `#${run.issue} base check has unrecognised completed conclusion ` +
            `${JSON.stringify(unknown.conclusion)} for ${unknown.name} — retrying next tick`,
        );
        continue;
      }
      d.store.updateRun(run.id, { baseCheck: "green", baseCheckAt: now });
      log(`#${run.issue} base ${run.baseRef} green at ${run.mergeSha}`);
      continue;
    }

    let previous;
    try {
      previous = await d.tracker.previousWorkflowRun(
        repoIdentity,
        failed.workflowId,
        run.baseRef,
        failed.createdAt,
      );
    } catch (err) {
      log(`#${run.issue} previous ${failed.name} run unavailable (${errText(err)}) — retrying next tick`);
      continue;
    }
    if (previous === undefined || (previous !== null && previous.status !== "completed")) continue;
    const preexisting =
      previous !== null &&
      previous.conclusion !== undefined &&
      FAILING_WORKFLOW_CONCLUSIONS.has(previous.conclusion);
    const detail =
      `${failed.name} failed at ${run.mergeSha} — ${failed.url}` +
      (preexisting ? " (already red before this merge)" : "");
    const flag: SettlementFlag = { kind: "base-branch-red", file: "(base branch)", detail };
    // Freeze merges to this repo while the base it merged into is red (#283).
    // The freeze is repo-scoped and sets independently of escalation delivery:
    // a merge that broke the base must not be followed by another merge onto
    // the same red base, even if paging the operator fails.
    if (
      d.store.setBaseFreeze(d.project.name, {
        repo: run.repo,
        culpritSha: run.mergeSha,
        detail,
        setAt: now,
      })
    ) {
      d.store.recordMaterialEvent({
        project: d.project.name,
        category: "base-red-freeze",
        summary: `merges to ${run.repo} frozen — base ${run.baseRef} red at ${run.mergeSha.slice(0, 8)}`,
        evidence:
          `${detail} This freeze names ${run.mergeSha.slice(0, 8)} as the suspected culprit merge. ` +
          "Merges to this repo are refused until the base is green again; reverting the culprit is the " +
          "likely remedy. The freeze lifts automatically on a green re-observation, or the operator can " +
          "override it with `omp-conductor unfreeze <repo>`.",
        occurredAt: now,
        recordedAt: now,
      });
    }
    const delivered = await safeEscalate(d, {
      tier: 1,
      project: d.project.name,
      issue: run.issue,
      runId: run.id,
      summary: `Base branch ${run.baseRef} is red after merge`,
      detail,
    });
    if (!delivered) continue;
    d.store.updateRun(run.id, {
      baseCheck: preexisting ? "red-preexisting" : "red",
      baseCheckAt: now,
      settlementFlags: appendSettlementFlag(run, flag),
    });
  }
}

/**
 * Refresh current base-branch health at each recently merged repository's live
 * head. This is status and release-gate evidence only; the per-merge audit
 * above remains the sole path that attributes and escalates a regression.
 */
export async function watchBaseHealth(
  d: Pick<Deps, "project" | "tracker" | "store">,
): Promise<void> {
  const now = Date.now();
  const previousByRepo = new Map(
    d.store.baseHealth(d.project.name).map((row) => [row.repo, row] as const),
  );
  for (const { repo, baseRef } of d.store.mergedRepoBranches(
    d.project.name,
    now - BASE_STATUS_WINDOW_MS,
  )) {
    const target = d.project.routing.repos[repo];
    if (target === undefined) {
      log(`base health skipped: routed repository ${repo} is no longer configured`);
      continue;
    }
    const identity = githubRepo(target.cloneUrl);
    if (identity === undefined) {
      log(`base health skipped: routed repository ${repo} has no GitHub identity`);
      continue;
    }
    const branch = baseRef ?? target.defaultBranch;

    let head: string | undefined;
    try {
      head = await d.tracker.branchHead(identity, branch);
    } catch (err) {
      log(`base ${repo}/${branch} head unavailable (${errText(err)}) — keeping previous health`);
      continue;
    }
    if (head === undefined) {
      log(`base ${repo}/${branch} head unavailable — keeping previous health`);
      continue;
    }

    const previous = previousByRepo.get(repo);
    const freeze = d.store.baseFreeze(d.project.name, repo);
    const frozen = freeze !== undefined && freeze.clearedAt === undefined;
    // The same-head/age shortcut exists to avoid re-querying GitHub when a
    // terminal verdict has not moved. It must NOT skip a frozen repo: a freeze
    // keyed on one red observation has to keep re-evaluating the same head so a
    // green rerun clears it without operator action (tonight's evidence — a
    // red/unknown read two minutes after the same SHA's CI succeeded — is why
    // it cannot be trusted as terminal).
    if (
      !frozen &&
      previous?.branch === branch &&
      previous.headSha === head &&
      (previous.verdict === "green" || previous.verdict === "red")
    ) {
      continue;
    }

    let runs;
    try {
      runs = await d.tracker.workflowRunsAt(identity, head, { event: "push", branch });
    } catch (err) {
      log(`base ${repo}/${branch} workflows unavailable (${errText(err)}) — keeping previous health`);
      continue;
    }
    if (runs === undefined) {
      log(`base ${repo}/${branch} workflows unavailable — keeping previous health`);
      continue;
    }

    let verdict: BaseHealth["verdict"];
    let detail: string | undefined;
    if (runs.length === 0) {
      verdict = "unknown";
      detail = `no push-triggered workflow run for ${head.slice(0, 8)}`;
    } else if (runs.some((run) => run.status !== "completed")) {
      verdict = "pending";
    } else {
      const failed = runs.find(
        (run) =>
          run.conclusion !== undefined &&
          FAILING_WORKFLOW_CONCLUSIONS.has(run.conclusion),
      );
      if (failed !== undefined) {
        verdict = "red";
        detail = `${failed.name} failed at ${head.slice(0, 8)} — ${failed.url}`;
      } else if (
        runs.some(
          (run) =>
            run.conclusion === undefined ||
            !SUCCESSFUL_WORKFLOW_CONCLUSIONS.has(run.conclusion),
        )
      ) {
        verdict = "pending";
      } else {
        verdict = "green";
      }
    }

    const health: BaseHealth = {
      repo,
      branch,
      headSha: head,
      verdict,
      runsCount: runs.length,
      checkedAt: now,
      ...(detail === undefined ? {} : { detail }),
    };
    d.store.upsertBaseHealth(d.project.name, health);
    previousByRepo.set(repo, health);

    // The freeze follows the live base verdict: red arms (or re-arms) the
    // repo-scoped freeze, green lifts it. pending/unknown leave it untouched —
    // a stale or in-flight reading must neither create a freeze nor clear one.
    if (verdict === "green") {
      if (d.store.clearBaseFreeze(d.project.name, repo, "daemon", "base-green", now)) {
        d.store.recordMaterialEvent({
          project: d.project.name,
          category: "base-recovered",
          summary: `merges to ${repo} unfrozen — base ${branch} observed green at ${head.slice(0, 8)}`,
          evidence: `The base-red freeze on ${repo} lifted automatically: ${branch} is green at ${head.slice(0, 8)}. Merges resume.`,
          occurredAt: now,
          recordedAt: now,
        });
      }
    } else if (verdict === "red") {
      if (
        d.store.setBaseFreeze(d.project.name, {
          repo,
          culpritSha: head,
          detail,
          setAt: now,
        })
      ) {
        d.store.recordMaterialEvent({
          project: d.project.name,
          category: "base-red-freeze",
          summary: `merges to ${repo} frozen — base ${branch} red at ${head.slice(0, 8)}`,
          evidence:
            `${detail ?? `workflow failed at ${head.slice(0, 8)}`} This freeze names ` +
            `${head.slice(0, 8)} as the suspected culprit. Reverting it is the likely remedy; the freeze ` +
            "lifts automatically on green or with `omp-conductor unfreeze <repo>`.",
          occurredAt: now,
          recordedAt: now,
        });
      }
    }
  }
}

export const RETAINED_CLEANUP_BATCH = 10;

export type CleanupRetainedWorktree = (
  mirrorPath: string,
  worktreePath: string,
  branch: string,
) => Promise<RetainedWorktreeCleanup>;

/**
 * Bounded, rotating cleanup for failure-path trees. Tracker state proves the
 * run is terminal; local git state independently proves deletion cannot erase
 * dirty or uniquely unpushed work.
 */
export async function cleanupRetainedRuns(
  d: Pick<Deps, "project" | "tracker" | "store" | "escalate">,
  queuedIssues: ReadonlySet<number>,
  cursor: RetainedCleanupCursor,
  cleanup: CleanupRetainedWorktree = cleanupRetainedWorktree,
): Promise<void> {
  const { project, tracker, store } = d;
  const candidates = store.retainedRuns(project.name);
  if (candidates.length === 0) {
    cursor.next = 0;
    return;
  }

  const occupied = new Set(store.activeRuns(project.name).map((run) => run.issue));
  const liveRepos = new Set(store.liveRuns(project.name).map((run) => run.repo));
  const start = cursor.next % candidates.length;
  const count = Math.min(RETAINED_CLEANUP_BATCH, candidates.length);
  const batch = Array.from({ length: count }, (_, offset) => candidates[(start + offset) % candidates.length]!);
  cursor.next = (start + count) % candidates.length;

  for (const run of batch) {
    if (occupied.has(run.issue) || queuedIssues.has(run.issue) || liveRepos.has(run.repo)) continue;
    // Attempts reuse one physical path and deterministic branch. An older row
    // cannot authorize deleting the newest failed attempt's evidence merely
    // because its own PR resolved first.
    const latest = store.latestRun(project.name, run.issue);
    if (
      latest !== undefined &&
      latest.id !== run.id &&
      latest.state !== "merged" &&
      latest.worktree !== ""
    ) {
      continue;
    }

    let terminal = false;
    try {
      if (run.prUrl !== undefined) {
        const pr = await tracker.prState(run.prUrl);
        if (pr === undefined) continue;
        if (pr === "merged" || pr === "closed") {
          terminal = true;
        } else {
          const issue = await tracker.issueState(run.issue);
          if (issue === undefined) continue;
          terminal = issue === "closed";
        }
      } else {
        const issue = await tracker.issueState(run.issue);
        if (issue === undefined) continue;
        terminal = issue === "closed";
      }
    } catch (err) {
      // The tracker throws exactly one classified error: `GhPrMissingError`,
      // an individual REST 404 it corroborated with a same-repository
      // pulls-list read, so the claimed PR definitively does not exist
      // (#779). That is a fact, not "could not tell": retrying can never
      // conjure the PR, so the missing PR is terminal and the retained row
      // settles in this same pass like a closed one (#1065). The row's
      // phantom URL is cleared with the settlement so a repeat pass neither
      // asks GitHub about it again nor logs about it again. Everything else —
      // a rate limit, a 5xx, a lost network — keeps the deferred retry below,
      // unchanged.
      if (err instanceof GhPrMissingError) {
        store.updateRun(run.id, { prUrl: null });
        log(`#${run.issue} retained cleanup settled: PR ${run.prUrl} does not exist`);
        terminal = true;
      } else {
        log(`#${run.issue} retained cleanup deferred: tracker state failed (${errText(err)})`);
        continue;
      }
    }
    if (!terminal) continue;

    const repo = Object.values(project.routing.repos).find((candidate) => candidate.name === run.repo);
    if (repo === undefined) {
      log(`#${run.issue} retained cleanup deferred: repo ${run.repo} is no longer configured`);
      continue;
    }

    const outcome = await cleanup(mirrorPathFor(repo, project.mirrorRoot), run.worktree, run.branch);
    if (outcome.kind === "removed") {
      store.updateRun(run.id, {
        worktree: "",
        ...(run.quarantineDetail === undefined ? {} : { quarantineDetail: null }),
      });
      log(`#${run.issue} retained worktree reaped: ${run.worktree} (${run.branch})`);
    } else if (outcome.reason === "quarantined") {
      // A tree whose object store cannot be made sound is potentially
      // stranded work: the daemon refuses to fetch into it, so its commits
      // cannot be verified against any remote. The row records the condition
      // so the status snapshot can name the tree, and the escalation ledger
      // dedupes on the stable summary below — a pass that keeps seeing the
      // same broken tree reports it once, never once per dispatch pass.
      store.updateRun(run.id, { quarantineDetail: outcome.detail });
      await safeEscalate(d, {
        tier: 1,
        project: project.name,
        issue: run.issue,
        summary: `#${run.issue} quarantined retained worktree — potentially stranded work`,
        detail: `${run.worktree} (${run.branch})\n${outcome.detail}`,
      });
    } else {
      // Any other retained reason means the tree is back under ordinary
      // retention: its alternates were repaired (or never needed it), so the
      // quarantine — if one was marked — is over and the snapshot must not
      // keep naming it as quarantined.
      if (run.quarantineDetail !== undefined) {
        store.updateRun(run.id, { quarantineDetail: null });
        log(`#${run.issue} retained worktree no longer quarantined: ${outcome.detail}`);
      }
      log(`#${run.issue} retained worktree kept (${outcome.reason}): ${outcome.detail}`);
    }
  }
}

/**
 * How many settled `ci-deterministic` rows one reconciliation pass may
 * re-examine beyond the persisted review cursor. Each candidate costs GitHub
 * calls to re-fetch the evidence its check log carried, so a fleet with a
 * long misclassified history works through it over several daemon starts
 * rather than spending one boot's budget on all of it (#638).
 */
export const HISTORICAL_INFRA_BATCH = 20;

/** How many workflow runs for the head commit one row's evidence pass reads —
 *  and the ceiling past which the row is refused as undecided rather than
 *  decided from a prefix of its head-pinned runs (review #654). */
export const HISTORICAL_INFRA_RUNS = 3;

/** How many attempts of one workflow run may be read for the failed log — and
 *  the ceiling past which the run is refused as undecided rather than read as
 *  a prefix that could hide a later attempt's real failure (review #654). A
 *  failed job rerun to green leaves the failure in an earlier attempt; the
 *  bound keeps one pathological run from costing the whole pass. */
export const HISTORICAL_INFRA_ATTEMPTS = 5;

/**
 * Repair settled `ci-deterministic` rows whose re-fetched check log carries a
 * closed infrastructure signature (#638). The forward classifier now names
 * the codeload setup 429 of #177 `ci-infra`, but a row already classified
 * `ci-deterministic`/`escalate` never re-enters the classification sweep, so
 * the old verdict charges an implementation attempt forever. Re-fetching the
 * head-pinned workflow-run attempt logs through the tracker, matching the
 * classifier's own closed signature list, and reclassifying `failureClass`
 * alone returns the attempt without re-animating a months-old run into
 * recovery.
 *
 * Bounded, idempotent and resumable: one batch per call, reclassifications
 * only, and the store's update is guarded by the row still reading
 * `ci-deterministic`, so a second pass touches nothing it already repaired.
 * The batch resumes below a persisted review cursor, so each row is evaluated
 * once rather than rescanning the newest non-matches forever and starving
 * older repairable rows. A row whose evidence could not be read is a
 * no-mutation refusal: the cursor never advances past it, so the next pass
 * asks again; a row that was read and shown *not* to be infrastructure
 * advances the cursor. Every repair requires *all* gathered failed logs to
 * carry an infra signature — a setup 429 in one attempt must not waive a
 * compile/test failure in a sibling attempt.
 *
 * Returns how many rows it repaired, for the boot log.
 */
export async function reconcileHistoricalInfra(d: Deps): Promise<number> {
  const { project, store } = d;
  const version = infraSignatureVersion();
  const persisted = store.historicalInfraCursor(project.name);
  // A cursor stamped by an older signature list is stale: the classifier now
  // recognises more evidence, so the pass restarts from the newest row rather
  // than skipping past newly repairable history (#638).
  const cursor =
    persisted !== undefined && persisted.classifierVersion === version
      ? { startedAt: persisted.startedAt, rowid: persisted.rowid }
      : undefined;
  const candidates = store.historicalInfraCandidates(project.name, HISTORICAL_INFRA_BATCH, cursor);
  let repaired = 0;
  let lastDecided: { startedAt: number; rowid: number } | undefined;
  for (const run of candidates) {
    const chunks = await historicalInfraEvidence(d, run);
    // Unreachable evidence is undecided exactly like a fresh classifier is: a
    // no-mutation refusal the next pass asks again, and the cursor stops here
    // so the row is re-offered rather than being skipped past.
    if (chunks === undefined) break;
    lastDecided = { startedAt: run.startedAt, rowid: run.rowid };
    // The head's runs were read and none holds a failing log — determinately
    // not infrastructure, so an old misclassified verdict stays charged.
    if (chunks.length === 0) continue;
    // Mixed guard: anything gathered that is not itself a closed infra
    // signature (a compile/test failure in a sibling attempt, a product 429)
    // refuses the whole row. One setup 429 is not permission to waive a real
    // implementation failure.
    if (!chunks.every((chunk) => infraLogSignature(chunk) !== undefined)) continue;
    if (store.reclassifyInfra(run.id)) {
      repaired += 1;
      // The every-guard above guarantees a signature and a first chunk; the
      // non-null assertions make the same fact readable to the type checker.
      log(
        `#${run.issue} repaired historical ${run.failureClass ?? "ci-deterministic"} → ci-infra (run ${run.id}): ` +
          `"${infraLogSignature(chunks[0]!)}" in the failed check log`,
      );
    }
  }
  if (lastDecided !== undefined) {
    store.setHistoricalInfraCursor(
      project.name,
      lastDecided.startedAt,
      lastDecided.rowid,
      version,
    );
  }
  return repaired;
}

/**
 * The failed check logs of a settled `ci-deterministic` row, head-pinned to
 * the exact commit the row ran against, or `undefined` when the evidence is
 * unreachable (#638). Only the workflow-run history at `run.headSha` is
 * evidence: the PR's *current* check rollup is not, because a later push's
 * checks must never classify an earlier run's row. `workflowRunsAt` is itself
 * head-scoped, and each run's failed attempts are read through the tracker's
 * guarded runner (call/refusal accounting and the rate-limit breaker apply).
 *
 * Returns the per-failed-job failed logs gathered across the head's runs (one
 * chunk per failed job, full and untruncated), or `undefined` when any read
 * could not be made — the run register, an attempt's job register, a failed
 * job's log, or the head-run list itself — or when a bounded evidence set
 * exceeded its limit (more head-pinned runs than `HISTORICAL_INFRA_RUNS`, or
 * more attempts than `HISTORICAL_INFRA_ATTEMPTS`). Undefined is a no-mutation
 * refusal the next pass asks again; the cursor never advances past it. `[]`
 * means the head's runs were read and none holds a failing log — a
 * determinately non-infrastructure answer.
 */
export async function historicalInfraEvidence(d: Deps, run: RunRecord): Promise<string[] | undefined> {
  // Without a head SHA there is no safe way to pin evidence to this row; a
  // head-less row is determinately not a repair candidate, so it advances the
  // cursor rather than stalling the pass.
  if (run.headSha === undefined) return [];
  const target = d.project.routing.repos[run.repo];
  const repoIdentity = target === undefined ? undefined : githubRepo(target.cloneUrl);
  if (repoIdentity === undefined) return [];
  let runs: WorkflowRun[] | undefined;
  try {
    runs = await d.tracker.workflowRunsAt(repoIdentity, run.headSha);
  } catch {
    return undefined;
  }
  if (runs === undefined) return undefined;
  // Refuse rather than decide from the first `HISTORICAL_INFRA_RUNS`: three
  // setup-429 runs must not waive a row whose fourth head-pinned run carries
  // the real compile/test failure (review #654). Undecided means the cursor
  // never advances past this row, so the next pass asks again.
  if (runs.length > HISTORICAL_INFRA_RUNS) return undefined;
  const chunks: string[] = [];
  for (const wf of runs) {
    const logs = await d.tracker.runFailedAttemptLogs(
      repoIdentity,
      wf.url,
      HISTORICAL_INFRA_ATTEMPTS,
    );
    if (logs === undefined) return undefined;
    chunks.push(...logs);
  }
  return chunks;
}