Runner updates: policy, test groups and the offer 8501fd5815b9

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

-- policy of the site (scope 'site', key ''), an owner ('owner', its name),

-- a test group ('group', its id) or a runner ('runner', its id). A NULL

-- inherits from the next level out; the nearest setting wins; off unless

-- some level turns updates on.

CREATE TABLE runner_update_policies (

    scope       TEXT NOT NULL,

    key         TEXT NOT NULL,

    updates     BOOLEAN,

    -- 'stable', 'canary' or a version.

    channel     TEXT,

    -- Hours (UTC) in which updates may start: from, to (to may wrap past midnight).

    window_from BIGINT,

    window_to   BIGINT,

    dry_run     BOOLEAN,

    PRIMARY KEY (scope, key)

);

-- Test groups: named sets of the site's runners (owner '') or one owner's.

-- A runner is in one when chosen, or when a stable hash of its id falls in

-- the group's percentage, so the same machines are in it every rollout.

CREATE TABLE runner_groups (

    id         BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,

    owner      TEXT NOT NULL,

    name       TEXT NOT NULL,

    percent    BIGINT,

    created_ms BIGINT NOT NULL,

    UNIQUE (owner, name)

);

CREATE TABLE runner_group_members (

    group_id  BIGINT NOT NULL REFERENCES runner_groups (id) ON DELETE CASCADE,

    runner_id BIGINT NOT NULL REFERENCES flow_runners (id) ON DELETE CASCADE,

    PRIMARY KEY (group_id, runner_id)

);

-- What the runner last said about its update, and when.

ALTER TABLE flow_runners ADD COLUMN update_note TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_runners ADD COLUMN update_ms BIGINT;

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

-- policy of the site (scope 'site', key ''), an owner ('owner', its name),

-- a test group ('group', its id) or a runner ('runner', its id). A NULL

-- inherits from the next level out; the nearest setting wins; off unless

-- some level turns updates on.

CREATE TABLE runner_update_policies (

    scope       TEXT NOT NULL,

    key         TEXT NOT NULL,

    updates     INTEGER,

    -- 'stable', 'canary' or a version.

    channel     TEXT,

    -- Hours (UTC) in which updates may start: from, to (to may wrap past midnight).

    window_from INTEGER,

    window_to   INTEGER,

    dry_run     INTEGER,

    PRIMARY KEY (scope, key)

);

-- Test groups: named sets of the site's runners (owner '') or one owner's.

-- A runner is in one when chosen, or when a stable hash of its id falls in

-- the group's percentage, so the same machines are in it every rollout.

CREATE TABLE runner_groups (

    id         INTEGER PRIMARY KEY,

    owner      TEXT NOT NULL,

    name       TEXT NOT NULL,

    percent    INTEGER,

    created_ms INTEGER NOT NULL,

    UNIQUE (owner, name)

);

CREATE TABLE runner_group_members (

    group_id  INTEGER NOT NULL REFERENCES runner_groups (id) ON DELETE CASCADE,

    runner_id INTEGER NOT NULL REFERENCES flow_runners (id) ON DELETE CASCADE,

    PRIMARY KEY (group_id, runner_id)

);

-- What the runner last said about its update, and when.

ALTER TABLE flow_runners ADD COLUMN update_note TEXT NOT NULL DEFAULT '';

ALTER TABLE flow_runners ADD COLUMN update_ms INTEGER;

//! Runner updates (oxygenhub#322): the policy at each level and the test

//! groups runners are put in.

use super::{now_ms, Pool};

use sqlx::FromRow;

/// One level's settings; `None` inherits from the next level out.

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

pub struct Policy {

    pub scope: String,

    pub key: String,

    pub updates: Option<bool>,

    pub channel: Option<String>,

    pub window_from: Option<i64>,

    pub window_to: Option<i64>,

    pub dry_run: Option<bool>,

}

impl Policy {

    /// Nothing set: the level inherits everything.

    pub fn is_empty(&self) -> bool {

        self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none()

    }

}

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

pub struct Group {

    pub id: i64,

    /// Empty: a group of the site's runners.

    pub owner: String,

    pub name: String,

    /// A share of the runners in reach, by a stable hash of their ids;

    /// `None`: only the chosen ones.

    pub percent: Option<i64>,

    pub created_ms: i64,

}

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

    sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await

}

pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> {

    let p: Option<Policy> = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2")

        .bind(scope)

        .bind(key)

        .fetch_optional(pool)

        .await?;

    Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() }))

}

/// Set a level's settings; one with nothing set is removed.

pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?;

    if p.is_empty() {

        return Ok(());

    }

    sqlx::query(

        "INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run)

         VALUES ($1, $2, $3, $4, $5, $6, $7)",

    )

    .bind(&p.scope)

    .bind(&p.key)

    .bind(p.updates)

    .bind(&p.channel)

    .bind(p.window_from)

    .bind(p.window_to)

    .bind(p.dry_run)

    .execute(pool)

    .await?;

    Ok(())

}

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

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

}

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

    sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await

}

pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> {

    sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *")

        .bind(owner)

        .bind(name)

        .bind(percent)

        .bind(now_ms())

        .fetch_one(pool)

        .await

}

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

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

    Ok(())

}

/// Remove a group, its members and its policy.

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

    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?;

    sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?;

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

    Ok(())

}

/// Every chosen member, as (group, runner).

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

    sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await

}

/// Make `runners` the group's chosen members.

pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> {

    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?;

    for r in runners {

        sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?;

    }

    Ok(())

}

/// What a runner said about its update.

pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> {

    sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3")

        .bind(note)

        .bind(now_ms())

        .bind(runner)

        .execute(pool)

        .await?;

    Ok(())

}

//! 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};

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";

/// 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 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 build `r` should install under `policy` at hour `hour` (UTC), or

/// why there is none.

pub async fn offer(pool: &Pool, r: &Runner, policy: &UpdatePolicy, hour: u8) -> 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)));

    }

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

    }

    let file = format!("flow-{}-{}", 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() }

    }

    #[test]

    fn a_smaller_share_sits_inside_a_larger_one_and_is_about_its_size() {

        let ten: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 10).collect();

        let fifty: Vec<i64> = (1..=1000).filter(|id| bucket(*id) < 50).collect();

        assert!(ten.iter().all(|id| fifty.contains(id)));

        assert!((60..=140).contains(&ten.len()), "{}", ten.len());

        assert_eq!(bucket(7), bucket(7), "stable");

    }

    #[test]

    fn the_nearest_level_wins_each_setting() {

        let mut l = Levels::default();

        let pol = |scope: &str, key: &str, f: &dyn Fn(&mut Policy)| {

            let mut p = Policy { scope: scope.into(), key: key.into(), ..Policy::default() };

            f(&mut p);

            ((scope.to_string(), key.to_string()), p)

        };

        l.policies.extend([

            pol("site", "", &|p| {

                p.updates = Some(false);

                p.channel = Some("stable".into());

            }),

            pol("owner", "acme", &|p| p.updates = Some(true)),

            pol("group", "1", &|p| p.channel = Some("canary".into())),

            pol("runner", "3", &|p| p.updates = Some(false)),

        ]);

        l.groups.push(Group { id: 1, owner: "acme".into(), name: "canary".into(), percent: None, created_ms: 0 });

        l.chosen.insert((1, 2));

        l.chosen.insert((1, 9));

        let site_runner = l.resolve(&runner(1, ""));

        assert!(!site_runner.updates, "off at the site");

        assert_eq!(site_runner.from, ["site", "site", "default", "default"]);

        let canary = l.resolve(&runner(2, "acme"));

        assert!(canary.updates && canary.channel == "canary");

        assert_eq!(canary.from[..2], ["owner acme".to_string(), "group canary".to_string()]);

        let held = l.resolve(&runner(3, "acme"));

        assert!(!held.updates, "the runner's own setting wins");

        assert_eq!(held.channel, "stable");

        assert!(l.groups_of(&runner(9, "other")).is_empty(), "an owner's group holds only its own runners");

        assert!(!Levels::default().resolve(&runner(4, "")).updates, "off unless chosen");

    }

    #[test]

    fn a_percentage_group_takes_its_share_of_the_runners_in_reach() {

        let mut l = Levels::default();

        l.groups.push(Group { id: 5, owner: String::new(), name: "farm 20%".into(), percent: Some(20), created_ms: 0 });

        let n = (1..=500).filter(|id| !l.groups_of(&runner(*id, "")).is_empty()).count();

        assert!((60..=140).contains(&n), "{n}");

    }

    #[test]

    fn windows_and_channels_read_as_written() {

        assert_eq!(parse_window("02-05"), Ok(Some((2, 5))));

        assert_eq!(parse_window(" 22:00-04:00 "), Ok(Some((22, 4))));

        assert_eq!(parse_window(""), Ok(None));

        assert!(parse_window("5").is_err() && parse_window("3-3").is_err() && parse_window("1-24").is_err());

        assert!(in_window((2, 5), 2) && in_window((2, 5), 4) && !in_window((2, 5), 5));

        assert!(in_window((22, 4), 23) && in_window((22, 4), 1) && !in_window((22, 4), 12));

        assert!(valid_channel("stable") && valid_channel("canary") && valid_channel("0.3.4"));

        assert!(!valid_channel("latest") && !valid_channel("0.3.") && !valid_channel(""));

        assert_eq!(version_of("v0.3.2"), "0.3.2");

        assert_eq!(version_of("0.3.1 (local build)"), "0.3.1");

    }

}

//! Admin → Runners → Updates (oxygenhub#322): where runner builds come

//! from, and who takes them: the site's, each owner's, each test group's

//! and each runner's settings, the nearest winning; test groups of chosen

//! runners or a percentage of them.

use super::settings::csrf;

use super::{filters, Error, Page, Shell};

use crate::auth::Viewer;

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

use crate::db::{accounts, runners};

use crate::runner_updates::{self as ru, Levels};

use crate::Shared;

use askama::Template;

use axum::extract::{Form, Path, State};

use flow_core::api::UpdatePolicy;

use serde::Deserialize;

fn require_admin(viewer: &Viewer, shell: &Shell) -> Result<accounts::Account, Error> {

    match viewer.account() {

        Some(a) if a.is_admin => Ok(a.clone()),

        _ => Err(Error::not_found(shell.clone())),

    }

}

/// One level's settings as its form shows them: `inherit`, `on` or `off`

/// for the switches, blank for an inherited channel or window.

pub struct PolicyView {

    pub scope: String,

    pub key: String,

    pub label: String,

    pub updates: &'static str,

    pub channel: String,

    pub window: String,

    pub dry_run: &'static str,

}

impl PolicyView {

    fn of(p: &Policy, label: String) -> PolicyView {

        let tri = |b: Option<bool>| match b {

            Some(true) => "on",

            Some(false) => "off",

            None => "inherit",

        };

        PolicyView {

            scope: p.scope.clone(),

            key: p.key.clone(),

            label,

            updates: tri(p.updates),

            channel: p.channel.clone().unwrap_or_default(),

            window: match (p.window_from, p.window_to) {

                (Some(a), Some(b)) => format!("{a:02}-{b:02}"),

                _ => String::new(),

            },

            dry_run: tri(p.dry_run),

        }

    }

    /// The settings in words, for a table.

    pub fn summary(&self) -> String {

        let mut s = vec![];

        if self.updates != "inherit" {

            s.push(format!("updates {}", self.updates));

        }

        if !self.channel.is_empty() {

            s.push(format!("channel {}", self.channel));

        }

        if !self.window.is_empty() {

            s.push(format!("window {} UTC", self.window));

        }

        if self.dry_run != "inherit" {

            s.push(format!("dry run {}", self.dry_run));

        }

        if s.is_empty() {

            "inherits everything".into()

        } else {

            s.join(", ")

        }

    }

}

pub struct GroupView {

    pub g: Group,

    /// Whom it can hold, in words.

    pub reach: String,

    /// Its members now, by name.

    pub members: Vec<String>,

    pub chosen: Vec<i64>,

    pub policy: PolicyView,

}

pub struct RunnerView {

    pub r: runners::Runner,

    pub groups: Vec<String>,

    pub effective: UpdatePolicy,

    /// What the hub would offer it now.

    pub offer: String,

    pub policy: PolicyView,

}

impl RunnerView {

    /// Its effective settings, each with where it came from.

    pub fn effective_words(&self) -> String {

        let e = &self.effective;

        let mut s = vec![format!("updates {} ({})", if e.updates { "on" } else { "off" }, e.from[0])];

        if e.updates {

            s.push(format!("channel {} ({})", e.channel, e.from[1]));

            if let Some((a, b)) = e.window {

                s.push(format!("window {a:02}–{b:02} UTC ({})", e.from[2]));

            }

            if e.dry_run {

                s.push(format!("dry run ({})", e.from[3]));

            }

        }

        s.join(" · ")

    }

}

#[derive(Template)]

#[template(path = "admin_runner_updates.html")]

pub struct UpdatesPage {

    shell: Shell,

    repo: String,

    site: PolicyView,

    owners: Vec<PolicyView>,

    groups: Vec<GroupView>,

    runners: Vec<RunnerView>,

    notice: Option<String>,

    error: Option<String>,

}

async fn page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Result<Page<UpdatesPage>, Error> {

    let e = |e| Error::from_db(shell.clone(), e);

    let levels = Levels::load(&app.db).await.map_err(e)?;

    let all = runners::list(&app.db).await.map_err(e)?;

    let repo = accounts::setting(&app.db, ru::SETTING_REPO).await.map_err(e)?.unwrap_or_default();

    let policy = |scope: &str, key: &str| {

        levels.policies.get(&(scope.to_string(), key.to_string())).cloned().unwrap_or(Policy {

            scope: scope.into(),

            key: key.into(),

            ..Policy::default()

        })

    };

    let mut owners: Vec<PolicyView> =

        levels.policies.values().filter(|p| p.scope == "owner").map(|p| PolicyView::of(p, format!("owner {}", p.key))).collect();

    owners.sort_by(|a, b| a.key.cmp(&b.key));

    let groups = levels

        .groups

        .iter()

        .map(|g| GroupView {

            g: g.clone(),

            reach: if g.owner.is_empty() { "the site's runners".into() } else { format!("{}'s runners", g.owner) },

            members: all.iter().filter(|r| levels.member(g, r)).map(|r| r.name.clone()).collect(),

            chosen: levels.chosen.iter().filter(|(gid, _)| *gid == g.id).map(|(_, r)| *r).collect(),

            policy: PolicyView::of(&policy("group", &g.id.to_string()), format!("group {}", g.name)),

        })

        .collect();

    let hour = time::OffsetDateTime::now_utc().hour();

    let mut views = vec![];

    for r in all {

        let effective = levels.resolve(&r);

        let (_, offer) = ru::offer(&app.db, &r, &effective, hour).await.map_err(e)?;

        views.push(RunnerView {

            groups: levels.groups_of(&r).iter().map(|g| g.name.clone()).collect(),

            policy: PolicyView::of(&policy("runner", &r.id.to_string()), format!("runner {}", r.name)),

            effective,

            offer,

            r,

        });

    }

    Ok(Page(UpdatesPage {

        site: PolicyView::of(&policy("site", ""), "the site".into()),

        shell,

        repo,

        owners,

        groups,

        runners: views,

        notice,

        error,

    }))

}

pub async fn show(State(app): State<Shared>, viewer: Viewer) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    require_admin(&viewer, &shell)?;

    page(&app, shell, None, None).await

}

#[derive(Deserialize)]

pub struct SourceForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    repo: String,

}

/// `POST /admin/runners/updates/source`: the repository whose releases carry runner builds.

pub async fn save_source(State(app): State<Shared>, viewer: Viewer, Form(f): Form<SourceForm>) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let repo = f.repo.trim().trim_matches('/').to_string();

    if !repo.is_empty() {

        let found = match repo.split_once('/') {

            Some((o, n)) => crate::db::repos::get(&app.db, o, n).await.map_err(|e| Error::from_db(shell.clone(), e))?,

            None => None,

        };

        if found.is_none() {

            return page(&app, shell, None, Some(format!("There is no repository {repo:?} here; give it as owner/name."))).await;

        }

    }

    accounts::set_setting(&app.db, ru::SETTING_REPO, &repo).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "runner.updates.source", &repo, "runner builds come from its releases").await.ok();

    let n = if repo.is_empty() {

        "Saved: no repository, so no runner is offered a build.".into()

    } else {

        format!("Saved: runner builds come from {repo}'s releases.")

    };

    page(&app, shell, Some(n), None).await

}

#[derive(Deserialize)]

pub struct PolicyForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    scope: String,

    #[serde(default)]

    key: String,

    #[serde(default)]

    updates: String,

    #[serde(default)]

    channel: String,

    #[serde(default)]

    window: String,

    #[serde(default)]

    dry_run: String,

}

/// `POST /admin/runners/updates/policy`: one level's settings.

pub async fn save_policy(State(app): State<Shared>, viewer: Viewer, Form(f): Form<PolicyForm>) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let key = f.key.trim().to_string();

    let label = match f.scope.as_str() {

        "site" if key.is_empty() => "the site".to_string(),

        "owner" => {

            let known = accounts::account_by_name(&app.db, &key).await.map_err(e)?.is_some()

                || accounts::org_by_name(&app.db, &key).await.map_err(e)?.is_some();

            if !known {

                return page(&app, shell, None, Some(format!("There is no account or organisation {key:?}."))).await;

            }

            format!("owner {key}")

        }

        "group" => match key.parse().ok().map(|id| rdb::group(&app.db, id)) {

            Some(g) => match g.await.map_err(e)? {

                Some(g) => format!("group {}", g.name),

                None => return page(&app, shell, None, Some("That test group is gone.".into())).await,

            },

            None => return page(&app, shell, None, Some("That test group is gone.".into())).await,

        },

        "runner" => match key.parse().ok().map(|id| runners::get(&app.db, id)) {

            Some(r) => match r.await.map_err(e)? {

                Some(r) => format!("runner {}", r.name),

                None => return page(&app, shell, None, Some("That runner is gone.".into())).await,

            },

            None => return page(&app, shell, None, Some("That runner is gone.".into())).await,

        },

        _ => return page(&app, shell, None, Some("Choose the site, an owner, a test group or a runner.".into())).await,

    };

    let tri = |v: &str| match v {

        "on" => Some(true),

        "off" => Some(false),

        _ => None,

    };

    let channel = f.channel.trim().to_string();

    if !channel.is_empty() && !ru::valid_channel(&channel) {

        return page(&app, shell, None, Some(format!("A channel is stable, canary or a version such as 0.3.4, not {channel:?}."))).await;

    }

    let window = match ru::parse_window(&f.window) {

        Ok(w) => w,

        Err(why) => return page(&app, shell, None, Some(format!("{why}."))).await,

    };

    let p = Policy {

        scope: f.scope.clone(),

        key,

        updates: tri(&f.updates),

        channel: Some(channel).filter(|c| !c.is_empty()),

        window_from: window.map(|w| w.0 as i64),

        window_to: window.map(|w| w.1 as i64),

        dry_run: tri(&f.dry_run),

    };

    rdb::set_policy(&app.db, &p).await.map_err(e)?;

    let words = PolicyView::of(&p, label.clone()).summary();

    accounts::audit(&app.db, Some(&me), "runner.updates.policy", &label, &words).await.ok();

    page(&app, shell, Some(format!("Saved for {label}: {words}.")), None).await

}

#[derive(Deserialize)]

pub struct GroupForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    #[serde(default)]

    owner: String,

    #[serde(default)]

    name: String,

    /// `percent` or `chosen`.

    #[serde(default)]

    mode: String,

    #[serde(default)]

    percent: String,

    #[serde(default)]

    runner: Vec<i64>,

}

/// The percentage a form gives, or why it cannot be one.

fn percent_of(f: &GroupForm) -> Result<Option<i64>, String> {

    if f.mode != "percent" {

        return Ok(None);

    }

    match f.percent.trim().trim_end_matches('%').parse::<i64>() {

        Ok(p) if (1..=100).contains(&p) => Ok(Some(p)),

        _ => Err("A percentage is a whole number from 1 to 100.".into()),

    }

}

/// `POST /admin/runners/updates/groups`: a new test group.

pub async fn create_group(

    State(app): State<Shared>,

    viewer: Viewer,

    axum_extra::extract::Form(f): axum_extra::extract::Form<GroupForm>,

) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let name = f.name.trim().to_string();

    if name.is_empty() || name.len() > 60 {

        return page(&app, shell, None, Some("A test group needs a name of up to 60 characters.".into())).await;

    }

    let owner = f.owner.trim().to_string();

    if !owner.is_empty()

        && accounts::account_by_name(&app.db, &owner).await.map_err(e)?.is_none()

        && accounts::org_by_name(&app.db, &owner).await.map_err(e)?.is_none()

    {

        return page(&app, shell, None, Some(format!("There is no account or organisation {owner:?}."))).await;

    }

    let percent = match percent_of(&f) {

        Ok(p) => p,

        Err(why) => return page(&app, shell, None, Some(why)).await,

    };

    if rdb::groups(&app.db).await.map_err(e)?.iter().any(|g| g.owner == owner && g.name == name) {

        return page(&app, shell, None, Some(format!("There is already a test group {name:?} for those runners."))).await;

    }

    let g = rdb::create_group(&app.db, &owner, &name, percent).await.map_err(e)?;

    if percent.is_none() {

        rdb::set_members(&app.db, g.id, &f.runner).await.map_err(e)?;

    }

    let what = match percent {

        Some(p) => format!("{p}% of {}", if owner.is_empty() { "the site's runners".into() } else { format!("{owner}'s runners") }),

        None => format!("{} chosen runners", f.runner.len()),

    };

    accounts::audit(&app.db, Some(&me), "runner.updates.group", &name, &format!("created: {what}")).await.ok();

    page(&app, shell, Some(format!("Test group {name} made: {what}. Give it its settings below.")), None).await

}

/// `POST /admin/runners/updates/groups/{id}`: who is in a test group.

pub async fn save_group(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    axum_extra::extract::Form(f): axum_extra::extract::Form<GroupForm>,

) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let Some(g) = rdb::group(&app.db, id).await.map_err(e)? else {

        return page(&app, shell, None, Some("That test group is gone.".into())).await;

    };

    let percent = match percent_of(&f) {

        Ok(p) => p,

        Err(why) => return page(&app, shell, None, Some(why)).await,

    };

    rdb::set_percent(&app.db, id, percent).await.map_err(e)?;

    rdb::set_members(&app.db, id, if percent.is_some() { &[] } else { &f.runner }).await.map_err(e)?;

    let what = match percent {

        Some(p) => format!("{p}% of its runners"),

        None => format!("{} chosen runners", f.runner.len()),

    };

    accounts::audit(&app.db, Some(&me), "runner.updates.group", &g.name, &what).await.ok();

    page(&app, shell, Some(format!("Test group {} is now {what}.", g.name)), None).await

}

#[derive(Deserialize)]

pub struct CsrfOnly {

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

/// `POST /admin/runners/updates/groups/{id}/delete`.

pub async fn delete_group(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<CsrfOnly>,

) -> Result<Page<UpdatesPage>, Error> {

    let shell = Shell::new(&app, "/admin/runners/updates", &viewer);

    let me = require_admin(&viewer, &shell)?;

    csrf(&viewer, &f.csrf_token, &shell)?;

    let e = |e| Error::from_db(shell.clone(), e);

    let Some(g) = rdb::group(&app.db, id).await.map_err(e)? else {

        return page(&app, shell, None, Some("That test group is gone.".into())).await;

    };

    rdb::delete_group(&app.db, id).await.map_err(e)?;

    accounts::audit(&app.db, Some(&me), "runner.updates.group", &g.name, "removed").await.ok();

    page(&app, shell, Some(format!("Test group {} removed; its runners follow the levels above it.", g.name)), None).await

}

{% extends "base.html" %}

{% import "admin_nav.html" as an %}

{# One level's update settings (oxygenhub#322); `top` is the site, which inherits from nothing. #}

{% macro policy_fields(p, top) %}

  <label class="pol">Updates <select name="updates">{% if !top %}<option value="inherit"{% if p.updates == "inherit" %} selected{% endif %}>inherit</option>{% endif %}<option value="off"{% if p.updates == "off" || (top && p.updates == "inherit") %} selected{% endif %}>off</option><option value="on"{% if p.updates == "on" %} selected{% endif %}>on</option></select></label>

  <label class="pol">Channel <input type="text" name="channel" value="{{ p.channel }}" placeholder="{% if top %}stable{% else %}inherit{% endif %}" size="9" list="update-channels"></label>

  <label class="pol">Window, UTC <input type="text" name="window" value="{{ p.window }}" placeholder="{% if top %}any time{% else %}inherit{% endif %}" size="7" title="Hours in which updates may start, from-to, e.g. 02-05"></label>

  <label class="pol">Dry run <select name="dry_run">{% if !top %}<option value="inherit"{% if p.dry_run == "inherit" %} selected{% endif %}>inherit</option>{% endif %}<option value="off"{% if p.dry_run == "off" || (top && p.dry_run == "inherit") %} selected{% endif %}>off</option><option value="on"{% if p.dry_run == "on" %} selected{% endif %}>on</option></select></label>

{% endmacro %}

{% block title %}Runner updates · Administration{% endblock %}

{% block content %}

<div class="head"><h1>Runners <small><a href="/admin">Administration</a></small></h1></div>

<div class="settings-layout">

{% call an::nav(crate::web::admin::AdminSection::RunnerUpdates) %}

<div class="sections settings-body">

<div class="bar"><span class="modes"><a href="/admin/runners">runners</a><a href="/admin/runners/images">images</a><a href="/admin/runners/settings">settings</a><a href="/admin/runners/updates" class="on">updates</a></span></div>

{% if let Some(n) = notice %}<p class="note">{{ n }}</p>{% endif %}

{% if let Some(e) = error %}<p class="error">{{ e }}</p>{% endif %}

<datalist id="update-channels"><option value="stable"><option value="canary"></datalist>

<section class="panel">

  <h2>Runners update themselves</h2>

  <p>A runner asks the hub which build it should run, fetches it, checks it as a release, finishes what it holds and swaps itself in; a build that does not reach the hub goes back to the one before. <strong>Off unless chosen.</strong> The site, an owner, a test group and a runner each may set any of the settings below, and the nearest setting wins: a runner's own, then its test groups', then its owner's, then the site's. A runner's <code>flow.toml</code> can still refuse every update (<code>[runner.update] allow = false</code>).</p>

  <p class="small mute">A rollout: turn updates on for a test group (two chosen runners, or 10% of them), watch it, then widen it to 50%, then turn updates on for everyone. A percentage is taken by a stable hash of each runner's id, so it is the same machines every time, and 10% sits inside 50%.</p>

</section>

<form class="card wide" method="post" action="/admin/runners/updates/source">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

  <h2>Where builds come from</h2>

  <label for="update-repo">The repository whose releases carry runner builds</label>

  <input type="text" id="update-repo" name="repo" value="{{ repo }}" placeholder="owner/name, e.g. andymccall/oxygenflow">

  <p class="small mute">Each release offers one file per machine, <code>flow-linux-x86_64</code>, <code>flow-linux-aarch64</code>, <code>flow-linux-riscv64</code>. <em>stable</em> is its newest release, <em>canary</em> its newest including pre-releases; a version (<code>0.3.4</code>) pins it, going back if need be.</p>

  <button type="submit">Save</button>

</form>

<form class="card wide" method="post" action="/admin/runners/updates/policy">

  <input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="scope" value="site">

  <h2>The site</h2>

  <p class="small mute">What every runner does unless a nearer level says otherwise: {{ site.summary() }}.</p>

  <div class="pol-row">{% call policy_fields(site, true) %}</div>

  <button type="submit">Save</button>

</form>

<section class="panel" id="owners">

  <h2>Owners</h2>

  <p class="small mute">For the runners an account or organisation registered for its own repositories.</p>

  {% if owners.is_empty() %}<p class="mute small">No owner sets anything; their runners follow the site.</p>{% endif %}

  {% for o in owners %}

  <form class="pol-form" method="post" action="/admin/runners/updates/policy">

    <input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="scope" value="owner"><input type="hidden" name="key" value="{{ o.key }}">

    <strong>{{ o.key }}</strong>{% call policy_fields(o, false) %}<button type="submit">Save</button>

  </form>

  {% endfor %}

  <form class="pol-form" method="post" action="/admin/runners/updates/policy">

    <input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="scope" value="owner">

    <label class="pol">Owner <input type="text" name="key" size="12" placeholder="account or org" required></label>{% call policy_fields(site, false) %}<button type="submit">Add</button>

  </form>

</section>

<section class="panel" id="groups">

  <h2>Test groups</h2>

  <p class="small mute">A named set of runners: those you choose, or a percentage of the site's runners (or one owner's). Give a group its own settings to try a build on it before the rest.</p>

  {% if groups.is_empty() %}<p class="mute small">No test groups yet.</p>{% endif %}

  {% for gv in groups %}{% let g = gv.g %}

  <div class="card wide update-group">

    <h3>{{ g.name }} <small class="mute">{{ gv.reach }} · {% if let Some(p) = g.percent %}{{ p }}%{% else %}chosen{% endif %} · {{ gv.members.len()|plural("runner", "runners") }} now</small></h3>

    <p class="small">{% if gv.members.is_empty() %}<span class="mute">No runners in it.</span>{% else %}{% for m in gv.members %}<span class="lbl">{{ m }}</span> {% endfor %}{% endif %}</p>

    <form class="pol-form" method="post" action="/admin/runners/updates/policy">

      <input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="scope" value="group"><input type="hidden" name="key" value="{{ g.id }}">

      {% call policy_fields(gv.policy, false) %}<button type="submit">Save settings</button>

    </form>

    <details><summary class="small">Who is in it</summary>

    <form method="post" action="/admin/runners/updates/groups/{{ g.id }}">

      <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

      <label class="pol"><input type="radio" name="mode" value="percent"{% if g.percent.is_some() %} checked{% endif %}> a percentage of {{ gv.reach }}: <input type="number" name="percent" min="1" max="100" size="4" value="{% if let Some(p) = g.percent %}{{ p }}{% endif %}">%</label>

      <label class="pol"><input type="radio" name="mode" value="chosen"{% if g.percent.is_none() %} checked{% endif %}> these runners:</label>

      <div class="pol-row">{% for rv in runners %}{% if g.owner.is_empty() || g.owner == rv.r.scope_owner %}<label class="pol"><input type="checkbox" name="runner" value="{{ rv.r.id }}"{% if gv.chosen.contains(rv.r.id) %} checked{% endif %}> {{ rv.r.name }}</label>{% endif %}{% endfor %}</div>

      <button type="submit">Save members</button>

    </form>

    <form class="tiny" method="post" action="/admin/runners/updates/groups/{{ g.id }}/delete" data-confirm="Remove the test group {{ g.name }}? Its runners follow the levels above it."><input type="hidden" name="_csrf" value="{{ shell.csrf }}"><button class="danger-btn">remove group</button></form>

    </details>

  </div>

  {% endfor %}

  <form class="card wide" method="post" action="/admin/runners/updates/groups">

    <input type="hidden" name="_csrf" value="{{ shell.csrf }}">

    <h3>New test group</h3>

    <div class="pol-row">

      <label class="pol">Name <input type="text" name="name" size="14" required placeholder="canary"></label>

      <label class="pol">Of <input type="text" name="owner" size="12" placeholder="the site's runners" title="Blank for the site's runners, or an owner for theirs"></label>

    </div>

    <label class="pol"><input type="radio" name="mode" value="percent"> a percentage: <input type="number" name="percent" min="1" max="100" size="4">%</label>

    <label class="pol"><input type="radio" name="mode" value="chosen" checked> these runners:</label>

    <div class="pol-row">{% for rv in runners %}<label class="pol"><input type="checkbox" name="runner" value="{{ rv.r.id }}"> {{ rv.r.name }}</label>{% endfor %}</div>

    <button type="submit">Make group</button>

  </form>

</section>

<section class="panel" id="runners">

  <h2>Runners</h2>

  {% if runners.is_empty() %}<p class="mute">No runners yet.</p>{% else %}

  <div class="table"><table>

  <tr><th>Runner</th><th>Runs</th><th>Groups</th><th>Its policy</th><th>Now</th><th>Last said</th><th>Its own settings</th></tr>

  {% for rv in runners %}{% let r = rv.r %}

  <tr>

    <td><a href="/admin/runners/{{ r.id }}"><strong>{{ r.name }}</strong></a>{% if !r.scope_owner.is_empty() %}<br><span class="small mute">{{ r.scope_owner }}'s</span>{% endif %}</td>

    <td class="small">{% if r.version.is_empty() %}<span class="mute">unknown</span>{% else %}flow {{ r.version }}{% endif %}</td>

    <td class="small">{% for g in rv.groups %}<span class="lbl">{{ g }}</span> {% endfor %}</td>

    <td class="small">{{ rv.effective_words() }}</td>

    <td class="small">{{ rv.offer }}</td>

    <td class="small">{% if r.update_note.is_empty() %}<span class="mute">nothing yet</span>{% else %}{{ r.update_note }}{% if let Some(t) = r.update_ms %} <time class="mute" title="{{ t|when_i }}">{{ t|ago_i }}</time>{% endif %}{% endif %}</td>

    <td><form class="pol-form tiny" method="post" action="/admin/runners/updates/policy">

      <input type="hidden" name="_csrf" value="{{ shell.csrf }}"><input type="hidden" name="scope" value="runner"><input type="hidden" name="key" value="{{ r.id }}">

      {% call policy_fields(rv.policy, false) %}<button type="submit">set</button></form></td>

  </tr>

  {% endfor %}

  </table></div>

  {% endif %}

</section>

</div>

</div>

{% endblock %}

- **Runner updates: policy, test groups and the offer (#322):** Admin › Runners › Updates sets where runner builds come from (a repository whose releases carry `flow-linux-ARCH` files) and who takes them. Updates are off unless chosen; the site, an owner, a test group and a runner may each set updates on or off, a channel (`stable`, `canary` or a version), a window of hours (UTC) and dry run, and the nearest setting wins, each setting on its own. A test group is chosen runners or a percentage of the site's (or an owner's) runners, by a stable hash of each runner's id, so a share is the same machines every rollout and 10% sits inside 50%. `GET /api/v1/flow/runner/update` tells a runner its policy, where each setting came from, and the build to install (its BLAKE3, to check the download against); `POST` takes what the runner did, shown on the page and in the audit log. Pins OxygenFlow's main.

        .route("/flow/runner/update", get(crate::flow::runner_update).post(crate::flow::report_update))

pub mod runner_updates;

    /// What it last said about its update (oxygenhub#322), and when.

    pub update_note: String,

    pub update_ms: Option<i64>,

    sqlx::query("DELETE FROM runner_group_members WHERE runner_id = $1").bind(id).execute(pool).await?;

    sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'runner' AND key = $1").bind(id.to_string()).execute(pool).await?;

/// `GET /api/v1/flow/runner/update` (oxygenhub#322): the update policy

/// that applies to this runner and the build it should run, if another.

pub async fn runner_update(State(app): State<Shared>, headers: HeaderMap) -> Result<Json<flow_core::api::UpdateAnswer>, ApiError> {

    let me = runner_auth(&app, &headers).await?;

    db::runners::seen(&app.db, me.id, None, None).await?;

    let policy = crate::runner_updates::Levels::load(&app.db).await?.resolve(&me);

    let hour = time::OffsetDateTime::now_utc().hour();

    let (offer, note) = crate::runner_updates::offer(&app.db, &me, &policy, hour).await?;

    Ok(Json(flow_core::api::UpdateAnswer { policy, offer, note }))

}

/// `POST /api/v1/flow/runner/update`: what the runner did about its update

/// (fetched, checked, installed, a dry run's verdict, a rollback), kept on

/// its page and in the audit log.

pub async fn report_update(

    State(app): State<Shared>,

    headers: HeaderMap,

    Json(r): Json<flow_core::api::UpdateReport>,

) -> Result<StatusCode, ApiError> {

    let me = runner_auth(&app, &headers).await?;

    let note: String = r.note.chars().map(|c| if c.is_control() { ' ' } else { c }).take(500).collect();

    let note = note.trim();

    if note.is_empty() {

        return Err(ApiError(StatusCode::BAD_REQUEST, "say what happened (note)".into()));

    }

    db::runner_updates::report(&app.db, me.id, note).await?;

    db::accounts::audit(&app.db, None, "runner.update", &me.name, note).await.ok();

    Ok(StatusCode::NO_CONTENT)

}

pub mod runner_updates;

    RunnerUpdates,

            AdminSection::RunnerUpdates,

            AdminSection::RunnerUpdates => "/admin/runners/updates",

            AdminSection::RunnerUpdates => "runner-updates",

            AdminSection::RunnerUpdates => "Runner updates",

pub mod runner_updates;

        .route("/admin/runners/updates", get(runner_updates::show))

        .route("/admin/runners/updates/source", post(runner_updates::save_source))

        .route("/admin/runners/updates/policy", post(runner_updates::save_policy))

        .route("/admin/runners/updates/groups", post(runner_updates::create_group))

        .route("/admin/runners/updates/groups/{id}", post(runner_updates::save_group))

        .route("/admin/runners/updates/groups/{id}/delete", post(runner_updates::delete_group))

/// oxygenhub#322: runners are offered the build their nearest update

/// setting asks for, off unless some level turns it on; a test group of

/// chosen runners tries it first.

#[tokio::test(flavor = "multi_thread")]

async fn runners_are_offered_updates_by_their_nearest_setting() {

    use oxy_core::identity::SecretKey;

    let h = hub().await;

    create(&h.app, "andy", "oxygenflow").await;

    let res = oxygenhub::router(h.app.clone())

        .oneshot(

            Request::post("/register")

                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")

                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery&setup=test-setup"))

                .unwrap(),

        )

        .await

        .unwrap();

    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();

    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();

    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();

    let web = |method: &'static str, path: &str, body: Option<String>| {

        let (app, cookie, path) = (h.app.clone(), cookie.clone(), path.to_string());

        async move {

            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);

            if body.is_some() {

                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");

            }

            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();

            let st = res.status();

            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())

        }

    };

    let (st, page) = web("GET", "/admin/runners/updates", None).await;

    assert_eq!(st, StatusCode::OK, "{page}");

    let csrf = {

        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

        page[i..].split('"').next().unwrap().to_string()

    };

    assert_eq!(call(&h.app, "GET", "/admin/runners/updates", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    let mut creds = vec![];

    for name in ["plex", "narvi"] {

        let (_, page) = web("POST", "/admin/runners/token", Some(format!("_csrf={csrf}"))).await;

        let i = page.find("--token ").unwrap() + "--token ".len();

        let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();

        let identity = SecretKey::generate().identity(&format!("{name} (OxygenFlow runner)"), "", 1);

        let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0 (build)", "max_jobs": 1 });

        let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });

        let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;

        assert_eq!(st, StatusCode::OK, "{body}");

        creds.push(serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string());

    }

    let ask = |who: usize| {

        let (app, cred) = (h.app.clone(), creds[who].clone());

        async move {

            let (st, body) = call(&app, "GET", "/api/v1/flow/runner/update", Some(&cred), None).await;

            assert_eq!(st, StatusCode::OK, "{body}");

            serde_json::from_str::<flow_core::api::UpdateAnswer>(&body).unwrap()

        }

    };

    let a = ask(0).await;

    assert!(!a.policy.updates && a.offer.is_none(), "off unless chosen: {a:?}");

    assert_eq!(a.note, "updates are off (default)");

    assert_eq!(call(&h.app, "GET", "/api/v1/flow/runner/update", Some("flowrunner_nope"), None).await.0, StatusCode::UNAUTHORIZED);

    // A release of builds, and the site taking them from it.

    let repo = db::repos::get(&h.app.db, "andy", "oxygenflow").await.unwrap().unwrap();

    let rel = db::releases::create(&h.app.db, repo.id, "v0.2.0", "0.2.0", "", false, None).await.unwrap();

    db::releases::add_asset(&h.app.db, rel.id, "flow-linux-x86_64", 42, &"ab".repeat(32), "application/octet-stream").await.unwrap();

    let (_, page) = web("POST", "/admin/runners/updates/source", Some(format!("_csrf={csrf}&repo=andy%2Fnope"))).await;

    assert!(page.contains("There is no repository"), "{page}");

    let (_, page) = web("POST", "/admin/runners/updates/source", Some(format!("_csrf={csrf}&repo=andy%2Foxygenflow"))).await;

    assert!(page.contains("runner builds come from andy/oxygenflow"), "{page}");

    // A test group of one chosen runner, with updates on: plex, not narvi.

    let plex = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();

    let (_, page) =

        web("POST", "/admin/runners/updates/groups", Some(format!("_csrf={csrf}&name=canary&mode=chosen&runner={}", plex.id))).await;

    assert!(page.contains("Test group canary made: 1 chosen runners"), "{page}");

    let group = db::runner_updates::groups(&h.app.db).await.unwrap().remove(0);

    let (_, page) =

        web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=group&key={}&updates=on", group.id))).await;

    assert!(page.contains("Saved for group canary: updates on."), "{page}");

    let a = ask(0).await;

    let offer = a.offer.clone().expect("plex is offered the build");

    assert_eq!((offer.version.as_str(), offer.asset.as_str(), offer.blake3.len()), ("0.2.0", "flow-linux-x86_64", 64));

    assert!(offer.url.starts_with("/andy/oxygenflow/asset/") && !offer.older, "{offer:?}");

    assert_eq!(a.policy.from[0], "group canary");

    assert!(ask(1).await.offer.is_none(), "narvi is not in the group");

    // The site's dry run reaches plex; narvi's own setting pins its version.

    let (_, page) = web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=site&updates=off&dry_run=on"))).await;

    assert!(page.contains("Saved for the site"), "{page}");

    let a = ask(0).await;

    assert!(a.policy.dry_run && a.policy.from[3] == "site" && a.note.contains("dry run"), "{a:?}");

    let narvi = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "narvi").unwrap();

    web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=runner&key={}&updates=on&channel=0.1.0", narvi.id)))

        .await;

    let b = ask(1).await;

    assert!(b.policy.updates && b.offer.is_none() && b.note.contains("has no release"), "{b:?}");

    // Outside its window it waits.

    let now = time::OffsetDateTime::now_utc().hour();

    let (from, to) = ((now + 2) % 24, (now + 4) % 24);

    web(

        "POST",

        "/admin/runners/updates/policy",

        Some(format!("_csrf={csrf}&scope=group&key={}&updates=on&window={from:02}-{to:02}", group.id)),

    )

    .await;

    let a = ask(0).await;

    assert!(a.offer.is_none() && a.note.contains("waits for its window"), "{a:?}");

    // Mistakes are refused, saying why.

    let (_, page) = web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=site&channel=latest"))).await;

    assert!(page.contains("A channel is stable, canary or a version"), "{page}");

    let (_, page) = web("POST", "/admin/runners/updates/policy", Some(format!("_csrf={csrf}&scope=site&window=9"))).await;

    assert!(page.contains("a window is two hours"), "{page}");

    let (_, page) = web("POST", "/admin/runners/updates/groups", Some(format!("_csrf={csrf}&name=farm&mode=percent&percent=0"))).await;

    assert!(page.contains("A percentage is a whole number from 1 to 100."), "{page}");

    // What a runner says shows on the page and in the audit log.

    let report = serde_json::json!({ "note": "dry run: flow 0.2.0 checks out" }).to_string();

    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runner/update", Some(&creds[0]), Some(report)).await.0, StatusCode::NO_CONTENT);

    let (_, page) = web("GET", "/admin/runners/updates", None).await;

    assert!(page.contains("dry run: flow 0.2.0 checks out") && page.contains(">canary</span>"), "{page}");