import { env } from "cloudflare:workers";
import { z } from "zod";
import { repoDb } from "../scoped-db";
import { locate } from "../scoped-db/locators";
import { recordJobArtifact } from "./job-records";
import { jobDb, runnerDb } from "./scope";
import { HttpError } from "../errors";
import { enforceRateLimit, requestIp } from "../limits";
import { loadRepoById } from "../repos";
import { credentialAllows, requestCredential } from "../token-scope";
import { cancelActiveJob, loadJobExecution, runnerStub } from "./jobs";
import type { WorkflowJobRow } from "./jobs";
import type { LogRow } from "./runner-meta";
import { claimOnce } from "./runner-claim";
import { checkRunnerBells } from "./job-doorbell";
import {
  deleteRunnerRow,
  MAX_RUNNER_JOBS,
  needsRotation,
  registerRunner,
  rotateRunnerToken,
  runnerForToken,
  runnerRepoIds,
  summarizeRunner,
} from "./runner-registration";
import type { RunnerRow } from "./runner-registration";

export 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 (
      await runnerDb(auth.runner.id)
    )?.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 perRepo = await Promise.all(
    (await runnerRepoIds(runner)).map((repoId) =>
      repoDb(repoId).all<{ id: string }>(
        "SELECT id FROM workflow_jobs WHERE runner_id = ? AND status = 'in_progress' LIMIT ?",
        runner.id,
        MAX_RUNNER_JOBS * 4,
      ),
    ),
  );
  return perRepo
    .flat()
    .slice(0, MAX_RUNNER_JOBS * 4)
    .map((row) => row.id);
}

async function ownedJob(runner: RunnerRow, jobId: string): Promise<WorkflowJobRow | null> {
  const db = await jobDb(jobId);
  const job = (await db?.first<WorkflowJobRow>("SELECT * FROM workflow_jobs WHERE id = ?", jobId)) ?? null;
  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 db = repoDb(job.repo_id);
  const existing = await db.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 db.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 recordJobArtifact(job.repo_id, { runId: job.run_id, jobId: job.id, name: safeName, size: length, r2Key: key });
  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) => cancelActiveJob({ id, runs_on: "[]" }, "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);
  const stub = runnerStub(match[1]);
  const assignment = await stub.assignment();
  if (assignment && assignment.runnerId !== runner.id)
    return json({ error: "job is not assigned to this runner" }, 409);
  if (match[2] === "heartbeat" && request.method === "POST") {
    if (assignment && !assignment.running) return json({ cancelled: true });
    // Some completions (workflow end, stale expiry) only touch D1, so a running object still checks the row. With a
    // matching object a replica read is enough: lag can only delay seeing "completed", never revoke ownership.
    const db = await jobDb(match[1]);
    if (assignment) {
      const row = await db?.first<{ status: string }>("SELECT status FROM workflow_jobs WHERE id = ?", match[1]);
      return json({ cancelled: row?.status === "completed" });
    }
    const row = await db?.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" });
  }
  if (assignment && (match[2] === "logs" || match[2] === "complete")) {
    if (!assignment.running) return json({ error: "job is not assigned to this runner" }, 409);
    return jobCall(stub, match[2], request);
  }
  const job = await ownedJob(runner, match[1]);
  if (!job) return json({ error: "job is not assigned to this runner" }, 409);
  if (match[3] && request.method === "PUT") return uploadArtifact(job, decodeURIComponent(match[3]), request);
  return jobCall(stub, match[2], request);
}

async function jobCall(stub: ReturnType<typeof runnerStub>, action: string, request: Request): Promise<Response> {
  if (action === "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 (action === "complete" && request.method === "POST") {
    const body = completeBody.parse(await request.json());
    await stub.completeExternal(body.exitCode);
    return json({ ok: true });
  }
  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 repoId = await locate("job_artifact", artifact[1]);
    const row = repoId
      ? await repoDb(repoId).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],
        )
      : null;
    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 db = await jobDb(match[1]);
  const job = await db?.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 });
}

/**
 * Safety net, run by the RepoState alarm while the repository has jobs on self-hosted runners: fails jobs whose
 * runner stopped heartbeating (or was deleted).
 */
export async function reapStaleRunnerJobsForRepo(repoId: string, now: number): Promise<void> {
  const running = await repoDb(repoId).all<{ id: string; runner_id: string; runs_on: string }>(
    "SELECT id, runner_id, runs_on FROM workflow_jobs WHERE status = 'in_progress' AND runner_id IS NOT NULL",
  );
  // null: the runner is confirmed gone (no scope or no row); a lookup that throws leaves its jobs alone this time.
  const lastSeen = new Map<string, number | null>();
  for (const runnerId of new Set(running.map((job) => job.runner_id))) {
    try {
      const db = await runnerDb(runnerId);
      const row = db
        ? await db.first<{ last_seen_at: number | null }>("SELECT last_seen_at FROM runners WHERE id = ?", runnerId)
        : null;
      lastSeen.set(runnerId, row?.last_seen_at ?? null);
    } catch (error) {
      console.warn("Runner lookup failed; its jobs are checked next time", runnerId, error);
    }
  }
  for (const job of running) {
    if (!lastSeen.has(job.runner_id)) continue;
    const seen = lastSeen.get(job.runner_id) ?? null;
    if (seen !== null && seen >= now - STALE_RUNNER_MS) continue;
    await cancelActiveJob({ id: job.id, runs_on: job.runs_on }, "runner stopped responding");
  }
}