import { all, run } from "../db";
import { randomId } from "../crypto";
import { authenticatedRemote } from "../git/artifacts";
import {
  completeJob,
  isCloudJob,
  loadJobExecution,
  markJobStarted,
  mintJobCloneToken,
  resolveSecrets,
  runnerStub,
} from "./jobs";
import type { WorkflowJobRow } from "./jobs";
import { buildJobScript } from "./script";
import { stepSecretNames } from "./compile";
import { jobTokenEnv, mintJobToken } from "./job-token";
import { CLOUD_RUNS_ON_VALUES, cloudRunsOnPlaceholders } from "./runner-sizes";
import { MAX_RUNNER_JOBS } from "./runner-registration";
import type { RunnerRow } from "./runner-registration";
import { LABELS_MATCH_SQL } from "./runner-policy";

const TOKEN_GRACE_SECONDS = 600;
const CLAIM_CANDIDATES = 25;
/** Values shorter than this are too common to mask without redacting ordinary log text. */
const MIN_MASK_LENGTH = 4;

/** Labels come only from the registration record; whatever the daemon sends with a claim is ignored. */
function runnerLabels(runner: RunnerRow): Set<string> {
  return new Set([...(JSON.parse(runner.labels) as string[]), "self-hosted"].map((label) => label.toLowerCase()));
}

function labelsMatch(required: string[], available: Set<string>): boolean {
  return required.length > 0 && required.every((label) => available.has(label.toLowerCase()));
}

/**
 * Candidate queued jobs inside the runner's scope. Fork pull request (untrusted) runs are only eligible when a
 * repo admin opted in, and then only for runners registered to that repo with admin-attested container isolation, so a shared
 * owner-wide or unisolated runner never executes fork code.
 */
async function candidateJobs(runner: RunnerRow): Promise<WorkflowJobRow[]> {
  const cloud = cloudRunsOnPlaceholders(6);
  return all<WorkflowJobRow>(
    `SELECT j.* FROM workflow_jobs j JOIN repos repo ON repo.id = j.repo_id JOIN workflow_runs w ON w.id = j.run_id
     JOIN (SELECT ?5 AS labels) r
     LEFT JOIN runner_policies p ON p.repo_id = repo.id
     WHERE j.status = 'queued' AND (repo.id = ?1 OR repo.owner_id = ?2) AND j.runs_on NOT IN (${cloud})
       AND ${LABELS_MATCH_SQL}
       AND ((w.source_repo_id = w.repo_id AND w.untrusted = 0)
         OR (p.allow_fork_pull_requests = 1 AND repo.id = ?1 AND ?3 = 1))
     ORDER BY j.created_at LIMIT ?4`,
    runner.repo_id ?? "",
    runner.owner_id ?? "",
    runner.isolation_attested === 1 ? 1 : 0,
    CLAIM_CANDIDATES,
    runner.labels,
    ...CLOUD_RUNS_ON_VALUES,
  );
}

/** Atomic assignment: the job must still be queued and the runner under its registered concurrency. */
async function assign(runner: RunnerRow, jobId: string): Promise<boolean> {
  const claimed = await run(
    `UPDATE workflow_jobs SET status = 'in_progress', runner_id = ?1, runner_name = ?2, started_at = ?3
     WHERE id = ?4 AND status = 'queued'
       AND (SELECT COUNT(*) FROM workflow_jobs WHERE runner_id = ?1 AND status = 'in_progress') < ?5
       AND EXISTS (SELECT 1 FROM runners WHERE id = ?1)`,
    runner.id,
    runner.name,
    Date.now(),
    jobId,
    Math.min(Math.max(runner.max_jobs, 1), MAX_RUNNER_JOBS),
  );
  return claimed.meta.changes === 1;
}

async function startJob(runner: RunnerRow, candidate: WorkflowJobRow) {
  const execution = await loadJobExecution(candidate.id);
  if (!execution) throw new Error("job vanished after claim");
  const ttl = execution.spec.timeoutMinutes * 60 + TOKEN_GRACE_SECONDS;
  const tokenSecret = await mintJobCloneToken(candidate.id, execution.sourceRepo.artifacts_name, ttl);
  const jobToken = await mintJobToken(execution);
  const secrets = await resolveSecrets(execution, stepSecretNames(execution.spec), jobToken);
  const cloneUrl = authenticatedRemote(execution.sourceRepo.artifacts_remote, tokenSecret);
  const nonce = randomId();
  const masks = [...Object.values(secrets), tokenSecret, jobToken, cloneUrl];
  await runnerStub(candidate.id).attachExternal(candidate.id, masks, nonce);
  await markJobStarted(execution, runner.id, runner.name);
  const jobEnv: Record<string, string> = {
    GITFOR_CLONE_URL: cloneUrl,
    GITFOR_SHA: execution.run.sha,
    GITFOR_REF: execution.run.ref,
    GITFOR_REPOSITORY: `${execution.repo.owner_name}/${execution.repo.name}`,
    GITFOR_RUN_ID: execution.run.id,
    GITFOR_RUN_NUMBER: String(execution.run.run_number),
    GIT_TERMINAL_PROMPT: "0",
    ...jobTokenEnv(jobToken, execution.spec.permissions),
  };
  for (const [name, value] of Object.entries(secrets)) jobEnv[`GITFOR_SECRET_${name}`] = value;
  return {
    id: candidate.id,
    name: candidate.name,
    repository: jobEnv.GITFOR_REPOSITORY,
    timeoutMinutes: execution.spec.timeoutMinutes,
    script: buildJobScript(execution.spec),
    markerNonce: nonce,
    env: jobEnv,
    masks: masks.filter((value) => value.length >= MIN_MASK_LENGTH),
  };
}

/** One claim attempt. `sawCandidates` is false only when the runner's scope had no matching queued job at all. */
export async function claimOnce(runner: RunnerRow) {
  const labels = runnerLabels(runner);
  const candidates = await candidateJobs(runner);
  for (const candidate of candidates) {
    const runsOn: string[] = JSON.parse(candidate.runs_on);
    if (isCloudJob(runsOn) || !labelsMatch(runsOn, labels)) continue;
    if (!(await assign(runner, candidate.id))) continue;
    try {
      return { job: await startJob(runner, candidate), sawCandidates: true };
    } catch (error) {
      console.error("Failed to start claimed job", candidate.id, error);
      await completeJob(candidate.id, "failure").catch((failure) =>
        console.error("Failed to fail claimed job", candidate.id, failure),
      );
    }
  }
  return { job: null, sawCandidates: candidates.length > 0 };
}