import { env } from "cloudflare:workers";
import { z } from "zod";
import { all, first, run } from "../db";
import { sha256Hex } from "../crypto";
import { loadRepoById } from "../repos";
import { credentialAllows, requestCredential } from "../token-scope";
import { authenticatedRemote, mintRepoToken } from "../git/artifacts";
import { isCloudJob, loadJobExecution, markJobStarted, resolveSecrets, runnerStub } from "./jobs";
import type { WorkflowJobRow } from "./jobs";
import { buildJobScript } from "./script";
import { stepSecretNames } from "./compile";
import type { LogRow } from "./runner-do";
import { randomId } from "../crypto";
import { jobTokenEnv, mintJobToken } from "./job-token";

const TOKEN_GRACE_SECONDS = 600;
const CLAIM_CANDIDATES = 25;
const STALE_RUNNER_MS = 10 * 60 * 1000;
const MAX_LOG_CHUNK_BYTES = 1024 * 1024;
const MAX_ARTIFACT_BYTES = 1024 * 1024 * 1024;
const MAX_ARTIFACTS_PER_JOB = 20;
export const RUNNER_TOKEN_PREFIX = "stig_runner";

type RunnerRow = {
  id: string;
  repo_id: string | null;
  owner_id: string | null;
  name: string;
  labels: string;
  last_seen_at: number | null;
};

const claimBody = z.object({ labels: z.array(z.string()).optional() });
const completeBody = z.object({ exitCode: z.number().int() });

function json(data: unknown, status = 200): Response {
  return Response.json(data, { status });
}

async function authenticateRunner(request: Request): Promise<RunnerRow | null> {
  const header = request.headers.get("authorization") ?? "";
  const token = header.replace(/^Bearer\s+/i, "").trim();
  if (!token.startsWith(`${RUNNER_TOKEN_PREFIX}_`)) return null;
  const runner = await first<RunnerRow>(
    "SELECT id, repo_id, owner_id, name, labels, last_seen_at FROM runners WHERE token_hash = ?",
    await sha256Hex(token),
  );
  if (runner) await run("UPDATE runners SET last_seen_at = ? WHERE id = ?", Date.now(), runner.id);
  return runner;
}

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

async function claim(runner: RunnerRow, request: Request): Promise<Response> {
  const body = claimBody.parse(await request.json().catch(() => ({})));
  const labels = new Set(
    [...JSON.parse(runner.labels), ...(body.labels ?? []), "self-hosted"].map((label: string) => label.toLowerCase()),
  );
  const candidates = await all<WorkflowJobRow>(
    `SELECT j.* FROM workflow_jobs j JOIN repos r ON r.id = j.repo_id
     WHERE j.status = 'queued' AND (r.id = ? OR r.owner_id = ?)
     ORDER BY j.created_at LIMIT ?`,
    runner.repo_id ?? "",
    runner.owner_id ?? "",
    CLAIM_CANDIDATES,
  );
  for (const candidate of candidates) {
    const runsOn: string[] = JSON.parse(candidate.runs_on);
    if (isCloudJob(runsOn) || !labelsMatch(runsOn, labels)) continue;
    const claimed = await run(
      "UPDATE workflow_jobs SET status = 'in_progress', runner_id = ?, runner_name = ?, started_at = ? WHERE id = ? AND status = 'queued'",
      runner.id,
      runner.name,
      Date.now(),
      candidate.id,
    );
    if (claimed.meta.changes !== 1) continue;
    const execution = await loadJobExecution(candidate.id);
    if (!execution) continue;
    const ttl = execution.spec.timeoutMinutes * 60 + TOKEN_GRACE_SECONDS;
    const tokenSecret = await mintRepoToken(execution.sourceRepo.artifacts_name, "read", ttl);
    const jobToken = await mintJobToken(execution);
    const secrets = await resolveSecrets(execution, stepSecretNames(execution.spec), jobToken);
    await runnerStub(candidate.id).attachExternal(candidate.id, [tokenSecret, jobToken]);
    await markJobStarted(execution, runner.id, runner.name);
    const jobEnv: Record<string, string> = {
      STIG_CLONE_URL: authenticatedRemote(execution.sourceRepo.artifacts_remote, tokenSecret),
      STIG_SHA: execution.run.sha,
      STIG_REF: execution.run.ref,
      STIG_REPOSITORY: `${execution.repo.owner_name}/${execution.repo.name}`,
      STIG_RUN_ID: execution.run.id,
      STIG_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[`STIG_SECRET_${name}`] = value;
    return json({
      job: {
        id: candidate.id,
        name: candidate.name,
        repository: jobEnv.STIG_REPOSITORY,
        timeoutMinutes: execution.spec.timeoutMinutes,
        script: buildJobScript(execution.spec),
        env: jobEnv,
      },
    });
  }
  return new Response(null, { status: 204 });
}

async function ownedJob(runner: RunnerRow, jobId: string): Promise<WorkflowJobRow | null> {
  const job = await first<WorkflowJobRow>("SELECT * FROM workflow_jobs WHERE id = ?", jobId);
  return job && job.runner_id === runner.id && job.status === "in_progress" ? job : null;
}

async function uploadArtifact(job: WorkflowJobRow, name: string, request: Request): Promise<Response> {
  const length = Number(request.headers.get("content-length") ?? "0");
  if (!request.body || !Number.isFinite(length) || length <= 0) return json({ error: "content-length required" }, 411);
  if (length > MAX_ARTIFACT_BYTES) return json({ error: "artifact too large" }, 413);
  const existing = await first<{ count: number }>(
    "SELECT COUNT(*) AS count FROM job_artifacts WHERE job_id = ?",
    job.id,
  );
  if ((existing?.count ?? 0) >= MAX_ARTIFACTS_PER_JOB) return json({ error: "too many artifacts" }, 429);
  const safeName = name.replace(/[^A-Za-z0-9._-]/g, "_");
  const key = `artifacts/${job.run_id}/${job.id}/${safeName}.tgz`;
  await env.BUCKET.put(key, request.body);
  await run(
    "INSERT INTO job_artifacts (id, run_id, job_id, name, size, r2_key, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
    randomId(),
    job.run_id,
    job.id,
    safeName,
    length,
    key,
    Date.now(),
  );
  return json({ ok: true });
}

export async function handleRunnerApi(request: Request, path: string): Promise<Response> {
  const runner = await authenticateRunner(request);
  if (!runner) return json({ error: "invalid runner token" }, 401);
  if (path === "claim" && request.method === "POST") return claim(runner, request);

  const match = /^jobs\/([a-z0-9]+)\/(logs|complete|artifacts\/([^/]+))$/.exec(path);
  if (!match) return json({ error: "not found" }, 404);
  const job = await ownedJob(runner, match[1]);
  if (!job) return json({ error: "job is not assigned to this runner" }, 409);
  const stub = runnerStub(job.id);

  if (match[2] === "logs" && request.method === "POST") {
    const declared = Number(request.headers.get("content-length") ?? "NaN");
    if (!Number.isFinite(declared)) return json({ error: "content-length required" }, 411);
    if (declared > MAX_LOG_CHUNK_BYTES) return json({ error: "log chunk too large" }, 413);
    const text = await request.text();
    if (text.length > MAX_LOG_CHUNK_BYTES) return json({ error: "log chunk too large" }, 413);
    await stub.ingest(text);
    return json({ ok: true });
  }
  if (match[2] === "complete" && request.method === "POST") {
    const body = completeBody.parse(await request.json());
    await stub.completeExternal(body.exitCode);
    return json({ ok: true });
  }
  if (match[3] && request.method === "PUT") return uploadArtifact(job, decodeURIComponent(match[3]), request);
  return json({ error: "not found" }, 404);
}

async function viewerCanReadJob(request: Request, repoId: string): Promise<boolean> {
  const repo = await loadRepoById(repoId);
  if (!repo) return false;
  return credentialAllows(repo, await requestCredential(request, true), "actions", "read");
}

/** Browser-facing job log and artifact endpoints (session or token authenticated). */
export async function handleJobApi(request: Request, path: string): Promise<Response> {
  const artifact = /^artifacts\/([a-z0-9]+)$/.exec(path);
  if (artifact) {
    const row = await first<{ repo_id: string; name: string; r2_key: string }>(
      "SELECT j.repo_id, a.name, a.r2_key FROM job_artifacts a JOIN workflow_jobs j ON j.id = a.job_id WHERE a.id = ?",
      artifact[1],
    );
    if (!row || !(await viewerCanReadJob(request, row.repo_id))) return json({ error: "not found" }, 404);
    const object = await env.BUCKET.get(row.r2_key);
    if (!object) return json({ error: "artifact expired" }, 410);
    return new Response(object.body, {
      headers: {
        "content-type": "application/gzip",
        "content-disposition": `attachment; filename="${row.name}.tgz"`,
        "content-length": String(object.size),
      },
    });
  }

  const match = /^jobs\/([a-z0-9]+)\/logs(\/stream)?$/.exec(path);
  if (!match) return json({ error: "not found" }, 404);
  const job = await first<{ id: string; repo_id: string; status: string; runs_on: string }>(
    "SELECT id, repo_id, status, runs_on FROM workflow_jobs WHERE id = ?",
    match[1],
  );
  if (!job || !(await viewerCanReadJob(request, job.repo_id))) return json({ error: "not found" }, 404);
  const stub = runnerStub(job.id, job.runs_on);

  if (match[2]) {
    const url = new URL(request.url);
    return stub.fetch(new Request(`https://runner/stream${url.search}`, { headers: request.headers }));
  }
  const live = await stub.logs(0);
  if (live.rows.length > 0 || !live.done) return json({ rows: live.rows, done: live.done });
  const archived = await env.BUCKET.get(`logs/${job.id}.json`);
  const rows: LogRow[] = archived ? await archived.json() : [];
  return json({ rows, done: true });
}

/** Cron safety net: fail jobs whose self-hosted runner stopped heartbeating. */
export async function reapStaleRunnerJobs(): Promise<void> {
  const stale = await all<{ id: string }>(
    `SELECT j.id FROM workflow_jobs j JOIN runners r ON r.id = j.runner_id
     WHERE j.status = 'in_progress' AND (r.last_seen_at IS NULL OR r.last_seen_at < ?)`,
    Date.now() - STALE_RUNNER_MS,
  );
  for (const job of stale) {
    await runnerStub(job.id).cancel("runner stopped responding");
  }
}