import { DurableObject } from "cloudflare:workers";
import { mintRepoToken, authenticatedRemote } from "../git/artifacts";
import { completeJob, loadJobExecution, markJobStarted, resolveSecrets, updateStep } from "./jobs";
import { buildJobScript, jobTimeoutCommand, MARKER_NONCE_FILE } from "./script";
import { declaredNames, maskValues, parseMarker, withOutput } from "./markers";
import { stepSecretNames } from "./compile";
import { run } from "../db";
import { jobTokenEnv, mintJobToken } from "./job-token";
import { randomId } from "../crypto";
import type { Conclusion } from "~/lib/types";

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 SSE_HEARTBEAT_MS = 15_000;
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_SUBSCRIBERS = 100;
const MAX_CLOUD_ARTIFACT_BYTES = 1024 * 1024 * 1024;
const MASKS_KEY = "masks";
const MASKING_UNAVAILABLE = "[masking unavailable; line hidden]";
const ARTIFACT_DIR = `${WORKSPACE}/.gitfor/artifacts`;

type JobMeta = {
  jobId: string;
  mode: "cloud" | "external";
  state: "running" | "finalizing" | "done";
  offset: number;
  step: number;
  artifacts: Array<{ name: string; path: string }>;
  conclusion: Conclusion | null;
  deadline: number;
  outputs?: Record<string, string>;
  nonce?: string;
  allowedArtifacts?: string[];
  allowedOutputs?: string[];
};

export type LogRow = { seq: number; step: number; text: string };

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

export class Runner extends DurableObject<Env> {
  private masks: string[] | null = null;
  private subscribers = new Set<WritableStreamDefaultWriter<Uint8Array>>();
  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> {
    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");
    if (!container.running) {
      container.start({ enableInternet: true });
    }
    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`);
  }

  private async storeMasks(values: string[]): Promise<void> {
    this.masks = maskValues(values);
    await this.ctx.storage.put(MASKS_KEY, this.masks);
  }

  private async ensureMasks(jobId: string): Promise<void> {
    this.masks ??= (await this.ctx.storage.get<string[]>(MASKS_KEY)) ?? null;
    const execution = this.masks === null ? await loadJobExecution(jobId) : null;
    if (execution)
      await this.storeMasks(Object.values(await resolveSecrets(execution, stepSecretNames(execution.spec))));
  }

  /** Fails closed: without the mask list a line could leak a secret, so it is hidden. */
  private mask(text: string): string {
    if (this.masks === null) return MASKING_UNAVAILABLE;
    let out = text;
    for (const value of this.masks) out = out.split(value).join("***");
    return out;
  }

  /** 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 mintRepoToken(execution.sourceRepo.artifacts_name, "read", 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.storeMasks([...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);
    await markJobStarted(execution, null, "cloudflare-container");

    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),
      };
      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): Promise<void> {
    const execution = await loadJobExecution(jobId);
    if (!execution) throw new Error(`Job ${jobId} not found`);
    await this.storeMasks(secretValues);
    await this.saveMeta({
      nonce,
      ...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,
    );
  }

  async ingest(text: string): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    const lines = text.split("\n");
    if (lines.at(-1) === "") lines.pop();
    await this.processLines(meta, 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 || 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.ensureMasks(meta.jobId);
    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);
      if (!marker) {
        if (line === "" && rows.length === 0 && lines.length > 1) continue;
        rows.push({ step: meta.step, text: this.mask(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.mask(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 {
    const frame = this.encoder.encode(`${id === undefined ? "" : `id: ${id}\n`}data: ${JSON.stringify(payload)}\n\n`);
    for (const writer of this.subscribers) {
      writer.write(frame).catch(() => this.subscribers.delete(writer));
    }
  }

  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 run(
        "INSERT INTO job_artifacts (id, run_id, job_id, name, size, r2_key, created_at) VALUES (?, ?, ?, ?, ?, ?, ?)",
        randomId(),
        execution.run.id,
        meta.jobId,
        artifact.name,
        size,
        key,
        Date.now(),
      );
    }
  }

  private async finalize(conclusion: Conclusion): Promise<void> {
    const meta = await this.meta();
    if (!meta || meta.state !== "running") return;
    meta.state = "finalizing";
    meta.conclusion = conclusion;
    await this.saveMeta(meta);
    try {
      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 run("UPDATE workflow_jobs SET outputs = ? WHERE id = ?", JSON.stringify(meta.outputs), meta.jobId);
    await completeJob(meta.jobId, conclusion);
    this.broadcast({ type: "done", conclusion });
    for (const writer of this.subscribers) writer.close().catch(() => undefined);
    this.subscribers.clear();
    meta.state = "done";
    await this.saveMeta(meta);
    this.ctx.storage.sql.exec("DELETE FROM log");
    if (meta.mode === "cloud" && this.ctx.container?.running) {
      await this.ctx.container.destroy().catch(() => undefined);
    }
  }

  /** 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 });
    if (this.subscribers.size >= MAX_SUBSCRIBERS) return new Response("Too many log viewers", { status: 429 });
    const meta = await this.meta();
    const lastId = Number(request.headers.get("last-event-id") ?? url.searchParams.get("after") ?? "0");
    const { readable, writable } = new TransformStream<Uint8Array, Uint8Array>();
    const writer = writable.getWriter();
    const backlog = this.ctx.storage.sql
      .exec<LogRow>("SELECT seq, step, text FROM log WHERE seq > ? ORDER BY seq", lastId)
      .toArray();
    for (const row of backlog) {
      writer.write(this.encoder.encode(`id: ${row.seq}\ndata: ${JSON.stringify({ type: "log", ...row })}\n\n`));
    }
    if (!meta || meta.state === "done") {
      writer.write(
        this.encoder.encode(`data: ${JSON.stringify({ type: "done", conclusion: meta?.conclusion ?? null })}\n\n`),
      );
      writer.close();
    } else {
      this.subscribers.add(writer);
      const heartbeat = setInterval(() => {
        writer.write(this.encoder.encode(": keepalive\n\n")).catch(() => {
          clearInterval(heartbeat);
          this.subscribers.delete(writer);
        });
      }, SSE_HEARTBEAT_MS);
    }
    return new Response(readable, {
      headers: { "content-type": "text/event-stream", "cache-control": "no-cache", connection: "keep-alive" },
    });
  }

  /** 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;
  }
}