//! Runner updates (oxygenhub#322, OxygenFlow docs/jetstream.md §8.2):

//! which build each runner should run, under the policy the site, its

//! owner, its test groups and the runner itself set, the nearest setting

//! winning. Off unless some level turns updates on.

use crate::db::runner_updates::{self as rdb, Group, Policy, Soak};

use crate::db::runners::Runner;

use crate::db::{self, Pool};

use flow_core::api::{UpdateOffer, UpdatePolicy};

use std::collections::{BTreeSet, HashMap};

/// The site setting naming the repository whose releases carry runner

/// builds (`owner/name`), each an asset `flow-OS-ARCH`.

pub const SETTING_REPO: &str = "runner_updates_repo";

/// The site settings for proving a version in a wave (oxygenflow#84): the

/// jobs each runner of the wave that took it must finish, or the hours

/// since its first job there after which one job is enough.

pub const SETTING_SOAK_JOBS: &str = "runner_updates_soak_jobs";

pub const SETTING_SOAK_HOURS: &str = "runner_updates_soak_hours";

pub const SOAK_JOBS: i64 = 5;

pub const SOAK_HOURS: i64 = 24;

/// Share an offered build with the swarm (oxygenflow#84): cut the release

/// asset into the hub's image chunks and keep its manifest, named, as an

/// image is, by the build's BLAKE3, so a runner in a swarm fetches it from

/// its site's peers and the hub together, as it fetches images, and the

/// hub serves it about once per site. Done once, in the background, the

/// first time the build is offered; a runner asking before it is done

/// fetches the build over HTTP.

pub fn share(app: &crate::Shared, offer: &UpdateOffer) {

    static SHARING: std::sync::Mutex<BTreeSet<String>> = std::sync::Mutex::new(BTreeSet::new());

    let store = app.images();

    let hash = offer.blake3.clone();

    if !jet::images::is_hash(&hash) || store.manifest(&hash).is_some() {

        return;

    }

    if !SHARING.lock().unwrap_or_else(|p| p.into_inner()).insert(hash.clone()) {

        return;

    }

    let (app, offer) = (app.clone(), offer.clone());

    tokio::spawn(async move {

        let path = crate::web::releases::blob_path(&app, &hash);

        let arch = offer.asset.rsplit('-').next().map(|a| a.trim_end_matches(".exe").to_string());

        let want = hash.clone();

        let made = tokio::task::spawn_blocking(move || -> Result<jet::images::Manifest, String> {

            let m = jet::images::scan_file(&path, arch, |c, bytes| Ok(store.put_chunk(c.hash(), bytes)?))

                .map_err(|e| format!("reading {}: {e}", path.display()))?;

            if m.hash != want {

                return Err(format!("the asset hashes to {}, not {}", &m.hash[..12], &want[..12]));

            }

            store.verify_and_keep(&m)?;

            Ok(m)

        })

        .await

        .unwrap_or_else(|e| Err(e.to_string()));

        match made {

            Ok(m) => match crate::images::record(&app.db, &m, &format!("runner build flow {}", offer.version)).await {

                Ok(()) => tracing::info!(

                    "runner updates: flow {} ({}) shared with the swarm: {} chunks",

                    offer.version,

                    offer.asset,

                    m.chunks.len()

                ),

                Err(e) => tracing::warn!("runner updates: recording flow {} for the swarm: {e}", offer.version),

            },

            Err(e) => tracing::warn!("runner updates: sharing flow {} ({}) with the swarm: {e}", offer.version, offer.asset),

        }

        SHARING.lock().unwrap_or_else(|p| p.into_inner()).remove(&hash);

    });

}

/// A runner's place, 0 to 99, for percentage groups: a stable hash of its

/// id, so a group's share is the same machines every rollout, and a

/// smaller share sits inside a larger one.

pub fn bucket(runner_id: i64) -> i64 {

    let h = blake3::hash(format!("runner {runner_id}").as_bytes());

    let mut first = [0u8; 8];

    first.copy_from_slice(&h.as_bytes()[..8]);

    (u64::from_le_bytes(first) % 100) as i64

}

/// Whether a group may hold the runner at all: a site group holds any, an

/// owner's group only that owner's runners.

pub fn in_reach(g: &Group, r: &Runner) -> bool {

    g.owner.is_empty() || g.owner == r.scope_owner

}

/// Every level's settings and the groups, read once.

#[derive(Clone, Debug, Default)]

pub struct Levels {

    pub policies: HashMap<(String, String), Policy>,

    pub groups: Vec<Group>,

    /// Chosen members, as (group, runner).

    pub chosen: BTreeSet<(i64, i64)>,

}

impl Levels {

    pub async fn load(pool: &Pool) -> sqlx::Result<Levels> {

        let policies = rdb::policies(pool).await?.into_iter().map(|p| ((p.scope.clone(), p.key.clone()), p)).collect();

        Ok(Levels { policies, groups: rdb::groups(pool).await?, chosen: rdb::members(pool).await?.into_iter().collect() })

    }

    pub fn member(&self, g: &Group, r: &Runner) -> bool {

        in_reach(g, r) && (self.chosen.contains(&(g.id, r.id)) || g.percent.is_some_and(|p| bucket(r.id) < p))

    }

    /// The runner's groups, nearest first: its owner's, then the site's.

    pub fn groups_of(&self, r: &Runner) -> Vec<&Group> {

        let mut gs: Vec<&Group> = self.groups.iter().filter(|g| self.member(g, r)).collect();

        gs.sort_by_key(|g| (g.owner.is_empty(), g.name.clone()));

        gs

    }

    fn get(&self, scope: &str, key: &str) -> Option<&Policy> {

        self.policies.get(&(scope.to_string(), key.to_string()))

    }

    /// The wave `r` takes a version in: the first of its groups' waves;

    /// `None` after every wave.

    pub fn wave_of(&self, r: &Runner) -> Option<i64> {

        self.groups_of(r).iter().filter_map(|g| g.wave).min()

    }

    /// The waves before `r`'s, in order, each with its runners among `all`:

    /// the waves of the groups that may hold `r` (the site's, and its

    /// owner's), a runner counted in its own first wave.

    pub fn waves_before<'a>(&self, r: &Runner, all: &'a [Runner]) -> Vec<(i64, Vec<&'a Runner>)> {

        let mine = self.wave_of(r);

        let mut waves: Vec<i64> =

            self.groups.iter().filter(|g| in_reach(g, r)).filter_map(|g| g.wave).filter(|w| mine.is_none_or(|m| *w < m)).collect();

        waves.sort();

        waves.dedup();

        waves

            .into_iter()

            .map(|w| {

                let members = all

                    .iter()

                    .filter(|o| self.wave_of(o) == Some(w) && self.groups_of(o).iter().any(|g| g.wave == Some(w) && in_reach(g, r)))

                    .collect();

                (w, members)

            })

            .collect()

    }

    /// The levels that apply to `r`, nearest first, each with its name.

    fn chain(&self, r: &Runner) -> Vec<(String, &Policy)> {

        let mut out = vec![];

        if let Some(p) = self.get("runner", &r.id.to_string()) {

            out.push(("runner".to_string(), p));

        }

        for g in self.groups_of(r) {

            if let Some(p) = self.get("group", &g.id.to_string()) {

                out.push((format!("group {}", g.name), p));

            }

        }

        if !r.scope_owner.is_empty() {

            if let Some(p) = self.get("owner", &r.scope_owner) {

                out.push((format!("owner {}", r.scope_owner), p));

            }

        }

        if let Some(p) = self.get("site", "") {

            out.push(("site".to_string(), p));

        }

        out

    }

    /// The policy for `r`: each setting from the nearest level that sets it.

    pub fn resolve(&self, r: &Runner) -> UpdatePolicy {

        let chain = self.chain(r);

        fn pick<T>(chain: &[(String, &Policy)], f: impl Fn(&Policy) -> Option<T>) -> (Option<T>, String) {

            chain.iter().find_map(|(name, p)| f(p).map(|v| (Some(v), name.clone()))).unwrap_or((None, "default".into()))

        }

        let (updates, u_from) = pick(&chain, |p| p.updates);

        let (channel, c_from) = pick(&chain, |p| p.channel.clone().filter(|c| !c.is_empty()));

        let (window, w_from) = pick(&chain, |p| Some((p.window_from? as u8, p.window_to? as u8)));

        let (dry_run, d_from) = pick(&chain, |p| p.dry_run);

        UpdatePolicy {

            updates: updates.unwrap_or(false),

            channel: channel.unwrap_or_else(|| "stable".into()),

            window,

            dry_run: dry_run.unwrap_or(false),

            from: vec![u_from, c_from, w_from, d_from],

        }

    }

}

/// Whether hour `h` (UTC) falls in `[from, to)`, which may wrap past midnight.

pub fn in_window((from, to): (u8, u8), h: u8) -> bool {

    if from <= to {

        from <= h && h < to

    } else {

        h >= from || h < to

    }

}

/// Parse `02-05` (hours, UTC) as a window; blank is none.

pub fn parse_window(s: &str) -> Result<Option<(u8, u8)>, String> {

    let s = s.trim();

    if s.is_empty() {

        return Ok(None);

    }

    let (a, b) = s.split_once('-').ok_or("a window is two hours, from-to, e.g. 02-05")?;

    let hour = |x: &str| x.trim().trim_end_matches(":00").parse::<u8>().ok().filter(|h| *h < 24);

    match (hour(a), hour(b)) {

        (Some(a), Some(b)) if a != b => Ok(Some((a, b))),

        _ => Err("a window is two different hours from 0 to 23, from-to, e.g. 02-05".into()),

    }

}

/// A channel as written: `stable`, `canary`, or a version.

pub fn valid_channel(c: &str) -> bool {

    matches!(c, "stable" | "canary") || (!c.is_empty() && c.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())))

}

/// The version in a runner's or a tag's words: `0.3.1 (build)` → `0.3.1`,

/// `v0.3.1` → `0.3.1`.

pub fn version_of(s: &str) -> String {

    s.split_whitespace().next().unwrap_or("").trim_start_matches('v').to_string()

}

fn older(a: &str, b: &str) -> bool {

    let parts = |v: &str| v.split('.').map(|p| p.parse::<u64>().unwrap_or(0)).collect::<Vec<_>>();

    parts(a) < parts(b)

}

/// The version a runner's report says rolled back: `rolled back from flow

/// 0.4.0 to 0.3.1: …` (the runner's words, oxygenflow spec §11.5.3) → `0.4.0`.

pub fn rolled_back_version(note: &str) -> Option<String> {

    let v = note.strip_prefix("rolled back from flow ")?.split_whitespace().next()?;

    v.split('.').all(|p| !p.is_empty() && p.bytes().all(|b| b.is_ascii_digit())).then(|| v.to_string())

}

/// The release asset a runner on `os` and `arch` runs: `flow-linux-x86_64`,

/// `flow-windows-x86_64.exe`.

pub fn asset_for(os: &str, arch: &str) -> String {

    format!("flow-{os}-{arch}{}", if os == "windows" { ".exe" } else { "" })

}

/// Where a wave stands with a version (oxygenflow#84).

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

pub enum WaveState {

    /// Every runner of the wave that took it has finished its soak jobs on

    /// it, or the soak time has passed since its first job there; or the

    /// wave has no runners.

    Proved,

    /// Not yet: `on` of `runners` run it, with `jobs` jobs between them.

    Soaking { runners: usize, on: usize, jobs: i64 },

    /// A job on it ended in error on this runner: the version stops here.

    Stopped { runner_id: i64 },

}

/// Where a wave of `members` stands with a version, from each runner's

/// [`Soak`] on it, at `now` (ms).

pub fn wave_state(members: &[i64], soaks: &[Soak], soak_jobs: i64, soak_ms: i64, now: i64) -> WaveState {

    let mine: Vec<&Soak> = soaks.iter().filter(|s| members.contains(&s.runner_id) && s.jobs > 0).collect();

    if let Some(s) = mine.iter().find(|s| s.errors > 0) {

        return WaveState::Stopped { runner_id: s.runner_id };

    }

    if members.is_empty() {

        return WaveState::Proved;

    }

    let jobs = mine.iter().map(|s| s.jobs).sum();

    let soaked = mine.iter().filter_map(|s| s.first_ms).min().is_some_and(|first| now - first >= soak_ms);

    if !mine.is_empty() && (soaked || mine.iter().all(|s| s.jobs >= soak_jobs)) {

        WaveState::Proved

    } else {

        WaveState::Soaking { runners: members.len(), on: mine.len(), jobs }

    }

}

/// What the waves before `r`'s say about `version`: `Ok(None)` go ahead;

/// `Ok(Some(why))` wait; `Err((runner, why))` it stopped on that runner.

async fn wave_gate(

    pool: &Pool,

    levels: &Levels,

    r: &Runner,

    version: &str,

    now: i64,

) -> sqlx::Result<Result<Option<String>, (String, String)>> {

    let all = db::runners::list(pool).await?;

    let waves = levels.waves_before(r, &all);

    if waves.is_empty() {

        return Ok(Ok(None));

    }

    let setting = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d);

    let soak_jobs = setting(db::accounts::setting(pool, SETTING_SOAK_JOBS).await?, SOAK_JOBS);

    let soak_hours = setting(db::accounts::setting(pool, SETTING_SOAK_HOURS).await?, SOAK_HOURS);

    let soaks = rdb::soaks(pool, version).await?;

    for (w, members) in waves {

        let ids: Vec<i64> = members.iter().map(|m| m.id).collect();

        match wave_state(&ids, &soaks, soak_jobs, soak_hours * 3_600_000, now) {

            WaveState::Proved => continue,

            WaveState::Stopped { runner_id } => {

                let name = members.iter().find(|m| m.id == runner_id).map(|m| m.name.clone()).unwrap_or_default();

                let job = rdb::first_error(pool, version, runner_id).await?.unwrap_or_else(|| "a job".into());

                return Ok(Err((name.clone(), format!("wave {w} stopped it on {name}: {job} ended in error"))));

            }

            WaveState::Soaking { runners, on, jobs } => {

                return Ok(Ok(Some(format!(

                    "flow {version} waits for wave {w} to prove it: {on} of {runners} runner{} on it, {jobs} job{} (each needs {soak_jobs} without an error, or {soak_hours} h)",

                    if runners == 1 { "" } else { "s" },

                    if jobs == 1 { "" } else { "s" },

                ))));

            }

        }

    }

    Ok(Ok(None))

}

/// The build `r` should install under `policy` at hour `hour` (UTC), or

/// why there is none. A version a wave stopped is held back from every

/// runner when `act` (a runner asking; a page only says so).

pub async fn offer(

    pool: &Pool,

    levels: &Levels,

    r: &Runner,

    policy: &UpdatePolicy,

    hour: u8,

    now: i64,

    act: bool,

) -> sqlx::Result<(Option<UpdateOffer>, String)> {

    if !policy.updates {

        return Ok((None, format!("updates are off ({})", policy.from[0])));

    }

    let Some(spec) = db::accounts::setting(pool, SETTING_REPO).await?.filter(|s| !s.trim().is_empty()) else {

        return Ok((None, "no runner releases repository is set (Admin › Runners › Updates)".into()));

    };

    let Some((owner, name)) = spec.trim().split_once('/') else {

        return Ok((None, format!("the runner releases repository {spec:?} is not owner/name")));

    };

    let Some(repo) = db::repos::get(pool, owner, name).await? else {

        return Ok((None, format!("the runner releases repository {spec} does not exist")));

    };

    let releases = db::releases::list(pool, repo.id).await?;

    let release = match policy.channel.as_str() {

        "stable" => releases.iter().find(|x| !x.prerelease),

        "canary" => releases.first(),

        v => releases.iter().find(|x| version_of(&x.tag) == v),

    };

    let Some(release) = release else {

        return Ok((None, format!("{spec} has no release for the {} channel", policy.channel)));

    };

    let (want, have) = (version_of(&release.tag), version_of(&r.version));

    if want == have {

        return Ok((None, format!("up to date: flow {have} ({} channel)", policy.channel)));

    }

    if let Some(h) = rdb::held(pool, &want).await? {

        return Ok((None, format!("flow {want} is held: {}; lift the hold on Admin › Runners › Updates", h.why)));

    }

    let going_back = older(&want, &have);

    if going_back && matches!(policy.channel.as_str(), "stable" | "canary") {

        return Ok((None, format!("runs flow {have}, newer than the {} channel's {want}", policy.channel)));

    }

    // Waves (oxygenflow#84): a later wave takes a version once those before it proved it.

    if !going_back {

        match wave_gate(pool, levels, r, &want, now).await? {

            Ok(None) => {}

            Ok(Some(why)) => return Ok((None, why)),

            Err((runner, why)) => {

                if act {

                    rdb::hold(pool, &want, &runner, &why).await?;

                    db::accounts::audit(pool, None, "runner.update.held", &want, &why).await.ok();

                }

                return Ok((None, format!("flow {want} is held: {why}; lift the hold on Admin › Runners › Updates")));

            }

        }

    }

    let file = asset_for(&r.os, &r.arch);

    let assets = db::releases::assets(pool, release.id).await?;

    let Some(a) = assets.iter().find(|a| a.name == file) else {

        return Ok((None, format!("{spec} {} has no {file}", release.tag)));

    };

    if let Some(w) = policy.window.filter(|w| !in_window(*w, hour)) {

        return Ok((None, format!("flow {want} waits for its window, {:02}:00–{:02}:00 UTC", w.0, w.1)));

    }

    let note = format!(

        "flow {want} ({} channel){}",

        policy.channel,

        if policy.dry_run { ": dry run, check it and say what would happen" } else { "" }

    );

    Ok((

        Some(UpdateOffer {

            version: want,

            tag: release.tag.clone(),

            asset: a.name.clone(),

            url: format!("/{owner}/{name}/asset/{}/{}", a.id, a.name),

            blake3: a.hash.clone(),

            size: a.size as u64,

            older: going_back,

        }),

        note,

    ))

}

#[cfg(test)]

mod tests {

    use super::*;

    fn runner(id: i64, owner: &str) -> Runner {

        Runner { id, scope_owner: owner.into(), version: "0.3.1 (build)".into(), ..Runner::default() }