import { DurableObject } from "cloudflare:workers";
import { LogSubscribers } from "./log-stream";
import { ensureRunnerContainer, INTERNAL_API_URL } from "../containers/ready";
import { authenticatedRemote } from "../git/artifacts";
import { completeJob, loadJobExecution, markJobStarted, resolveSecrets, mintJobCloneToken, updateStep } from "./jobs";
import { buildJobScript, jobTimeoutCommand, MARKER_NONCE_FILE } from "./script";
import { addMaskValue, declaredNames, parseMarker, withOutput } from "./markers";
import { JobMasks, splitLogChunk } from "./job-masks";
import { stepSecretNames } from "./compile";
import { recordJobArtifact, saveJobOutputs } from "./job-records";
import { jobTokenEnv, mintJobToken } from "./job-token";
import { randomId } from "../crypto";
import type { Conclusion } from "~/lib/types";
import { PURGE_DELAY_MS } from "./runner-meta";
import type { JobMeta, LogRow } from "./runner-meta";

const POLL_INTERVAL_MS = 1500;
const READ_CHUNK_BYTES = 256 * 1024;
const TOKEN_GRACE_SECONDS = 600;
const SETUP_STEP = -1;
const WORKSPACE = "/workspace";
const SCRIPT_PATH = `${WORKSPACE}/.gitfor-job.sh`;
const LOG_PATH = `${WORKSPACE}/.gitfor-job.log`;
const EXIT_PATH = `${WORKSPACE}/.gitfor-job.exit`;
const GIT_TASK_TIMEOUT_SECONDS = 900;
const GIT_TASK_IDLE_MS = 5 * 60 * 1000;
const MAX_LOG_BYTES = 20 * 1024 * 1024;
const MAX_LINE_CHARS = 16 * 1024;
const LOG_BYTES_KEY = "logBytes";
const MAX_CLOUD_ARTIFACTS = 20;
const MAX_CLOUD_ARTIFACT_BYTES = 1024 * 1024 * 1024;
const ARTIFACT_DIR = `${WORKSPACE}/.gitfor/artifacts`;

async function readAll(stream: ReadableStream | null): Promise<string> {
  return stream ? new Response(stream).text() : "";
}

export class Runner extends DurableObject<Env> {
  private readonly masks = new JobMasks(this.ctx.storage);
  private readonly streams = new LogSubscribers();
  private gitQueue: Promise<unknown> = Promise.resolve();
  private readonly encoder = new TextEncoder();

  constructor(ctx: DurableObjectState, env: Env) {
    super(ctx, env);
    ctx.blockConcurrencyWhile(async () => {
      this.ctx.storage.sql.exec(
        "CREATE TABLE IF NOT EXISTS log (seq INTEGER PRIMARY KEY AUTOINCREMENT, step INTEGER NOT NULL, text TEXT NOT NULL)",
      );
    });
  }

  private async meta(): Promise<JobMeta | null> {
    return (await this.ctx.storage.get<JobMeta>("meta")) ?? null;
  }

  private async saveMeta(meta: JobMeta): Promise<void> {
    if (meta.state === "running" && ((await this.meta())?.state ?? "running") !== "running") return;
    await this.ctx.storage.put("meta", meta);
  }

  private async container(): Promise<Container> {
    const container = this.ctx.container;
    if (!container) throw new Error("Runner Durable Object has no container binding");
    await ensureRunnerContainer(container, this.ctx.exports.JobApiGateway);
    return container;
  }

  private async exec(
    command: string[],
    options: ContainerExecOptions = {},
  ): Promise<{ code: number; stdout: string; stderr: string }> {
    const container = await this.container();
    const process = await container.exec(command, options);
    const [stdout, stderr, code] = await Promise.all([
      readAll(process.stdout),
      readAll(process.stderr),
      process.exitCode,
    ]);
    return { code, stdout, stderr };
  }

  private async writeFile(path: string, content: string): Promise<void> {
    const container = await this.container();
    const bytes = this.encoder.encode(content);
    const stdin = new ReadableStream<Uint8Array>({
      start(controller) {
        controller.enqueue(bytes);
        controller.close();
      },
    });
    const process = await container.exec(["sh", "-c", `mkdir -p "$(dirname "$1")" && cat > "$1"`, "sh", path], {
      stdin,
      stdout: "ignore",
    });
    const code = await process.exitCode;
    if (code !== 0) throw new Error(`Failed to write ${path} into the container`);
  }

  /** Launches a cloud job: boots the container, writes the generated script and starts polling. */
  async startJob(jobId: string): Promise<void> {
    const existing = await this.meta();
    if (existing) return;
    const execution = await loadJobExecution(jobId);
    if (!execution) throw new Error(`Job ${jobId} not found`);

    const ttl = execution.spec.timeoutMinutes * 60 + TOKEN_GRACE_SECONDS;
    const tokenSecret = await mintJobCloneToken(jobId, execution.sourceRepo.artifacts_name, ttl);
    const cloneUrl = authenticatedRemote(execution.sourceRepo.artifacts_remote, tokenSecret);
    const jobToken = await mintJobToken(execution);
    const secrets = await resolveSecrets(execution, stepSecretNames(execution.spec), jobToken);
    await this.masks.store([...Object.values(secrets), tokenSecret, jobToken, cloneUrl]);
    const nonce = randomId();

    const meta: JobMeta = {
      jobId,
      mode: "cloud",
      state: "running",
      offset: 0,
      step: SETUP_STEP,
      artifacts: [],
      conclusion: null,
      deadline: Date.now() + ttl * 1000,
      nonce,
      ...declaredNames(execution),
    };
    await this.saveMeta(meta);
    const started = await markJobStarted(execution, null, "cloudflare-container");
    if (!started) return this.finalize("cancelled");
    try {
      const container = await this.container();
      await container.setInactivityTimeout(execution.spec.timeoutMinutes * 60 * 1000 + GIT_TASK_IDLE_MS);
      await this.writeFile(`${WORKSPACE}/${MARKER_NONCE_FILE}`, nonce);
      await this.writeFile(SCRIPT_PATH, buildJobScript(execution.spec));
      const jobEnv: Record<string, string> = {
        HOME: "/root",
        GITFOR_CLONE_URL: cloneUrl,
        GITFOR_SHA: execution.run.sha,
        GITFOR_REF: execution.run.ref,
        GITFOR_REPOSITORY: `${execution.repo.owner_name}/${execution.repo.name}`,
        GITFOR_WORKSPACE: WORKSPACE,
        GITFOR_RUN_ID: execution.run.id,
        GITFOR_RUN_NUMBER: String(execution.run.run_number),
        GIT_TERMINAL_PROMPT: "0",
        ...jobTokenEnv(jobToken, execution.spec.permissions),
        GITFOR_API_URL: INTERNAL_API_URL,
      };
      for (const [name, value] of Object.entries(secrets)) jobEnv[`GITFOR_SECRET_${name}`] = value;
      const command = jobTimeoutCommand(execution.spec, SCRIPT_PATH)
        .map((part) => `'${part.replace(/'/g, "'\\''")}'`)
        .join(" ");
      const launcher = await container.exec(
        ["sh", "-c", `setsid sh -c "${command} > ${LOG_PATH} 2>&1; echo \\$? > ${EXIT_PATH}" > /dev/null 2>&1 &`],
        { env: jobEnv, stdout: "ignore", stderr: "ignore" },
      );
      await launcher.exitCode;
    } catch (error) {
      this.appendLines([{ step: SETUP_STEP, text: `Failed to start runner: ${String(error)}` }]);
      await this.finalize("failure");
      return;
    }
    await this.ctx.storage.setAlarm(Date.now() + POLL_INTERVAL_MS);
  }

  /** Prepares log ingestion for a job claimed by a self-hosted runner. */
  async attachExternal(jobId: string, secretValues: string[], nonce: string, runnerId?: string): Promise<void> {
    const execution = await loadJobExecution(jobId);
    if (!execution) throw new Error(`Job ${jobId} not found`);
    await this.masks.store(secretValues);
    await this.saveMeta({
      nonce,
      runnerId,
      ...declaredNames(execution),
      jobId,
      mode: "external",
      state: "running",
      offset: 0,
      step: SETUP_STEP,
      artifacts: [],
      conclusion: null,
      deadline: Date.now() + execution.spec.timeoutMinutes * 60 * 1000 + TOKEN_GRACE_SECONDS * 1000,
    });
    await this.ctx.storage.setAlarm(
      Date.now() + execution.spec.timeoutMinutes * 60 * 1000 + TOKEN_GRACE_SECONDS * 1000,
    );
  }

  /** Which self-hosted runner holds this job and whether it is still running; null before a runner claimed it. */
  async assignment(): Promise<{ runnerId: string; running: boolean } | null> {
    const meta = await this.meta();
    return meta?.runnerId ? { runnerId: meta.runnerId, running: meta.state === "running" } : null;
  }

  async ingest(text: string): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    const split = splitLogChunk(meta.logCarry ?? "", text);
    meta.logCarry = split.carry;
    await this.processLines(meta, split.lines);
  }

  async completeExternal(exitCode: number): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    await this.finalize(exitCode === 0 ? "success" : "failure");
  }

  async cancel(reason: string): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    this.appendLines([{ step: meta.step, text: `Job cancelled: ${reason}` }]);
    if (meta.mode === "cloud" && this.ctx.container?.running) {
      await this.ctx.container.destroy("cancelled").catch(() => undefined);
    }
    await this.finalize("cancelled");
  }

  override async alarm(): Promise<void> {
    const meta = await this.meta();
    if (!meta) return;
    if (meta.state === "done") {
      await this.purge();
      return;
    }
    // A finalize that failed after marking the job `finalizing` (D1/R2 error, isolate eviction) must still reach
    // completeJob, or the job stays in_progress and its token stays usable until expiry (CI-A6).
    if (meta.state === "finalizing") {
      await this.settle(meta);
      return;
    }
    if (meta.state !== "running") return;
    if (meta.mode === "external") {
      if (Date.now() >= meta.deadline) await this.cancel("exceeded the job timeout");
      return;
    }
    try {
      await this.poll(meta);
    } catch (error) {
      this.appendLines([{ step: meta.step, text: `Runner error: ${String(error)}` }]);
      await this.finalize("failure");
      return;
    }
    const current = await this.meta();
    if (current?.state === "running") {
      if (Date.now() > current.deadline) {
        await this.cancel("exceeded the job timeout");
        return;
      }
      await this.ctx.storage.setAlarm(Date.now() + POLL_INTERVAL_MS);
    }
  }

  private async poll(meta: JobMeta): Promise<void> {
    const read = await this.exec([
      "sh",
      "-c",
      `tail -c +${meta.offset + 1} ${LOG_PATH} 2>/dev/null | head -c ${READ_CHUNK_BYTES}`,
    ]);
    const exit = await this.exec(["sh", "-c", `cat ${EXIT_PATH} 2>/dev/null || true`]);
    const finished = exit.stdout.trim() !== "";
    let chunk = read.stdout;
    if (!finished) {
      const lastNewline = chunk.lastIndexOf("\n");
      chunk =
        lastNewline === -1 && chunk.length < READ_CHUNK_BYTES ? "" : chunk.slice(0, lastNewline + 1 || chunk.length);
    }
    if (chunk.length > 0) {
      meta.offset += this.encoder.encode(chunk).byteLength;
      const lines = chunk.split("\n");
      if (lines.at(-1) === "") lines.pop();
      await this.processLines(meta, lines);
    }
    if (finished && (await this.meta())?.state === "running") {
      const fullyRead = this.encoder.encode(read.stdout).byteLength < READ_CHUNK_BYTES;
      if (!fullyRead) return;
      const code = Number(exit.stdout.trim());
      await this.finalize(code === 0 ? "success" : "failure");
    }
  }

  private async processLines(meta: JobMeta, lines: string[]): Promise<void> {
    await this.masks.ensure(async () => {
      const execution = await loadJobExecution(meta.jobId);
      return execution ? Object.values(await resolveSecrets(execution, stepSecretNames(execution.spec))) : null;
    }, meta);
    const [artifactNames, outputNames] = [new Set(meta.allowedArtifacts), new Set(meta.allowedOutputs)];
    const rows: Array<{ step: number; text: string }> = [];
    for (const line of lines) {
      const marker = parseMarker(line, meta.nonce ?? null);
      const value = marker ? null : addMaskValue(line);
      if (value !== null) await this.masks.add(meta, value);
      if (!marker) {
        if (line === "" && rows.length === 0 && lines.length > 1) continue;
        rows.push({ step: meta.step, text: this.masks.apply(line) });
        continue;
      }
      if (rows.length > 0 && rows.at(-1)?.text === "") rows.pop();
      switch (marker.type) {
        case "setup-start":
          meta.step = SETUP_STEP;
          break;
        case "step-start":
          meta.step = marker.index;
          await updateStep(meta.jobId, marker.index, "in_progress", null);
          this.broadcast({ type: "step", index: marker.index, status: "in_progress", conclusion: null });
          break;
        case "step-end": {
          const conclusion: Conclusion = marker.code === 0 ? "success" : "failure";
          await updateStep(meta.jobId, marker.index, "completed", conclusion);
          this.broadcast({ type: "step", index: marker.index, status: "completed", conclusion });
          break;
        }
        case "step-skip":
          await updateStep(meta.jobId, marker.index, "completed", "skipped");
          this.broadcast({ type: "step", index: marker.index, status: "completed", conclusion: "skipped" });
          break;
        case "artifact":
          if (artifactNames.has(marker.name) && !meta.artifacts.some((item) => item.name === marker.name))
            meta.artifacts.push({ name: marker.name, path: `${ARTIFACT_DIR}/${marker.name}.tgz` });
          break;
        case "output":
          if (outputNames.has(marker.name))
            meta.outputs = withOutput(meta.outputs ?? {}, marker.name, this.masks.apply(marker.value));
          break;
        case "job-end":
        case "setup-end":
          break;
      }
    }
    this.appendLines(rows);
    await this.saveMeta(meta);
  }

  private appendLines(rows: Array<{ step: number; text: string }>): void {
    if (rows.length === 0) return;
    let stored = this.ctx.storage.kv.get<number>(LOG_BYTES_KEY) ?? 0;
    if (stored >= MAX_LOG_BYTES) return;
    const merged: Array<{ step: number; text: string }> = [];
    for (const original of rows) {
      const row =
        original.text.length > MAX_LINE_CHARS
          ? { ...original, text: `${original.text.slice(0, MAX_LINE_CHARS)} …[line truncated]` }
          : original;
      stored += row.text.length;
      if (stored >= MAX_LOG_BYTES) {
        merged.push({ step: row.step, text: "[log output exceeded the size limit; further output is discarded]" });
        break;
      }
      const last = merged.at(-1);
      if (last && last.step === row.step) last.text += `\n${row.text}`;
      else merged.push({ ...row });
    }
    this.ctx.storage.kv.put(LOG_BYTES_KEY, stored);
    for (const row of merged) {
      const cursor = this.ctx.storage.sql.exec<{ seq: number }>(
        "INSERT INTO log (step, text) VALUES (?, ?) RETURNING seq",
        row.step,
        row.text,
      );
      const seq = cursor.one().seq;
      this.broadcast({ type: "log", seq, step: row.step, text: row.text }, seq);
    }
  }

  private broadcast(payload: Record<string, unknown>, id?: number): void {
    this.streams.broadcast(payload, id);
  }

  private async uploadArtifacts(meta: JobMeta): Promise<void> {
    if (meta.mode !== "cloud" || meta.artifacts.length === 0) return;
    const execution = await loadJobExecution(meta.jobId);
    if (!execution) return;
    const container = await this.container();
    for (const artifact of meta.artifacts.slice(0, MAX_CLOUD_ARTIFACTS)) {
      const size = Number((await this.exec(["stat", "-c", "%s", artifact.path])).stdout.trim());
      if (!Number.isFinite(size) || size <= 0 || size > MAX_CLOUD_ARTIFACT_BYTES) continue;
      const process = await container.exec(["cat", artifact.path], { stderr: "ignore" });
      if (!process.stdout) continue;
      const key = `artifacts/${execution.run.id}/${meta.jobId}/${artifact.name}.tgz`;
      const sized = new FixedLengthStream(size);
      const upload = this.env.BUCKET.put(key, sized.readable);
      await process.stdout.pipeTo(sized.writable);
      await upload;
      await recordJobArtifact(execution.repo.id, {
        runId: execution.run.id,
        jobId: meta.jobId,
        name: artifact.name,
        size,
        r2Key: key,
      });
    }
  }

  private async finalize(conclusion: Conclusion): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    if (meta.logCarry) {
      const carry = meta.logCarry;
      meta.logCarry = "";
      await this.processLines(meta, [carry]);
    }
    meta.state = "finalizing";
    meta.conclusion = conclusion;
    await this.saveMeta(meta);
    await this.settle(meta);
  }

  /** Persists artifacts, logs and outputs, then records the terminal state; re-armed via the alarm if it throws. */
  private async settle(meta: JobMeta): Promise<void> {
    const conclusion = meta.conclusion ?? "failure";
    try {
      if (!meta.persisted) {
        try {
          if (conclusion !== "cancelled") await this.uploadArtifacts(meta);
        } catch (error) {
          this.appendLines([{ step: meta.step, text: `Failed to upload artifacts: ${String(error)}` }]);
        }
        const rows = this.ctx.storage.sql.exec<LogRow>("SELECT seq, step, text FROM log ORDER BY seq").toArray();
        await this.env.BUCKET.put(`logs/${meta.jobId}.json`, JSON.stringify(rows), {
          httpMetadata: { contentType: "application/json" },
        });
        if (meta.outputs) await saveJobOutputs(meta.jobId, meta.outputs);
        meta.persisted = true;
        await this.saveMeta(meta);
      }
      await completeJob(meta.jobId, conclusion);
    } catch (error) {
      console.error("Job finalization failed; retrying", meta.jobId, error);
      await this.ctx.storage.setAlarm(Date.now() + POLL_INTERVAL_MS);
      throw error;
    }
    this.streams.finish({ type: "done", conclusion });
    meta.state = "done";
    await this.saveMeta(meta);
    this.ctx.storage.sql.exec("DELETE FROM log");
    await this.ctx.storage.setAlarm(Date.now() + PURGE_DELAY_MS);
    if (meta.mode === "cloud" && this.ctx.container?.running) {
      await this.ctx.container.destroy().catch(() => undefined);
    }
  }

  /** Drops all storage once a finished job's grace period ends; logs already live in R2. */
  private async purge(): Promise<void> {
    await this.ctx.storage.deleteAlarm();
    await this.ctx.storage.deleteAll();
    this.ctx.storage.sql.exec(
      "CREATE TABLE IF NOT EXISTS log (seq INTEGER PRIMARY KEY AUTOINCREMENT, step INTEGER NOT NULL, text TEXT NOT NULL)",
    );
  }

  /** Tells open log streams the job row changed: it started, or it finished without ever reaching a runner. */
  async announce(event: "started" | "finished"): Promise<void> {
    if (event === "started")
      return this.broadcast({ type: "step", index: SETUP_STEP, status: "in_progress", conclusion: null });
    if (!(await this.meta())) this.streams.finish({ type: "done", conclusion: null });
  }

  /** Returns buffered log rows while the job runs; null once logs moved to R2. */
  async logs(afterSeq: number): Promise<{ rows: LogRow[]; done: boolean }> {
    const meta = await this.meta();
    const rows = this.ctx.storage.sql
      .exec<LogRow>("SELECT seq, step, text FROM log WHERE seq > ? ORDER BY seq", afterSeq)
      .toArray();
    return { rows, done: !meta || meta.state === "done" };
  }

  override async fetch(request: Request): Promise<Response> {
    const url = new URL(request.url);
    if (url.pathname !== "/stream") return new Response("Not found", { status: 404 });
    const meta = await this.meta();
    const lastId = Number(request.headers.get("last-event-id") ?? url.searchParams.get("after") ?? "0");
    const backlog = this.ctx.storage.sql
      .exec<LogRow>("SELECT seq, step, text FROM log WHERE seq > ? ORDER BY seq", lastId)
      .toArray();
    const over = meta?.state === "done" || (!meta && url.searchParams.get("pending") !== "1");
    return this.streams.open(backlog, over ? { type: "done", conclusion: meta?.conclusion ?? null } : null);
  }

  /** Runs a privileged git script (merges, mergeability checks) in this repo's container, one at a time. */
  async runGitTask(
    script: string,
    env: Record<string, string>,
  ): Promise<{ code: number; stdout: string; stderr: string }> {
    const task = this.gitQueue.then(async () => {
      const container = await this.container();
      await container.setInactivityTimeout(GIT_TASK_IDLE_MS);
      const path = `/tmp/gitfor-git-${randomId(8)}.sh`;
      await this.writeFile(path, script);
      return this.exec(["timeout", String(GIT_TASK_TIMEOUT_SECONDS), "bash", path], {
        env: { HOME: "/root", GIT_TERMINAL_PROMPT: "0", ...env },
      });
    });
    this.gitQueue = task.catch(() => undefined);
    return task;
  }
}