import { env } from "cloudflare:workers";
import { batch, first, stmt } from "../db";
import { randomId } from "../crypto";
import { setCommitStatus } from "../statuses";
import { CAPS } from "../limits";
import { publicUrl } from "../events/emit";
import { withStore } from "../git/store";
import { compileWorkflow } from "./compile";
import type { CompiledJob, EventContext } from "./compile";
import { parseWorkflow, WORKFLOW_DIR } from "./spec";
import { workflowMatches } from "./triggers";
import type { Trigger } from "./triggers";
import type { RepoRow } from "~/lib/types";

const WORKFLOW_FILE = /\.ya?ml$/;
const MAX_WORKFLOW_FILES = 30;
const MAX_WORKFLOW_BYTES = 128 * 1024;
const MAX_JOBS_PER_RUN = 50;
const MAX_STEP_ROWS_PER_RUN = 800;

type WorkflowFile = { path: string; source: string; tooLarge: boolean };

async function readWorkflowFiles(trigger: Trigger): Promise<WorkflowFile[]> {
  return withStore(trigger.sourceRepo.artifacts_name, async (store) => {
    const commit = await store.commit(trigger.sha);
    if (!commit) return [];
    const directory = await store.entryAtPath(commit.treeSha, WORKFLOW_DIR);
    if (!directory || directory.type !== "tree") return [];
    const entries = (await store.listTree(directory.sha, WORKFLOW_DIR))
      .filter((entry) => entry.type !== "tree" && WORKFLOW_FILE.test(entry.name))
      .slice(0, MAX_WORKFLOW_FILES);
    const files = await Promise.all(
      entries.map(async (entry) => {
        const bytes = await store.blob(entry.sha);
        if (!bytes) return null;
        if (bytes.byteLength > MAX_WORKFLOW_BYTES) return { path: entry.path, source: "", tooLarge: true };
        return { path: entry.path, source: new TextDecoder().decode(bytes), tooLarge: false };
      }),
    );
    return files.filter((file): file is WorkflowFile => file !== null);
  });
}

function eventContext(
  repo: RepoRow,
  trigger: Trigger,
  runId: string,
  runNumber: number,
  workflowName: string,
): EventContext {
  const refName = trigger.ref.replace(/^refs\/(heads|tags)\//, "");
  return {
    sha: trigger.sha,
    ref: trigger.ref,
    ref_name: refName,
    ref_type: trigger.ref.startsWith("refs/tags/") ? "tag" : "branch",
    repository: `${repo.owner_name}/${repo.name}`,
    repository_owner: repo.owner_name,
    event_name: trigger.event,
    actor: trigger.actorName,
    run_id: runId,
    run_number: runNumber,
    workflow: workflowName,
    server_url: env.PUBLIC_URL,
    head_ref: trigger.headRef ?? "",
    base_ref: trigger.baseRef ?? "",
    event: {
      action: trigger.action ?? "",
      number: trigger.prNumber ?? 0,
      inputs: trigger.inputs,
    },
  };
}

async function allocateRunNumber(repoId: string): Promise<number> {
  const row = await first<{ run_number: number }>(
    "UPDATE repos SET next_run_number = next_run_number + 1 WHERE id = ? RETURNING next_run_number - 1 AS run_number",
    repoId,
  );
  if (!row) throw new Error("Repository disappeared while scheduling workflows");
  return row.run_number;
}

export function statusContext(workflowName: string, jobName: string): string {
  return `${workflowName} / ${jobName}`;
}

export function runUrl(repo: Pick<RepoRow, "owner_name" | "name">, runNumber: number): string {
  return publicUrl(`/${repo.owner_name}/${repo.name}/actions/runs/${runNumber}`);
}

async function insertRun(input: {
  repo: RepoRow;
  trigger: Trigger;
  path: string;
  workflowName: string;
  runId: string;
  runNumber: number;
  jobs: CompiledJob[];
  error: string | null;
  approvalRequired: boolean;
}): Promise<void> {
  const now = Date.now();
  const { repo, trigger } = input;
  const statements = [
    stmt(
      `INSERT INTO workflow_runs (id, repo_id, source_repo_id, run_number, workflow_path, workflow_name, event, ref, sha,
         title, actor_id, pr_number, approval_required, status, conclusion, created_at, completed_at)
       VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
      input.runId,
      repo.id,
      trigger.sourceRepo.id,
      input.runNumber,
      input.path,
      input.workflowName,
      trigger.event,
      trigger.ref,
      trigger.sha,
      input.error ? `Invalid workflow: ${input.error}`.slice(0, 500) : trigger.title,
      trigger.actorId,
      trigger.prNumber,
      input.approvalRequired ? 1 : 0,
      input.error ? "completed" : "queued",
      input.error ? "failure" : null,
      now,
      input.error ? now : null,
    ),
  ];
  for (const job of input.jobs) {
    const jobId = randomId();
    statements.push(
      stmt(
        `INSERT INTO workflow_jobs (id, run_id, repo_id, job_key, name, runs_on, needs, spec, status, created_at)
         VALUES (?, ?, ?, ?, ?, ?, ?, ?, 'pending', ?)`,
        jobId,
        input.runId,
        repo.id,
        job.key,
        job.name,
        JSON.stringify(job.runsOn),
        JSON.stringify(job.needs),
        JSON.stringify(job),
        now,
      ),
    );
    for (const step of job.steps) {
      statements.push(
        stmt(
          "INSERT INTO job_steps (job_id, idx, name, status) VALUES (?, ?, ?, 'pending')",
          jobId,
          step.index,
          step.name,
        ),
      );
    }
  }
  await batch(statements);
}

/** Discovers matching workflow files at the trigger commit and enqueues a durable run for each. */
export async function scheduleWorkflows(
  repo: RepoRow,
  trigger: Trigger,
  options: { forceWorkflowPath?: boolean; requireApproval?: boolean } = {},
): Promise<string[]> {
  if (repo.has_actions !== 1) return [];
  const discovered = await readWorkflowFiles(trigger);
  const files = options.forceWorkflowPath
    ? discovered.filter((file) => file.path === trigger.workflowPath)
    : discovered;
  const runIds: string[] = [];

  for (const file of files) {
    if (await atRunCap(repo.id, options.requireApproval === true)) {
      console.warn("Skipping workflow scheduling: too many active runs", repo.id);
      break;
    }
    const parsed = file.tooLarge
      ? { ok: false as const, error: `workflow files must be smaller than ${MAX_WORKFLOW_BYTES / 1024} KB` }
      : parseWorkflow(file.source);
    if (!parsed.ok) {
      if (trigger.event === "workflow_dispatch" && trigger.workflowPath !== file.path) continue;
      const runId = randomId();
      const runNumber = await allocateRunNumber(repo.id);
      await insertRun({
        repo,
        trigger,
        path: file.path,
        workflowName: file.path,
        runId,
        runNumber,
        jobs: [],
        error: parsed.error,
        approvalRequired: false,
      });
      await setCommitStatus({
        repoId: repo.id,
        sha: trigger.sha,
        context: file.path,
        state: "error",
        description: "Invalid workflow file",
        targetUrl: runUrl(repo, runNumber),
      });
      continue;
    }
    if (!options.forceWorkflowPath && !workflowMatches(parsed.spec, trigger, file.path)) continue;

    const workflowName = parsed.spec.name ?? file.path.split("/").pop() ?? file.path;
    const runId = randomId();
    const runNumber = await allocateRunNumber(repo.id);
    let jobs: CompiledJob[] = [];
    let error: string | null = null;
    try {
      jobs = compileWorkflow(parsed.spec, eventContext(repo, trigger, runId, runNumber, workflowName), trigger.inputs);
      const stepRows = jobs.reduce((sum, job) => sum + job.steps.length, 0);
      if (jobs.length > MAX_JOBS_PER_RUN) throw new Error(`a run may contain at most ${MAX_JOBS_PER_RUN} jobs`);
      if (stepRows > MAX_STEP_ROWS_PER_RUN)
        throw new Error(`a run may contain at most ${MAX_STEP_ROWS_PER_RUN} steps in total`);
    } catch (compileError) {
      error = compileError instanceof Error ? compileError.message : "Failed to compile workflow";
    }
    const approvalRequired = options.requireApproval === true;
    if (approvalRequired && trigger.prNumber !== null)
      await supersedePendingApprovals(repo.id, trigger.prNumber, file.path);
    await insertRun({ repo, trigger, path: file.path, workflowName, runId, runNumber, jobs, error, approvalRequired });
    if (error) continue;

    await Promise.all(
      jobs.map((job) =>
        setCommitStatus({
          repoId: repo.id,
          sha: trigger.sha,
          context: statusContext(workflowName, job.name),
          state: "pending",
          description: approvalRequired ? "Awaiting maintainer approval" : "Queued",
          targetUrl: runUrl(repo, runNumber),
        }),
      ),
    );
    if (!approvalRequired) await startRun(runId);
    runIds.push(runId);
  }
  return runIds;
}

export async function startRun(runId: string): Promise<void> {
  await env.CI_WORKFLOW.create({ id: runId, params: { runId } });
}

async function atRunCap(repoId: string, approval: boolean): Promise<boolean> {
  const active = await first<{ count: number }>(
    "SELECT COUNT(*) AS count FROM workflow_runs WHERE repo_id = ? AND status != 'completed' AND approval_required = ?",
    repoId,
    approval ? 1 : 0,
  );
  return (active?.count ?? 0) >= (approval ? CAPS.pendingApprovalsPerRepo : CAPS.activeRunsPerRepo);
}

/** A new push to a fork PR replaces its older unapproved runs so one author can't fill the approval queue. */
async function supersedePendingApprovals(repoId: string, prNumber: number, workflowPath: string): Promise<void> {
  const now = Date.now();
  await batch([
    stmt(
      `UPDATE workflow_jobs SET status = 'completed', conclusion = 'cancelled', completed_at = ?
       WHERE run_id IN (SELECT id FROM workflow_runs WHERE repo_id = ? AND pr_number = ? AND workflow_path = ?
         AND approval_required = 1 AND status = 'queued')`,
      now,
      repoId,
      prNumber,
      workflowPath,
    ),
    stmt(
      `UPDATE workflow_runs SET status = 'completed', conclusion = 'cancelled', completed_at = ?
       WHERE repo_id = ? AND pr_number = ? AND workflow_path = ? AND approval_required = 1 AND status = 'queued'`,
      now,
      repoId,
      prNumber,
      workflowPath,
    ),
  ]);
}