import { repoDb } from "../scoped-db";
import { repoOwnerId, runnerAccountId, runnerDb } from "./scope";
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 { busyRunnerCounts, MAX_RUNNER_JOBS, runnerRepoIds } 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, repoIds: string[]): Promise<WorkflowJobRow[]> {
  const cloud = cloudRunsOnPlaceholders(5);
  const perRepo = await Promise.all(
    repoIds.map((repoId) =>
      repoDb(repoId).all<WorkflowJobRow>(
        `SELECT j.* FROM workflow_jobs j JOIN workflow_runs w ON w.id = j.run_id
         JOIN (SELECT ?4 AS labels) r
         LEFT JOIN runner_policies p ON p.repo_id = j.repo_id
         WHERE j.status = 'queued' 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 j.repo_id = ?1 AND ?2 = 1))
         ORDER BY j.created_at LIMIT ?3`,
        runner.repo_id ?? "",
        runner.isolation_attested === 1 ? 1 : 0,
        CLAIM_CANDIDATES,
        runner.labels,
        ...CLOUD_RUNS_ON_VALUES,
      ),
    ),
  );
  return perRepo
    .flat()
    .sort((left, right) => left.created_at - right.created_at)
    .slice(0, CLAIM_CANDIDATES);
}

/**
 * Assignment: the job must still be queued and the runner (still registered) under its concurrency. The update
 * is atomic within the job's repository; jobs the runner holds in other repositories are counted beforehand.
 */
async function assign(runner: RunnerRow, job: WorkflowJobRow, repoIds: string[]): Promise<boolean> {
  const db = await runnerDb(runner.id);
  if (!(await db?.first("SELECT 1 AS ok FROM runners WHERE id = ?", runner.id))) return false;
  const busy = await busyRunnerCounts(
    repoIds.filter((repoId) => repoId !== job.repo_id),
    [runner.id],
  );
  const claimed = await repoDb(job.repo_id).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') + ?6 < ?5`,
    runner.id,
    runner.name,
    Date.now(),
    job.id,
    Math.min(Math.max(runner.max_jobs, 1), MAX_RUNNER_JOBS),
    busy.get(runner.id) ?? 0,
  );
  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];
  const stub = runnerStub(candidate.id);
  await stub.attachExternal(candidate.id, masks, nonce, runner.id);
  if (!(await markJobStarted(execution, runner.id, runner.name))) {
    // Cancelled between assignment and attach: wind the object down to `done` so it purges instead of idling.
    await stub.cancel("cancelled before the runner started it");
    throw new Error("job was cancelled before it started");
  }
  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),
  };
}

/**
 * Repositories a runner may take jobs from right now. A repository runner lives in the AccountState of the owner
 * that registered it; once the repository belongs to someone else it must never be handed that repository's jobs.
 */
async function claimableRepoIds(runner: RunnerRow): Promise<string[]> {
  if (!runner.repo_id) return runnerRepoIds(runner);
  const [ownerId, accountId] = await Promise.all([repoOwnerId(runner.repo_id), runnerAccountId(runner.id)]);
  return ownerId !== null && ownerId === accountId ? [runner.repo_id] : [];
}

/**
 * Re-checked after the claim, so a transfer that commits between reading the runner's repositories and claiming
 * can't hand the new owner's job (and secrets) to the previous owner's runner. A runner lives in the AccountState
 * of the account it serves, which must still own the job's repository; otherwise the job goes back to the queue.
 */
async function stillServes(runner: RunnerRow, job: WorkflowJobRow): Promise<boolean> {
  const [ownerId, accountId] = await Promise.all([repoOwnerId(job.repo_id), runnerAccountId(runner.id)]);
  if (ownerId !== null && ownerId === accountId) return true;
  await repoDb(job.repo_id).run(
    `UPDATE workflow_jobs SET status = 'queued', runner_id = NULL, runner_name = NULL, started_at = NULL
     WHERE id = ? AND runner_id = ? AND status = 'in_progress'`,
    job.id,
    runner.id,
  );
  return false;
}

/** 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 repoIds = await claimableRepoIds(runner);
  const candidates = await candidateJobs(runner, repoIds);
  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, repoIds))) continue;
    if (!(await stillServes(runner, candidate))) 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),
      );
    }
  }
  // The doorbell is shared by every runner in the scope, so only silence it when nothing at all is queued
  // there; a runner whose labels match none of the queued jobs must not mute the bell for one whose do.
  return { job: null, sawCandidates: candidates.length > 0 || (await scopeHasQueuedJobs(repoIds)) };
}

async function scopeHasQueuedJobs(repoIds: string[]): Promise<boolean> {
  const cloud = cloudRunsOnPlaceholders(1);
  for (const repoId of repoIds) {
    const row = await repoDb(repoId).first<{ found: number }>(
      `SELECT 1 AS found FROM workflow_jobs WHERE status = 'queued' AND runs_on NOT IN (${cloud}) LIMIT 1`,
      ...CLOUD_RUNS_ON_VALUES,
    );
    if (row) return true;
  }
  return false;
}