crates/hub/src/flow.rs at 53a0fb2e4ac6 · andymccall/oxygenhub

# 53a0fb2e4ac6 oxygenhub / crates / hub / src / flow.rs
114.1 KiB · Rust · last changed Flow priorities: site, organisation and repository levels, and expediting a run 19 days ago source blame history raw
//! OxygenFlow on the hub (hub spec §13.2): runs started by pushes, the
//! runner API, job tokens and the checks that mirror jobs. Pipelines are
//! parsed by `flow-core`, the same code `flow check` runs.

use crate::api::ApiError;
use crate::db::{self, flow as fdb};
use crate::repos::RepoName;
use crate::Shared;
use axum::extract::{Path, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::{IntoResponse, Response};
use axum::Json;
use flow_core::api::{Claim, ClaimRequest, Finish, LogBatch, MAX_RECORDS, MAX_RECORD_TEXT};
use flow_core::{Pipeline, PIPELINE_PATH};
use oxy_core::hash::ChangeHash;

/// Prefix of job tokens, so the wire protocol can tell them from personal tokens.
pub const JOB_TOKEN_PREFIX: &str = "flowjob_";
/// A job token is refused this long after its job started, whatever its state.
pub const JOB_TOKEN_MAX_AGE_MS: i64 = 2 * 3600 * 1000;

/// Whether pushes to this repository start runs: its own setting when a
/// site administrator made one, else hub.toml's `[flow] repos`.
pub fn enabled(app: &Shared, row: &db::repos::RepoRow) -> bool {
    match row.flow.as_str() {
        // `own`: on, run only by the repository's or its owner's runners (#225).
        "on" | "own" => true,
        "off" => false,
        _ => app.config.flow.allows(&row.owner, &row.name),
    }
}

/// A bookmark moved by a push: start a run when the repository is allowed
/// and the new state has a pipeline.
pub async fn on_push(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, state: &str, heads: &[ChangeHash], actor: &str) {
    if heads.is_empty() || !enabled(app, row) {
        return;
    }
    // A pushed tag runs the jobs whose `on` names it (oxygenflow#24).
    let started = match bookmark.strip_prefix("tags/") {
        Some(tag) => start_run_with(app, row, bookmark, state, heads, actor, &format!("tag:{tag}"), None).await,
        None => {
            let touched = touched_since_last_run(app, row, bookmark, heads).await;
            let started = start_run_scoped(app, row, bookmark, state, heads, actor, "push", touched.as_deref()).await;
            if let Ok(Some(n)) = started {
                supersede(app, row, bookmark, n).await;
            }
            started
        }
    };
    if let Err(e) = started {
        tracing::warn!("flow: starting a run of {}/{}: {e}", row.owner, row.name);
    }
}

/// "Run pipeline" (hub spec §13.2): run a bookmark's current state without
/// moving it. Returns the new run's number.
/// `jobs`, when not empty, are the only jobs to start (oxygenhub#130), with
/// the jobs they depend on.
pub async fn run_bookmark(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, actor: &str, jobs: &[String]) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let b = bookmark.to_string();
    let found = app
        .repos
        .read(&repo, move |r, _| {
            let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
            Ok(Some((st, r.store().get_state(&st)?.heads)))
        })
        .await
        .map_err(|e| e.to_string())?;
    let Some((state, heads)) = found else { return Err(format!("no bookmark {bookmark}")) };
    if heads.is_empty() {
        return Err(format!("{bookmark} has no changes yet"));
    }
    start_run_at(app, row, bookmark, &state.to_hex(), &heads, actor, "manual", None, None, jobs)
        .await?
        .ok_or_else(|| format!("{bookmark} has no {PIPELINE_PATH}"))
}

/// "Run pipeline" on one committed change: the state whose only head is
/// `change`, whatever bookmarks hold it. Returns the new run's number.
pub async fn run_change(app: &Shared, row: &db::repos::RepoRow, change: &str, actor: &str, jobs: &[String]) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    let h = ChangeHash::from_hex(change.trim().trim_start_matches('#')).ok_or("give the full hash of a change")?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let known = app.repos.read(&repo, move |r, _| Ok(r.store().get_change_meta(&h).is_ok())).await.map_err(|e| e.to_string())?;
    if !known {
        return Err(format!("no change {} in this repository", &change.trim()[..12.min(change.trim().len())]));
    }
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(vec![h]).encode());
    start_run_at(app, row, "", &state.to_hex(), &[h], actor, "manual", None, None, jobs)
        .await?
        .ok_or_else(|| format!("that change's state has no {PIPELINE_PATH}"))
}

/// Re-run a finished run: the same state, bookmark and pipeline, as a new
/// run whose checks replace the old ones. Returns the new run's number.
pub async fn rerun(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<i64, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    if run.live() {
        return Err(format!("run #{} is still {}", run.number, run.status));
    }
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    if heads.is_empty() {
        return Err("the run names no changes".into());
    }
    // A rerun repeats its run's event: the jobs it had are the jobs it gets.
    let trigger = format!("rerun:{}:{}", run.number, original_event(run));
    start_run(app, row, &run.bookmark, &run.state, &heads, actor, &trigger)
        .await?
        .ok_or_else(|| format!("that state has no {PIPELINE_PATH}"))
}

/// Run one job of a finished run again (oxygenflow#31), with the jobs
/// that depend on it; the jobs it depends on keep their results, so a
/// flaky test reruns without the build before it. Its checks go back to
/// queued. Returns how many jobs were queued again.
pub async fn rerun_job(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, job: &str, actor: &str) -> Result<usize, String> {
    if !enabled(app, row) {
        return Err("OxygenFlow is not enabled for this repository".into());
    }
    if run.live() {
        return Err(format!("run #{} is still {}: wait for it, or cancel it, first", run.number, run.status));
    }
    let changed = fdb::rerun_job(&app.db, run.id, job).await.map_err(|e| {
        tracing::warn!("flow: rerunning {job} of run #{} of {}/{}: {e}", run.number, row.owner, row.name);
        "the job could not be queued again".to_string()
    })?;
    if changed.is_empty() {
        return Err(format!("run #{} has no job {job:?}", run.number));
    }
    for j in &changed {
        report_check(app, row, &run.state, run.number, &j.name, "queued", &format!("queued again by {actor}")).await;
    }
    Ok(changed.len())
}

/// Cancel a run still going (oxygenflow#32), its checks with it.
pub async fn cancel(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, actor: &str) -> Result<(), String> {
    if !run.live() {
        return Err(format!("run #{} is already {}", run.number, run.status));
    }
    cancel_with(app, run, &format!("cancelled by {actor}")).await.map_err(|e| {
        tracing::warn!("flow: cancelling run #{} of {}/{}: {e}", run.number, row.owner, row.name);
        "the run could not be cancelled".to_string()
    })
}

async fn cancel_with(app: &Shared, run: &fdb::Run, why: &str) -> sqlx::Result<()> {
    let changed = fdb::cancel_run(&app.db, run.id, why).await?;
    report_changed(app, run.id, changed).await
}

/// Why job `job` of `run` cannot be decided on by `who` (oxygenflow#57),
/// or its approval: it must be waiting, the jobs it depends on done and
/// passed, `who` must have the role it names on the repository, and with
/// others-only must not be the person whose push started the run.
pub async fn may_decide(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    job: &fdb::Job,
    who: &db::accounts::Account,
) -> Result<flow_core::Approval, String> {
    let spec: flow_core::Job = serde_json::from_str(&job.spec).map_err(|e| format!("stored job: {e}"))?;
    let Some(approval) = spec.approval else { return Err(format!("{} does not wait for approval", job.name)) };
    if job.status != "queued" || job.approval != "waiting" {
        return Err(format!("{} is not waiting for approval", job.name));
    }
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| e.to_string())?;
    if let Some(d) = job.depends_on().iter().find(|d| !jobs.iter().any(|j| &j.name == *d && j.status == "passed")) {
        return Err(format!("{} waits for {d} first", job.name));
    }
    let need = crate::auth::roles::Role::parse(&approval.by).unwrap_or(crate::auth::roles::Role::Maintain);
    let role = crate::auth::roles::role_for(&app.db, Some(who), row).await.map_err(|e| e.to_string())?;
    if role.is_none_or(|r| r < need) {
        return Err(format!("approving {} takes the {} role on {}/{}", job.name, approval.by, row.owner, row.name));
    }
    if approval.others_only && who.username == run.actor {
        return Err(format!("{} must be approved by someone other than {}, whose push started the run", job.name, run.actor));
    }
    Ok(approval)
}

/// Approve or reject job `name` of `run` as `who` (oxygenflow#57): the
/// decision goes on the job, its log and the audit log; a rejected job ends
/// as cancelled and the jobs after it are skipped. What happened, or why not.
pub async fn decide(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    name: &str,
    who: &db::accounts::Account,
    approve: bool,
) -> Result<String, String> {
    decide_with(app, row, run, name, who, approve, None).await
}

/// [`decide`], from a verified seal (`Some(approval ref)`) or without one;
/// a gate that takes a sealed approval is not approved without.
async fn decide_with(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    name: &str,
    who: &db::accounts::Account,
    approve: bool,
    seal: Option<&str>,
) -> Result<String, String> {
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| e.to_string())?;
    let Some(job) = jobs.into_iter().find(|j| j.name == name) else { return Err(format!("run #{} has no job {name}", run.number)) };
    let approval = may_decide(app, row, run, &job, who).await?;
    if approve && approval.sealed && seal.is_none() {
        return Err(format!(
            "{name} takes an approval signed with your key: in a checkout of {}/{}, run oxy seal --flow {}/{}/{}/flow/{} --job {name}",
            row.owner,
            row.name,
            app.config.base_url(),
            row.owner,
            row.name,
            run.number
        ));
    }
    let target = format!("{}/{}#{} {name}", row.owner, row.name, run.number);
    let sealed = seal.map(|r| format!(", sealed (approval {})", &r[..12.min(r.len())])).unwrap_or_default();
    let said = if approve {
        if !fdb::approve_job(&app.db, job.id, &who.username, seal.unwrap_or("")).await.map_err(|e| e.to_string())? {
            return Err(format!("{name} was decided on meanwhile"));
        }
        db::accounts::audit(&app.db, Some(who), "flow.approved", &target, seal.unwrap_or("")).await.ok();
        format!("approved by {}{sealed}", who.username)
    } else {
        fdb::note_rejection(&app.db, job.id, &who.username).await.map_err(|e| e.to_string())?;
        let why = format!("rejected by {}{sealed}", who.username);
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "cancelled", exit_code: None, error: &why, receipt: "", receipt_hash: "" })
                .await
                .map_err(|e| e.to_string())?;
        report_changed(app, run.id, changed).await.ok();
        db::accounts::audit(&app.db, Some(who), "flow.rejected", &target, "").await.ok();
        why
    };
    let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said.clone() };
    fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
    Ok(format!("{name}: {said}"))
}

/// Tell the people who may approve a job of `run` that has just become
/// ready (its dependencies all passed) and waits for them (oxygenflow#57).
/// Told once: the job's log notes it.
async fn tell_approvers(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) {
    const TOLD: &str = "waiting for approval; the people who may approve were told";
    let Ok(jobs) = fdb::jobs(&app.db, run.id).await else { return };
    for job in jobs.iter().filter(|j| j.status == "queued" && j.approval == "waiting") {
        if !job.depends_on().iter().all(|d| jobs.iter().any(|j| &j.name == d && j.status == "passed")) {
            continue;
        }
        let already = fdb::log(&app.db, job.id, 0, 50).await.unwrap_or_default().iter().any(|l| l.text == TOLD);
        if already {
            continue;
        }
        let mut targets = Vec::new();
        for a in db::accounts::accounts(&app.db).await.unwrap_or_default().iter().filter(|a| !a.blocked()) {
            if may_decide(app, row, run, job, a).await.is_ok() {
                targets.push((a.id, db::notify::Reason::Participating));
            }
        }
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: TOLD.into() };
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
        let title = format!("{}/{} run #{}: {} waits for your approval", row.owner, row.name, run.number, job.name);
        let url = format!("/{}/{}/flow/{}", row.owner, row.name, run.number);
        crate::notify::tell(app, "flow.approval", &title, &url, &run.actor, Some(row.id), targets).await;
    }
}

/// What approving job `name` of `run` means, for the approver to sign
/// (oxygenhub#235): this repository, the run's URL, the job's hash with its
/// sandbox, the run's state, the blobs it takes, its environment.
pub async fn approval_statement(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    name: &str,
) -> Result<(oxy_core::approval::Statement, fdb::Job), String> {
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| e.to_string())?;
    let Some(job) = jobs.into_iter().find(|j| j.name == name) else { return Err(format!("run #{} has no job {name}", run.number)) };
    let spec: flow_core::Job = serde_json::from_str(&job.spec).map_err(|e| format!("stored job: {e}"))?;
    let sandbox: flow_core::Sandbox = serde_json::from_str(&job.sandbox).map_err(|e| format!("stored sandbox: {e}"))?;
    let repo_name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let repo = app.repos.read(&repo_name, |r, _| Ok(r.id()?)).await.map_err(|e| e.to_string())?;
    let state = oxy_core::hash::ObjectRef::from_hex(&run.state).ok_or("the run has no state")?;
    let inputs = artifact_inputs(app, &job, &spec).await?.iter().filter_map(|i| oxy_core::hash::ObjectRef::from_hex(&i.blob)).collect();
    let statement = oxy_core::approval::Statement {
        repo,
        run: format!("{}/{}/{}/flow/{}", app.config.base_url(), row.owner, row.name, run.number),
        job: name.to_string(),
        job_hash: oxy_core::hash::ObjectRef(spec.hash(&sandbox)),
        state,
        inputs,
        environment: spec.environment.clone().unwrap_or_default(),
        approve: true,
    };
    Ok((statement, job))
}

/// Decide on job `name` of `run` from a signed approval (oxygenhub#235): it
/// must verify, cover exactly this job's statement, and be signed by a key
/// linked to an account (a delegated agent key is not) that may decide; it
/// is kept in the repository. What happened, or why not.
pub async fn decide_sealed(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run, name: &str, signed_hex: &str) -> Result<String, String> {
    let bytes = hex::decode(signed_hex.trim()).map_err(|_| "the signed approval is not hex".to_string())?;
    let sa = oxy_core::approval::SignedApproval::decode(&bytes).map_err(|e| format!("not a signed approval: {e}"))?;
    sa.verify().map_err(|e| format!("it does not verify: {e}"))?;
    let (mut expected, _) = approval_statement(app, row, run, name).await?;
    let given = sa.statement().map_err(|e| e.to_string())?;
    expected.approve = given.approve;
    if !sa.covers(&expected) {
        return Err(format!(
            "it approves something else than job {name} of run #{} as it is now (another run, state, plan or environment): sign again",
            run.number
        ));
    }
    let key = sa.signer.hash().to_hex();
    let links = db::accounts::identity_accounts(&app.db).await.map_err(|e| e.to_string())?;
    let Some((_, username)) = links.iter().find(|(k, _)| *k == key) else {
        return Err(format!(
            "key {} is not linked to an account here (link it under Settings › Keys; an agent's delegated key cannot approve)",
            &key[..12]
        ));
    };
    let who = db::accounts::account_by_name(&app.db, username).await.map_err(|e| e.to_string())?.ok_or("the key's account is gone")?;
    let repo_name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let jobs = fdb::jobs(&app.db, run.id).await.map_err(|e| e.to_string())?;
    let job = jobs.iter().find(|j| j.name == name).ok_or("no such job")?;
    may_decide(app, row, run, job, &who).await?;
    let kept = app.repos.write(&repo_name, move |r, _| Ok(r.add_approval(&sa)?)).await.map_err(|e| e.to_string())?;
    decide_with(app, row, run, name, &who, given.approve, Some(&kept.to_hex())).await
}

/// End the approvals that waited too long (oxygenflow#57): from when the
/// jobs a waiting job depends on finished (or its run began).
pub async fn expire_approvals(app: &Shared, now: i64) -> sqlx::Result<usize> {
    let mut ended = 0;
    for job in fdb::waiting_for_approval(&app.db).await? {
        let Ok(spec) = serde_json::from_str::<flow_core::Job>(&job.spec) else { continue };
        let Some(approval) = spec.approval else { continue };
        let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? else { continue };
        let jobs = fdb::jobs(&app.db, run.id).await?;
        let wanted = job.depends_on();
        let deps: Vec<&fdb::Job> = jobs.iter().filter(|j| wanted.contains(&j.name)).collect();
        if deps.iter().any(|d| d.status != "passed") {
            continue;
        }
        let since = deps.iter().filter_map(|d| d.finished_ms).max().unwrap_or(run.created_ms);
        if now - since < approval.expires_s as i64 * 1000 {
            continue;
        }
        let hours = approval.expires_s as f64 / 3600.0;
        let why =
            format!("no approval within {}", if hours >= 1.0 { format!("{hours:.0}h") } else { format!("{}m", approval.expires_s / 60) });
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "cancelled", exit_code: None, error: &why, receipt: "", receipt_hash: "" })
                .await?;
        report_changed(app, run.id, changed).await?;
        ended += 1;
    }
    Ok(ended)
}

/// A push makes the push runs still going on its bookmark moot: they are
/// cancelled, and the new run's scope reaches back past them.
async fn supersede(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, number: i64) {
    let Ok(older) = fdb::live_push_runs_before(&app.db, row.id, bookmark, number).await else { return };
    for run in older {
        if let Err(e) = cancel_with(app, &run, &format!("superseded by run #{number}")).await {
            tracing::warn!("flow: superseding run #{} of {}/{}: {e}", run.number, row.owner, row.name);
        }
    }
}

/// The event a run's trigger stands for: `push`, `tag:NAME`, `manual`,
/// `schedule`, and `rerun:N:EVENT` repeating EVENT.
fn event_of<'a>(trigger: &'a str, bookmark: &'a str) -> flow_core::pipeline::Event<'a> {
    use flow_core::pipeline::Event;
    let t = match trigger.strip_prefix("rerun:") {
        Some(rest) => rest.split_once(':').map(|(_, e)| e).unwrap_or("manual"),
        None => trigger,
    };
    if let Some(tag) = t.strip_prefix("tag:") {
        Event::Tag(tag)
    } else if t == "push" {
        Event::Push(bookmark)
    } else if t == "schedule" {
        Event::Schedule
    } else {
        Event::Manual
    }
}

/// The event to repeat when rerunning `run`.
fn original_event(run: &fdb::Run) -> String {
    let t = run.trigger.as_str();
    match t.strip_prefix("rerun:") {
        Some(rest) => rest.split_once(':').map(|(_, e)| e.to_string()).unwrap_or_else(|| "manual".into()),
        None => t.to_string(),
    }
}

/// The paths the changes a push brought touch: the changes in `heads`'
/// state that the bookmark's previous run did not have, and the files their
/// edits and tree operations name, as the new state places them. `None`
/// (run every job) for the bookmark's first run, or when it cannot be told.
async fn touched_since_last_run(app: &Shared, row: &db::repos::RepoRow, bookmark: &str, heads: &[ChangeHash]) -> Option<Vec<String>> {
    let last = fdb::latest_on(&app.db, row.id, bookmark).await.ok()??;
    let old: Vec<ChangeHash> = last.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    if old.is_empty() {
        return None;
    }
    let repo = RepoName::new(&row.owner, &row.name)?;
    let new = heads.to_vec();
    app.repos
        .read(&repo, move |r, cache| {
            let fresh = r.negotiate(&old, &new)?;
            let loaded = cache.load(r, &new)?;
            let by_node: std::collections::HashMap<oxy_core::ids::NodeId, &String> =
                loaded.snapshot.files.iter().map(|(p, f)| (f.node, p)).collect();
            let mut paths = std::collections::BTreeSet::new();
            for h in fresh {
                let (c, _) = r.store().get_change_lazy(&h)?;
                let nodes = c.edits.iter().map(|e| e.node()).chain(c.tree_ops.iter().map(|o| o.node()));
                for n in nodes {
                    if let Some(p) = by_node.get(&n) {
                        paths.insert((*p).clone());
                    }
                }
            }
            Ok(paths.into_iter().collect::<Vec<String>>())
        })
        .await
        .ok()
}

/// Whether a job runs for a push that touched `touched`: always without a
/// scope or without knowing what was touched, else when a path matches.
fn in_scope(scope: &[String], touched: Option<&[String]>) -> bool {
    let Some(touched) = touched else { return true };
    if scope.is_empty() {
        return true;
    }
    let mut b = globset::GlobSetBuilder::new();
    for g in scope {
        match globset::Glob::new(g) {
            Ok(g) => {
                b.add(g);
            }
            Err(_) => return true,
        }
    }
    let Ok(set) = b.build() else { return true };
    touched.iter().any(|p| set.is_match(p))
}

/// [`start_run`] for a push: jobs whose `scope` the push did not touch are
/// left out of the run, unless a job that runs depends on them.
#[allow(clippy::too_many_arguments)]
pub async fn start_run_scoped(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
) -> Result<Option<i64>, String> {
    start_run_with(app, row, bookmark, state, heads, actor, trigger, touched).await
}

/// Start a run of the pipeline in the state `heads` make (whose state
/// hash is `state`), recorded on `bookmark`: for a push, "Run pipeline"
/// (`trigger` = `manual`) and re-runs (`rerun:N`). `actor` is the account
/// whose access the run's subrepo reads are checked against. Returns the
/// run's number, or `None` when the state has no pipeline. The caller
/// checks that Flow is enabled.
pub async fn start_run(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
) -> Result<Option<i64>, String> {
    start_run_with(app, row, bookmark, state, heads, actor, trigger, None).await
}

#[allow(clippy::too_many_arguments)]
async fn start_run_with(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
) -> Result<Option<i64>, String> {
    start_run_at(app, row, bookmark, state, heads, actor, trigger, touched, None, &[]).await
}

/// [`start_run_with`], for a scheduled run due at `at` on the hub's clock;
/// `only`, when not empty, the jobs a manual run starts (oxygenhub#130).
#[allow(clippy::too_many_arguments)]
async fn start_run_at(
    app: &Shared,
    row: &db::repos::RepoRow,
    bookmark: &str,
    state: &str,
    heads: &[ChangeHash],
    actor: &str,
    trigger: &str,
    touched: Option<&[String]>,
    at: Option<flow_core::pipeline::When>,
    only: &[String],
) -> Result<Option<i64>, String> {
    let repo = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let h = heads.to_vec();
    let text = app
        .repos
        .read(&repo, move |r, cache| {
            let loaded = cache.load(r, &h)?;
            Ok(match loaded.snapshot.files.get(PIPELINE_PATH).map(|f| &f.content) {
                Some(oxy_core::state::FileContent::Text { bytes, .. }) => Some(String::from_utf8_lossy(bytes).into_owned()),
                _ => None,
            })
        })
        .await
        .map_err(|e| format!("reading the pipeline: {e}"))?;
    let Some(text) = text else { return Ok(None) };
    let mut hex: Vec<String> = heads.iter().map(|h| h.to_hex()).collect();
    hex.sort();
    let heads_s = hex.join("+");
    let base = db::flow::NewRun { repo_id: row.id, bookmark, state, heads: &heads_s, pipeline_hash: "", actor, trigger, error: None };
    let id = match Pipeline::parse(&text) {
        Err(problems) => {
            let msg = problems.to_string();
            db::flow::create_run(&app.db, db::flow::NewRun { error: Some(&msg), ..base }, &[]).await.map_err(|e| e.to_string())?
        }
        Ok(p) => {
            let depths = p.depths();
            // Jobs out of the push's scope stay out, unless a kept job
            // needs them.
            // The jobs this run's event starts (`on`), out of those its
            // push touched (`scope`); then what they depend on.
            let event = match at {
                Some(w) => flow_core::pipeline::Event::ScheduleAt(w),
                None => event_of(trigger, bookmark),
            };
            let starts = |j: &flow_core::Job| p.on_of(j).matches(event) && in_scope(&j.scope, touched);
            let mut keep: std::collections::HashSet<String> = if only.is_empty() {
                p.order().into_iter().filter(|j| starts(j)).map(|j| j.name.clone()).collect()
            } else {
                // Named jobs (oxygenhub#130): each one this event starts.
                let startable: Vec<&str> = p.order().into_iter().filter(|j| starts(j)).map(|j| j.name.as_str()).collect();
                for name in only {
                    if !startable.contains(&name.as_str()) {
                        let what = if p.jobs.contains_key(name) {
                            "is not started by hand (no `on { manual }`)"
                        } else {
                            "is not in the pipeline"
                        };
                        return Err(format!("job {name:?} {what}; jobs a manual run can start: {}", startable.join(", ")));
                    }
                }
                only.iter().cloned().collect()
            };
            loop {
                let needed: Vec<String> = p
                    .order()
                    .into_iter()
                    .filter(|j| keep.contains(&j.name))
                    .flat_map(|j| j.depends_on.iter().cloned())
                    .filter(|d| !keep.contains(d))
                    .collect();
                if needed.is_empty() {
                    break;
                }
                keep.extend(needed);
            }
            let mut jobs: Vec<fdb::NewJob> = p
                .order()
                .into_iter()
                .filter(|j| keep.contains(&j.name))
                .map(|j| {
                    let sb = &p.sandboxes[&j.sandbox];
                    fdb::NewJob {
                        name: j.name.clone(),
                        depth: depths[j.name.as_str()] as i64,
                        depends: j.depends_on.clone(),
                        spec: serde_json::to_string(j).unwrap_or_default(),
                        sandbox: serde_json::to_string(sb).unwrap_or_default(),
                        runtime: sb.runtime.as_str().into(),
                        need_disk_mib: None,
                    }
                })
                .collect();
            // What each job needs, learned from its past runs (oxygenflow#41).
            for j in &mut jobs {
                j.need_disk_mib = fdb::learned_disk_need(&app.db, row.id, &j.name).await.map_err(|e| e.to_string())?;
            }
            if jobs.is_empty() {
                // Nothing this event starts (or this push touched).
                return Ok(None);
            }
            let hash = hex::encode(p.hash());
            let id =
                db::flow::create_run(&app.db, db::flow::NewRun { pipeline_hash: &hash, ..base }, &jobs).await.map_err(|e| e.to_string())?;
            if let Ok(Some(run)) = fdb::run_by_id(&app.db, id).await {
                for j in &jobs {
                    report_check(app, row, state, run.number, &j.name, "queued", "waiting for a runner").await;
                }
                images_check(app, row, state, run.number, &jobs).await;
            }
            id
        }
    };
    let run = fdb::run_by_id(&app.db, id).await.map_err(|e| e.to_string())?.ok_or("the run vanished")?;
    Ok(Some(run.number))
}

/// The `pipeline/images` check (#137): set failing when an image the run's
/// jobs pin is held by neither the hub nor a runner, so a proposal pinning
/// one says so before it lands rather than its runs queueing for good;
/// passed again by a later run of the state once every image is held.
async fn images_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, jobs: &[fdb::NewJob]) {
    let mut pinned: Vec<String> = jobs
        .iter()
        .filter_map(|j| serde_json::from_str::<serde_json::Value>(&j.sandbox).ok())
        .filter_map(|v| v["rootfs"].as_str().map(String::from))
        .filter(|r| !r.is_empty())
        .collect();
    pinned.sort();
    pinned.dedup();
    if pinned.is_empty() {
        return;
    }
    let mut held = crate::images::held(&app.db).await.unwrap_or_default();
    // Nobody can be said to lack an image unless some runner is known and
    // every one reports what it holds.
    let runners: Vec<db::runners::Runner> =
        db::runners::list(&app.db).await.unwrap_or_default().into_iter().filter(|r| r.state != "disabled").collect();
    if runners.is_empty() || runners.iter().any(|r| r.image_list().is_none()) {
        return;
    }
    for r in &runners {
        held.extend(r.image_list().unwrap_or_default());
    }
    let missing: Vec<&String> = pinned.iter().filter(|h| !held.contains(*h)).collect();
    // Only a missing image makes the check; one that was failing on this
    // state turns to passed once its images are held.
    let failing_before = db::hooks::checks_for_state(&app.db, row.id, state)
        .await
        .unwrap_or_default()
        .iter()
        .any(|c| c.name == "pipeline/images" && c.status != "passed");
    if missing.is_empty() && !failing_before {
        return;
    }
    let (status, summary) = if missing.is_empty() {
        (
            "passed",
            format!("{} now held", if pinned.len() == 1 { "its image is".to_string() } else { format!("all {} images are", pinned.len()) }),
        )
    } else {
        let names: Vec<String> = missing.iter().map(|h| h[..12.min(h.len())].to_string()).collect();
        ("failed", format!("neither the hub nor a runner holds image {}: send it with flow rootfs push HASH", names.join(", ")))
    };
    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);
    if let Ok(c) = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state, name: "pipeline/images", status, summary: &summary, url: &url, reporter: None, seal: "" },
    )
    .await
    {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
}

/// Record a job's check; it links to the run's page by the run's number.
async fn report_check(app: &Shared, row: &db::repos::RepoRow, state: &str, run_number: i64, job: &str, status: &str, summary: &str) {
    let url = format!("{}/{}/{}/flow/{run_number}", app.config.base_url(), row.owner, row.name);
    let name = format!("flow/{job}");
    let c = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state, name: &name, status, summary, url: &url, reporter: None, seal: "" },
    )
    .await;
    if let Ok(c) = c {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
}

/// The check status a job's status shows as.
fn check_status(job_status: &str) -> &'static str {
    match job_status {
        "queued" => "queued",
        "running" => "running",
        "passed" => "passed",
        "skipped" => "skipped",
        "cancelled" => "cancelled",
        _ => "failed",
    }
}

/// The runner a bearer token belongs to: a registered runner, or one
/// configured with a hub.toml token (tracked from its first request). A
/// disabled runner is refused.
pub(crate) async fn runner_auth(app: &Shared, headers: &HeaderMap) -> Result<db::runners::Runner, ApiError> {
    let Some(t) = oxy_server::bearer_token(headers) else {
        return Err(ApiError(StatusCode::UNAUTHORIZED, "a runner token is required".into()));
    };
    let hash = blake3::hash(t.as_bytes()).to_hex().to_string();
    let runner = match db::runners::by_token(&app.db, &hash).await? {
        Some(r) => r,
        None if app.config.flow.runner_tokens.iter().any(|x| crate::auth::mac_equal(x, t)) => {
            db::runners::configured(&app.db, &hash, "runner").await?
        }
        None => return Err(ApiError(StatusCode::UNAUTHORIZED, "not a runner token".into())),
    };
    if runner.state == "disabled" {
        return Err(ApiError(StatusCode::FORBIDDEN, format!("runner {} is disabled on this hub", runner.name)));
    }
    Ok(runner)
}

/// `POST /api/v1/flow/runners/register`: a one-time registration token
/// from Admin → Runners buys a runner credential, binding the runner's
/// signing identity (hub spec §13.2).
pub async fn register(
    State(app): State<Shared>,
    Json(req): Json<flow_core::api::Register>,
) -> Result<Json<flow_core::api::Registered>, ApiError> {
    let bytes = hex::decode(req.identity.trim()).map_err(|_| ApiError(StatusCode::BAD_REQUEST, "identity: not hex".into()))?;
    let identity = oxy_core::identity::Identity::decode(&bytes).map_err(|e| ApiError(StatusCode::BAD_REQUEST, format!("identity: {e}")))?;
    let name: String = req.name.chars().filter(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.')).take(64).collect();
    if name.is_empty() {
        return Err(ApiError(StatusCode::BAD_REQUEST, "give the runner a name of letters, digits, - _ or .".into()));
    }
    let credential = format!("flowrunner_{}", crate::auth::random_hex(32));
    let hash = blake3::hash(credential.as_bytes()).to_hex().to_string();
    let reg = blake3::hash(req.token.trim().as_bytes()).to_hex().to_string();
    let runner =
        db::runners::register(&app.db, &reg, &name, &hash, &identity.hash().to_hex(), &identity.name, "").await?.ok_or_else(|| {
            ApiError(StatusCode::FORBIDDEN, "that registration token is unknown, used or expired; create one on Admin → Runners".into())
        })?;
    db::runners::seen(&app.db, runner.id, None, Some(&req.info)).await?;
    tracing::info!("flow: runner {} registered (key {})", runner.name, &runner.key[..12]);
    Ok(Json(flow_core::api::Registered { id: runner.id, name: runner.name, credential }))
}

/// `POST /api/v1/flow/claim`.
pub async fn claim(State(app): State<Shared>, headers: HeaderMap, Json(req): Json<ClaimRequest>) -> Result<Response, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    let claimed: String = req.runner.chars().filter(|c| !c.is_control()).take(64).collect();
    db::runners::seen(&app.db, me.id, Some(&claimed), req.info.as_ref()).await?;
    if let Some(holding) = &req.holding {
        end_jobs_not_held(&app, &me, holding).await?;
    }
    // A draining runner finishes what it has and takes nothing new.
    if me.state == "draining" {
        return Ok(StatusCode::NO_CONTENT.into_response());
    }
    let runner = if me.source == "registered" { me.name.clone() } else { claimed };
    let secret = format!("{JOB_TOKEN_PREFIX}{}", crate::auth::random_hex(32));
    // What it holds and what it is (OxygenFlow spec §11): the images'
    // architectures it reports are remembered for every runner's claims.
    if let Some(archs) = &req.image_archs {
        fdb::record_image_archs(&app.db, archs).await?;
    }
    let arch = req.info.as_ref().map(|i| i.arch.clone()).filter(|a| !a.is_empty()).unwrap_or_else(|| me.arch.clone());
    // What it offers, kept for the placement pass; a job it declined is
    // not offered to it again while this stays the same (spec §11.5).
    let profile = profile_of(&req, &arch);
    db::runners::set_offer(&app.db, me.id, &req.runtimes, req.images.as_deref(), req.fetches_images, &profile).await?;
    if let Some(f) = &req.facts {
        db::runners::set_facts(&app.db, me.id, f).await?;
    }
    app.swarm.claimed(me.id, req.swarm.as_ref());
    // The build caches it holds (oxygenflow#51), and when each last kept a
    // disk (oxygenflow#54; 0 from an older runner), for placing jobs by.
    if let Some(c) = &req.caches {
        let kept = |slot: &str| req.cache_times.as_ref().and_then(|t| t.get(slot)).map_or(0, |ms| *ms as i64);
        CACHES
            .lock()
            .unwrap_or_else(|p| p.into_inner())
            .get_or_insert_with(Default::default)
            .insert((app.repos.root().display().to_string(), me.id), c.iter().map(|s| (s.clone(), kept(s))).collect());
    }
    let room = fdb::Room {
        declined: fdb::declined_by(&app.db, me.id, &profile).await?,
        memory_mib: req.free_memory_mib,
        cpus: req.free_cpus,
        images: req.images.as_ref().map(|i| i.iter().map(|h| h.to_ascii_lowercase()).collect()),
        arch,
        image_archs: fdb::image_archs(&app.db).await?,
        fetchable: if req.fetches_images { crate::images::held(&app.db).await? } else { Default::default() },
        free_disk_mib: req.facts.as_ref().and_then(|f| f.free_disk_mib),
    };
    // The labels it reports now, or those it registered with, and those
    // an administrator gave it (#98).
    let mut labels: Vec<String> = match &req.info {
        Some(i) => i.labels.clone(),
        None => me.reported_label_list().into_iter().map(String::from).collect(),
    };
    for l in me.admin_label_list() {
        if !labels.iter().any(|x| x == l) {
            labels.push(l.to_string());
        }
    }
    // Ranking (oxygenflow#42): a job another free runner would do better
    // (it holds the image, or is faster) is held for it briefly.
    let runners = db::runners::list(&app.db).await?;
    let me_now = runners.iter().find(|r| r.id == me.id).cloned().unwrap_or_else(|| me.clone());
    let running: std::collections::HashMap<i64, i64> = db::runners::running(&app.db).await?.into_iter().collect();
    let queue = fdb::queue(&app.db).await?;
    let owners: std::collections::HashMap<i64, (String, i64)> = queue.iter().map(|q| (q.job_id, (q.owner.clone(), q.repo_id))).collect();
    // Each job's build cache slot (oxygenflow#51), as runners report theirs.
    let slots: std::collections::HashMap<i64, String> =
        queue.iter().map(|q| (q.job_id, flow_core::api::cache_slot(&format!("{}/{}", q.owner, q.name), &q.job))).collect();
    let (archs, held) = (fdb::image_archs(&app.db).await?, crate::images::held(&app.db).await?);
    let now = db::now_ms();
    let fleet = Fleet { image_archs: &archs, hub_images: &held, now };
    let hub_key = app.repos.root().display().to_string();
    let better = |j: &fdb::Job| -> bool {
        let Some((owner, repo_id)) = owners.get(&j.id) else { return false };
        let rootfs = rootfs_of(j);
        let mut holds = HELD.lock().unwrap_or_else(|p| p.into_inner());
        let holds = holds.get_or_insert_with(Default::default);
        holds.retain(|_, t| now - *t < 600_000);
        if holds.get(&(hub_key.clone(), j.id)).is_some_and(|t| now - *t >= BETTER_HOLD_MS) {
            return false;
        }
        let slot = slots.get(&j.id).map(String::as_str).unwrap_or("");
        let mine = fit(&hub_key, &me_now, &rootfs, slot);
        let other = runners.iter().any(|r| {
            r.id != me.id
                && r.state == "active"
                && r.last_seen_ms.is_some_and(|t| now - t < POLLING_MS)
                && running.get(&r.id).copied().unwrap_or(0) < r.max_jobs.max(1)
                && fit(&hub_key, r, &rootfs, slot).beats(&mine)
                && runner_lacks(j, owner, *repo_id, r, &fleet, &[]).is_none()
        });
        if other {
            holds.entry((hub_key.clone(), j.id)).or_insert(now);
        }
        other
    };
    let Some(job) =
        fdb::claim_ranked(&app.db, &req.runtimes, &runner, &crate::auth::token_hash(&secret), room, &me.scope(), &labels, &better).await?
    else {
        return Ok(StatusCode::NO_CONTENT.into_response());
    };
    // Why here (oxygenflow#42).
    let waited =
        HELD.lock().unwrap_or_else(|p| p.into_inner()).as_mut().and_then(|h| h.remove(&(hub_key.clone(), job.id))).map(|t| now - t);
    let rootfs = rootfs_of(&job);
    let mut why = vec![if me_now.image_list().is_some_and(|have| have.contains(&rootfs)) {
        "holds the image".to_string()
    } else {
        "fetches the image from the hub".to_string()
    }];
    if let Some(kept) = slots.get(&job.id).and_then(|slot| cache_kept(&hub_key, me.id, slot)) {
        why.insert(
            0,
            if kept > 0 { format!("holds its build cache (kept {} ago)", ago(now - kept)) } else { "holds its build cache".into() },
        );
    }
    match running.get(&me.id).copied().unwrap_or(0) {
        0 => why.push("idle".into()),
        n => why.push(format!("running {n} other job{}", if n == 1 { "" } else { "s" })),
    }
    if let Some(sp) = me_now.speed {
        why.push(format!("speed {sp} (class {})", me_now.speed_class.unwrap_or(0)));
    }
    if let Some(w) = waited {
        why.push(format!("after {} s held for a better runner", w / 1000));
    }
    fdb::set_placed(&app.db, job.id, &why.join(", ")).await?;
    db::runners::set_job_runner(&app.db, job.id, me.id).await?;
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    report_check(&app, &row, &run.state, run.number, &job.name, "running", &format!("running on {runner}")).await;
    let bad = |e: serde_json::Error| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}"));
    // The job's limit: its own timeout, else its sandbox's, else the site
    // default, never past the site maximum; recorded for the reaper.
    let limits = limits(&app).await;
    let spec: flow_core::pipeline::Job = serde_json::from_str(&job.spec).map_err(bad)?;
    let sandbox: flow_core::pipeline::Sandbox = serde_json::from_str(&job.sandbox).map_err(bad)?;
    let mut timeout_s = spec.timeout_s.or(sandbox.timeout_s).unwrap_or(limits.default_s).min(limits.max_s).max(60);
    // ...and never past the runner's own limit, when it has one (#98).
    if let Some(m) = me.timeout_minutes.filter(|m| *m > 0) {
        timeout_s = timeout_s.min(m as u64 * 60).max(60);
    }
    fdb::set_job_timeout(&app.db, job.id, timeout_s).await?;
    let claim = Claim {
        job_id: job.id,
        run_id: run.id,
        repo: format!("{}/{}", row.owner, row.name),
        clone_url: format!("{}/{}/{}", app.config.base_url(), row.owner, row.name),
        token: secret,
        heads: run.head_list().into_iter().map(String::from).collect(),
        pipeline_hash: run.pipeline_hash.clone(),
        job: spec,
        sandbox,
        timeout_s,
        subrepos: vec![],
        state: run.state.clone(),
        bookmark: run.bookmark.clone(),
        run_number: run.number,
        // The site's, the owner's and the repository's variables (#102).
        env: crate::variables::env_for_repo(&app.db, &row).await?,
        secrets: vec![],
        inputs: vec![],
    };
    let mut claim = claim;
    let (inputs, ids, mut left_out) = subrepo_inputs(&app, &row, &run).await;
    // The secrets the job declares, as files (#107); one not set is said
    // at the top of the log and left out, so the job fails on the file.
    let names: Vec<String> = claim.job.secrets.iter().map(|s| s.name.clone()).collect();
    let found = crate::secrets::for_job(&app, &row, &names).await?;
    // The identities it asks for (#103): a token each, as a file.
    // An environment goes in its tokens only once the job was approved
    // (oxygenflow#59): the approval is the environment's protection.
    let environment = claim.job.environment.clone().filter(|_| job.approval == "approved");
    if let (Some(e), None, false) = (&claim.job.environment, &environment, claim.job.identities.is_empty()) {
        left_out.push(format!(
            "environment {e}: not in the job's identity tokens, because the job was not approved (give it an approval block)"
        ));
    }
    if !claim.job.identities.is_empty() {
        let issuer = app.oidc().await.map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("oidc: {e}")))?;
        for i in &claim.job.identities {
            let token = issuer.mint(&crate::oidc::JobClaims {
                issuer: app.config.base_url(),
                audience: &i.audience,
                owner: &row.owner,
                name: &row.name,
                visibility: &row.visibility,
                bookmark: &run.bookmark,
                run_number: run.number,
                job: &job.name,
                state: &run.state,
                runner: &runner,
                actor: &run.actor,
                environment: environment.as_deref(),
                approved_by: (job.approval == "approved").then_some(job.approved_by.as_str()),
                approval: (!job.approval_seal.is_empty()).then_some(job.approval_seal.as_str()),
                now_ms: db::now_ms(),
                ttl_s: timeout_s,
            });
            claim.secrets.push(flow_core::api::SecretInput {
                name: format!("identity:{}", i.audience),
                path: i.mount_path.clone(),
                value: token,
            });
        }
    }
    for s in &claim.job.secrets {
        match found.iter().find(|(n, _)| *n == s.name) {
            Some((name, value)) => {
                claim.secrets.push(flow_core::api::SecretInput { name: name.clone(), path: s.mount_path.clone(), value: value.clone() })
            }
            None => left_out.push(format!(
                "secret {:?} is not set for this repository, its owner or the site; {} will be missing",
                s.name, s.mount_path
            )),
        }
    }
    if !ids.is_empty() {
        fdb::set_also_read(&app.db, job.id, &ids).await?;
    }
    // Say at the top of the job's log what it will not have (oxygenhub#56),
    // rather than let it fail later on missing files.
    if !left_out.is_empty() {
        let records: Vec<flow_core::api::Record> =
            left_out.into_iter().map(|text| flow_core::api::Record { t: 0, stream: "flow".into(), text }).collect();
        fdb::append_log(&app.db, job.id, &records, MAX_RECORDS as i64).await?;
    }
    claim.subrepos = inputs;
    // Artifacts of jobs it depends on (oxygenflow#58), with the blob each
    // producer's receipt names; the runner checks what it fetches against it.
    match artifact_inputs(&app, &job, &claim.job).await {
        Ok(found) => claim.inputs = found,
        Err(why) => {
            let error = format!("inputs: {why}");
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
            return Err(ApiError(StatusCode::CONFLICT, error));
        }
    }
    Ok(Json(claim).into_response())
}

/// The artifacts a job takes (oxygenflow#58), each from a job of the same
/// run that passed with a receipt; a passing job's receipt names exactly the
/// artifacts it uploaded (checked at finish), so their blobs are attested.
async fn artifact_inputs(app: &Shared, job: &fdb::Job, spec: &flow_core::Job) -> Result<Vec<flow_core::api::ArtifactInput>, String> {
    if spec.inputs.is_empty() {
        return Ok(vec![]);
    }
    let jobs = fdb::jobs(&app.db, job.run_id).await.map_err(|e| e.to_string())?;
    let mut out = Vec::new();
    for i in &spec.inputs {
        let Some(producer) = jobs.iter().find(|j| j.name == i.from) else {
            return Err(format!("{:?} is not a job of this run", i.from));
        };
        if producer.status != "passed" || producer.receipt_hash.is_empty() {
            return Err(format!("{} has not passed with a receipt, so {} cannot be taken from it", i.from, i.artifact));
        }
        let artifacts = fdb::artifacts(&app.db, producer.id).await.map_err(|e| e.to_string())?;
        let Some(a) = artifacts.iter().find(|a| a.name == i.artifact) else {
            return Err(format!("{} kept no artifact {}", i.from, i.artifact));
        };
        out.push(flow_core::api::ArtifactInput {
            artifact: i.artifact.clone(),
            from: i.from.clone(),
            path: i.path.clone(),
            blob: a.blob.clone(),
        });
    }
    Ok(out)
}

/// `GET /api/v1/flow/jobs/{id}/inputs/{index}`: the bytes of input `index`
/// of a running job (oxygenflow#58), for the runner holding it.
pub async fn input(State(app): State<Shared>, headers: HeaderMap, Path((id, index)): Path<(i64, usize)>) -> Result<Response, ApiError> {
    let runner = runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    // A registered runner reads only the inputs of a job it holds (a
    // configured token's runner names itself when it claims).
    if runner.source == "registered" && job.runner != runner.name {
        return Err(ApiError(StatusCode::FORBIDDEN, "this job is held by another runner".into()));
    }
    let spec: flow_core::Job =
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
    let found = artifact_inputs(&app, &job, &spec).await.map_err(|why| ApiError(StatusCode::CONFLICT, why))?;
    let Some(i) = found.get(index) else {
        return Err(ApiError(StatusCode::NOT_FOUND, format!("job {} has no input {index}", job.name)));
    };
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
    let blob = oxy_core::hash::ObjectRef::from_hex(&i.blob)
        .ok_or_else(|| ApiError(StatusCode::INTERNAL_SERVER_ERROR, "a kept artifact has a bad blob".into()))?;
    let bytes = app
        .repos
        .read(&repo, move |r, _| Ok(r.read_blob(&blob)?))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the artifact: {e}")))?;
    Ok(([(axum::http::header::CONTENT_TYPE, "application/octet-stream")], bytes).into_response())
}

/// BLAKE3 over what a claiming runner offers: tiers, images, architecture,
/// labels and capacity, each list sorted.
fn profile_of(req: &ClaimRequest, arch: &str) -> String {
    let sorted = |v: &[String]| {
        let mut v = v.to_vec();
        v.sort();
        v
    };
    let info = req.info.as_ref();
    let v = serde_json::json!({
        "arch": arch,
        "tiers": sorted(&req.runtimes),
        "images": req.images.as_deref().map(sorted),
        "labels": info.map(|i| sorted(&i.labels)),
        "memory_mib": info.and_then(|i| i.memory_mib),
        "cpus": info.and_then(|i| i.cpus),
    });
    blake3::hash(v.to_string().as_bytes()).to_hex().to_string()
}

/// A job handed back by this many different runners is unplaceable.
pub const UNPLACEABLE_AFTER: usize = 3;

/// `POST /api/v1/flow/jobs/{id}/decline` (OxygenFlow spec §11.5): the
/// runner will not run a job it was handed. The job is queued again and
/// not offered to that runner while its profile stays the same (and for
/// [`fdb::DECLINE_HOLDS_MS`] at most); once [`UNPLACEABLE_AFTER`]
/// different runners have declined it, it ends as unplaceable with their
/// reasons. A decline is never a failed check.
/// `POST /api/v1/flow/runners/heartbeat` (oxygenhub#243): a runner whose
/// job slots are all full claims nothing, so it says it is there.
pub async fn heartbeat(State(app): State<Shared>, headers: HeaderMap) -> Result<StatusCode, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    db::runners::seen(&app.db, me.id, None, None).await?;
    Ok(StatusCode::NO_CONTENT)
}

pub async fn decline(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(d): Json<flow_core::api::Decline>,
) -> Result<StatusCode, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    db::runners::seen(&app.db, me.id, None, None).await?;
    let reason: String = d.reason.chars().map(|c| if c.is_control() { ' ' } else { c }).take(500).collect();
    let reason = match reason.trim() {
        "" => "no reason given".to_string(),
        r => r.to_string(),
    };
    let Some(all) = fdb::decline(&app.db, id, me.id, &me.name, &me.profile, &reason).await? else {
        return Err(ApiError(StatusCode::CONFLICT, "that job is not running on this runner".into()));
    };
    tracing::info!("flow: runner {} declined job {id}: {reason}", me.name);
    let job = fdb::job(&app.db, id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such job".into()))?;
    let runners: std::collections::BTreeSet<i64> = all.iter().map(|d| d.runner_id).collect();
    if runners.len() >= UNPLACEABLE_AFTER {
        let error = format!(
            "unplaceable: declined by {} runners: {}",
            runners.len(),
            all.iter().map(|d| format!("{}: {}", d.runner, d.reason)).collect::<Vec<_>>().join("; ")
        );
        let changed =
            fdb::finish(&app.db, id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
        report_changed(&app, job.run_id, changed).await?;
    } else if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
            let summary = format!("declined by {}: {reason}; waiting for another runner", me.name);
            report_check(&app, &row, &run.state, run.number, &job.name, "queued", &summary).await;
            fdb::set_waiting(&app.db, id, &summary).await?;
        }
    }
    Ok(StatusCode::NO_CONTENT)
}

/// Why no runner can take `job` of a repository of `owner` (id `repo_id`)
/// now (spec §11.4), from what each runner offered at its last claim:
/// empty when one could (it waits its turn or a free slot), else what
/// each one lacks. `declines` are the job's.
/// What placement knows of the fleet: images' architectures, the images
/// the hub holds, and the time.
pub struct Fleet<'a> {
    pub image_archs: &'a std::collections::HashMap<String, String>,
    pub hub_images: &'a std::collections::HashSet<String>,
    pub now: i64,
}

/// Why runner `r` cannot take `job` of a repository of `owner` (id
/// `repo_id`), from what it offered at its last claim (spec §11.4); `None`
/// when it can. `declines` are the job's.
pub fn runner_lacks(
    job: &fdb::Job,
    owner: &str,
    repo_id: i64,
    r: &db::runners::Runner,
    fleet: &Fleet,
    declines: &[fdb::DeclineRow],
) -> Option<String> {
    let sandbox: serde_json::Value = serde_json::from_str(&job.sandbox).unwrap_or_default();
    let rootfs = sandbox["rootfs"].as_str().unwrap_or("");
    let wanted_arch = job.arch().or_else(|| fleet.image_archs.get(rootfs).cloned());
    let need = |k: &str| sandbox[k].as_u64().unwrap_or(0);
    if r.state == "draining" {
        return Some("is draining".into());
    }
    let scope = r.scope();
    if (!scope.owner.is_empty() && scope.owner != owner) || scope.repo_id.is_some_and(|id| id != repo_id) {
        return Some("serves other repositories".into());
    }
    let tiers = r.tier_list();
    if !tiers.is_empty() && !tiers.contains(&job.runtime.as_str()) {
        return Some(format!("has no {} tier (it offers {})", job.runtime, tiers.join(", ")));
    }
    if let Some(w) = wanted_arch.as_deref().filter(|w| !r.arch.is_empty() && *w != r.arch) {
        return Some(format!("is {}; the job needs {w}", r.arch));
    }
    let fetches = r.fetches_images && fleet.hub_images.contains(rootfs);
    if !fetches && r.image_list().is_some_and(|have| !have.iter().any(|h| h == rootfs)) {
        let why = if r.fetches_images { " (nor does the hub: send it with flow rootfs push)" } else { "" };
        return Some(format!("does not hold image {}{why}", &rootfs[..12.min(rootfs.len())]));
    }
    let labels = r.label_list();
    if let Some(l) = job.needs().iter().find(|l| !labels.contains(&l.as_str())) {
        return Some(format!("lacks label {l}"));
    }
    if let Some(m) = r.memory_mib.filter(|m| (*m as u64) < need("memory_mib")) {
        return Some(format!("has {m} MiB; the job needs {} MiB", need("memory_mib")));
    }
    if let Some(c) = r.cpus.filter(|c| (*c as u64) < need("cpus")) {
        return Some(format!("has {c} CPUs; the job needs {}", need("cpus")));
    }
    // The disk it learned it needs (oxygenflow#41).
    if let (Some(n), Some(free)) = (job.need_disk_mib, r.free_disk_mib) {
        if free < n {
            return Some(format!(
                "has {} free; the job needs ~{} (its recent runs took up to {})",
                mib(free),
                mib(n),
                mib((n - 64) * 4 / 5)
            ));
        }
    }
    if let Some(d) = declines.iter().find(|d| d.runner_id == r.id && fleet.now - d.ms < fdb::DECLINE_HOLDS_MS) {
        return Some(format!("declined it: {}", d.reason));
    }
    None
}

/// Why no runner can take `job` of a repository of `owner` (id `repo_id`)
/// now (spec §11.4), from what each runner offered at its last claim:
/// empty when one could (it waits its turn or a free slot), else what
/// each one lacks. `declines` are the job's.
#[allow(clippy::too_many_arguments)]
pub fn why_waiting(
    job: &fdb::Job,
    owner: &str,
    repo_id: i64,
    runners: &[db::runners::Runner],
    image_archs: &std::collections::HashMap<String, String>,
    declines: &[fdb::DeclineRow],
    now: i64,
    hub_images: &std::collections::HashSet<String>,
) -> String {
    let fleet = Fleet { image_archs, hub_images, now };
    let online: Vec<&db::runners::Runner> = runners.iter().filter(|r| r.state != "disabled" && r.online()).collect();
    if online.is_empty() {
        return "no runner is online".into();
    }
    let mut why = Vec::new();
    for r in &online {
        match runner_lacks(job, owner, repo_id, r, &fleet, declines) {
            None => return String::new(),
            Some(w) => why.push(format!("{} {w}", r.name)),
        }
    }
    let more = why.len().saturating_sub(4);
    why.truncate(4);
    let mut text = format!("no runner can take it: {}", why.join("; "));
    if more > 0 {
        text.push_str(&format!("; and {more} more"));
    }
    text
}

/// How long a job is held for a better runner before any runner that fits
/// takes it (oxygenflow#42).
const BETTER_HOLD_MS: i64 = 15_000;
/// A runner counts as there to take a job when it asked this recently.
const POLLING_MS: i64 = 10_000;

/// When each job was first passed over for a better runner.
/// Keyed by hub (tests run several in one process) and job.
static HELD: std::sync::Mutex<Option<std::collections::HashMap<(String, i64), i64>>> = std::sync::Mutex::new(None);

/// The build cache slots each runner said at its last claim it holds
/// (oxygenflow#51). Keyed by hub (tests run several in one process) and
/// runner; a restart forgets them until each runner's next claim.
/// Each slot maps to when it last kept a disk (Unix ms, 0 unknown).
type CacheSlots = std::collections::HashMap<(String, i64), std::collections::HashMap<String, i64>>;
static CACHES: std::sync::Mutex<Option<CacheSlots>> = std::sync::Mutex::new(None);

/// When runner `runner` last kept a disk in build cache slot `slot` (0 when
/// it did not say); `None` when it holds none there.
fn cache_kept(hub: &str, runner: i64, slot: &str) -> Option<i64> {
    CACHES
        .lock()
        .unwrap_or_else(|p| p.into_inner())
        .as_ref()
        .and_then(|c| c.get(&(hub.to_string(), runner)))
        .and_then(|held| held.get(slot).copied())
}

/// A copy of a build cache kept at least this much later is the fresher
/// (oxygenflow#54): closer than that, the two are as good as each other.
const FRESHER_MS: i64 = 60_000;

/// How well a runner fits a job (oxygenflow#42, #51, #54).
struct Fit {
    /// When it last kept the job's build cache, if it holds it.
    cache: Option<i64>,
    image: bool,
    speed: i64,
}

impl Fit {
    /// Whether this fit is better than `other`: holding the job's build
    /// cache counts most (a warm build beats any image fetch), and of two
    /// that hold it the fresher copy; then holding the image, then speed.
    fn beats(&self, other: &Fit) -> bool {
        match (self.cache, other.cache) {
            (Some(_), None) => return true,
            (None, Some(_)) => return false,
            (Some(a), Some(b)) if a > 0 && b > 0 && (a - b).abs() >= FRESHER_MS => return a > b,
            _ => {}
        }
        (self.image, self.speed) > (other.image, other.speed)
    }
}

/// How runner `r` fits a job on image `rootfs` whose build cache slot is
/// `slot`.
fn fit(hub: &str, r: &db::runners::Runner, rootfs: &str, slot: &str) -> Fit {
    Fit {
        cache: cache_kept(hub, r.id, slot),
        image: r.image_list().is_some_and(|have| have.iter().any(|h| h == rootfs)),
        speed: r.speed_class.unwrap_or(0),
    }
}

/// A span as people say it: 40 s, 12 min, 3 h, 2 days.
fn ago(ms: i64) -> String {
    let s = ms.max(0) / 1000;
    match s {
        0..=59 => format!("{s} s"),
        60..=3599 => format!("{} min", s / 60),
        3600..=172_799 => format!("{} h", s / 3600),
        _ => format!("{} days", s / 86_400),
    }
}

fn rootfs_of(job: &fdb::Job) -> String {
    serde_json::from_str::<serde_json::Value>(&job.sandbox).ok().and_then(|v| v["rootfs"].as_str().map(String::from)).unwrap_or_default()
}

/// A size in MiB as people say it: 9.4 GiB, 740 MiB.
fn mib(m: i64) -> String {
    if m >= 1024 {
        format!("{:.1} GiB", m as f64 / 1024.0)
    } else {
        format!("{m} MiB")
    }
}

/// One placement pass: say why each ready queued job waits.
pub async fn explain_waits(app: &Shared) -> sqlx::Result<()> {
    let ready = fdb::ready_queued(&app.db).await?;
    if ready.is_empty() {
        return Ok(());
    }
    let runners = db::runners::list(&app.db).await?;
    let archs = fdb::image_archs(&app.db).await?;
    let held = crate::images::held(&app.db).await?;
    let now = db::now_ms();
    for (job, owner, repo_id) in ready {
        let declines = fdb::declines_of_job(&app.db, job.id).await?;
        let why = why_waiting(&job, &owner, repo_id, &runners, &archs, &declines, now, &held);
        if why == job.waiting {
            continue;
        }
        fdb::set_waiting(&app.db, job.id, &why).await?;
        // The job's check says it too, so a proposal shows why it is stuck
        // (oxygenhub#28).
        if let (Some(run), Some(row)) = (fdb::run_by_id(&app.db, job.run_id).await?, db::repos::get_by_id(&app.db, repo_id).await?) {
            let summary = if why.is_empty() { "waiting for a runner" } else { why.as_str() };
            report_check(app, &row, &run.state, run.number, &job.name, "queued", summary).await;
        }
    }
    Ok(())
}

/// Run a placement pass every 15 seconds.
pub async fn placement(app: Shared) {
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
        if let Err(e) = explain_waits(&app).await {
            tracing::warn!("flow: placement pass: {e}");
        }
    }
}

/// The Oxygen subrepo pins in `name`'s state `heads`: (path, repository id, pinned heads).
pub(crate) async fn pins_in(
    app: &Shared,
    name: crate::repos::RepoName,
    heads: Vec<oxy_core::hash::ChangeHash>,
) -> Result<Vec<(String, [u8; 16], Vec<oxy_core::hash::ChangeHash>)>, String> {
    app.repos
        .read(&name, move |repo, cache| {
            let loaded = cache.load(repo, &heads)?;
            let mut out = Vec::new();
            for (path, f) in &loaded.snapshot.files {
                if f.kind != oxy_core::objects::NodeKind::Subrepo {
                    continue;
                }
                if let oxy_core::state::FileContent::Opaque { blobs } = &f.content {
                    if let Some(b) = blobs.first() {
                        if let Ok(oxy_core::subrepo::Pin::Oxy { repo: id, heads, .. }) = repo
                            .read_blob(b)
                            .map_err(crate::repos::RepoError::Oxy)
                            .and_then(|bytes| oxy_core::subrepo::Pin::decode(&bytes).map_err(|e| crate::repos::RepoError::Oxy(e.into())))
                        {
                            out.push((path.clone(), id, heads));
                        }
                    }
                }
            }
            Ok(out)
        })
        .await
        .map_err(|e| e.to_string())
}

/// How deep subrepos of subrepos are followed.
const MAX_SUBREPO_DEPTH: usize = 3;

/// The Oxygen subrepos the run's state pins, and those their states pin in
/// turn (parents first, paths joined), that live on this hub and that the
/// run's pusher may read: what the job's workspace needs beside its own
/// tree. A pin the hub cannot serve is left out, with what it pins, and
/// the third list says why, for the job's log.
async fn subrepo_inputs(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
) -> (Vec<flow_core::api::SubrepoInput>, Vec<i64>, Vec<String>) {
    let Some(name) = crate::repos::RepoName::new(&row.owner, &row.name) else { return (vec![], vec![], vec![]) };
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
    let rows = db::repos::list(&app.db).await.unwrap_or_default();
    let mut inputs = Vec::new();
    let mut ids = Vec::new();
    let mut left_out = Vec::new();
    let actor = if run.actor.is_empty() { "the run's starter".to_string() } else { run.actor.clone() };
    // (path prefix, repository, heads, depth), parents before children.
    let mut queue = std::collections::VecDeque::from([(String::new(), name, heads, 0usize)]);
    while let Some((prefix, name, heads, depth)) = queue.pop_front() {
        if depth == MAX_SUBREPO_DEPTH {
            continue;
        }
        for (path, id, pinned) in pins_in(app, name, heads).await.unwrap_or_default() {
            let path = if prefix.is_empty() { path } else { format!("{prefix}/{path}") };
            let mut hosted = false;
            for other in &rows {
                let Some(other_name) = crate::repos::RepoName::new(&other.owner, &other.name) else { continue };
                let matches = app.repos.read(&other_name, move |r, _| Ok(r.id().ok() == Some(id))).await.unwrap_or(false);
                if !matches {
                    continue;
                }
                hosted = true;
                let role = crate::auth::roles::role_for(&app.db, pusher.as_ref(), other).await.ok().flatten();
                // A Diffusion bump's run reads the upstream it bumps: the one
                // the consumer pins (someone who may write it committed the
                // pin), and no other (#150).
                let diffused = run.actor == crate::diffusion::ACCOUNT
                    && run.bookmark.starts_with("diffusion/")
                    && sqlx::query_scalar::<_, i64>("SELECT COUNT(*) FROM diffusion_pins WHERE consumer_id = $1 AND upstream_repo_id = $2")
                        .bind(run.repo_id)
                        .bind(other.id)
                        .fetch_one(&app.db)
                        .await
                        .unwrap_or(0)
                        > 0;
                if diffused || crate::auth::roles::allows(role, crate::auth::Access::Read) {
                    inputs.push(flow_core::api::SubrepoInput {
                        path: path.clone(),
                        repo: format!("{}/{}", other.owner, other.name),
                        clone_url: format!("{}/{}/{}", app.config.base_url(), other.owner, other.name),
                        heads: pinned.iter().map(|h| h.to_hex()).collect(),
                    });
                    if !ids.contains(&other.id) {
                        ids.push(other.id);
                    }
                    queue.push_back((path.clone(), other_name, pinned.clone(), depth + 1));
                } else {
                    left_out.push(format!(
                        "subrepo {path} ({}/{}) is left out: {actor} cannot read it, so the job has no files there",
                        other.owner, other.name
                    ));
                }
                break;
            }
            if !hosted {
                left_out.push(format!("subrepo {path} is left out: the repository it pins is not on this hub"));
            }
        }
    }
    (inputs, ids, left_out)
}

async fn running_job(app: &Shared, id: i64) -> Result<fdb::Job, ApiError> {
    match fdb::job(&app.db, id).await? {
        Some(j) if j.status == "running" => Ok(j),
        Some(_) => Err(ApiError(StatusCode::CONFLICT, "this job is not running".into())),
        None => Err(ApiError(StatusCode::NOT_FOUND, "no such job".into())),
    }
}

/// `POST /api/v1/flow/jobs/{id}/log`.
pub async fn log(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(mut batch): Json<LogBatch>,
) -> Result<StatusCode, ApiError> {
    runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    // An empty batch asks only whether the job still runs: a cancelled
    // one answers 409 and its runner stops it (oxygenflow#32).
    if batch.records.is_empty() {
        return Ok(StatusCode::NO_CONTENT);
    }
    // A secret's value never reaches the stored log (#107).
    let values: Vec<String> = match serde_json::from_str::<flow_core::Job>(&job.spec) {
        Ok(spec) if !spec.secrets.is_empty() => {
            let run = fdb::run_by_id(&app.db, job.run_id).await?;
            let row = match run {
                Some(run) => db::repos::get_by_id(&app.db, run.repo_id).await?,
                None => None,
            };
            match row {
                Some(row) => {
                    let names: Vec<String> = spec.secrets.iter().map(|s| s.name.clone()).collect();
                    crate::secrets::for_job(&app, &row, &names).await?.into_iter().map(|(_, v)| v).collect()
                }
                None => vec![],
            }
        }
        _ => vec![],
    };
    let kid = app.oidc.get().map(|i| i.kid.clone());
    for r in &mut batch.records {
        if !values.is_empty() {
            r.text = crate::secrets::mask(&r.text, &values);
        }
        if let Some(kid) = &kid {
            r.text = crate::oidc::mask(&r.text, kid);
        }
        if r.text.len() > MAX_RECORD_TEXT {
            let mut cut = MAX_RECORD_TEXT;
            while !r.text.is_char_boundary(cut) {
                cut -= 1;
            }
            r.text.truncate(cut);
        }
        if !matches!(r.stream.as_str(), "out" | "err" | "console" | "flow") {
            r.stream = "console".into();
        }
    }
    let kept = fdb::append_log(&app.db, job.id, &batch.records, MAX_RECORDS as i64).await?;
    if kept < batch.records.len() && job.log_count < MAX_RECORDS as i64 {
        tracing::info!("flow: job {} reached the log limit", job.id);
    }
    Ok(StatusCode::NO_CONTENT)
}

/// Largest artifact the hub keeps; the runner refuses larger ones first.
pub const MAX_ARTIFACT_BYTES: usize = 512 << 20;

/// `POST /api/v1/flow/jobs/{id}/artifacts/{name}`: the bytes of an artifact
/// the job declares (OxygenFlow spec §5.2), kept as a blob of the run's
/// repository. Answers `{"blob"}`.
pub async fn artifact(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path((id, name)): Path<(i64, String)>,
    body: axum::body::Bytes,
) -> Result<Response, ApiError> {
    runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    let spec: flow_core::Job =
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
    let Some(out) = spec.outputs.iter().find(|o| o.name == name) else {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no artifact {name:?}", job.name)));
    };
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
    // A report's artifact is its manifest (#127): checked, and its files
    // found in the store, before it is kept.
    let report = match out.kind {
        flow_core::OutputKind::Report => Some(report_manifest(&app, &repo, &name, &body).await?),
        flow_core::OutputKind::Artifact => None,
    };
    let content = blake3::hash(&body).to_hex().to_string();
    let size = body.len() as i64;
    let artifact = report.is_none();
    let report_bytes = (artifact && name == crate::coverage::ARTIFACT).then(|| body.clone());
    let junit_bytes = (artifact && name == crate::junit::ARTIFACT).then(|| body.clone());
    let blob = app
        .repos
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the artifact: {e}")))?;
    let a = fdb::Artifact { job_id: job.id, name, path: out.path.clone(), blob: blob.to_hex(), size, content, created_ms: db::now_ms() };
    fdb::add_artifact(&app.db, &a).await?;
    if let Some((manifest, declaration)) = report {
        let origin = crate::reports::Origin {
            repo_id: row.id,
            run_id: run.id,
            run_number: run.number,
            job_id: job.id,
            job: &job.name,
            bookmark: &run.bookmark,
            state: &run.state,
        };
        let said = match declaration {
            Ok(d) => match crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await {
                Ok(id) => {
                    report_threshold_checks(&app, &row, &run, &a.name, id, &d.metrics).await;
                    let s = if d.metrics.len() == 1 { "" } else { "s" };
                    format!("report {}: {} \"{}\", {} metric{s}", a.name, d.kind, d.title, d.metrics.len())
                }
                Err(e) => format!("report {}: not indexed: {e}", a.name),
            },
            Err(why) => {
                let d = crate::reports::Declaration::unreadable(&a.name, &why);
                crate::reports::keep(&app, &origin, &a.name, &d, &manifest, &a.blob).await.ok();
                format!("report {}: report.toml: {why}", a.name)
            }
        };
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
    }
    // A coverage report (#113) is read now; what it says, or why it could
    // not be read, goes in the job's log. It never fails the upload.
    if let Some(bytes) = report_bytes {
        let said = match read_coverage(&app, &repo, &row, &run, &job, &a.blob, bytes).await {
            Ok(text) | Err(text) => text,
        };
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
    }
    // A test report (#114) likewise.
    if let Some(bytes) = junit_bytes {
        let said = match read_junit(&app, &row, &run, &job, &a.blob, bytes).await {
            Ok(text) | Err(text) => text,
        };
        let record = flow_core::api::Record { t: 0, stream: "flow".into(), text: said };
        fdb::append_log(&app.db, job.id, &[record], MAX_RECORDS as i64).await.ok();
    }
    Ok(Json(serde_json::json!({ "blob": a.blob })).into_response())
}

/// A check per metric with a threshold (#127, docs/reports.md §5), named
/// `report/NAME/METRIC` on the run's state: failed when the value crosses
/// `fail_above` or `fail_below`, which a proposal may require.
async fn report_threshold_checks(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    name: &str,
    id: i64,
    metrics: &[crate::reports::Metric],
) {
    let url = format!("{}/{}/{}/reports/{name}?id={id}", app.config.base_url(), row.owner, row.name);
    for m in metrics.iter().filter(|m| m.fail_above.is_some() || m.fail_below.is_some()) {
        let unit = if m.unit.is_empty() { String::new() } else { format!(" {}", m.unit) };
        let above = m.fail_above.filter(|a| m.value > *a);
        let below = m.fail_below.filter(|b| m.value < *b);
        let (status, summary) = match (above, below) {
            (Some(a), _) => ("failed", format!("{}{unit}, above {a}", m.value)),
            (_, Some(b)) => ("failed", format!("{}{unit}, below {b}", m.value)),
            _ => {
                let limits: Vec<String> = [m.fail_above.map(|a| format!("at most {a}")), m.fail_below.map(|b| format!("at least {b}"))]
                    .into_iter()
                    .flatten()
                    .collect();
                ("passed", format!("{}{unit} ({})", m.value, limits.join(", ")))
            }
        };
        let check = format!("report/{name}/{}", m.key);
        if let Ok(c) = db::hooks::upsert_check(
            &app.db,
            row.id,
            db::hooks::CheckReport { state: &run.state, name: &check, status, summary: &summary, url: &url, reporter: None, seal: "" },
        )
        .await
        {
            let api = crate::api::check_of(&c);
            app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
            crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
        }
    }
}

/// Check report manifest `body` (#127) and that the repository holds every
/// file it names; answers it with the report's declaration, or why that
/// could not be read.
async fn report_manifest(
    app: &Shared,
    repo: &RepoName,
    name: &str,
    body: &[u8],
) -> Result<(crate::reports::Manifest, Result<crate::reports::Declaration, String>), ApiError> {
    let manifest = crate::reports::parse_manifest(body).map_err(|e| ApiError(StatusCode::BAD_REQUEST, e))?;
    let refs: Vec<(String, oxy_core::hash::ObjectRef)> =
        manifest.files.iter().filter_map(|f| Some((f.path.clone(), oxy_core::hash::ObjectRef::from_hex(&f.blob)?))).collect();
    let declaration = manifest.blob("report.toml").and_then(oxy_core::hash::ObjectRef::from_hex);
    let found = app
        .repos
        .read(repo, move |r, _| {
            if let Some((path, _)) = refs.iter().find(|(_, b)| !r.store().has(b)) {
                return Ok(Err(path.clone()));
            }
            Ok(Ok(declaration.map(|b| r.read_blob(&b)).transpose()?))
        })
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("read the report: {e}")))?;
    let toml = found.map_err(|path| ApiError(StatusCode::BAD_REQUEST, format!("report file {path} was not uploaded; upload it first")))?;
    let declaration = crate::reports::parse_declaration(name, &toml.unwrap_or_default());
    Ok((manifest, declaration))
}

/// `POST /api/v1/flow/jobs/{id}/reports/{name}/files`: one file of a report
/// the job declares (#127), its path in `X-Flow-Report-Path`, kept as a
/// blob of the run's repository. Answers `{"blob"}`; the manifest naming
/// the files follows as the report's artifact.
pub async fn report_file(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path((id, name)): Path<(i64, String)>,
    body: axum::body::Bytes,
) -> Result<Response, ApiError> {
    runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    let spec: flow_core::Job =
        serde_json::from_str(&job.spec).map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("stored job: {e}")))?;
    if !spec.outputs.iter().any(|o| o.name == name && o.kind == flow_core::OutputKind::Report) {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("job {} declares no report {name:?}", job.name)));
    }
    let path = headers.get("x-flow-report-path").and_then(|v| v.to_str().ok()).unwrap_or_default();
    if !crate::reports::path_ok(path) {
        return Err(ApiError(StatusCode::BAD_REQUEST, format!("report file {path:?}: give its path in X-Flow-Report-Path")));
    }
    let run = fdb::run_by_id(&app.db, job.run_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such run".into()))?;
    let row =
        db::repos::get_by_id(&app.db, run.repo_id).await?.ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "no such repository".into()))?;
    let repo = RepoName::new(&row.owner, &row.name).ok_or_else(|| ApiError(StatusCode::NOT_FOUND, "bad repository name".into()))?;
    let blob = app
        .repos
        .write(&repo, move |r, _| Ok(r.store_blob(&body)?))
        .await
        .map_err(|e| ApiError(StatusCode::INTERNAL_SERVER_ERROR, format!("keep the report file: {e}")))?;
    Ok(Json(serde_json::json!({ "blob": blob.to_hex() })).into_response())
}

/// Parse and keep a job's JUnit report: what it says, or why not.
async fn read_junit(
    app: &Shared,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    job: &fdb::Job,
    blob: &str,
    bytes: axum::body::Bytes,
) -> Result<String, String> {
    if bytes.len() > crate::junit::MAX_REPORT {
        return Err(format!("tests: the report is {} MiB; at most {} MiB is read", bytes.len() >> 20, crate::junit::MAX_REPORT >> 20));
    }
    let text = String::from_utf8(bytes.to_vec()).map_err(|_| "tests: the report is not UTF-8 text (JUnit XML is expected)".to_string())?;
    let cases = tokio::task::spawn_blocking(move || crate::junit::parse(&text))
        .await
        .map_err(|e| format!("tests: {e}"))?
        .map_err(|e| format!("tests: {e}"))?;
    let origin = crate::junit::Origin {
        repo_id: row.id,
        run_id: run.id,
        run_number: run.number,
        job_id: job.id,
        job: &job.name,
        bookmark: &run.bookmark,
        state: &run.state,
        blob,
    };
    crate::junit::keep(app, &origin, &cases).await.map_err(|e| format!("tests: {e}"))?;
    let r = crate::junit::for_run(&app.db, run.id)
        .await
        .map_err(|e| format!("tests: {e}"))?
        .into_iter()
        .find(|r| r.job_id == job.id)
        .ok_or_else(|| "tests: the report was not kept".to_string())?;
    // The `tests` check on the run's state: what fails, and which of those
    // passed in the same job's latest report on the default bookmark, so a
    // proposal names what it broke.
    let base = crate::junit::baseline(&app.db, &r, &row.default_bookmark).await.map_err(|e| format!("tests: {e}"))?;
    let newly = match &base {
        Some(b) if r.broken() > 0 => crate::junit::newly_failing(&app.db, r.id, b.id).await.map_err(|e| format!("tests: {e}"))?,
        _ => vec![],
    };
    let (status, summary) = if r.broken() == 0 {
        ("passed", format!("{} passed", r.passed))
    } else if let Some(b) = &base {
        let names: Vec<&str> = newly.iter().take(3).map(String::as_str).collect();
        let more = if newly.len() > 3 { format!(", +{}", newly.len() - 3) } else { String::new() };
        let new = if newly.is_empty() {
            format!("none new against {} run #{}", row.default_bookmark, b.run_number)
        } else {
            format!("{} new against {}: {}{more}", newly.len(), row.default_bookmark, names.join(", "))
        };
        ("failed", format!("{} failing; {new}", r.broken()))
    } else {
        ("failed", format!("{} failing", r.broken()))
    };
    let url = format!("{}/{}/{}/tests?report={}", app.config.base_url(), row.owner, row.name, r.id);
    if let Ok(c) = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state: &run.state, name: "tests", status, summary: &summary, url: &url, reporter: None, seal: "" },
    )
    .await
    {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
    Ok(format!("tests: {} in {}", r.summary(), human_ms(r.time_ms)))
}

/// Parse and keep a job's coverage report: what it says, or why not.
async fn read_coverage(
    app: &Shared,
    repo: &RepoName,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    job: &fdb::Job,
    blob: &str,
    bytes: axum::body::Bytes,
) -> Result<String, String> {
    if bytes.len() > crate::coverage::MAX_REPORT {
        return Err(format!(
            "coverage: the report is {} MiB; at most {} MiB is read",
            bytes.len() >> 20,
            crate::coverage::MAX_REPORT >> 20
        ));
    }
    let text = String::from_utf8(bytes.to_vec())
        .map_err(|_| "coverage: the report is not text (LCOV or Cobertura XML is expected)".to_string())?;
    let (files, outside, format) = tokio::task::spawn_blocking(move || crate::coverage::parse(&text))
        .await
        .map_err(|e| format!("coverage: {e}"))?
        .map_err(|e| format!("coverage: {e}"))?;
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let in_state: std::collections::HashSet<String> = app
        .repos
        .read(repo, move |r, cache| Ok(cache.load(r, &heads)?.snapshot.files.keys().cloned().collect()))
        .await
        .map_err(|e| format!("coverage: the run's files could not be read: {e}"))?;
    let heads: Vec<String> = run.head_list().into_iter().map(String::from).collect();
    let origin = crate::coverage::Origin {
        repo_id: row.id,
        run_id: Some(run.id),
        run_number: run.number,
        job_id: Some(job.id),
        uploader: "",
        job: &job.name,
        bookmark: &run.bookmark,
        state: &run.state,
        heads: &heads,
        blob,
        format,
    };
    let id = crate::coverage::keep(app, &origin, files, outside, &in_state).await.map_err(|e| format!("coverage: {e}"))?;
    let Some(r) = crate::coverage::report(&app.db, row.id, id).await.map_err(|e| format!("coverage: {e}"))? else {
        return Err("coverage: the report was not kept".into());
    };
    // The `coverage` check on the run's state (#113): against the default
    // bookmark, under the repository's rules.
    let base = crate::coverage::baseline(&app.db, &r, &row.default_bookmark).await.map_err(|e| format!("coverage: {e}"))?;
    let rules = crate::coverage::rules(&app.db, row.id).await.map_err(|e| format!("coverage: {e}"))?;
    let (status, summary) = crate::coverage::verdict(&r, base.as_ref(), &row.default_bookmark, rules);
    let url = format!("{}/{}/{}/coverage?report={}", app.config.base_url(), row.owner, row.name, r.id);
    if let Ok(c) = db::hooks::upsert_check(
        &app.db,
        row.id,
        db::hooks::CheckReport { state: &run.state, name: "coverage", status, summary: &summary, url: &url, reporter: None, seal: "" },
    )
    .await
    {
        let api = crate::api::check_of(&c);
        app.events.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });
        crate::hooks::emit(app, row, "check", "flow", serde_json::json!({ "check": api })).await;
    }
    let pct = r.percent().map(|p| format!("{p:.1}%")).unwrap_or_else(|| "no lines".into());
    let dropped =
        if r.dropped > 0 { format!(" ({} named files are not in this state and were left out)", r.dropped) } else { String::new() };
    Ok(format!("coverage: {pct} of lines ({} of {}) in {} files{dropped}", r.lines_hit, r.lines_found, r.files))
}

/// `POST /api/v1/flow/jobs/{id}/finish`.
pub async fn finish(
    State(app): State<Shared>,
    headers: HeaderMap,
    Path(id): Path<i64>,
    Json(f): Json<Finish>,
) -> Result<StatusCode, ApiError> {
    let me = runner_auth(&app, &headers).await?;
    let job = running_job(&app, id).await?;
    let status = match (f.exit_code, &f.error) {
        (Some(0), None) => "passed",
        (Some(_), _) => "failed",
        (None, _) => "error",
    };
    let (receipt, receipt_hash) = match &f.receipt {
        Some(r) if r.job_name == job.name && Some(r.exit_code as i32) == f.exit_code => {
            // A passing job's receipt names exactly the artifacts uploaded,
            // in name order; a failing one's names none, though it may have
            // kept `when "always"` artifacts (oxygenflow#37).
            let kept: Vec<String> =
                if r.exit_code == 0 { fdb::artifacts(&app.db, job.id).await?.into_iter().map(|a| a.blob).collect() } else { vec![] };
            if r.outputs.iter().map(hex::encode).collect::<Vec<_>>() != kept {
                return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt names other outputs than the artifacts uploaded".into()));
            }
            (serde_json::to_string(r).unwrap_or_default(), hex::encode(r.hash()))
        }
        Some(_) => return Err(ApiError(StatusCode::BAD_REQUEST, "the receipt does not match the job".into())),
        None => (String::new(), String::new()),
    };
    let error = f.error.clone().unwrap_or_default();
    fdb::set_peaks(&app.db, job.id, f.peak_disk_mib, f.peak_memory_mib).await?;
    let changed = fdb::finish(
        &app.db,
        job.id,
        fdb::Outcome { status, exit_code: f.exit_code.map(i64::from), error: &error, receipt: &receipt, receipt_hash: &receipt_hash },
    )
    .await?;
    if let Some(run) = fdb::run_by_id(&app.db, job.run_id).await? {
        if let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? {
            if let (Some(r), Some(signed)) = (&f.receipt, &f.signed_receipt) {
                let (attest, signer, note) = match keep_receipt(&app, &me, &row, &run, r, signed).await {
                    Ok((attest, signer)) => (attest, signer, String::new()),
                    Err(e) => {
                        tracing::warn!("flow: job {} receipt not kept: {e}", job.id);
                        (String::new(), String::new(), e)
                    }
                };
                fdb::set_receipt(&app.db, job.id, &attest, &signer, &note).await?;
            }
        }
    }
    report_changed(&app, job.run_id, changed).await?;
    Ok(StatusCode::NO_CONTENT)
}

/// Report the checks of jobs whose status changed.
async fn report_changed(app: &Shared, run_id: i64, changed: Vec<fdb::Job>) -> sqlx::Result<()> {
    let Some(run) = fdb::run_by_id(&app.db, run_id).await? else {
        return Ok(());
    };
    let Some(row) = db::repos::get_by_id(&app.db, run.repo_id).await? else {
        return Ok(());
    };
    let mut errored_on: Vec<i64> = Vec::new();
    let some_passed = changed.iter().any(|j| j.status == "passed");
    if !changed.is_empty() {
        app.events.live.publish(&crate::events::topic::run(run_id), "jobs");
    }
    for j in changed {
        let summary = match j.status.as_str() {
            "passed" => format!("passed in {}", duration(&j)),
            "failed" => format!("exit {} after {}", j.exit_code.unwrap_or(-1), duration(&j)),
            "skipped" => "skipped: a dependency did not pass".into(),
            "cancelled" => j.error.clone(),
            _ => format!("could not run: {}", j.error),
        };
        report_check(app, &row, &run.state, run.number, &j.name, check_status(&j.status), &summary).await;
        if j.status == "error" {
            if let Some(r) = runner_of(app, j.id).await {
                errored_on.push(r);
            }
        }
    }
    errored_on.dedup();
    for r in errored_on {
        drain_if_erroring(app, r).await;
    }
    if some_passed {
        tell_approvers(app, &row, &run).await;
    }
    if let Some(done) = fdb::run_by_id(&app.db, run_id).await? {
        if done.status == "failed" && !done.live() {
            tell_failed(app, &row, &done).await;
        }
        // A pass after a failure on the same bookmark: fixed (#183).
        if done.status == "passed" {
            let before: Option<String> = sqlx::query_scalar(
                "SELECT status FROM flow_runs WHERE repo_id = $1 AND bookmark = $2 AND id < $3 AND status IN ('passed', 'failed')
                 ORDER BY id DESC LIMIT 1",
            )
            .bind(done.repo_id)
            .bind(&done.bookmark)
            .bind(done.id)
            .fetch_optional(&app.db)
            .await?;
            // Or this very run failed and a job of it was run again.
            let url = format!("/{}/{}/flow/{}", row.owner, row.name, done.number);
            let failed_here: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM notifications WHERE kind = 'flow.failed' AND url = $1")
                .bind(&url)
                .fetch_one(&app.db)
                .await?;
            if before.as_deref() == Some("failed") || failed_here > 0 {
                tell_fixed(app, &row, &done).await;
            }
        }
    }
    // A passing run of an upstream's default bookmark: consumers following
    // it when green are bumped (Diffusion, #148).
    if let Some(done) = fdb::run_by_id(&app.db, run_id).await? {
        if done.status == "passed" && done.bookmark == row.default_bookmark {
            let heads: Vec<ChangeHash> = done.head_list().iter().filter_map(|h| ChangeHash::from_hex(h)).collect();
            let (app, row) = (app.clone(), row.clone());
            tokio::spawn(async move { crate::diffusion::propagate(&app, &row, heads, true).await });
        }
        // A bump's run passed: a consumer that lands its bumps itself may
        // land it now (docs/diffusion.md §4.2).
        if let Some(path) = done.bookmark.strip_prefix("diffusion/").filter(|_| done.status == "passed") {
            tokio::spawn(crate::diffusion::try_land(app.clone(), row.clone(), path.to_string()));
        }
    }
    Ok(())
}

/// The account OxygenFlow's notices come from, made on first use with no
/// way to sign in.
pub const ACCOUNT: &str = "oxygenflow";

async fn flow_account(pool: &db::Pool) -> Option<db::accounts::Account> {
    if let Ok(Some(a)) = db::accounts::account_by_name(pool, ACCOUNT).await {
        return Some(a);
    }
    db::accounts::create_account(pool, ACCOUNT, "oxygenflow@localhost", "!", "", false).await.ok()
}

async fn runner_of(app: &Shared, job_id: i64) -> Option<i64> {
    sqlx::query_scalar("SELECT runner_id FROM flow_jobs WHERE id = $1").bind(job_id).fetch_optional(&app.db).await.ok().flatten().flatten()
}

/// A failed run (oxygenflow#30): its pusher and the repository's watchers
/// are told, in their inboxes and by email as they chose, naming the jobs
/// that did not pass.
async fn tell_failed(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) {
    let Some(me) = flow_account(&app.db).await else { return };
    let jobs = fdb::jobs(&app.db, run.id).await.unwrap_or_default();
    let broken: Vec<String> = jobs.iter().filter(|j| matches!(j.status.as_str(), "failed" | "error")).map(|j| j.name.clone()).collect();
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
    let on = if run.bookmark.is_empty() { String::new() } else { format!(" on {}", run.bookmark) };
    crate::notify::deliver(
        app,
        crate::notify::Event {
            kind: "flow.failed",
            title: format!("{}/{}: run #{}{on} failed ({})", row.owner, row.name, run.number, broken.join(", ")),
            url: format!("/{}/{}/flow/{}", row.owner, row.name, run.number),
            actor: &me,
            repo_id: row.id,
            thread: None,
            mention_text: "",
            direct: pusher.into_iter().map(|a| a.id).collect(),
        },
    )
    .await;
}

/// A run that passes after its bookmark's last one failed (oxygenhub#183):
/// its pusher and the repository's watchers hear it is green again.
async fn tell_fixed(app: &Shared, row: &db::repos::RepoRow, run: &fdb::Run) {
    let Some(me) = flow_account(&app.db).await else { return };
    let pusher = db::accounts::account_by_name(&app.db, &run.actor).await.ok().flatten();
    let on = if run.bookmark.is_empty() { String::new() } else { format!(" on {}", run.bookmark) };
    crate::notify::deliver(
        app,
        crate::notify::Event {
            kind: "flow.fixed",
            title: format!("{}/{}: run #{}{on} passed: fixed", row.owner, row.name, run.number),
            url: format!("/{}/{}/flow/{}", row.owner, row.name, run.number),
            actor: &me,
            repo_id: row.id,
            thread: None,
            mention_text: "",
            direct: pusher.into_iter().map(|a| a.id).collect(),
        },
    )
    .await;
}

/// A runner whose last jobs all ended in infrastructure errors is drained,
/// so it stops taking jobs it fails before running them, and the
/// administrators are told (oxygenflow#30).
async fn drain_if_erroring(app: &Shared, runner_id: i64) {
    let n = app.config.flow.drain_after_errors;
    let Ok(Some(last)) = db::runners::erroring(&app.db, runner_id, n).await else { return };
    let Ok(Some(r)) = db::runners::get(&app.db, runner_id).await else { return };
    if r.state != "active" || db::runners::set_state(&app.db, runner_id, "draining").await.is_err() {
        return;
    }
    tracing::warn!("flow: runner {} drained after {n} errors in a row (last: {last})", r.name);
    let Some(me) = flow_account(&app.db).await else { return };
    let admins: Vec<i64> = sqlx::query_scalar("SELECT id FROM accounts WHERE is_admin = TRUE").fetch_all(&app.db).await.unwrap_or_default();
    crate::notify::deliver(
        app,
        crate::notify::Event {
            kind: "flow.runner_drained",
            title: format!("Runner {} drained: its last {n} jobs ended in errors (last: {last})", r.name),
            url: format!("/admin/runners/{runner_id}"),
            actor: &me,
            repo_id: 0,
            thread: None,
            mention_text: "",
            direct: admins,
        },
    )
    .await;
}

/// The site's job time limits (oxygenhub#89): set on Admin → Runners →
/// Settings, else from `hub.toml`.
pub struct Limits {
    /// For a job whose pipeline names none.
    pub default_s: u64,
    /// The most a pipeline may ask for.
    pub max_s: u64,
    /// Grace past a job's limit before the hub gives it up: a runner stops
    /// a job at its limit and reports it within moments, so one still
    /// running this long after is one the runner no longer reports (it was
    /// restarted, or lost the hub while the job ended, oxygenhub#71).
    pub grace_ms: i64,
}

pub const SETTING_DEFAULT: &str = "flow.job_timeout_minutes";
pub const SETTING_MAX: &str = "flow.job_timeout_max_minutes";
pub const SETTING_GRACE: &str = "flow.lost_grace_minutes";

pub async fn limits(app: &Shared) -> Limits {
    let minutes = |key: &'static str| async move {
        db::accounts::setting(&app.db, key).await.ok().flatten().and_then(|v| v.parse::<u64>().ok()).filter(|m| *m > 0)
    };
    let default_m = minutes(SETTING_DEFAULT).await.unwrap_or(app.config.flow.job_timeout_minutes.max(1));
    let max_m = minutes(SETTING_MAX).await.unwrap_or(default_m.max(6 * 60)).max(default_m);
    let grace_m = minutes(SETTING_GRACE).await.unwrap_or(5);
    Limits { default_s: default_m * 60, max_s: max_m * 60, grace_ms: grace_m as i64 * 60_000 }
}

/// End every job still running past its time limit plus the grace:
/// it is marked an error with the reason, the jobs that depended on it are
/// skipped and its run settled, as when a runner reports an error.
pub async fn reap_lost_jobs(app: &Shared) -> sqlx::Result<usize> {
    let limits = limits(app).await;
    let lost = fdb::running_past_limit(&app.db, db::now_ms(), limits.default_s, limits.grace_ms).await?;
    let n = lost.len();
    for job in lost {
        let error = format!(
            "lost: still running {} minutes after it started, past the time limit; the runner{} stopped reporting it",
            (db::now_ms() - job.started_ms.unwrap_or(0)) / 60_000,
            if job.runner.is_empty() { String::new() } else { format!(" {}", job.runner) },
        );
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
        tracing::warn!("flow: job {} {error}", job.id);
        report_changed(app, job.run_id, changed).await?;
    }
    Ok(n)
}

/// End the jobs running on `runner` that it does not hold (oxygenhub#71):
/// it restarted since claiming them, or never received the claim. A
/// runner claims one job at a time, so a job it holds is always listed.
async fn end_jobs_not_held(app: &Shared, runner: &db::runners::Runner, holding: &[i64]) -> sqlx::Result<()> {
    for job in fdb::running_on(&app.db, runner.id).await? {
        if holding.contains(&job.id) {
            continue;
        }
        let error = format!("lost: the runner {} no longer holds this job (it restarted, or never received it)", runner.name);
        let changed =
            fdb::finish(&app.db, job.id, fdb::Outcome { status: "error", exit_code: None, error: &error, receipt: "", receipt_hash: "" })
                .await?;
        tracing::warn!("flow: job {} {error}", job.id);
        report_changed(app, job.run_id, changed).await?;
    }
    Ok(())
}

/// Start the scheduled runs due at `when` (oxygenflow#24): on each
/// repository that runs Flow, the jobs of its default bookmark's pipeline
/// with a `schedule` line due then. Returns how many runs started.
pub async fn run_due(app: &Shared, when: flow_core::pipeline::When) -> usize {
    let Ok(rows) = db::repos::list(&app.db).await else { return 0 };
    let mut started = 0;
    for row in rows.into_iter().filter(|r| enabled(app, r)) {
        let Some(repo) = RepoName::new(&row.owner, &row.name) else { continue };
        let b = row.default_bookmark.clone();
        let found = app
            .repos
            .read(&repo, move |r, _| {
                let Some(st) = r.view()?.bookmarks.get(&b).copied() else { return Ok(None) };
                Ok(Some((st, r.store().get_state(&st)?.heads)))
            })
            .await;
        let Ok(Some((state, heads))) = found else { continue };
        if heads.is_empty() {
            continue;
        }
        match start_run_at(app, &row, &row.default_bookmark, &state.to_hex(), &heads, &row.owner, "schedule", None, Some(when), &[]).await {
            Ok(Some(_)) => started += 1,
            Ok(None) => {}
            Err(e) => tracing::warn!("flow: scheduled run of {}/{}: {e}", row.owner, row.name),
        }
    }
    started
}

/// The hub's clock as a [`When`](flow_core::pipeline::When).
pub fn when_now() -> flow_core::pipeline::When {
    let now = time::OffsetDateTime::now_utc();
    flow_core::pipeline::When {
        minute: now.minute() as u32,
        hour: now.hour() as u32,
        day: now.day() as u32,
        month: u8::from(now.month()) as u32,
        weekday: now.weekday().number_days_from_sunday() as u32,
    }
}

/// Start the scheduled runs due each minute, once a minute.
pub async fn scheduler(app: Shared) {
    let mut last = None;
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(15)).await;
        let minute = time::OffsetDateTime::now_utc().unix_timestamp() / 60;
        if last == Some(minute) {
            continue;
        }
        last = Some(minute);
        let n = run_due(&app, when_now()).await;
        if n > 0 {
            tracing::info!("flow: {n} scheduled run(s) started");
        }
    }
}

/// Look for lost jobs once a minute.
pub async fn reaper(app: Shared) {
    loop {
        tokio::time::sleep(std::time::Duration::from_secs(60)).await;
        if let Err(e) = reap_lost_jobs(&app).await {
            tracing::warn!("flow: looking for lost jobs: {e}");
        }
        match expire_approvals(&app, db::now_ms()).await {
            Ok(n) if n > 0 => tracing::info!("flow: {n} approvals expired"),
            Ok(_) => {}
            Err(e) => tracing::warn!("flow: expiring approvals: {e}"),
        }
    }
}

/// Check a runner's signed receipt for a job of `run` and keep it in the
/// repository (Oxygen spec §14.6): the attestation's hex and the runner's
/// name, or why it was not kept. It must be the job's receipt, for the
/// run's state and repository, signed by the key the runner's token first
/// signed with.
pub async fn keep_receipt(
    app: &Shared,
    runner: &db::runners::Runner,
    row: &db::repos::RepoRow,
    run: &fdb::Run,
    receipt: &flow_core::Receipt,
    signed_hex: &str,
) -> Result<(String, String), String> {
    use oxy_core::receipt::SignedReceipt;
    let bytes = hex::decode(signed_hex).map_err(|_| "the signed receipt is not hex".to_string())?;
    let sr = SignedReceipt::decode(&bytes).map_err(|e| format!("not a signed receipt: {e}"))?;
    sr.verify().map_err(|e| format!("it does not verify: {e}"))?;
    if sr.receipt.hash().0 != receipt.hash() {
        return Err("it signs another receipt than the job's".into());
    }
    let heads: Vec<oxy_core::hash::ChangeHash> = run.head_list().into_iter().filter_map(oxy_core::hash::ChangeHash::from_hex).collect();
    let state = oxy_core::hash::hash_object(oxy_core::hash::ObjectType::State, &oxy_core::objects::StateObject::new(heads).encode());
    if sr.state() != state {
        return Err("it is for another state than the run's".into());
    }
    let key = sr.runner.hash().to_hex();
    let bound = db::runners::bind_key(&app.db, runner.id, &key, &sr.runner.name).await.map_err(|e| e.to_string())?;
    if bound != key {
        return Err(format!("signed by {}, not the key this runner signs with ({})", &key[..12], &bound[..12.min(bound.len())]));
    }
    let name = RepoName::new(&row.owner, &row.name).ok_or("bad repository name")?;
    let signer = sr.runner.name.clone();
    let attest = app.repos.write(&name, move |repo, _| Ok(repo.add_receipt(&sr)?)).await.map_err(|e| e.to_string())?;
    Ok((attest.to_hex(), signer))
}

/// A job's wall time, as the pages show it.
pub fn duration(j: &fdb::Job) -> String {
    match (j.started_ms, j.finished_ms) {
        (Some(a), Some(b)) => human_ms(b - a),
        (Some(a), None) => human_ms(db::now_ms() - a),
        _ => "–".into(),
    }
}

pub fn human_ms(ms: i64) -> String {
    let ms = ms.max(0);
    if ms < 1000 {
        format!("{ms} ms")
    } else if ms < 60_000 {
        format!("{:.1} s", ms as f64 / 1000.0)
    } else {
        format!("{} min {} s", ms / 60_000, (ms % 60_000) / 1000)
    }
}

/// Where a live run stands among the runners' work: what its viewers may
/// know of it. Jobs of other repositories are named only when those are
/// public; the rest are counted.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct QueueInfo {
    /// 1-based place of the run's first queued job among all queued jobs.
    pub position: Option<usize>,
    /// Jobs running now, and queued jobs ahead of the run's first one.
    pub running: Vec<String>,
    pub ahead: Vec<String>,
    /// Of those, jobs in repositories the page may not name.
    pub hidden: usize,
}

impl QueueInfo {
    /// "3rd in line" and the like.
    pub fn place(&self) -> Option<String> {
        self.position.map(|n| {
            let suffix = match (n % 10, n % 100) {
                (1, 11) | (2, 12) | (3, 13) => "th",
                (1, _) => "st",
                (2, _) => "nd",
                (3, _) => "rd",
                _ => "th",
            };
            format!("{n}{suffix} in line")
        })
    }
}

/// The queue as `run` (of repository `repo_id`) sees it, from `queue`.
pub fn queue_info(queue: &[fdb::QueueEntry], repo_id: i64, run_id: i64) -> QueueInfo {
    let mut info = QueueInfo::default();
    let first = queue.iter().filter(|e| e.status == "queued").position(|e| e.run_id == run_id);
    info.position = first.map(|i| i + 1);
    let label = |e: &fdb::QueueEntry| format!("{}/{} #{} {}", e.owner, e.name, e.run_number, e.job);
    let mut queued_seen = 0;
    for e in queue {
        if e.run_id == run_id {
            continue;
        }
        let ahead = match e.status.as_str() {
            "running" => true,
            _ => {
                queued_seen += 1;
                first.is_some_and(|f| queued_seen <= f)
            }
        };
        if !ahead {
            continue;
        }
        if e.repo_id != repo_id && e.visibility != "public" {
            info.hidden += 1;
        } else if e.status == "running" {
            info.running.push(label(e));
        } else {
            info.ahead.push(label(e));
        }
    }
    info
}

#[cfg(test)]
mod queue_tests {
    use super::*;

    fn e(job_id: i64, run_id: i64, repo_id: i64, vis: &str, job: &str, status: &str) -> fdb::QueueEntry {
        fdb::QueueEntry {
            job_id,
            run_id,
            repo_id,
            owner: "o".into(),
            name: format!("r{repo_id}"),
            visibility: vis.into(),
            run_number: run_id,
            job: job.into(),
            status: status.into(),
            runner: String::new(),
            spec: String::new(),
            sandbox: String::new(),
            started_ms: None,
            priority_rank: 2,
            priority_note: String::new(),
        }
    }

    #[test]
    fn a_run_learns_its_place_and_what_is_ahead_of_it() {
        // Running first, then queued in claim order (as `fdb::queue` returns them).
        let q = vec![
            e(1, 10, 1, "public", "build", "running"),
            e(2, 11, 2, "private", "test", "running"),
            e(3, 10, 1, "public", "test", "queued"),
            e(4, 12, 3, "private", "lint", "queued"),
            e(5, 13, 4, "public", "build", "queued"),
            e(6, 13, 4, "public", "test", "queued"),
            e(7, 14, 5, "public", "late", "queued"),
        ];
        let info = queue_info(&q, 4, 13);
        assert_eq!(info.position, Some(3));
        assert_eq!(info.place().as_deref(), Some("3rd in line"));
        assert_eq!(info.running, ["o/r1 #10 build"]);
        assert_eq!(info.ahead, ["o/r1 #10 test"], "queued ahead, not behind");
        assert_eq!(info.hidden, 2, "private repositories are counted, not named");
        // A private repository's own jobs are named to its viewers.
        assert_eq!(queue_info(&q, 3, 12).ahead, ["o/r1 #10 test"]);
        assert_eq!(queue_info(&q, 3, 12).position, Some(2));
        // A run with nothing queued has no place.
        assert_eq!(queue_info(&q, 1, 99).position, None);
        let ordinal = |n| QueueInfo { position: Some(n), ..Default::default() }.place().unwrap();
        assert_eq!(
            [ordinal(1), ordinal(2), ordinal(11), ordinal(22), ordinal(113)],
            ["1st in line", "2nd in line", "11th in line", "22nd in line", "113th in line"]
        );
    }
}

#[cfg(test)]
mod placement_tests {
    use super::*;

    #[test]
    fn a_waiting_job_says_what_each_runner_lacks() {
        use crate::db::runners::Runner;
        let rootfs = "ab".repeat(32);
        let job = fdb::Job {
            id: 7,
            runtime: "syscall".into(),
            sandbox: serde_json::json!({ "rootfs": rootfs, "memory_mib": 4096, "cpus": 2 }).to_string(),
            spec: serde_json::json!({ "name": "build" }).to_string(),
            ..Default::default()
        };
        let now = db::now_ms();
        let runner = |id: i64, name: &str| Runner {
            id,
            name: name.into(),
            state: "active".into(),
            last_seen_ms: Some(now),
            arch: "x86_64".into(),
            tiers: "microvm,syscall".into(),
            images: serde_json::to_string(&[&rootfs]).unwrap(),
            memory_mib: Some(16384),
            cpus: Some(8),
            ..Default::default()
        };
        let archs = std::collections::HashMap::new();
        let why = |runners: &[Runner], declines: &[fdb::DeclineRow]| {
            why_waiting(&job, "andy", 1, runners, &archs, declines, now, &Default::default())
        };
        assert_eq!(why(&[], &[]), "no runner is online");
        assert_eq!(why(&[runner(1, "earth-1")], &[]), "", "one can take it: it only waits its turn");
        let microvm_only = Runner { tiers: "microvm".into(), ..runner(1, "earth-1") };
        let riscv = Runner { arch: "riscv64".into(), ..runner(2, "narvi") };
        let small = Runner { memory_mib: Some(2048), ..runner(3, "pi") };
        let other = Runner { scope_owner: "bob".into(), ..runner(4, "bobs") };
        let offline = Runner { last_seen_ms: Some(now - 3_600_000), ..runner(5, "gone") };
        assert_eq!(
            why(&[microvm_only, small, other, offline], &[]),
            "no runner can take it: earth-1 has no syscall tier (it offers microvm); pi has 2048 MiB; the job needs 4096 MiB; bobs serves other repositories"
        );
        let archs_known: std::collections::HashMap<String, String> = [(rootfs.clone(), "x86_64".to_string())].into();
        assert_eq!(
            why_waiting(&job, "andy", 1, &[riscv], &archs_known, &[], now, &Default::default()),
            "no runner can take it: narvi is riscv64; the job needs x86_64"
        );
        let without = Runner { images: "[]".into(), ..runner(6, "new") };
        assert_eq!(why(&[without], &[]), "no runner can take it: new does not hold image abababababab");
        let d = fdb::DeclineRow {
            job_id: 7,
            runner_id: 1,
            runner: "earth-1".into(),
            reason: "only 512 MiB of memory is free here".into(),
            ms: now,
        };
        assert_eq!(
            why(&[runner(1, "earth-1")], std::slice::from_ref(&d)),
            "no runner can take it: earth-1 declined it: only 512 MiB of memory is free here"
        );
        let old = fdb::DeclineRow { ms: now - fdb::DECLINE_HOLDS_MS - 1, ..d };
        assert_eq!(why(&[runner(1, "earth-1")], &[old]), "", "an old decline no longer holds");
    }
}
`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet
Keyboard
