//! Admin → Runners (hub spec §13.2): the OxygenFlow runners of this hub,

//! what they are and are running, their history, registration tokens, and

//! draining, disabling and removing them.

use super::settings::csrf;

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

use crate::auth::Viewer;

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

use crate::Shared;

use askama::Template;

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

use axum::response::{IntoResponse, Redirect, Response};

use serde::Deserialize;

use std::collections::HashMap;

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

    }

}

pub struct RunnerRow {

    pub r: runners::Runner,

    /// Jobs it is running now.

    pub running: i64,

    /// The repositories it serves, in words.

    pub scope: String,

    /// Where its peers reach it in the image swarm, and its site

    /// (oxygenhub#144); empty when it is not in one.

    pub swarm: String,

    /// The jobs it runs now: (repository, run, job, started).

    pub jobs: Vec<(String, i64, String, Option<i64>)>,

}

impl RunnerRow {

    /// What it is doing (oxygenhub#243): `disabled`, `draining`, `idle`,

    /// or `running N of M`.

    pub fn doing(&self) -> String {

        match self.r.state.as_str() {

            "disabled" => "disabled".into(),

            "draining" if self.running > 0 => format!("draining, {} left", self.running),

            "draining" => "draining".into(),

            _ if self.running > 0 => format!("running {} of {}", self.running, self.r.max_jobs),

            _ => "idle".into(),

        }

    }

    /// The jobs it runs, for the status's tooltip.

    pub fn jobs_line(&self) -> String {

        let now = crate::db::now_ms();

        self.jobs

            .iter()

            .map(|(repo, run, job, started)| match started {

                Some(t) => format!("{repo} #{run} {job}, {} min", (now - t).max(0) / 60_000),

                None => format!("{repo} #{run} {job}"),

            })

            .collect::<Vec<_>>()

            .join("; ")

    }

}

/// The queue's filters (oxygenhub#243), kept in the URL.

#[derive(Deserialize, Default, Clone)]

pub struct QueueFilter {

    #[serde(default)]

    pub repo: String,

    #[serde(default)]

    pub status: String,

    #[serde(default)]

    pub runner: String,

}

impl QueueFilter {

    fn keeps(&self, q: &db::flow::QueueEntry) -> bool {

        (self.repo.is_empty() || format!("{}/{}", q.owner, q.name) == self.repo)

            && (self.status.is_empty() || q.status == self.status)

            && (self.runner.is_empty() || q.runner == self.runner)

    }

    pub fn any(&self) -> bool {

        !(self.repo.is_empty() && self.status.is_empty() && self.runner.is_empty())

    }

}

/// A scope in words: the whole site, an owner's repositories, or one

/// repository.

async fn scope_words(app: &Shared, s: &runners::Scope) -> String {

    match s.repo_id {

        Some(id) => match db::repos::get_by_id(&app.db, id).await.ok().flatten() {

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

            None => "a removed repository".into(),

        },

        None if s.owner.is_empty() => "the whole site".into(),

        None => format!("{}'s repositories", s.owner),

    }

}

/// A scope as typed: empty for the whole site, an owner, or

/// `owner/repository`.

async fn parse_scope(app: &Shared, text: &str) -> Result<runners::Scope, String> {

    let text = text.trim().trim_matches('/');

    if text.is_empty() {

        return Ok(runners::Scope::default());

    }

    if let Some((owner, name)) = text.split_once('/') {

        return match db::repos::get(&app.db, owner, name).await {

            Ok(Some(r)) => Ok(runners::Scope { owner: r.owner.clone(), repo_id: Some(r.id) }),

            _ => Err(format!("There is no repository {text}.")),

        };

    }

    let user = accounts::account_by_name(&app.db, text).await.ok().flatten().is_some();

    let org = accounts::org_by_name(&app.db, text).await.ok().flatten().is_some();

    if user || org {

        Ok(runners::Scope { owner: text.to_string(), repo_id: None })

    } else {

        Err(format!("There is no user or organisation {text}."))

    }

}

#[derive(Template)]

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

pub struct RunnersPage {

    shell: Shell,

    runners: Vec<RunnerRow>,

    /// Registration tokens waiting to be used: (created by, created, expires).

    pending: Vec<(String, i64, i64)>,

    /// A token just made, shown once, with the command that uses it.

    new_token: Option<String>,

    hub: String,

    notice: Option<String>,

    error: Option<String>,

    /// Running jobs, then queued ones in the order runners claim them,

    /// as the filter keeps them.

    queue: Vec<db::flow::QueueEntry>,

    /// How many there are, unfiltered.

    queue_total: usize,

    filter: QueueFilter,

    /// The repositories and runners in the queue, for the filter's choices.

    queue_repos: Vec<String>,

    queue_runners: Vec<String>,

}

async fn page(

    app: &Shared,

    shell: Shell,

    new_token: Option<String>,

    notice: Option<String>,

    error: Option<String>,

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

    page_filtered(app, shell, new_token, notice, error, QueueFilter::default()).await

}

async fn page_filtered(

    app: &Shared,

    shell: Shell,

    new_token: Option<String>,

    notice: Option<String>,

    error: Option<String>,

    filter: QueueFilter,

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

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

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

    let running: HashMap<i64, i64> = runners::running(&app.db).await.map_err(e)?.into_iter().collect();

    let pending = runners::pending_registrations(&app.db).await.map_err(e)?;

    let all = db::flow::queue(&app.db).await.map_err(e)?;

    let mut rows = Vec::with_capacity(list.len());

    for r in list {

        let scope = scope_words(app, &r.scope()).await;

        let swarm = app

            .swarm

            .addr(r.id)

            .map(|s| if s.site.is_empty() { s.addr } else { format!("{} · site {}", s.addr, s.site) })

            .unwrap_or_default();

        let jobs = all

            .iter()

            .filter(|q| q.status == "running" && q.runner == r.name)

            .map(|q| (format!("{}/{}", q.owner, q.name), q.run_number, q.job.clone(), q.started_ms))

            .collect();

        rows.push(RunnerRow { running: running.get(&r.id).copied().unwrap_or(0), scope, swarm, jobs, r });

    }

    let mut queue_repos: Vec<String> = all.iter().map(|q| format!("{}/{}", q.owner, q.name)).collect();

    queue_repos.sort();

    queue_repos.dedup();

    let mut queue_runners: Vec<String> = all.iter().filter(|q| !q.runner.is_empty()).map(|q| q.runner.clone()).collect();

    queue_runners.sort();

    queue_runners.dedup();

    let queue_total = all.len();

    let queue = all.into_iter().filter(|q| filter.keeps(q)).collect();

    Ok(Page(RunnersPage {

        hub: app.config.base_url().to_string(),

        shell,

        runners: rows,

        pending,

        new_token,

        notice,

        error,

        queue,

        queue_total,

        filter,

        queue_repos,

        queue_runners,

    }))

}

/// `GET /admin/runners`.

pub async fn list(State(app): State<Shared>, viewer: Viewer, Query(filter): Query<QueueFilter>) -> Result<Page<RunnersPage>, Error> {

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

    require_admin(&viewer, &shell)?;

    page_filtered(&app, shell, None, None, None, filter).await

}

#[derive(Deserialize)]

pub struct CsrfOnly {

    #[serde(rename = "_csrf")]

    csrf_token: String,

}

#[derive(Deserialize)]

pub struct ScopeForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// Empty for the whole site, an owner, or `owner/repository`.

    #[serde(default)]

    scope: String,

}

/// `POST /admin/runners/token`: a one-time registration token, shown once,

/// for a runner of the scope given.

pub async fn new_token(State(app): State<Shared>, viewer: Viewer, Form(f): Form<ScopeForm>) -> Result<Page<RunnersPage>, Error> {

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

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

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

    let scope = match parse_scope(&app, &f.scope).await {

        Ok(s) => s,

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

    };

    let token = format!("flowreg_{}", crate::auth::random_hex(24));

    runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), &me.username, &scope)

        .await

        .map_err(|e| Error::from_db(shell.clone(), e))?;

    let words = scope_words(&app, &scope).await;

    accounts::audit(&app.db, Some(&me), "flow.runner.token", "", &format!("registration token created for {words}")).await.ok();

    page(&app, shell, Some(token), Some(format!("The runner it registers will take jobs from {words}.")), None).await

}

/// `POST /admin/runners/{id}/scope`: change the repositories a runner serves.

pub async fn set_scope(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<ScopeForm>,

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

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

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

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

    let r =

        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let scope = match parse_scope(&app, &f.scope).await {

        Ok(s) => s,

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

    };

    runners::set_scope(&app.db, id, &scope).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let words = scope_words(&app, &scope).await;

    accounts::audit(&app.db, Some(&me), "flow.runner.scope", &r.name, &words).await.ok();

    page(&app, shell, None, Some(format!("{} now takes jobs from {words}.", r.name)), None).await

}

#[derive(Deserialize)]

pub struct LimitsForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// Comma-separated.

    #[serde(default)]

    labels: String,

    /// Minutes; empty for the site's limit.

    #[serde(default)]

    timeout_minutes: String,

}

/// `POST /admin/runners/{id}/limits`: the labels an administrator gives a

/// runner and its own job time limit (#98).

pub async fn set_limits(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<LimitsForm>,

) -> Result<Response, Error> {

    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);

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

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

    let r =

        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let mut labels: Vec<String> = Vec::new();

    for l in f.labels.split([',', ' ']).map(str::trim).filter(|l| !l.is_empty()) {

        if !l.chars().all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')) || l.len() > 40 {

            return Err(Error::bad(shell, format!("label {l:?}: letters, digits, - _ and . only, up to 40 characters")));

        }

        if !labels.iter().any(|x| x == l) {

            labels.push(l.to_string());

        }

    }

    let timeout = match f.timeout_minutes.trim() {

        "" => None,

        t => match t.parse::<i64>() {

            Ok(m) if (1..=24 * 60).contains(&m) => Some(m),

            _ => return Err(Error::bad(shell, "a time limit is a whole number of minutes from 1 to 1440")),

        },

    };

    runners::set_limits(&app.db, id, &labels.join(","), timeout).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    let words = format!(

        "labels {}; time limit {}",

        if labels.is_empty() { "none".to_string() } else { labels.join(", ") },

        timeout.map(|m| format!("{m} min")).unwrap_or_else(|| "the site's".into())

    );

    accounts::audit(&app.db, Some(&me), "flow.runner.limits", &r.name, &words).await.ok();

    Ok(Redirect::to(&format!("/admin/runners/{id}")).into_response())

}

#[derive(Deserialize)]

pub struct StateForm {

    #[serde(rename = "_csrf")]

    csrf_token: String,

    /// `active`, `draining` or `disabled`.

    state: String,

}

/// `POST /admin/runners/{id}/state`: drain, disable or resume a runner.

pub async fn set_state(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<StateForm>,

) -> Result<Response, Error> {

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

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

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

    if !matches!(f.state.as_str(), "active" | "draining" | "disabled") {

        return Err(Error::bad(shell, "a runner is active, draining or disabled"));

    }

    let r =

        runners::get(&app.db, id).await.map_err(|e| Error::from_db(shell.clone(), e))?.ok_or_else(|| Error::not_found(shell.clone()))?;

    runners::set_state(&app.db, id, &f.state).await.map_err(|e| Error::from_db(shell.clone(), e))?;

    accounts::audit(&app.db, Some(&me), "flow.runner.state", &r.name, &f.state).await.ok();

    Ok(Redirect::to("/admin/runners").into_response())

}

/// `POST /admin/runners/{id}/remove`: forget a runner that runs nothing.

pub async fn remove(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Form(f): Form<CsrfOnly>,

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

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

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

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

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

    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let busy = runners::running(&app.db).await.map_err(e)?.into_iter().any(|(rid, n)| rid == id && n > 0);

    if busy {

        return page(

            &app,

            shell,

            None,

            None,

            Some(format!("{} is running jobs: drain it first, then remove it when they finish.", r.name)),

        )

        .await;

    }

    runners::remove(&app.db, id).await.map_err(e)?;

    accounts::audit(&app.db, Some(&me), "flow.runner.remove", &r.name, "").await.ok();

    let also = if r.source == "config" {

        " It is configured in hub.toml; remove its token there too, or it comes back on its next claim."

    } else {

        ""

    };

    page(&app, shell, None, Some(format!("Removed {}.{also}", r.name)), None).await

}

#[derive(Template)]

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

pub struct RunnerPage {

    shell: Shell,

    r: runners::Runner,

    running: i64,

    history: Vec<runners::RunnerJob>,

    declines: Vec<crate::db::flow::RunnerDecline>,

    older: Option<i64>,

    jobs: i64,

    passed: i64,

    busy_ms: i64,

}

impl RunnerPage {

    /// Time spent running jobs, as pages show durations.

    pub fn busy(&self) -> String {

        crate::flow::human_ms(self.busy_ms)

    }

}

#[derive(Deserialize, Default)]

pub struct HistoryQuery {

    before: Option<i64>,

}

const PER_PAGE: i64 = 50;

/// `GET /admin/runners/{id}`: a runner and the jobs it ran.

pub async fn show(

    State(app): State<Shared>,

    viewer: Viewer,

    Path(id): Path<i64>,

    Query(q): Query<HistoryQuery>,

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

    let shell = Shell::new(&app, &format!("/admin/runners/{id}"), &viewer);

    require_admin(&viewer, &shell)?;

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

    let r = runners::get(&app.db, id).await.map_err(e)?.ok_or_else(|| Error::not_found(shell.clone()))?;

    let mut history = runners::history(&app.db, id, PER_PAGE + 1, q.before).await.map_err(e)?;

    let older = (history.len() as i64 > PER_PAGE).then(|| history[PER_PAGE as usize - 1].started_ms).flatten();

    history.truncate(PER_PAGE as usize);

    let (jobs, passed, busy_ms) = runners::totals(&app.db, id).await.map_err(e)?;

    let running = runners::running(&app.db).await.map_err(e)?.into_iter().find(|(rid, _)| *rid == id).map(|(_, n)| n).unwrap_or(0);

    let declines = crate::db::flow::declines_of_runner(&app.db, id, 20).await.map_err(e)?;

    Ok(Page(RunnerPage { shell, r, running, history, declines, older, jobs, passed, busy_ms }))

}

pub fn duration(j: &runners::RunnerJob) -> String {

    match (j.started_ms, j.finished_ms) {

        (Some(a), Some(b)) => crate::flow::human_ms(b - a),

        (Some(a), None) => crate::flow::human_ms(db::now_ms() - a),

        _ => "–".into(),

    }

}

// ---------------------------------------------------------------- settings

#[derive(Template)]

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

pub struct SettingsPage {

    shell: Shell,

    default_m: u64,

    max_m: u64,

    grace_m: i64,

    /// What hub.toml says, shown beside a setting the page overrides.

    file_default_m: u64,

    flow_repos: Vec<String>,

    notice: Option<String>,

    error: Option<String>,

}

async fn settings_page(app: &Shared, shell: Shell, notice: Option<String>, error: Option<String>) -> Page<SettingsPage> {

    let l = crate::flow::limits(app).await;

    Page(SettingsPage {

        shell,

        default_m: l.default_s / 60,

        max_m: l.max_s / 60,

        grace_m: l.grace_ms / 60_000,

        file_default_m: app.config.flow.job_timeout_minutes,

        flow_repos: app.config.flow.repos.clone(),

        notice,

        error,

    })

}

/// `GET /admin/runners/settings` (oxygenhub#89).

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