//! 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`, `manual` or `rerun:N` (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>,

}

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>,

    /// 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,

}

impl Job {

    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>,

}

/// 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 = ? AND j.name = ? AND j.peak_disk_mib IS NOT NULL ORDER BY j.id DESC LIMIT ?)",

    )

    .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 = ? WHERE id = ?").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 = ?, peak_memory_mib = ? WHERE id = ?")

        .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(())

}

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`, `manual` or `rerun:N`.

    pub trigger: &'a str,

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

    pub error: Option<&'a str>,

}

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 (?, (SELECT COALESCE(MAX(number), 0) + 1 FROM flow_runs WHERE repo_id = ?), ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) 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 {

        sqlx::query(

            "INSERT INTO flow_jobs (run_id, name, depth, depends, spec, sandbox, runtime, status, need_disk_mib) VALUES (?, ?, ?, ?, ?, ?, ?, 'queued', ?)",

        )

        .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)

        .execute(&mut *tx)

        .await?;

    }

    tx.commit().await?;

    Ok(id)

}

/// 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,

}

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(

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

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

         WHERE j.status IN ('queued', 'running') AND r.status IN ('queued', 'running') \

         ORDER BY CASE j.status WHEN 'running' THEN 0 ELSE 1 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 = ? AND number < ? ORDER BY number DESC LIMIT ?")

        .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 = ? AND bookmark = ? AND heads != '' AND status != 'cancelled'

         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 = ? AND id < ? ORDER BY id DESC LIMIT ?")

        .bind(repo_id)

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

        .bind(limit)

        .fetch_all(pool)

        .await

}

/// A repository's run by its number there.

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

    sqlx::query_as("SELECT * FROM flow_runs WHERE repo_id = ? AND number = ?").bind(repo_id).bind(number).fetch_optional(pool).await

}

pub async fn run_by_id(pool: &Pool, id: i64) -> sqlx::Result<Option<Run>> {

    sqlx::query_as("SELECT * FROM flow_runs WHERE id = ?").bind(id).fetch_optional(pool).await

}

pub async fn jobs(pool: &Pool, run_id: i64) -> sqlx::Result<Vec<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE run_id = ? ORDER BY depth, name").bind(run_id).fetch_all(pool).await

}

pub async fn job(pool: &Pool, id: i64) -> sqlx::Result<Option<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE id = ?").bind(id).fetch_optional(pool).await

}

/// The running job a job token belongs to, while it may still be used.

pub async fn job_by_token(pool: &Pool, token_hash: &str, max_age_ms: i64) -> sqlx::Result<Option<Job>> {

    sqlx::query_as("SELECT * FROM flow_jobs WHERE token_hash = ? AND status = 'running' AND started_ms >= ?")

        .bind(token_hash)

        .bind(now_ms() - max_age_ms)

        .fetch_optional(pool)

        .await

}

/// Take the oldest queued job whose dependencies all passed and whose tier

/// the runner offers, and mark it running for `runner` with a token.

/// Let a job's token also read these repositories (its pinned subrepos).

pub async fn set_also_read(pool: &Pool, job_id: i64, repo_ids: &[i64]) -> sqlx::Result<()> {

    let list: Vec<String> = repo_ids.iter().map(|i| i.to_string()).collect();

    sqlx::query("UPDATE flow_jobs SET also_read = ? WHERE id = ?").bind(list.join(",")).bind(job_id).execute(pool).await?;

    Ok(())

}

impl Job {

    pub fn also_read_ids(&self) -> Vec<i64> {

        self.also_read.split(',').filter_map(|x| x.trim().parse().ok()).collect()

    }

}

/// What a runner can take on: free memory (MiB) and vCPUs, `None` for no

/// limit, and the rootfs images it holds, `None` when it does not say.

#[derive(Clone, Debug, Default)]

pub struct Room {

    pub memory_mib: Option<u32>,

    pub cpus: Option<u32>,

    pub images: Option<std::collections::HashSet<String>>,

    /// The runner's architecture (`x86_64`, `riscv64`, …), empty when unknown.

    pub arch: String,

    /// Each image's architecture as runners have reported it (`flow_images`).

    pub image_archs: std::collections::HashMap<String, String>,

    /// Jobs this runner declined, not to be offered to it again (spec §11.5).

    pub declined: std::collections::HashSet<i64>,

    /// Images the hub holds that this runner would fetch when it lacks

    /// them (oxygenflow#39); empty for a runner that does not fetch.

    pub fetchable: std::collections::HashSet<String>,

    /// Its free disk now, from its claim's facts (oxygenflow#40); `None`

    /// when it does not say.

    pub free_disk_mib: Option<u64>,

}

impl Room {

    /// Whether a job needing `need` MiB of disk fits in its free disk.

    pub fn disk_fits(&self, need: Option<i64>) -> bool {

        match (need, self.free_disk_mib) {

            (Some(n), Some(free)) => n.max(0) as u64 <= free,

            _ => true,

        }

    }

}

impl Room {

    /// Whether a job's sandbox (its stored JSON) fits: its memory and vCPUs,

    /// and its rootfs image among the runner's (oxygenhub#73).

    pub fn fits(&self, sandbox_json: &str) -> bool {

        let v: serde_json::Value = serde_json::from_str(sandbox_json).unwrap_or_default();

        let need = |k: &str| v[k].as_u64().unwrap_or(0);

        let rootfs = v["rootfs"].as_str().unwrap_or("");

        // The architecture: the sandbox's own ask, else the image's as known;

        // an unknown one matches, as before (OxygenFlow spec §11.3).

        let wanted = v["arch"].as_str().map(String::from).or_else(|| self.image_archs.get(rootfs).cloned());

        let arch_ok = self.arch.is_empty() || wanted.as_deref().is_none_or(|w| w == self.arch);

        arch_ok

            && self.memory_mib.is_none_or(|m| need("memory_mib") <= m as u64)

            && self.cpus.is_none_or(|c| need("cpus") <= c as u64)

            && (rootfs.is_empty() || self.images.as_ref().is_none_or(|have| have.contains(rootfs)) || self.fetchable.contains(rootfs))

    }

}

/// Remember which architecture each image is for, as a runner reports it.

pub async fn record_image_archs(pool: &Pool, archs: &std::collections::BTreeMap<String, String>) -> sqlx::Result<()> {

    for (hash, arch) in archs.iter().take(500) {

        if hash.len() != 64 || !flow_core::arch_ok(arch) {

            continue;

        }

        sqlx::query("INSERT INTO flow_images (hash, arch, seen_ms) VALUES (?, ?, ?) ON CONFLICT (hash) DO UPDATE SET arch = excluded.arch, seen_ms = excluded.seen_ms")

        .bind(hash.to_ascii_lowercase())

        .bind(arch)

        .bind(now_ms())

        .execute(pool)

        .await?;

    }

    Ok(())

}

/// Every image's architecture the hub knows.

pub async fn image_archs(pool: &Pool) -> sqlx::Result<std::collections::HashMap<String, String>> {

    let rows: Vec<(String, String)> = sqlx::query_as("SELECT hash, arch FROM flow_images").fetch_all(pool).await?;

    Ok(rows.into_iter().collect())

}

pub async fn claim(pool: &Pool, runtimes: &[String], runner: &str, token_hash: &str) -> sqlx::Result<Option<Job>> {

    claim_within(pool, runtimes, runner, token_hash, Room::default(), &Default::default(), &[]).await

}

/// The oldest runnable job that fits in `room`.

/// The first queued job a runner with these tiers, this room and this

/// scope may take, now running under `runner`. A scoped runner never sees

/// another scope's jobs (oxygenhub#41).

pub async fn claim_within(

    pool: &Pool,

    runtimes: &[String],

    runner: &str,

    token_hash: &str,

    room: Room,

    scope: &super::runners::Scope,

    labels: &[String],

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

    claim_ranked(pool, runtimes, runner, token_hash, room, scope, labels, &|_| false).await

}

/// [`claim_within`], passing over a job `elsewhere` says a better runner

/// should take (oxygenflow#42).

#[allow(clippy::too_many_arguments)]

pub async fn claim_ranked(

    pool: &Pool,

    runtimes: &[String],

    runner: &str,

    token_hash: &str,

    room: Room,

    scope: &super::runners::Scope,

    labels: &[String],

    elsewhere: &(dyn Fn(&Job) -> bool + Sync),

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

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

    let queued: Vec<Job> = sqlx::query_as(

        "SELECT j.* 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 = 'queued' AND r.status IN ('queued', 'running') \

         AND (? = '' OR rep.owner = ?) AND (? IS NULL OR r.repo_id = ?) \

         ORDER BY j.run_id, j.depth, j.id LIMIT 200",

    )

    .bind(&scope.owner)

    .bind(&scope.owner)

    .bind(scope.repo_id)

    .bind(scope.repo_id)

    .fetch_all(&mut *tx)

    .await?;

    for j in queued {

        if !runtimes.iter().any(|r| r == &j.runtime)

            || !room.fits(&j.sandbox)

            || !room.disk_fits(j.need_disk_mib)

            || room.declined.contains(&j.id)

        {

            continue;

        }

        // Every label the job and its sandbox ask for (`runs-on`).

        if !j.needs().iter().all(|l| labels.contains(l)) {

            continue;

        }

        let deps = j.depends_on();

        if !deps.is_empty() {

            let mut passed = 0i64;

            for d in &deps {

                let st: Option<String> = sqlx::query_scalar("SELECT status FROM flow_jobs WHERE run_id = ? AND name = ?")

                    .bind(j.run_id)

                    .bind(d)

                    .fetch_optional(&mut *tx)

                    .await?;

                if st.as_deref() == Some("passed") {

                    passed += 1;

                }

            }

            if passed != deps.len() as i64 {

                continue;

            }

        }

        if elsewhere(&j) {

            continue;

        }

        let now = now_ms();

        let done = sqlx::query(

            "UPDATE flow_jobs SET status = 'running', runner = ?, token_hash = ?, started_ms = ? WHERE id = ? AND status = 'queued'",

        )

        .bind(runner)

        .bind(token_hash)

        .bind(now)

        .bind(j.id)

        .execute(&mut *tx)

        .await?;

        if done.rows_affected() == 1 {

            sqlx::query("UPDATE flow_runs SET status = 'running' WHERE id = ? AND status = 'queued'")

                .bind(j.run_id)

                .execute(&mut *tx)

                .await?;

            tx.commit().await?;

            return job(pool, j.id).await;

        }

    }

    tx.commit().await?;

    Ok(None)

}

/// How long a decline keeps a job from its runner when the runner's

/// profile stays the same: a reason such as memory in use passes.

pub const DECLINE_HOLDS_MS: i64 = 10 * 60_000;

/// The jobs runner `runner_id` declined with this profile, recently.

pub async fn declined_by(pool: &Pool, runner_id: i64, profile: &str) -> sqlx::Result<std::collections::HashSet<i64>> {

    let rows: Vec<(i64,)> = sqlx::query_as("SELECT job_id FROM flow_declines WHERE runner_id = ? AND profile = ? AND ms > ?")