runner_updates.rs 93.0%

387 of 416 lines run, 65 of 152 functions · run #544 on main · run missed
//! 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>,
    /// Builds held back after rolling back on a runner (#326).
    holds: Vec<rdb::Hold>,
    /// Proving a version in a wave (oxygenflow#84).
    soak_jobs: i64,
    soak_hours: i64,
    notice: Option<String>,
    error: Option<String>,
}

/// Who changes runner updates: the site's administrators anything; an
/// organisation's owners their organisation's setting, its test groups and
/// its runners (Andy: an organisation tries a build on some of its runners
/// before the rest).
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Reach {
    Site,
    Org(String),
}

impl Reach {
    fn holds_runner(&self, r: &runners::Runner) -> bool {
        match self {
            Reach::Site => true,
            Reach::Org(o) => r.scope_owner == *o,
        }
    }

    fn holds_group(&self, g: &Group) -> bool {
        match self {
            Reach::Site => true,
            Reach::Org(o) => g.owner == *o,
        }
    }
}

/// What a page shows of runner updates, within `reach`.
pub struct UpdateViews {
    pub site: PolicyView,
    pub owners: Vec<PolicyView>,
    pub groups: Vec<GroupView>,
    pub runners: Vec<RunnerView>,
}

pub async fn views(app: &Shared, reach: &Reach) -> sqlx::Result<UpdateViews> {
    let levels = Levels::load(&app.db).await?;
    let all = runners::list(&app.db).await?;
    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> = match reach {
        Reach::Site => {
            levels.policies.values().filter(|p| p.scope == "owner").map(|p| PolicyView::of(p, format!("owner {}", p.key))).collect()
        }
        Reach::Org(o) => vec![PolicyView::of(&policy("owner", o), format!("owner {o}"))],
    };
    owners.sort_by(|a, b| a.key.cmp(&b.key));
    let groups = levels
        .groups
        .iter()
        .filter(|g| reach.holds_group(g))
        .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 out = vec![];
    for r in all.into_iter().filter(|r| reach.holds_runner(r)) {
        let effective = levels.resolve(&r);
        let (_, offer) = ru::offer(&app.db, &levels, &r, &effective, hour, crate::db::now_ms(), false).await?;
        out.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(UpdateViews { site: PolicyView::of(&policy("site", ""), "the site".into()), owners, groups, runners: out })
}

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 repo = accounts::setting(&app.db, ru::SETTING_REPO).await.map_err(e)?.unwrap_or_default();
    let v = views(app, &Reach::Site).await.map_err(e)?;
    let holds = rdb::holds(&app.db).await.map_err(e)?;
    let number = |v: Option<String>, d: i64| v.and_then(|s| s.trim().parse::<i64>().ok()).filter(|n| *n > 0).unwrap_or(d);
    let soak_jobs = number(accounts::setting(&app.db, ru::SETTING_SOAK_JOBS).await.map_err(e)?, ru::SOAK_JOBS);
    let soak_hours = number(accounts::setting(&app.db, ru::SETTING_SOAK_HOURS).await.map_err(e)?, ru::SOAK_HOURS);
    Ok(Page(UpdatesPage {
        shell,
        repo,
        site: v.site,
        owners: v.owners,
        groups: v.groups,
        runners: v.runners,
        holds,
        soak_jobs,
        soak_hours,
        notice,
        error,
    }))
}

/// What a change did, to say on the page: `Ok` a notice, `Err` what was wrong.
pub type Outcome = Result<String, String>;

/// [`page`] saying what `outcome` did.
async fn page_after(app: &Shared, shell: Shell, outcome: Outcome) -> Result<Page<UpdatesPage>, Error> {
    match outcome {
        Ok(n) => page(app, shell, Some(n), None).await,
        Err(why) => page(app, shell, None, Some(why)).await,
    }
}

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,
    #[serde(default)]
    soak_jobs: String,
    #[serde(default)]
    soak_hours: 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;
        }
    }
    // Waves (oxygenflow#84): blank keeps the default.
    for (key, value, max, what) in [
        (ru::SETTING_SOAK_JOBS, &f.soak_jobs, 1000, "jobs each runner of a wave finishes"),
        (ru::SETTING_SOAK_HOURS, &f.soak_hours, 720, "hours a wave soaks"),
    ] {
        let v = value.trim();
        if !v.is_empty() && !v.parse::<i64>().is_ok_and(|n| (1..=max).contains(&n)) {
            return page(&app, shell, None, Some(format!("The {what} is a whole number from 1 to {max}."))).await;
        }
        accounts::set_setting(&app.db, key, v).await.map_err(|e| Error::from_db(shell.clone(), e))?;
    }
    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")]
    pub(crate) 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,
}

impl PolicyForm {
    /// The form an API setting stands for.
    pub fn of(u: &hub_api::UpdateSetting) -> PolicyForm {
        let tri = |b: Option<bool>| match b {
            Some(true) => "on",
            Some(false) => "off",
            None => "inherit",
        };
        PolicyForm {
            csrf_token: String::new(),
            scope: u.scope.clone(),
            key: u.key.clone(),
            updates: tri(u.updates).into(),
            channel: u.channel.clone().unwrap_or_default(),
            window: u.window.clone().unwrap_or_default(),
            dry_run: tri(u.dry_run).into(),
        }
    }
}

/// Set one level's settings, within `reach`.
pub async fn apply_policy(app: &Shared, me: &accounts::Account, reach: &Reach, f: &PolicyForm) -> sqlx::Result<Outcome> {
    let key = f.key.trim().to_string();
    let label = match (f.scope.as_str(), reach) {
        ("site", Reach::Site) if key.is_empty() => "the site".to_string(),
        ("owner", Reach::Org(o)) if key != *o => return Ok(Err("An organisation sets its own runners' updates only.".into())),
        ("owner", _) => {
            let known = accounts::account_by_name(&app.db, &key).await?.is_some() || accounts::org_by_name(&app.db, &key).await?.is_some();
            if !known {
                return Ok(Err(format!("There is no account or organisation {key:?}.")));
            }
            format!("owner {key}")
        }
        ("group", _) => match key.parse().ok() {
            Some(id) => match rdb::group(&app.db, id).await?.filter(|g| reach.holds_group(g)) {
                Some(g) => format!("group {}", g.name),
                None => return Ok(Err("That test group is gone.".into())),
            },
            None => return Ok(Err("That test group is gone.".into())),
        },
        ("runner", _) => match key.parse().ok() {
            Some(id) => match runners::get(&app.db, id).await?.filter(|r| reach.holds_runner(r)) {
                Some(r) => format!("runner {}", r.name),
                None => return Ok(Err("That runner is gone.".into())),
            },
            None => return Ok(Err("That runner is gone.".into())),
        },
        _ => return Ok(Err("Choose the site, an owner, a test group or a runner.".into())),
    };
    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 Ok(Err(format!("A channel is stable, canary or a version such as 0.3.4, not {channel:?}.")));
    }
    let window = match ru::parse_window(&f.window) {
        Ok(w) => w,
        Err(why) => return Ok(Err(format!("{why}."))),
    };
    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?;
    let words = PolicyView::of(&p, label.clone()).summary();
    accounts::audit(&app.db, Some(me), "runner.updates.policy", &label, &words).await.ok();
    Ok(Ok(format!("Saved for {label}: {words}.")))
}

/// `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 outcome = apply_policy(&app, &me, &Reach::Site, &f).await.map_err(|e| Error::from_db(shell.clone(), e))?;
    page_after(&app, shell, outcome).await
}

#[derive(Deserialize)]
pub struct GroupForm {
    #[serde(rename = "_csrf")]
    pub(crate) 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>,
    /// Its wave (oxygenflow#84), 1 to 9; blank for none.
    #[serde(default)]
    wave: String,
}

/// The wave a form gives, or why it cannot be one.
fn wave_of(f: &GroupForm) -> Result<Option<i64>, String> {
    let w = f.wave.trim();
    if w.is_empty() {
        return Ok(None);
    }
    match w.parse::<i64>() {
        Ok(n) if (1..=9).contains(&n) => Ok(Some(n)),
        _ => Err("A wave is a whole number from 1 to 9, or blank for after every wave.".into()),
    }
}

/// 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()),
    }
}

/// The chosen runners a form names that `owner`'s group may hold.
async fn chosen_in_reach(app: &Shared, owner: &str, f: &GroupForm) -> sqlx::Result<Vec<i64>> {
    let all = runners::list(&app.db).await?;
    Ok(f.runner.iter().copied().filter(|id| all.iter().any(|r| r.id == *id && (owner.is_empty() || r.scope_owner == owner))).collect())
}

/// Make a test group, within `reach` (an organisation's are its own).
pub async fn apply_create_group(app: &Shared, me: &accounts::Account, reach: &Reach, f: &GroupForm) -> sqlx::Result<Outcome> {
    let name = f.name.trim().to_string();
    if name.is_empty() || name.len() > 60 {
        return Ok(Err("A test group needs a name of up to 60 characters.".into()));
    }
    let owner = match reach {
        Reach::Site => f.owner.trim().to_string(),
        Reach::Org(o) => o.clone(),
    };
    if !owner.is_empty()
        && accounts::account_by_name(&app.db, &owner).await?.is_none()
        && accounts::org_by_name(&app.db, &owner).await?.is_none()
    {
        return Ok(Err(format!("There is no account or organisation {owner:?}.")));
    }
    let percent = match percent_of(f) {
        Ok(p) => p,
        Err(why) => return Ok(Err(why)),
    };
    if rdb::groups(&app.db).await?.iter().any(|g| g.owner == owner && g.name == name) {
        return Ok(Err(format!("There is already a test group {name:?} for those runners.")));
    }
    let wave = match wave_of(f) {
        Ok(w) => w,
        Err(why) => return Ok(Err(why)),
    };
    let chosen = chosen_in_reach(app, &owner, f).await?;
    let g = rdb::create_group(&app.db, &owner, &name, percent).await?;
    rdb::set_wave(&app.db, g.id, wave).await?;
    if percent.is_none() {
        rdb::set_members(&app.db, g.id, &chosen).await?;
    }
    let mut 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", chosen.len()),
    };
    if let Some(w) = wave {
        what.push_str(&format!(", wave {w}"));
    }
    accounts::audit(&app.db, Some(me), "runner.updates.group", &name, &format!("created: {what}")).await.ok();
    Ok(Ok(format!("Test group {name} made: {what}. Give it its settings below.")))
}

/// `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 outcome = apply_create_group(&app, &me, &Reach::Site, &f).await.map_err(|e| Error::from_db(shell.clone(), e))?;
    page_after(&app, shell, outcome).await
}

/// Set who is in test group `id`, within `reach`.
pub async fn apply_save_group(app: &Shared, me: &accounts::Account, reach: &Reach, id: i64, f: &GroupForm) -> sqlx::Result<Outcome> {
    let Some(g) = rdb::group(&app.db, id).await?.filter(|g| reach.holds_group(g)) else {
        return Ok(Err("That test group is gone.".into()));
    };
    let percent = match percent_of(f) {
        Ok(p) => p,
        Err(why) => return Ok(Err(why)),
    };
    let wave = match wave_of(f) {
        Ok(w) => w,
        Err(why) => return Ok(Err(why)),
    };
    let chosen = if percent.is_some() { vec![] } else { chosen_in_reach(app, &g.owner, f).await? };
    rdb::set_percent(&app.db, id, percent).await?;
    rdb::set_members(&app.db, id, &chosen).await?;
    rdb::set_wave(&app.db, id, wave).await?;
    let mut what = match percent {
        Some(p) => format!("{p}% of its runners"),
        None => format!("{} chosen runners", chosen.len()),
    };
    if let Some(w) = wave {
        what.push_str(&format!(" in wave {w}"));
    }
    accounts::audit(&app.db, Some(me), "runner.updates.group", &g.name, &what).await.ok();
    Ok(Ok(format!("Test group {} is now {what}.", g.name)))
}

/// `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 outcome = apply_save_group(&app, &me, &Reach::Site, id, &f).await.map_err(|e| Error::from_db(shell.clone(), e))?;
    page_after(&app, shell, outcome).await
}

#[derive(Deserialize)]
pub struct CsrfOnly {
    #[serde(rename = "_csrf")]
    pub(crate) csrf_token: String,
}

/// Remove test group `id`, within `reach`.
pub async fn apply_delete_group(app: &Shared, me: &accounts::Account, reach: &Reach, id: i64) -> sqlx::Result<Outcome> {
    let Some(g) = rdb::group(&app.db, id).await?.filter(|g| reach.holds_group(g)) else {
        return Ok(Err("That test group is gone.".into()));
    };
    rdb::delete_group(&app.db, id).await?;
    accounts::audit(&app.db, Some(me), "runner.updates.group", &g.name, "removed").await.ok();
    Ok(Ok(format!("Test group {} removed; its runners follow the levels above it.", g.name)))
}

/// `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 outcome = apply_delete_group(&app, &me, &Reach::Site, id).await.map_err(|e| Error::from_db(shell.clone(), e))?;
    page_after(&app, shell, outcome).await
}

#[derive(Deserialize)]
pub struct LiftForm {
    #[serde(rename = "_csrf")]
    csrf_token: String,
    version: String,
}

/// `POST /admin/runners/updates/holds/lift`: offer a held build again (#326).
pub async fn lift_hold(State(app): State<Shared>, viewer: Viewer, Form(f): Form<LiftForm>) -> 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 v = f.version.trim();
    if !rdb::lift(&app.db, v).await.map_err(|e| Error::from_db(shell.clone(), e))? {
        return page(&app, shell, None, Some(format!("flow {v} is not held."))).await;
    }
    accounts::audit(&app.db, Some(&me), "runner.update.lifted", v, "offered again").await.ok();
    page(&app, shell, Some(format!("flow {v} is offered again where the settings say so.")), None).await
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Coverage / crates/hub/src/web/runner_updates.rs 93.0%

Keyboard
