import { env } from "cloudflare:workers";
import { z } from "zod";
import { all, first, run } from "../db";
import { HttpError } from "../errors";
import { enforceRateLimit, requestIp } from "../limits";
import { loadRepoById } from "../repos";
import { credentialAllows, requestCredential } from "../token-scope";
import { loadJobExecution, runnerStub } from "./jobs";
import type { WorkflowJobRow } from "./jobs";
import type { LogRow } from "./runner-do";
import { randomId } from "../crypto";
import { claimOnce } from "./runner-claim";
import { checkRunnerBells } from "./job-doorbell";
import {
  deleteRunnerRow,
  MAX_RUNNER_JOBS,
  needsRotation,
  purgeExpiredRegistrations,
  registerRunner,
  rotateRunnerToken,
  runnerForToken,
  summarizeRunner,
} from "./runner-registration";
import type { RunnerRow } from "./runner-registration";

const STALE_RUNNER_MS = 3 * 60 * 1000;
const LAST_SEEN_WRITE_MS = 15 * 1000;
const LONG_POLL_MAX_MS = 25 * 1000;
const CLAIM_RETRY_MS = 2 * 1000;
const MAX_LOG_CHUNK_BYTES = 1024 * 1024;
const MAX_ARTIFACT_BYTES = 1024 * 1024 * 1024;
const MAX_ARTIFACTS_PER_JOB = 20;
const MAX_REGISTER_BODY_BYTES = 16 * 1024;
const ROTATE_HEADER = "x-gitfor-rotate";

const registerBody = z.object({
  token: z.string().min(1).max(200),
  name: z.string().min(1).max(64),
  labels: z.array(z.string().max(32)).max(20).default([]),
  os: z.string().max(16),
  arch: z.string().max(16),
  version: z.string().max(64).default(""),
  isolation: z.string().max(16),
  maxJobs: z.number().int().default(1),
});
const completeBody = z.object({ exitCode: z.number().int() });
const rotateBody = z.object({ token_sha256: z.string().length(64) });

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

type RunnerAuth = { runner: RunnerRow; previous: boolean };

async function authenticateRunner(request: Request): Promise<RunnerAuth | null> {
  const header = request.headers.get("authorization") ?? "";
  const token = header.replace(/^Bearer\s+/i, "").trim();
  const auth = await runnerForToken(token);
  if (!auth) return null;
  const now = Date.now();
  if (auth.runner.last_seen_at === null || now - auth.runner.last_seen_at > LAST_SEEN_WRITE_MS) {
    await run("UPDATE runners SET last_seen_at = ? WHERE id = ?", now, auth.runner.id);
  }
  return auth;
}

/**
 * Long-poll: holds the request up to `wait` ms, re-checking the queue, so idle runners make few requests.
 * The scope doorbells keep idle re-checks off D1 until a job is queued.
 */
async function claim(runner: RunnerRow, request: Request): Promise<Response> {
  await enforceRateLimit("GIT_LIMITER", `runner-claim:${runner.id}`);
  const requested = Number(new URL(request.url).searchParams.get("wait") ?? "0");
  const wait = Number.isFinite(requested) ? Math.min(Math.max(requested, 0), LONG_POLL_MAX_MS) : 0;
  const deadline = Date.now() + wait;
  for (;;) {
    const bells = await checkRunnerBells(runner);
    if (bells.ringing) {
      const { job, sawCandidates } = await claimOnce(runner);
      if (job) return json({ job });
      if (!sawCandidates) await bells.silence();
    }
    if (request.signal.aborted || Date.now() + CLAIM_RETRY_MS > deadline) break;
    await new Promise((resolve) => setTimeout(resolve, CLAIM_RETRY_MS));
  }
  return new Response(null, { status: 204 });
}

async function register(request: Request): Promise<Response> {
  await enforceRateLimit("AUTH_LIMITER", `runner-register:${requestIp(request)}`);
  const declared = Number(request.headers.get("content-length") ?? "0");
  if (declared > MAX_REGISTER_BODY_BYTES) return json({ error: "body too large" }, 413);
  const text = await request.text();
  if (new TextEncoder().encode(text).byteLength > MAX_REGISTER_BODY_BYTES)
    return json({ error: "body too large" }, 413);
  const parsed = registerBody.safeParse(JSON.parse(text || "{}"));
  if (!parsed.success) return json({ error: "invalid registration request" }, 422);
  const { runner, token } = await registerRunner(parsed.data);
  return json({ runner: summarizeRunner(runner, 0), token }, 201);
}

async function busyJobIds(runner: RunnerRow): Promise<string[]> {
  const rows = await all<{ id: string }>(
    "SELECT id FROM workflow_jobs WHERE runner_id = ? AND status = 'in_progress' LIMIT ?",
    runner.id,
    MAX_RUNNER_JOBS * 4,
  );
  return rows.map((row) => row.id);
}

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 execution = await loadJobExecution(job.id);
  const declared = (execution?.spec.steps ?? []).some((step) => step.artifact?.name === name);
  if (!declared) return json({ error: "artifact is not declared by an upload-artifact step" }, 400);
  const duplicate = await first<{ id: string }>(
    "SELECT id FROM job_artifacts WHERE job_id = ? AND name = ?",
    job.id,
    name,
  );
  if (duplicate) return json({ error: "artifact already uploaded" }, 409);
  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 });
}

async function runnerRoute(auth: RunnerAuth, request: Request, path: string): Promise<Response> {
  const { runner } = auth;
  if (path === "claim" && request.method === "POST") return claim(runner, request);
  if (path === "rotate" && request.method === "POST") {
    if (auth.previous) return json({ error: "rotate with the current credential" }, 401);
    const body = rotateBody.parse(await request.json());
    await rotateRunnerToken(runner, body.token_sha256);
    return json({ ok: true });
  }
  if (path === "self" && request.method === "GET") {
    const busy = await busyJobIds(runner);
    return json({ runner: summarizeRunner(runner, busy.length), jobs: busy });
  }
  if (path === "self" && request.method === "DELETE") {
    await deleteRunnerRow(runner.id);
    return new Response(null, { status: 204 });
  }
  if (path === "reset" && request.method === "POST") {
    // A restarted daemon lost its sandboxes, so whatever it held is failed now rather than at the stale reaper.
    const busy = await busyJobIds(runner);
    await Promise.all(busy.map((id) => runnerStub(id).cancel("the runner restarted before the job finished")));
    return json({ cancelled: busy });
  }

  const match = /^jobs\/([a-z0-9]+)\/(logs|complete|heartbeat|artifacts\/([^/]+))$/.exec(path);
  if (!match) return json({ error: "not found" }, 404);
  if (match[2] === "heartbeat" && request.method === "POST") {
    const row = await first<{ status: string; runner_id: string | null }>(
      "SELECT status, runner_id FROM workflow_jobs WHERE id = ?",
      match[1],
    );
    if (!row || row.runner_id !== runner.id) return json({ error: "job is not assigned to this runner" }, 409);
    return json({ cancelled: row.status !== "in_progress" });
  }
  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);
}

export async function handleRunnerApi(request: Request, path: string): Promise<Response> {
  try {
    if (path === "register" && request.method === "POST") return await register(request);
    const auth = await authenticateRunner(request);
    if (!auth) return json({ error: "invalid runner token" }, 401);
    const response = await runnerRoute(auth, request, path);
    if (!auth.previous && needsRotation(auth.runner) && path !== "rotate") {
      const headers = new Headers(response.headers);
      headers.set(ROTATE_HEADER, "1");
      return new Response(response.body, { status: response.status, headers });
    }
    return response;
  } catch (error) {
    if (error instanceof HttpError) return json({ error: error.message }, error.status);
    if (error instanceof SyntaxError || error instanceof z.ZodError) return json({ error: "invalid request" }, 400);
    throw error;
  }
}

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);

  // A queued job has no runner state yet, so the runner alone can't tell "not started" from "finished";
  // the job row decides, and a queued job's stream stays open until the runner picks it up.
  const finished = job.status === "completed";
  if (match[2]) {
    const after = Number(new URL(request.url).searchParams.get("after") ?? "0");
    const search = new URLSearchParams({ after: String(Number.isFinite(after) ? after : 0) });
    if (!finished) search.set("pending", "1");
    return stub.fetch(new Request(`https://runner/stream?${search}`, { headers: request.headers }));
  }
  const live = await stub.logs(0);
  // Finished per the row but not yet archived by the runner: keep serving its live rows until it is.
  if (live.rows.length > 0 || !finished || !live.done) return json({ rows: live.rows, done: finished && 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, and drop dead registrations. */
export async function reapStaleRunnerJobs(): Promise<void> {
  await purgeExpiredRegistrations();
  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");
  }
}