//! OxygenFlow runners (hub spec §13.2): who may claim jobs, what they are,

//! whether they are taking work, and what they have run.

use super::{now_ms, Pool};

use sqlx::FromRow;

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

pub struct Runner {

    pub id: i64,

    pub name: String,

    pub token_hash: String,

    pub source: String,

    pub key: String,

    pub key_name: String,

    pub arch: String,

    pub os: String,

    pub version: String,

    pub labels: String,

    pub max_jobs: i64,

    pub memory_mib: Option<i64>,

    pub cpus: Option<i64>,

    pub state: String,

    pub created_ms: i64,

    pub created_by: String,

    pub last_seen_ms: Option<i64>,

    /// The owner whose repositories it serves; empty for the whole site.

    pub scope_owner: String,

    /// The one repository it serves, when it serves one.

    pub scope_repo_id: Option<i64>,

    /// Labels an administrator gave it, beside those it reports (#98).

    pub admin_labels: String,

    /// A time limit for its jobs in minutes, below the site's, when set.

    pub timeout_minutes: Option<i64>,

    /// It fetches images it lacks from the hub (oxygenflow#39).

    pub fetches_images: bool,

    /// What it measured at its last claim (oxygenflow#40): free disk (MiB),

    /// the 1-minute load ×100, its speed score and class, its kernel.

    pub free_disk_mib: Option<i64>,

    pub load_x100: Option<i64>,

    pub speed: Option<i64>,

    pub speed_class: Option<i64>,

    pub kernel: String,

    /// The tiers it offered at its last claim, comma-separated.

    pub tiers: String,

    /// The images it held then (a JSON array); empty when it did not say.

    pub images: String,

    /// BLAKE3 of what it offers; its declines hold until this changes.

    pub profile: String,

}

/// Which repositories' jobs a runner may take (oxygenhub#41).

#[derive(Clone, Debug, Default, PartialEq, Eq)]

pub struct Scope {

    /// Empty: any owner.

    pub owner: String,

    /// `None`: any of the owner's repositories.

    pub repo_id: Option<i64>,

}

impl Scope {

    pub fn is_site(&self) -> bool {

        self.owner.is_empty() && self.repo_id.is_none()

    }

}

/// A runner that has not polled for this long is shown offline.

pub const ONLINE_MS: i64 = 90_000;

impl Runner {

    pub fn online(&self) -> bool {

        self.last_seen_ms.is_some_and(|t| now_ms() - t < ONLINE_MS)

    }

    /// `online`, `offline`, `draining` or `disabled`.

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

        match self.state.as_str() {

            "disabled" => "disabled",

            "draining" => "draining",

            _ if self.online() => "online",

            _ => "offline",

        }

    }

    /// Every label it carries: what it reports, and what an administrator

    /// gave it.

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

        let mut out: Vec<&str> = self.labels.split(',').filter(|l| !l.is_empty()).collect();

        for l in self.admin_label_list() {

            if !out.contains(&l) {

                out.push(l);

            }

        }

        out

    }

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

        self.admin_labels.split(',').filter(|l| !l.is_empty()).collect()

    }

    /// The labels it reported itself, as the admin page distinguishes them.

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

        self.labels.split(',').filter(|l| !l.is_empty()).collect()

    }

    pub fn scope(&self) -> Scope {

        Scope { owner: self.scope_owner.clone(), repo_id: self.scope_repo_id }

    }

}

pub async fn by_token(pool: &Pool, token_hash: &str) -> sqlx::Result<Option<Runner>> {

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

}

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

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

}

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

    sqlx::query_as("SELECT * FROM flow_runners ORDER BY name, id").fetch_all(pool).await

}

/// The row of a runner configured with a hub.toml token, made on first sight.

pub async fn configured(pool: &Pool, token_hash: &str, name: &str) -> sqlx::Result<Runner> {

    sqlx::query("INSERT OR IGNORE INTO flow_runners (name, token_hash, source, created_ms) VALUES (?, ?, 'config', ?)")

        .bind(name)

        .bind(token_hash)

        .bind(now_ms())

        .execute(pool)

        .await?;

    Ok(by_token(pool, token_hash).await?.expect("just made"))

}

/// What a claim says about the runner, and that it was seen now. A

/// configured runner takes the name it claims with.

/// The administrator's settings for a runner (#98).

pub async fn set_limits(pool: &Pool, id: i64, admin_labels: &str, timeout_minutes: Option<i64>) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET admin_labels = ?, timeout_minutes = ? WHERE id = ?")

        .bind(admin_labels)

        .bind(timeout_minutes)

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

pub async fn seen(pool: &Pool, id: i64, name: Option<&str>, info: Option<&flow_core::api::RunnerInfo>) -> sqlx::Result<()> {

    let now = now_ms();

    match info {

        Some(i) => {

            sqlx::query(

                "UPDATE flow_runners SET arch = ?, os = ?, version = ?, labels = ?, max_jobs = ?, memory_mib = ?, cpus = ?, last_seen_ms = ? WHERE id = ?",

            )

            .bind(&i.arch)

            .bind(&i.os)

            .bind(&i.version)

            .bind(i.labels.join(","))

            .bind(i.max_jobs as i64)

            .bind(i.memory_mib.map(i64::from))

            .bind(i.cpus.map(i64::from))

            .bind(now)

            .bind(id)

            .execute(pool)

            .await?;

        }

        None => {

            sqlx::query("UPDATE flow_runners SET last_seen_ms = ? WHERE id = ?").bind(now).bind(id).execute(pool).await?;

        }

    }

    if let Some(n) = name {

        sqlx::query("UPDATE flow_runners SET name = ? WHERE id = ? AND source = 'config'").bind(n).bind(id).execute(pool).await?;

    }

    Ok(())

}

/// Keep what a runner measured about itself at a claim (oxygenflow#40).

pub async fn set_facts(pool: &Pool, id: i64, f: &flow_core::api::Facts) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET free_disk_mib = ?, load_x100 = ?, speed = ?, speed_class = ?, kernel = ? WHERE id = ?")

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

        .bind(f.load_x100.map(i64::from))

        .bind(f.speed.map(i64::from))

        .bind(f.speed_class.map(i64::from))

        .bind(f.kernel.clone().unwrap_or_default())

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

impl Runner {

    /// Its facts as words for the Runners page: "speed 100 (class 3) · load

    /// 1.12 · 41 GiB free"; empty when it has sent none.

    pub fn facts_line(&self) -> String {

        let mut out = Vec::new();

        if let Some(s) = self.speed {

            out.push(format!("speed {s} (class {})", self.speed_class.unwrap_or(0)));

        }

        if let Some(l) = self.load_x100 {

            out.push(format!("load {:.2}", l as f64 / 100.0));

        }

        if let Some(d) = self.free_disk_mib {

            out.push(if d >= 1024 { format!("{} GiB free", d / 1024) } else { format!("{d} MiB free") });

        }

        out.join(" · ")

    }

}

/// What a runner offers at a claim (OxygenFlow spec §11.1): its tiers and

/// images, and the profile hash over those, its architecture, labels and

/// capacity.

pub async fn set_offer(

    pool: &Pool,

    id: i64,

    tiers: &[String],

    images: Option<&[String]>,

    fetches_images: bool,

    profile: &str,

) -> sqlx::Result<()> {

    let images = images.map(|i| serde_json::to_string(i).unwrap_or_default()).unwrap_or_default();

    sqlx::query("UPDATE flow_runners SET tiers = ?, images = ?, fetches_images = ?, profile = ? WHERE id = ?")

        .bind(tiers.join(","))

        .bind(images)

        .bind(fetches_images)

        .bind(profile)

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

impl Runner {

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

        self.tiers.split(',').filter(|t| !t.is_empty()).collect()

    }

    /// The images it holds, `None` when it does not say.

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

        if self.images.is_empty() {

            None

        } else {

            serde_json::from_str(&self.images).ok()

        }

    }

}

pub async fn set_state(pool: &Pool, id: i64, state: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET state = ?, state_ms = ? WHERE id = ?")

        .bind(state)

        .bind(super::now_ms())

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Whether runner `id`'s last `n` jobs since its state last changed all

/// ended in an infrastructure error (oxygenflow#30), with the last one's.

pub async fn erroring(pool: &Pool, id: i64, n: u32) -> sqlx::Result<Option<String>> {

    if n == 0 {

        return Ok(None);

    }

    let last: Vec<(String, String)> = sqlx::query_as(

        "SELECT j.status, j.error FROM flow_jobs j JOIN flow_runners r ON r.id = j.runner_id

         WHERE j.runner_id = ? AND j.finished_ms IS NOT NULL AND j.finished_ms >= r.state_ms

         ORDER BY j.finished_ms DESC, j.id DESC LIMIT ?",

    )

    .bind(id)

    .bind(n as i64)

    .fetch_all(pool)

    .await?;

    Ok((last.len() == n as usize && last.iter().all(|(s, _)| s == "error")).then(|| last[0].1.clone()))

}

/// Forget a runner: its credential stops working. Its jobs keep its name.

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

    sqlx::query("DELETE FROM flow_runners WHERE id = ?").bind(id).execute(pool).await?;

    Ok(())

}

/// The key a runner signs receipts with: the one bound, or `key` (named

/// `name`) when none is yet.

pub async fn bind_key(pool: &Pool, id: i64, key: &str, name: &str) -> sqlx::Result<String> {

    sqlx::query("UPDATE flow_runners SET key = ?, key_name = ? WHERE id = ? AND key = ''")

        .bind(key)

        .bind(name)

        .bind(id)

        .execute(pool)

        .await?;

    sqlx::query_scalar("SELECT key FROM flow_runners WHERE id = ?").bind(id).fetch_one(pool).await

}

/// Keys that runners of this hub sign with.

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

    sqlx::query_scalar("SELECT key FROM flow_runners WHERE key != ''").fetch_all(pool).await

}

/// A one-time registration token, valid for a day, for a runner of `scope`.

pub async fn add_registration(pool: &Pool, token_hash: &str, by: &str, scope: &Scope) -> sqlx::Result<()> {

    let now = now_ms();

    sqlx::query(

        "INSERT INTO flow_runner_registrations (token_hash, created_by, created_ms, expires_ms, scope_owner, scope_repo_id) VALUES (?, ?, ?, ?, ?, ?)",

    )

    .bind(token_hash)

    .bind(by)

    .bind(now)

    .bind(now + 24 * 3_600_000)

    .bind(&scope.owner)

    .bind(scope.repo_id)

    .execute(pool)

    .await?;

    Ok(())

}

/// The runners that may take a repository's jobs: the site's, its owner's

/// and its own (oxygenhub#41).

pub async fn serving(pool: &Pool, owner: &str, repo_id: i64) -> sqlx::Result<Vec<Runner>> {

    sqlx::query_as(

        "SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) \

         OR (scope_owner = ? AND scope_repo_id IS NULL) OR scope_repo_id = ? ORDER BY name",

    )

    .bind(owner)

    .bind(repo_id)

    .fetch_all(pool)

    .await

}

/// The runners that serve an owner: the site's, and those scoped to the

/// owner or one of its repositories.

pub async fn serving_owner(pool: &Pool, owner: &str) -> sqlx::Result<Vec<Runner>> {

    sqlx::query_as("SELECT * FROM flow_runners WHERE (scope_owner = '' AND scope_repo_id IS NULL) OR scope_owner = ? ORDER BY name")

        .bind(owner)

        .fetch_all(pool)

        .await

}

/// Change the repositories a runner serves.

pub async fn set_scope(pool: &Pool, id: i64, scope: &Scope) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET scope_owner = ?, scope_repo_id = ? WHERE id = ?")

        .bind(&scope.owner)

        .bind(scope.repo_id)

        .bind(id)

        .execute(pool)

        .await?;

    Ok(())

}

/// Registration tokens not used yet and not expired: (created by, created, expires).

pub async fn pending_registrations(pool: &Pool) -> sqlx::Result<Vec<(String, i64, i64)>> {

    sqlx::query_as("SELECT created_by, created_ms, expires_ms FROM flow_runner_registrations WHERE used_ms IS NULL AND expires_ms > ? ORDER BY created_ms DESC")

        .bind(now_ms())

        .fetch_all(pool)

        .await

}

/// Use registration token `token_hash` to register a runner: `None` when

/// the token is unknown, used or expired.

#[allow(clippy::too_many_arguments)]

pub async fn register(

    pool: &Pool,

    registration: &str,

    name: &str,

    token_hash: &str,

    key: &str,

    key_name: &str,

    by_default: &str,

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

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

    let now = now_ms();

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

        "SELECT created_by, scope_owner, scope_repo_id FROM flow_runner_registrations WHERE token_hash = ? AND used_ms IS NULL AND expires_ms > ?",

    )

    .bind(registration)

    .bind(now)

    .fetch_optional(&mut *tx)

    .await?;

    let Some((who, scope_owner, scope_repo_id)) = found else { return Ok(None) };

    let id: i64 = sqlx::query_scalar(

        "INSERT INTO flow_runners (name, token_hash, source, key, key_name, created_ms, created_by, scope_owner, scope_repo_id) VALUES (?, ?, 'registered', ?, ?, ?, ?, ?, ?) RETURNING id",

    )

    .bind(name)

    .bind(token_hash)

    .bind(key)

    .bind(key_name)

    .bind(now)

    .bind(if who.is_empty() { by_default } else { &who })

    .bind(scope_owner)

    .bind(scope_repo_id)

    .fetch_one(&mut *tx)

    .await?;

    sqlx::query("UPDATE flow_runner_registrations SET used_ms = ?, runner_id = ? WHERE token_hash = ?")

        .bind(now)

        .bind(id)

        .bind(registration)

        .execute(&mut *tx)

        .await?;

    tx.commit().await?;

    get(pool, id).await

}

/// A job a runner ran, for its history.

#[derive(Clone, Debug, FromRow)]

pub struct RunnerJob {

    pub owner: String,

    pub repo: String,

    pub run_number: i64,

    pub job: String,

    pub status: String,

    pub started_ms: Option<i64>,

    pub finished_ms: Option<i64>,

    pub receipt_attest: String,

}

/// Jobs runner `id` ran or is running, newest first.

pub async fn history(pool: &Pool, id: i64, limit: i64, before: Option<i64>) -> sqlx::Result<Vec<RunnerJob>> {

    sqlx::query_as(

        "SELECT p.owner AS owner, p.name AS repo, r.number AS run_number, j.name AS job, j.status AS status, j.started_ms AS started_ms, \

         j.finished_ms AS finished_ms, j.receipt_attest AS receipt_attest \

         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.runner_id = ? AND (? IS NULL OR j.started_ms < ?) ORDER BY j.started_ms DESC LIMIT ?",

    )

    .bind(id)

    .bind(before)

    .bind(before)

    .bind(limit)

    .fetch_all(pool)

    .await

}

/// (jobs run, passed, busy milliseconds) of runner `id`.

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

    sqlx::query_as(

        "SELECT COUNT(*), COALESCE(SUM(status = 'passed'), 0), COALESCE(SUM(finished_ms - started_ms), 0) \

         FROM flow_jobs WHERE runner_id = ? AND finished_ms IS NOT NULL",

    )

    .bind(id)

    .fetch_one(pool)

    .await

}

/// Jobs running on each runner now, by runner id.

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

    sqlx::query_as("SELECT runner_id, COUNT(*) FROM flow_jobs WHERE status = 'running' AND runner_id IS NOT NULL GROUP BY runner_id")

        .fetch_all(pool)

        .await

}

pub async fn set_job_runner(pool: &Pool, job_id: i64, runner_id: i64) -> sqlx::Result<()> {

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

    Ok(())

}