//! OxygenFlow runs, jobs and logs (hub spec §13.2).

use super::{now_ms, Pool};

use sqlx::FromRow;

#[derive(Clone, Debug, FromRow)]

pub struct Run {

    pub id: i64,

    pub repo_id: i64,

    /// The run's number within its repository, from 1 (migration 0019).

    #[sqlx(default)]

    pub number: i64,

    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or

    /// `rerun:N:EVENT` (migration 0020).

    #[sqlx(default)]

    pub trigger: String,

    pub bookmark: String,

    pub state: String,

    pub heads: String,

    pub pipeline_hash: String,

    pub status: String,

    pub error: String,

    pub actor: String,

    pub created_ms: i64,

    pub finished_ms: Option<i64>,

    /// `urgent` when expedited (oxygenhub#244), else empty: the run follows

    /// its repository's, organisation's or the site's level.

    #[sqlx(default)]

    pub priority: String,

    /// Who expedited it and why.

    #[sqlx(default)]

    pub priority_note: String,

}

/// The priority levels, most urgent first (oxygenhub#244).

pub const PRIORITIES: [&str; 4] = ["urgent", "high", "normal", "low"];

/// SQL for the level a run's jobs get from where they are, as a rank

/// (0 urgent … 3 low): the run's own (expedited), else its repository's,

/// its organisation's, the site's (`settings.flow_priority`), else normal.

macro_rules! level_rank {

    () => {

        "CASE COALESCE(NULLIF(r.priority, ''), NULLIF(rep.flow_priority, ''), \

           NULLIF((SELECT o.flow_priority FROM organisations o WHERE o.name = rep.owner), ''), \

           NULLIF((SELECT s.value FROM settings s WHERE s.key = 'flow_priority'), ''), 'normal') \

         WHEN 'urgent' THEN 0 WHEN 'high' THEN 1 WHEN 'low' THEN 3 ELSE 2 END"

    };

}

/// SQL for a queued job's level as a rank (0 urgent … 3 low): the level

/// its run gets from where it is, lowered by the job's own priority from

/// its pipeline (`low`, `normal`), never raised by it; an expedited run is

/// urgent whatever its jobs say (oxygenhub#244). Needs `j` (the job), `r`

/// (its run) and `rep` (its repository) in the query.

pub const PRIORITY_RANK: &str = concat!(

    "CASE WHEN r.priority <> '' THEN ",

    level_rank!(),

    " WHEN j.priority = 'low' THEN 3 WHEN j.priority = 'normal' AND (",

    level_rank!(),

    ") < 2 THEN 2 ELSE ",

    level_rank!(),

    " END"

);

/// SQL for why a job stands where it does, beside [`PRIORITY_RANK`]: an

/// expedited run's note, else the pipeline's word for a job it lowered.

pub const PRIORITY_NOTE: &str =

    "CASE WHEN r.priority <> '' THEN r.priority_note WHEN j.priority <> '' THEN 'pipeline: ' || j.priority ELSE r.priority_note END";

/// How many runs of `owner`'s repositories are expedited and still live.

pub async fn expedited_live(pool: &Pool, owner: &str) -> sqlx::Result<i64> {

    sqlx::query_scalar(

        "SELECT COUNT(*) FROM flow_runs r JOIN repositories rep ON rep.id = r.repo_id \

         WHERE rep.owner = $1 AND r.priority = 'urgent' AND r.status IN ('queued', 'running')",

    )

    .bind(owner)

    .fetch_one(pool)

    .await

}

/// Raise run `run_id` to urgent, saying who and why; a run already ended

/// is left alone. Whether it changed.

pub async fn expedite(pool: &Pool, run_id: i64, note: &str) -> sqlx::Result<bool> {

    let r = sqlx::query("UPDATE flow_runs SET priority = 'urgent', priority_note = $1 WHERE id = $2 AND status IN ('queued', 'running')")

        .bind(note)

        .bind(run_id)

        .execute(pool)

        .await?;

    Ok(r.rows_affected() > 0)

}

impl Run {

    /// How the run page says what started it.

    pub fn started_by(&self) -> String {

        match self.trigger.as_str() {

            "manual" => format!("started by {}", self.actor),

            "schedule" => "started on schedule".to_string(),

            t if t.starts_with("tag:") => format!("tag {} pushed by {}", &t[4..], self.actor),

            t if t.starts_with("rerun:") => format!("re-run of #{} by {}", self.rerun_of().unwrap_or(""), self.actor),

            _ => format!("pushed by {}", self.actor),

        }

    }

    /// The run a rerun repeats: `N` of `rerun:N` or `rerun:N:EVENT`.

    pub fn rerun_of(&self) -> Option<&str> {

        let rest = self.trigger.strip_prefix("rerun:")?;

        Some(rest.split(':').next().unwrap_or(rest))

    }

    /// The trigger as people read it: `rerun:N`, without the event it repeats.

    pub fn trigger_label(&self) -> String {

        match self.rerun_of() {

            Some(n) => format!("rerun:{n}"),

            None => self.trigger.clone(),

        }

    }

    pub fn head_list(&self) -> Vec<&str> {

        self.heads.split('+').filter(|h| !h.is_empty()).collect()

    }

    pub fn live(&self) -> bool {

        matches!(self.status.as_str(), "queued" | "running")

    }

}

#[derive(Clone, Debug, Default, FromRow)]

pub struct Job {

    pub id: i64,

    pub run_id: i64,

    pub name: String,

    pub depth: i64,

    pub depends: String,

    pub spec: String,

    pub sandbox: String,

    pub runtime: String,

    pub status: String,

    pub runner: String,

    pub token_hash: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    pub exit_code: Option<i64>,

    pub error: String,

    pub receipt: String,

    pub receipt_hash: String,

    pub log_count: i64,

    /// Comma-separated ids of other repositories the job token may read.

    pub also_read: String,

    /// The receipt attestation kept in the repository (Oxygen spec §14.6).

    pub receipt_attest: String,

    /// Who signed it: the runner's identity name.

    pub receipt_signer: String,

    /// Why a signed receipt was not kept.

    pub receipt_note: String,

    /// Why a queued job no runner can take waits (spec §11.4), from the

    /// last placement pass; empty when it is only waiting its turn.

    pub waiting: String,

    /// What it took, as its runner reported (oxygenflow#41), in MiB.

    pub peak_disk_mib: Option<i64>,

    pub peak_memory_mib: Option<i64>,

    /// Where its time went, JSON of phase to milliseconds (oxygenflow#35).

    #[sqlx(default)]

    pub phases: String,

    /// The free disk it needs, learned from past runs of the same job.

    pub need_disk_mib: Option<i64>,

    /// Why it ran on its runner (oxygenflow#42).

    pub placed: String,

    /// `waiting` for a person, `approved`, or empty (oxygenflow#57).

    pub approval: String,

    /// Who approved or rejected it, and when.

    pub approved_by: String,

    pub approved_ms: Option<i64>,

    /// The agent token it was decided through, if one (oxygenhub#229).

    #[sqlx(default)]

    pub approved_via: String,

    /// The environment it deploys to, or empty (oxygenflow#59).

    pub environment: String,

    /// The approval attestation that released it, when sealed (#235).

    pub approval_seal: String,

    /// When it entered the queue (made, or run again), and when it became

    /// ready to run: set at its claim (migration 0080, oxygenhub#257).

    #[sqlx(default)]

    pub queued_ms: Option<i64>,

    #[sqlx(default)]

    pub ready_ms: Option<i64>,

    /// When a runner last handed it back, drained or with its lease lapsed

    /// (oxygenflow#76, migration 0094), and how many times: such a job is

    /// queued ahead of new work.

    #[sqlx(default)]

    pub handed_back_ms: Option<i64>,

    #[sqlx(default)]

    pub handbacks: i64,

}

impl Job {

    /// Where its time went, as its runner reported (oxygenflow#35): each

    /// phase it took, in the order it took them, with its milliseconds.

    pub fn phase_list(&self) -> Vec<(String, u64)> {

        let m: std::collections::BTreeMap<String, u64> = serde_json::from_str(&self.phases).unwrap_or_default();

        ["image", "prepare", "run", "finish"].iter().filter_map(|p| m.get(*p).map(|ms| (p.to_string(), *ms))).collect()

    }

    /// The run whose pass this job reused (#265), if it did.

    pub fn reused_from(&self) -> Option<i64> {

        self.error.strip_prefix("reused from run #")?.parse().ok()

    }

    pub fn depends_on(&self) -> Vec<String> {

        serde_json::from_str(&self.depends).unwrap_or_default()

    }

    pub fn done(&self) -> bool {

        matches!(self.status.as_str(), "passed" | "failed" | "skipped" | "error" | "cancelled")

    }

}

#[derive(Clone, Debug, FromRow)]

pub struct LogLine {

    pub seq: i64,

    pub t: i64,

    pub stream: String,

    pub text: String,

}

/// A job as the trigger creates it.

pub struct NewJob {

    pub name: String,

    pub depth: i64,

    pub depends: Vec<String>,

    pub spec: String,

    pub sandbox: String,

    pub runtime: String,

    /// The free disk it needs, from [`learned_disk_need`].

    pub need_disk_mib: Option<i64>,

    /// A pass on the same state to reuse instead of running it (#265): the

    /// job is created passed, with that pass's receipt.

    pub reused: Option<Reused>,

}

/// A job's earlier pass on the same state, spec and sandbox (#265).

#[derive(Clone, Debug)]

pub struct Reused {

    /// The run it passed in.

    pub run_number: i64,

    pub receipt: String,

    pub receipt_hash: String,

    pub receipt_attest: String,

    pub receipt_signer: String,

    pub receipt_note: String,

}

impl Reused {

    /// What the reused job says, and how [`Job::reused_from`] reads it back.

    pub fn note(&self) -> String {

        format!("reused from run #{}", self.run_number)

    }

}

/// The newest pass of job `name` with exactly this spec and sandbox on

/// `state` in this repository (#265).

pub async fn passed_on_state(

    pool: &Pool,

    repo_id: i64,

    state: &str,

    name: &str,

    spec: &str,

    sandbox: &str,

) -> sqlx::Result<Option<Reused>> {

    let row: Option<(i64, String, String, String, String, String)> = sqlx::query_as(

        "SELECT r.number, j.receipt, j.receipt_hash, j.receipt_attest, j.receipt_signer, j.receipt_note \

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE r.repo_id = $1 AND r.state = $2 AND j.name = $3 AND j.spec = $4 AND j.sandbox = $5 AND j.status = 'passed' \

         ORDER BY j.id DESC LIMIT 1",

    )

    .bind(repo_id)

    .bind(state)

    .bind(name)

    .bind(spec)

    .bind(sandbox)

    .fetch_optional(pool)

    .await?;

    Ok(row.map(|(run_number, receipt, receipt_hash, receipt_attest, receipt_signer, receipt_note)| Reused {

        run_number,

        receipt,

        receipt_hash,

        receipt_attest,

        receipt_signer,

        receipt_note,

    }))

}

/// Runs of a job whose peaks set its next need.

const NEED_FROM_RUNS: i64 = 5;

/// The free disk a new run of job `name` in repository `repo_id` needs

/// (oxygenflow#41): the most its last runs took, plus a quarter; `None`

/// for a job that has not run here yet.

pub async fn learned_disk_need(pool: &Pool, repo_id: i64, name: &str) -> sqlx::Result<Option<i64>> {

    let peak: Option<i64> = sqlx::query_scalar(

        "SELECT MAX(peak_disk_mib) FROM (SELECT j.peak_disk_mib FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id \

         WHERE r.repo_id = $1 AND j.name = $2 AND j.peak_disk_mib IS NOT NULL ORDER BY j.id DESC LIMIT $3) recent",

    )

    .bind(repo_id)

    .bind(name)

    .bind(NEED_FROM_RUNS)

    .fetch_one(pool)

    .await?;

    Ok(peak.map(|p| p + p / 4 + 64))

}

/// Record why job `job_id` ran where it did.

pub async fn set_placed(pool: &Pool, job_id: i64, placed: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET placed = $1 WHERE id = $2").bind(placed).bind(job_id).execute(pool).await?;

    Ok(())

}

/// Keep what a job took, as its runner reported.

pub async fn set_peaks(pool: &Pool, job_id: i64, disk_mib: Option<u64>, memory_mib: Option<u64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_jobs SET peak_disk_mib = $1, peak_memory_mib = $2 WHERE id = $3")

        .bind(disk_mib.map(|v| v.min(i64::MAX as u64) as i64))

        .bind(memory_mib.map(|v| v.min(i64::MAX as u64) as i64))

        .bind(job_id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Where a finished job's time went, as its runner reported (oxygenflow#35).

pub async fn set_phases(pool: &Pool, job_id: i64, phases: &std::collections::BTreeMap<String, u64>) -> sqlx::Result<()> {

    let json = if phases.is_empty() { String::new() } else { serde_json::to_string(phases).unwrap_or_default() };

    sqlx::query("UPDATE flow_jobs SET phases = $1 WHERE id = $2").bind(json).bind(job_id).execute(pool).await?;

    Ok(())

}

pub struct NewRun<'a> {

    pub repo_id: i64,

    pub bookmark: &'a str,

    pub state: &'a str,

    pub heads: &'a str,

    pub pipeline_hash: &'a str,

    pub actor: &'a str,

    /// `push`, `proposal`, `tag:NAME`, `manual`, `schedule` or

    /// `rerun:N:EVENT`.

    pub trigger: &'a str,

    /// Problems when the pipeline is invalid: the run is recorded in error.

    pub error: Option<&'a str>,

}

/// A push that asked for no run (#284): recorded, finished, as skipped,

/// with `why` (which change asked, and who pushed it) as its note.

pub async fn create_skipped_run(pool: &Pool, r: NewRun<'_>, why: &str) -> sqlx::Result<i64> {

    let mut tx = super::begin_write(pool).await?;

    let now = now_ms();

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \

         VALUES ($1, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = $2), $3, $4, $5, $6, 'skipped', $7, $8, $9, $10, $11) RETURNING id",

    )

    .bind(r.repo_id)

    .bind(r.repo_id)

    .bind(r.bookmark)

    .bind(r.state)

    .bind(r.heads)

    .bind(r.pipeline_hash)

    .bind(why)

    .bind(r.actor)

    .bind(r.trigger)

    .bind(now)

    .bind(now)

    .fetch_one(&mut *tx)

    .await?;

    tx.commit().await?;

    Ok(id)

}

pub async fn create_run(pool: &Pool, r: NewRun<'_>, jobs: &[NewJob]) -> sqlx::Result<i64> {

    let mut tx = super::begin_write(pool).await?;

    let now = now_ms();

    let status = if r.error.is_some() { "error" } else { "queued" };

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runs (repo_id, number, bookmark, state, heads, pipeline_hash, status, error, actor, trigger, created_ms, finished_ms) \

         VALUES ($1, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = $2), $3, $4, $5, $6, $7, $8, $9, $10, $11, $12) RETURNING id",

    )

    .bind(r.repo_id)

    .bind(r.repo_id)

    .bind(r.bookmark)

    .bind(r.state)

    .bind(r.heads)

    .bind(r.pipeline_hash)

    .bind(status)

    .bind(r.error.unwrap_or(""))

    .bind(r.actor)

    .bind(r.trigger)

    .bind(now)

    .bind(r.error.map(|_| now))

    .fetch_one(&mut *tx)

    .await?;

    for j in jobs {

        let job_id: i64 = sqlx::query_scalar(

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib, approval, environment, priority, queued_ms) \

             VALUES ($1, $2, $3, $4, $5, $6, $7, 'queued', $8, $9, $10, $11, $12) RETURNING id",

        )

        .bind(id)

        .bind(&j.name)

        .bind(j.depth)

        .bind(serde_json::to_string(&j.depends).unwrap_or_else(|_| "[]".into()))

        .bind(&j.spec)

        .bind(&j.sandbox)

        .bind(&j.runtime)

        .bind(j.need_disk_mib)

        .bind(if needs_approval(&j.spec) { "waiting" } else { "" })

        .bind(environment_of(&j.spec))

        .bind(priority_of(&j.spec))

        .bind(now)

        .fetch_one(&mut *tx)

        .await?;

        // A pass reused (#265): done before any runner can claim it.

        if let Some(r) = &j.reused {

            sqlx::query(

                "UPDATE flow_jobs SET status = 'passed', exit_code = 0, error = $1, receipt = $2, receipt_hash = $3, receipt_attest = $4, \

                 receipt_signer = $5, receipt_note = $6, started_ms = $7, finished_ms = $7 WHERE id = $8",

            )

            .bind(r.note())

            .bind(&r.receipt)

            .bind(&r.receipt_hash)

            .bind(&r.receipt_attest)

            .bind(&r.receipt_signer)

            .bind(&r.receipt_note)

            .bind(now)

            .bind(job_id)

            .execute(&mut *tx)

            .await?;

        }

    }

    // Every job reused: the run has passed already.

    if !jobs.is_empty() && jobs.iter().all(|j| j.reused.is_some()) {

        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = $1 WHERE id = $2").bind(now).bind(id).execute(&mut *tx).await?;

    }

    tx.commit().await?;

    Ok(id)

}

/// The environment a compiled job deploys to, or empty (oxygenflow#59).

fn environment_of(spec: &str) -> String {

    serde_json::from_str::<serde_json::Value>(spec).ok().and_then(|v| v["environment"].as_str().map(String::from)).unwrap_or_default()

}

/// A compiled job's own priority (`low`, `normal`), or empty; anything

/// else is not one a pipeline may give (oxygenhub#244).

fn priority_of(spec: &str) -> String {

    serde_json::from_str::<serde_json::Value>(spec)

        .ok()

        .and_then(|v| v["priority"].as_str().map(String::from))

        .filter(|p| p == "low" || p == "normal")

        .unwrap_or_default()

}

/// Whether a compiled job asks for approval (oxygenflow#57).

fn needs_approval(spec: &str) -> bool {

    serde_json::from_str::<serde_json::Value>(spec).is_ok_and(|v| !v["approval"].is_null())

}

/// The labels a compiled job and its sandbox ask a runner for (`runs-on`).

fn labels_in(spec: &str, sandbox: &str) -> Vec<String> {

    let mut out: Vec<String> = [spec, sandbox]

        .iter()

        .filter_map(|j| serde_json::from_str::<serde_json::Value>(j).ok())

        .flat_map(|v| v["runs_on"].as_array().cloned().unwrap_or_default())

        .filter_map(|l| l.as_str().map(String::from))

        .collect();

    out.sort();

    out.dedup();

    out

}

impl Job {

    /// The runner labels this job needs.

    pub fn needs(&self) -> Vec<String> {

        labels_in(&self.spec, &self.sandbox)

    }

    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).

    pub fn arch(&self) -> Option<String> {

        arch_in(&self.sandbox)

    }

}

/// The architecture a sandbox's JSON asks for, if any.

fn arch_in(sandbox: &str) -> Option<String> {

    serde_json::from_str::<serde_json::Value>(sandbox).ok()?["arch"].as_str().map(String::from)

}

/// A job waiting for or held by a runner, with where it comes from.

#[derive(Clone, Debug, FromRow)]

pub struct QueueEntry {

    pub job_id: i64,

    pub run_id: i64,

    pub repo_id: i64,

    pub owner: String,

    pub name: String,

    pub visibility: String,

    pub run_number: i64,

    pub job: String,

    pub status: String,

    /// The runner holding it, when running.

    pub runner: String,

    pub spec: String,

    pub sandbox: String,

    /// When it started running.

    pub started_ms: Option<i64>,

    /// Its level (0 urgent … 3 low, oxygenhub#244) and, when expedited, who

    /// did it and why.

    pub priority_rank: i64,

    pub priority_note: String,

}

impl QueueEntry {

    pub fn priority(&self) -> &'static str {

        PRIORITIES.get(self.priority_rank as usize).copied().unwrap_or("normal")

    }

}

impl QueueEntry {

    /// The runner labels the job needs.

    pub fn needs(&self) -> Vec<String> {

        labels_in(&self.spec, &self.sandbox)

    }

    /// The architecture the job's sandbox asks for (OxygenFlow spec §11.3).

    pub fn arch(&self) -> Option<String> {

        arch_in(&self.sandbox)

    }

}

/// Every job of a live run that is running or queued: running ones first,

/// then queued ones in the order runners claim them (hub spec §13.2).

pub async fn queue(pool: &Pool) -> sqlx::Result<Vec<QueueEntry>> {

    sqlx::query_as(&format!(

        "SELECT j.id AS job_id, j.run_id, r.repo_id, rep.owner, rep.name, rep.visibility, r.number AS run_number, j.name AS job, j.status, j.runner, j.spec, j.sandbox, j.started_ms, \

           CAST({PRIORITY_RANK} AS BIGINT) AS priority_rank, {PRIORITY_NOTE} AS priority_note \

         FROM flow_jobs j JOIN flow_runs r ON r.id = j.run_id JOIN repositories rep ON rep.id = r.repo_id \

         WHERE j.status IN ('queued', 'running') AND j.approval != 'waiting' AND r.status IN ('queued', 'running') \

         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 END, \

           CASE j.status WHEN 'running' THEN 0 ELSE {PRIORITY_RANK} END, \

           CASE j.status WHEN 'running' THEN 0 ELSE ROW_NUMBER() OVER (PARTITION BY r.repo_id, j.status ORDER BY j.run_id, j.depth, j.id) \

           + (SELECT COUNT(*) FROM flow_jobs x JOIN flow_runs xr ON xr.id = x.run_id WHERE x.status = 'running' AND xr.repo_id = r.repo_id) END, \

           j.run_id, j.depth, j.id"

    ))

    .fetch_all(pool)

    .await

}

/// A repository's runs, newest first, numbered below `before`.

pub async fn runs_by_number(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND number < $2 ORDER BY number DESC LIMIT $3")

        .bind(repo_id)

        .bind(before.unwrap_or(i64::MAX))

        .bind(limit)

        .fetch_all(pool)

        .await

}

/// The latest run on this bookmark of this repository, for what a push

/// changed since (a job's scope).

/// The run a push to `bookmark` is scoped against: the newest that was not

/// cancelled, nor a push run still going (a new push supersedes it), so

/// that what a cancelled run would have checked is checked by the next.

pub async fn latest_on(pool: &Pool, repo_id: i64, bookmark: &str) -> sqlx::Result<Option<Run>> {

    sqlx::query_as(

        "SELECT * FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND heads != '' AND status NOT IN ('cancelled', 'skipped')

         AND NOT (trigger = 'push' AND status IN ('queued', 'running')) ORDER BY id DESC LIMIT 1",

    )

    .bind(repo_id)

    .bind(bookmark)

    .fetch_optional(pool)

    .await

}

pub async fn runs(pool: &Pool, repo_id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = $1 AND id < $2 ORDER BY id DESC LIMIT $3")

        .bind(repo_id)

        .bind(before.unwrap_or(i64::MAX))

        .bind(limit)