import { env, WorkflowEntrypoint } from "cloudflare:workers";
import { numberVar } from "../limits";
import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers";
import { all, first, run } from "../db";
import { loadRepoById } from "../repos";
import { emitNotification, emitWebhook, publicUrl } from "../events/emit";
import { completeJob, isCloudJob, runnerStub } from "./jobs";
import type { WorkflowJobRow, WorkflowRunRow } from "./jobs";
import type { CompiledJob } from "./compile";
import type { Conclusion } from "~/lib/types";

export type CiRunParams = { runId: string };

type JobState = {
  id: string;
  key: string;
  baseKey: string;
  needs: string[];
  status: string;
  conclusion: Conclusion | null;
  cloud: boolean;
  runIfOk: boolean;
  runIfFailed: boolean;
};

const WAIT_TIMEOUT = "5 minutes";
const MAX_ITERATIONS = 1000;
const QUEUE_DEADLINE_MS = 24 * 60 * 60 * 1000;
const DAY_MS = 24 * 60 * 60 * 1000;
const MINUTE_MS = 60 * 1000;
const CAPACITY_RETRY = "1 minute";
const DEFAULT_JOBS_PER_OWNER = 4;
const DEFAULT_MINUTES_PER_DAY = 600;
const CLOUD_RUNS_ON = ['["cloudflare"]', '["ubuntu-latest"]'];

type OwnerQuota = { ownerId: string; quotaExceeded: boolean; createdAt: number };

/**
 * Daily Container-minute budget per owner. Running and queued jobs are charged their full timeout
 * up front so an owner just under budget can't start a burst of long jobs.
 */
async function ownerQuota(runId: string): Promise<OwnerQuota | null> {
  const owner = await first<{ owner_id: string; created_at: number }>(
    "SELECT r.owner_id, w.created_at FROM workflow_runs w JOIN repos r ON r.id = w.repo_id WHERE w.id = ?",
    runId,
  );
  if (!owner) return null;
  const now = Date.now();
  const usage = await first<{ used_ms: number | null }>(
    `SELECT SUM(CASE WHEN j.status = 'completed' THEN COALESCE(j.completed_at - j.started_at, 0)
                     ELSE json_extract(j.spec, '$.timeoutMinutes') * ?1 END) AS used_ms
     FROM workflow_jobs j JOIN repos r ON r.id = j.repo_id
     WHERE r.owner_id = ?2 AND j.runs_on IN (?3, ?4) AND j.status IN ('completed', 'queued', 'in_progress')
       AND (j.status != 'completed' OR j.completed_at > ?5)`,
    MINUTE_MS,
    owner.owner_id,
    CLOUD_RUNS_ON[0],
    CLOUD_RUNS_ON[1],
    now - DAY_MS,
  );
  const minutes = numberVar(env.CLOUD_MINUTES_PER_DAY, DEFAULT_MINUTES_PER_DAY);
  return {
    ownerId: owner.owner_id,
    quotaExceeded: (usage?.used_ms ?? 0) >= minutes * MINUTE_MS,
    createdAt: owner.created_at,
  };
}

/** Atomically moves a cloud job to queued only while the owner is under its Container concurrency limit. */
async function claimCloudSlot(jobId: string, ownerId: string): Promise<boolean> {
  const result = await run(
    `UPDATE workflow_jobs SET status = 'queued' WHERE id = ?1 AND status = 'pending' AND (
       SELECT COUNT(*) FROM workflow_jobs j JOIN repos r ON r.id = j.repo_id
       WHERE r.owner_id = ?2 AND j.runs_on IN (?3, ?4) AND j.status IN ('queued', 'in_progress')
     ) < ?5`,
    jobId,
    ownerId,
    CLOUD_RUNS_ON[0],
    CLOUD_RUNS_ON[1],
    numberVar(env.CLOUD_JOBS_PER_OWNER, DEFAULT_JOBS_PER_OWNER),
  );
  return result.meta.changes === 1;
}

async function loadStates(runId: string): Promise<JobState[]> {
  const rows = await all<WorkflowJobRow>(
    "SELECT * FROM workflow_jobs WHERE run_id = ? ORDER BY created_at, job_key",
    runId,
  );
  return rows.map((row) => {
    const spec: CompiledJob = JSON.parse(row.spec);
    return {
      id: row.id,
      key: row.job_key,
      baseKey: spec.baseKey,
      needs: spec.needs,
      status: row.status,
      conclusion: row.conclusion,
      cloud: isCloudJob(spec.runsOn),
      runIfOk: spec.runIfOk,
      runIfFailed: spec.runIfFailed,
    };
  });
}

function dependencyOutcome(job: JobState, jobs: JobState[]): "waiting" | "ok" | "failed" {
  const dependencies = jobs.filter((candidate) => job.needs.includes(candidate.baseKey));
  if (dependencies.some((dependency) => dependency.status !== "completed")) return "waiting";
  return dependencies.every((dependency) => dependency.conclusion === "success" || dependency.conclusion === "skipped")
    ? "ok"
    : "failed";
}

function runConclusion(jobs: JobState[]): Conclusion {
  if (jobs.length === 0) return "skipped";
  if (jobs.some((job) => job.conclusion === "failure")) return "failure";
  if (jobs.some((job) => job.conclusion === "cancelled")) return "cancelled";
  if (jobs.every((job) => job.conclusion === "skipped")) return "skipped";
  return "success";
}

/** Durable orchestration of one workflow run: dispatches jobs as their `needs` resolve and waits for completion events. */
export class CiRunWorkflow extends WorkflowEntrypoint<Env, CiRunParams> {
  override async run(event: Readonly<WorkflowEvent<CiRunParams>>, step: WorkflowStep): Promise<void> {
    const { runId } = event.payload;
    const started = await step.do("start", async () => {
      const row = await first<{ status: string }>("SELECT status FROM workflow_runs WHERE id = ?", runId);
      if (!row || row.status === "completed") return false;
      await run("UPDATE workflow_runs SET status = 'in_progress', started_at = ? WHERE id = ?", Date.now(), runId);
      return true;
    });
    if (!started) return;

    for (let iteration = 0; iteration < MAX_ITERATIONS; iteration++) {
      const progress = await step.do(`schedule-${iteration}`, async () => {
        const jobs = await loadStates(runId);
        let dispatched = 0;
        let deferred = 0;
        const quota = jobs.some((job) => job.cloud && job.status === "pending") ? await ownerQuota(runId) : null;
        const pastDeadline = quota !== null && Date.now() - quota.createdAt > QUEUE_DEADLINE_MS;
        for (const job of jobs.filter((candidate) => candidate.status === "pending")) {
          const outcome = dependencyOutcome(job, jobs);
          if (outcome === "waiting") continue;
          const shouldRun = outcome === "ok" ? job.runIfOk : job.runIfFailed;
          if (!shouldRun) {
            await completeJob(job.id, "skipped");
            job.status = "completed";
            job.conclusion = "skipped";
            continue;
          }
          if (job.cloud && (quota === null || quota.quotaExceeded || pastDeadline)) {
            await completeJob(job.id, "cancelled");
            job.status = "completed";
            job.conclusion = "cancelled";
            continue;
          }
          if (job.cloud && quota) {
            if (!(await claimCloudSlot(job.id, quota.ownerId))) {
              deferred++;
              continue;
            }
          } else {
            await run("UPDATE workflow_jobs SET status = 'queued' WHERE id = ? AND status = 'pending'", job.id);
          }
          job.status = "queued";
          if (job.cloud) {
            try {
              await runnerStub(job.id).startJob(job.id);
            } catch (error) {
              console.error("Failed to start cloud job", job.id, error);
              await completeJob(job.id, "failure");
            }
          }
          dispatched++;
        }
        const refreshed = await loadStates(runId);
        const active = refreshed.filter((job) => job.status !== "completed");
        const running = active.filter((job) => job.status === "queued" || job.status === "in_progress");
        if (deferred > 0 && running.length === 0) return { active: active.length, dispatched, deferred };
        if (active.length > 0 && running.length === 0) {
          for (const job of active) await completeJob(job.id, "cancelled");
          return { active: 0, dispatched, deferred };
        }
        return { active: active.length, dispatched, deferred };
      });
      if (progress.active === 0) break;
      if (progress.deferred > 0 && progress.dispatched === 0) {
        await step.sleep(`capacity-${iteration}`, CAPACITY_RETRY);
        continue;
      }
      try {
        await step.waitForEvent(`wait-${iteration}`, { type: "job-done", timeout: WAIT_TIMEOUT });
      } catch {
        await step.do(`expire-${iteration}`, async () => {
          const stale = await all<{ id: string }>(
            "SELECT id FROM workflow_jobs WHERE run_id = ? AND status = 'queued' AND created_at < ?",
            runId,
            Date.now() - QUEUE_DEADLINE_MS,
          );
          for (const job of stale) await completeJob(job.id, "cancelled");
        });
      }
    }

    await step.do("finish", async () => {
      for (const job of await loadStates(runId)) {
        if (job.status === "completed") continue;
        if (job.cloud && job.status === "in_progress") await runnerStub(job.id).cancel("workflow ended");
        else await completeJob(job.id, "cancelled");
      }
      const jobs = await loadStates(runId);
      const conclusion = runConclusion(jobs);
      await run(
        "UPDATE workflow_runs SET status = 'completed', conclusion = ?, completed_at = ? WHERE id = ? AND status != 'completed'",
        conclusion,
        Date.now(),
        runId,
      );
      const runRow = await first<WorkflowRunRow>("SELECT * FROM workflow_runs WHERE id = ?", runId);
      const repo = runRow ? await loadRepoById(runRow.repo_id) : null;
      if (!runRow || !repo) return;
      const url = publicUrl(`/${repo.owner_name}/${repo.name}/actions/runs/${runRow.run_number}`);
      await emitWebhook(repo.id, "workflow_run", {
        action: "completed",
        workflow_run: {
          id: runRow.id,
          run_number: runRow.run_number,
          name: runRow.workflow_name,
          event: runRow.event,
          head_sha: runRow.sha,
          head_branch: runRow.ref.replace(/^refs\/heads\//, ""),
          status: "completed",
          conclusion,
          html_url: url,
        },
      });
      if (runRow.actor_id && conclusion === "failure") {
        await emitNotification({
          repoId: repo.id,
          issueId: null,
          actorId: null,
          reason: "ci",
          title: `${runRow.workflow_name} failed on ${runRow.ref.replace(/^refs\/heads\//, "")}`,
          url,
          mentionText: null,
          directRecipients: [runRow.actor_id],
        });
      }
    });
  }
}