//! OxygenFlow on the hub (hub spec §13.2), end to end: a real `oxy` push
//! starts a run, a runner claims jobs through the API in dependency order,
//! job tokens read only their repository, results become checks and pages.

use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use http_body_util::BodyExt;
use oxy_core::protocol::{open_remote, RemoteOptions};
use oxy_core::repo::Repo;
use oxygenhub::config::Config;
use oxygenhub::repos::RepoName;
use oxygenhub::{db, Shared};
use std::path::Path;
use tower::ServiceExt;

const ROOTFS: &str = "36fc4e3238044e424219a2b9002bf61bbe1805fec14d52cf37e17fae77c9ff16";

fn hermetic() {
    static ONCE: std::sync::Once = std::sync::Once::new();
    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();
    ONCE.call_once(|| {
        let d = tempfile::tempdir().unwrap();
        std::env::set_var("OXY_HOME", d.path());
        DIR.set(d).ok();
    });
}

fn write(root: &Path, rel: &str, content: &str) {
    let p = root.join(rel);
    std::fs::create_dir_all(p.parent().unwrap()).unwrap();
    std::fs::write(p, content).unwrap();
}

struct Hub {
    app: Shared,
    base: String,
    _dir: tempfile::TempDir,
}

async fn hub() -> Hub {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let mut config = Config::load(dir.path()).unwrap();
    config.auth.push_tokens = vec!["site".into()];
    config.flow.runner_tokens = vec!["runner-secret".into(), "runner-two".into(), "runner-three".into()];
    config.flow.repos = vec!["andy/*".into()];
    config.registration.mode = "open".into();
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    config.server.base_url = format!("http://127.0.0.1:{port}");
    let app = oxygenhub::build(config).await.unwrap();
    let router = oxygenhub::router(app.clone());
    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
    Hub { app, base: format!("http://127.0.0.1:{port}"), _dir: dir }
}

async fn call(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<String>) -> (StatusCode, String) {
    let mut req = Request::builder().method(method).uri(path);
    if let Some(b) = bearer {
        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));
    }
    let body = match json {
        Some(j) => {
            req = req.header(header::CONTENT_TYPE, "application/json");
            Body::from(j)
        }
        None => Body::empty(),
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let status = res.status();
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    (status, String::from_utf8_lossy(&bytes).into_owned())
}

/// A GET's status, `Location` and `Content-Type`.
async fn head(app: &Shared, path: &str) -> (StatusCode, String, String) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let h = |k: header::HeaderName| res.headers().get(k).and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
    (res.status(), h(header::LOCATION), h(header::CONTENT_TYPE))
}

async fn claim(app: &Shared) -> Option<serde_json::Value> {
    let (st, body) =
        call(app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"test-1","runtimes":["microvm"]}"#.into())).await;
    match st {
        StatusCode::NO_CONTENT => None,
        StatusCode::OK => Some(serde_json::from_str(&body).unwrap()),
        other => panic!("claim: {other} {body}"),
    }
}

fn receipt(job: &str, pipeline_hash: &str, exit: u32) -> serde_json::Value {
    serde_json::json!({
        "job_name": job, "pipeline_hash": pipeline_hash, "scope_tree_hash": "11".repeat(32),
        "inputs": [], "environment": "22".repeat(32), "outputs": [], "exit_code": exit, "duration_ms": 640
    })
}

async fn finish(app: &Shared, id: i64, body: serde_json::Value) -> StatusCode {
    call(app, "POST", &format!("/api/v1/flow/jobs/{id}/finish"), Some("runner-secret"), Some(body.to_string())).await.0
}

async fn create(app: &Shared, owner: &str, name: &str) {
    app.repos.create(&RepoName::parse(&format!("{owner}/{name}")).unwrap(), None).unwrap();
    db::repos::create(&app.db, owner, name, "", hub_api::Visibility::Public).await.unwrap();
}

fn client(url: &str, work: &Path) -> Repo {
    let mut c = Repo::init(work).unwrap();
    c.config.set("remote.origin.url", url).unwrap();
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();
    c
}

fn pipeline(extra: &str) -> String {
    format!(
        "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    depends-on \"build\"\n    exec \"sh\" \"-eu\" \"test.sh\"\n}}\njob \"lint\" {{\n    sandbox \"alpine\"\n    exec \"true\"\n}}\n{extra}"
    )
}

async fn runs_of(app: &Shared, owner: &str, name: &str) -> Vec<db::flow::Run> {
    let row = db::repos::get(&app.db, owner, name).await.unwrap().unwrap();
    for _ in 0..100 {
        let r = db::flow::runs(&app.db, row.id, 10, None).await.unwrap();
        if !r.is_empty() {
            return r;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    vec![]
}

/// The checks on a state once there are `n` of them (they follow the run).
async fn checks_when(app: &Shared, repo_id: i64, state: &str, n: usize) -> Vec<db::hooks::Check> {
    for _ in 0..100 {
        let c = db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap();
        if c.len() >= n {
            return c;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    db::hooks::checks_for_state(&app.db, repo_id, state).await.unwrap()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_push_runs_the_pipeline_through_a_runner() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "secret").await;
    create(&h.app, "other", "plain").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The push started a run with three queued jobs, and checks for them.
    let runs = runs_of(&h.app, "andy", "demo").await;
    assert_eq!(runs.len(), 1);
    let run = &runs[0];
    assert_eq!((run.bookmark.as_str(), run.status.as_str(), run.actor.as_str()), ("main", "queued", "site token"));
    assert_eq!(run.pipeline_hash.len(), 64);
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert_eq!(jobs.iter().map(|j| (j.name.as_str(), j.depth)).collect::<Vec<_>>(), [("build", 0), ("lint", 0), ("test", 1)]);
    // The checks are reported just after the run is recorded.
    let checks = checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "queued"), ("flow/lint", "queued"), ("flow/test", "queued")]
    );

    // Runners need a runner token.
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", Some("site"), Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/claim", None, Some(r#"{"runner":"x","runtimes":["microvm"]}"#.into())).await.0,
        StatusCode::UNAUTHORIZED
    );
    // A runner without the tier gets nothing.
    let (st, _) =
        call(&h.app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(r#"{"runner":"x","runtimes":["wasm"]}"#.into())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);

    // Claims come in dependency order; `test` waits for `build`.
    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    assert_eq!((a["job"]["name"].as_str(), b["job"]["name"].as_str()), (Some("build"), Some("lint")));
    assert!(claim(&h.app).await.is_none(), "test waits for build");
    assert_eq!(a["repo"], "andy/demo");
    assert_eq!(a["clone_url"], url.as_str());
    assert_eq!(a["sandbox"]["rootfs"], ROOTFS);
    assert_eq!(a["job"]["exec"], serde_json::json!(["make"]));
    assert_eq!(a["timeout_s"], 3600);
    // The job is told what it builds (FLOW_STATE, FLOW_BOOKMARK, FLOW_RUN).
    assert_eq!(
        (a["state"].as_str(), a["bookmark"].as_str(), a["run_number"].as_i64()),
        (Some(run.state.as_str()), Some("main"), Some(run.number))
    );
    let heads: Vec<String> = a["heads"].as_array().unwrap().iter().map(|v| v.as_str().unwrap().to_string()).collect();
    assert_eq!(heads.join("+"), run.heads);
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(checks.iter().find(|c| c.name == "flow/build").unwrap().status, "running");

    // The job token reads this repository over the wire, and nothing else.
    let token = a["token"].as_str().unwrap().to_string();
    assert!(token.starts_with("flowjob_"));
    let opts = RemoteOptions { token: Some(token.clone()), ca: None, repo_id: None, cache_dir: None };
    open_remote(&url, &opts).unwrap().refs().unwrap();
    assert!(open_remote(&format!("{}/andy/secret", h.base), &opts).unwrap().refs().is_err(), "another repository");
    let bad = RemoteOptions { token: Some("flowjob_00".into()), ca: None, repo_id: None, cache_dir: None };
    assert!(open_remote(&url, &bad).unwrap().refs().is_err(), "an unknown job token");
    // … and cannot push.
    c.config.set("remote.origin.token", &token).unwrap();
    c.save_config().unwrap();
    write(work.path(), "more.txt", "x\n");
    c.snapshot("test").unwrap();
    c.commit("More", &[]).unwrap();
    let pushed = c.push("origin", false);
    assert!(pushed.is_err() || pushed.unwrap().iter().all(|r| !r.ok), "a job token cannot push");
    c.config.set("remote.origin.token", "site").unwrap();
    c.save_config().unwrap();

    // Logs, then results with receipts.
    let a_id = a["job_id"].as_i64().unwrap();
    let b_id = b["job_id"].as_i64().unwrap();
    let log = serde_json::json!({ "records": [
        { "t": 5, "stream": "flow", "text": "booting" },
        { "t": 120, "stream": "out", "text": "cc -o demo demo.c" },
        { "t": 130, "stream": "weird", "text": "x".repeat(20_000) },
    ]});
    assert_eq!(
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{a_id}/log"), Some("runner-secret"), Some(log.to_string())).await.0,
        StatusCode::NO_CONTENT
    );
    let lines = db::flow::log(&h.app.db, a_id, 0, 100).await.unwrap();
    assert_eq!(lines.len(), 3);
    assert_eq!((lines[2].stream.as_str(), lines[2].text.len()), ("console", 16 * 1024), "streams are known ones, text is capped");
    let ph = a["pipeline_hash"].as_str().unwrap();
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::BAD_REQUEST,
        "a receipt for another job"
    );
    assert_eq!(
        finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": receipt("build", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, a_id, serde_json::json!({ "exit_code": 0, "duration_ms": 700 })).await, StatusCode::CONFLICT, "only once");
    assert!(open_remote(&url, &opts).unwrap().refs().is_err(), "the token ends with its job");
    assert_eq!(
        finish(&h.app, b_id, serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": receipt("lint", ph, 0) })).await,
        StatusCode::NO_CONTENT
    );
    let t = claim(&h.app).await.unwrap();
    assert_eq!(t["job"]["name"], "test");
    let t_id = t["job_id"].as_i64().unwrap();
    assert_eq!(
        finish(&h.app, t_id, serde_json::json!({ "exit_code": 3, "duration_ms": 500, "receipt": receipt("test", ph, 3) })).await,
        StatusCode::NO_CONTENT
    );

    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();
    assert_eq!(run.status, "failed");
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let build = jobs.iter().find(|j| j.name == "build").unwrap();
    assert_eq!((build.status.as_str(), build.exit_code, build.receipt_hash.len()), ("passed", Some(0), 64));
    assert_eq!(jobs.iter().find(|j| j.name == "test").unwrap().status, "failed");
    let checks = db::hooks::checks_for_state(&h.app.db, run.repo_id, &run.state).await.unwrap();
    assert_eq!(
        checks.iter().map(|c| (c.name.as_str(), c.status.as_str())).collect::<Vec<_>>(),
        [("flow/build", "passed"), ("flow/lint", "passed"), ("flow/test", "failed")]
    );
    assert!(checks[2].summary.contains("exit 3") && checks[2].url.ends_with(&format!("/andy/demo/flow/{}", run.number)));

    // Pages: the list, the run with its graph and logs, and the live view.
    let (st, list) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(list.contains(&format!("/andy/demo/flow/{}", run.number)) && list.contains("st-failed"), "{list}");
    let (st, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<svg class=\"flow-graph\""), "graph");
    assert_eq!(page.matches("data-job-node=").count(), 3);
    assert_eq!(page.matches("<path class=\"edge").count(), 1, "one dependency edge");
    assert!(page.contains("cc -o demo demo.c"), "log shown");
    assert!(!page.contains("data-flow-live"), "a finished run does not poll");
    let (st, live) = call(&h.app, "GET", &format!("/andy/demo/flow/{}/live?after={a_id}:1", run.number), None, None).await;
    assert_eq!(st, StatusCode::OK);
    let live: serde_json::Value = serde_json::from_str(&live).unwrap();
    assert_eq!(live["status"], "failed");
    let build_live = live["jobs"].as_array().unwrap().iter().find(|j| j["name"] == "build").unwrap().clone();
    assert_eq!(build_live["lines"].as_array().unwrap().len(), 2, "lines after seq 1");
    assert_eq!(call(&h.app, "GET", "/andy/demo/flow/999", None, None).await.0, StatusCode::NOT_FOUND);

    // A failed dependency skips what follows; an unrun job is an error.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo changed\n");
    c.snapshot("test").unwrap();
    c.commit("Change the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut second = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 2 {
            second = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let second = second.expect("a second run");
    let x = claim(&h.app).await.unwrap();
    let y = claim(&h.app).await.unwrap();
    let (build_id, lint_id) = if x["job"]["name"] == "build" {
        (x["job_id"].as_i64().unwrap(), y["job_id"].as_i64().unwrap())
    } else {
        (y["job_id"].as_i64().unwrap(), x["job_id"].as_i64().unwrap())
    };
    assert_eq!(
        finish(&h.app, build_id, serde_json::json!({ "exit_code": null, "duration_ms": 10, "error": "rootfs is not on this runner" }))
            .await,
        StatusCode::NO_CONTENT
    );
    assert_eq!(finish(&h.app, lint_id, serde_json::json!({ "exit_code": 0, "duration_ms": 10 })).await, StatusCode::NO_CONTENT);
    assert!(claim(&h.app).await.is_none(), "test was skipped");
    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();
    assert_eq!(
        jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect::<Vec<_>>(),
        [("build", "error"), ("lint", "passed"), ("test", "skipped")]
    );
    assert_eq!(db::flow::run_by_id(&h.app.db, second.id).await.unwrap().unwrap().status, "failed");

    // An invalid pipeline is a run in error with the line.
    write(work.path(), ".flow/pipeline.kdl", &pipeline("job \"broken\" {\n    sandbox \"missing\"\n    exec \"x\"\n}\n"));
    c.snapshot("test").unwrap();
    c.commit("Break the pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut third = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 3 {
            third = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let third = third.expect("a third run");
    assert_eq!(third.status, "error");
    assert!(third.error.contains("line 18") && third.error.contains("\"missing\""), "{}", third.error);
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", third.number), None, None).await;
    assert!(page.contains("The pipeline is not valid"));

    // State stamps (hub spec §13.2) and images in READMEs.
    write(work.path(), "README.md", "# Demo\n\n[![flow](.flow/stamp)](.flow/run)\n\n![logo](img/logo.png)\n");
    std::fs::create_dir_all(work.path().join("img")).unwrap();
    std::fs::write(work.path().join("img/logo.png"), b"\x89PNG\r\n\x1a\n\0\0\0\rIHDR\0\0\0\x01\0\0\0\x01\x08\x06\0\0\0\x1f\x15\xc4\x89")
        .unwrap();
    c.snapshot("test").unwrap();
    c.commit("A README with a stamp", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut fourth = None;
    for _ in 0..100 {
        let r = runs_of(&h.app, "andy", "demo").await;
        if r.len() == 4 {
            fourth = Some(r[0].clone());
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    let fourth = fourth.expect("a fourth run");
    let (_, overview) = call(&h.app, "GET", "/andy/demo", None, None).await;
    assert!(overview.contains("src=\"/andy/demo/raw/main/.flow/stamp\""), "{overview}");
    assert!(overview.contains("href=\"/andy/demo/file/main/.flow/run\""));
    assert!(overview.contains("src=\"/andy/demo/raw/main/img/logo.png\""), "images use the raw route");
    let (st, _, ct) = head(&h.app, "/andy/demo/raw/main/img/logo.png").await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/png"));
    let (st, loc, _) = head(&h.app, "/andy/demo/raw/main/.flow/stamp").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/stamp/main", "{st} {loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/stamp/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/stamp/{}", fourth.heads), "a bookmark's stamp names its state: {loc}");
    let (st, _, ct) = head(&h.app, &loc).await;
    assert_eq!((st, ct.as_str()), (StatusCode::OK, "image/svg+xml; charset=utf-8"));
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.starts_with("<svg") && svg.contains("error ·") && svg.contains(&fourth.heads[..12]), "{svg}");
    let (_, svg1) = call(&h.app, "GET", &format!("/andy/demo/flow/stamp/{}", run.heads), None, None).await;
    assert!(svg1.contains("failed ·") && svg1.contains(&format!("run #{} failed", run.id)), "an older state keeps its own result: {svg1}");
    assert!(svg1.contains(&build.receipt_hash), "receipt hashes in the metadata");
    assert_eq!(svg1.matches("text-anchor=\"middle\"").count(), 3, "one glyph per job");
    let (st, loc, _) = head(&h.app, "/andy/demo/file/main/.flow/run").await;
    assert!(st.is_redirection() && loc == "/andy/demo/flow/at/main", "{loc}");
    let (st, loc, _) = head(&h.app, "/andy/demo/flow/at/main").await;
    assert!(st.is_redirection() && loc == format!("/andy/demo/flow/{}", fourth.id), "{loc}");

    // A repository outside [flow] repos starts nothing.
    let work2 = tempfile::tempdir().unwrap();
    let c2 = client(&format!("{}/other/plain", h.base), work2.path());
    write(work2.path(), ".flow/pipeline.kdl", &pipeline(""));
    c2.snapshot("test").unwrap();
    c2.commit("Pipeline", &[]).unwrap();
    assert!(c2.push("origin", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    assert!(db::flow::runs(&h.app.db, row.id, 10, None).await.unwrap().is_empty());
    let (_, page) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(page.contains("not enabled for this repository"));
    let (_, loc, _) = head(&h.app, "/other/plain/flow/stamp/main").await;
    let (_, svg) = call(&h.app, "GET", &loc, None, None).await;
    assert!(svg.contains("no run ·"), "{svg}");
    let (_, loc, _) = head(&h.app, "/other/plain/flow/at/main").await;
    assert_eq!(loc, "/other/plain/flow");
}

#[tokio::test(flavor = "multi_thread")]
async fn claims_carry_pinned_subrepos_and_the_job_token_reads_only_those() {
    let h = hub().await;
    for n in ["base", "lib", "vault", "other", "app"] {
        create(&h.app, "andy", n).await;
    }
    let push = |name: &str, file: &str| {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/{name}", h.base), work.path());
        write(work.path(), file, "x\n");
        c.snapshot("t").unwrap();
        c.commit("first", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        let heads = c.bookmark_heads().unwrap();
        (work, heads)
    };
    let (_b, base_heads) = push("base", "base.rs");
    // lib pins base in turn: a subrepo of a subrepo.
    let lib_work = tempfile::tempdir().unwrap();
    let lib = client(&format!("{}/andy/lib", h.base), lib_work.path());
    lib.submodule_add(&format!("{}/andy/base", h.base), "deps/base", Some("main")).unwrap();
    write(lib_work.path(), "lib.rs", "x\n");
    lib.snapshot("t").unwrap();
    lib.commit("lib on base", &[]).unwrap();
    assert!(lib.push("origin", false).unwrap().iter().all(|r| r.ok));
    let lib_heads = lib.bookmark_heads().unwrap();
    let (_v, _) = push("vault", "secret.rs");
    let (_o, _) = push("other", "o.rs");

    // The app pins lib and vault as subrepos and pushes as the site token.
    let work = tempfile::tempdir().unwrap();
    let app = client(&format!("{}/andy/app", h.base), work.path());
    app.submodule_add(&format!("{}/andy/lib", h.base), "deps/lib", Some("main")).unwrap();
    // Vault is private: a checkout with no token for this hub cannot add
    // it, but the app's remote on the same hub lends its token (oxygen#32).
    // The subrepo keeps no copy: its own fetches find the parent's (oxygen#47).
    sqlx::query("UPDATE repositories SET visibility = 'private' WHERE name = 'vault'").execute(&h.app.db).await.unwrap();
    let bare_work = tempfile::tempdir().unwrap();
    let mut bare = Repo::init(bare_work.path()).unwrap();
    bare.config.set("remote.origin.url", &format!("{}/andy/bare", h.base)).unwrap();
    bare.save_config().unwrap();
    assert!(bare.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).is_err(), "not without a token");
    app.submodule_add(&format!("{}/andy/vault", h.base), "deps/vault", Some("main")).unwrap();
    let nested = Repo::open(&work.path().join("deps/vault")).unwrap();
    assert!(nested.config.remote_tokens.is_empty(), "no token is copied into a subrepo");
    nested.fetch("origin").unwrap();
    // The pusher below is the site token, which has no account: it cannot read vault.
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    app.snapshot("t").unwrap();
    app.commit("app with deps", &[]).unwrap();
    assert!(app.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "app").await.is_empty());

    let job = claim(&h.app).await.unwrap();
    let subs = job["subrepos"].as_array().cloned().unwrap_or_default();
    assert_eq!(subs.len(), 2, "lib and what lib pins; vault is private and the pusher has no account: {subs:?}");
    assert_eq!(subs[1]["path"], "deps/lib/deps/base", "nested under its parent, after it");
    assert_eq!(subs[1]["heads"], serde_json::json!(base_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    let _ = std::fs::remove_dir_all(work.path().join("deps/vault2"));
    assert_eq!(subs[0]["path"], "deps/lib");
    assert_eq!(subs[0]["repo"], "andy/lib");
    assert_eq!(subs[0]["heads"], serde_json::json!(lib_heads.iter().map(|h| h.to_hex()).collect::<Vec<_>>()));
    assert!(subs[0]["clone_url"].as_str().unwrap().ends_with("/andy/lib"));
    // The job's log says vault was left out, and why (oxygenhub#56).
    let job_id = job["job_id"].as_i64().unwrap();
    let lines = db::flow::log(&h.app.db, job_id, 0, 100).await.unwrap();
    assert!(
        lines.iter().any(|l| l.stream == "flow" && l.text.contains("subrepo deps/vault (andy/vault) is left out")),
        "{:?}",
        lines.iter().map(|l| &l.text).collect::<Vec<_>>()
    );

    // The job token reads its repository and the pinned lib, nothing else.
    let token = job["token"].as_str().unwrap().to_string();
    let read = |name: &str| {
        let opts = RemoteOptions { token: Some(token.clone()), ..Default::default() };
        open_remote(&format!("{}/andy/{name}", h.base), &opts).and_then(|r| r.refs()).is_ok()
    };
    assert!(read("app") && read("lib") && read("base"));
    assert!(!read("vault"), "not pinned for this pusher");
    assert!(!read("other"), "public, but not the job's to read");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
async fn concurrent_log_appends_never_fail_on_a_locked_database() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "Makefile", "all:\n");
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let job = db::flow::jobs(&h.app.db, run.id).await.unwrap().remove(0);
    // Many writers at once, each reading the count and then inserting: a
    // deferred transaction would fail with "database is locked".
    let mut tasks = Vec::new();
    for w in 0..16 {
        let pool = h.app.db.clone();
        tasks.push(tokio::spawn(async move {
            let records: Vec<flow_core::api::Record> =
                (0..20).map(|i| flow_core::api::Record { t: i, stream: "out".into(), text: format!("writer {w} line {i}") }).collect();
            db::flow::append_log(&pool, job.id, &records, 100_000).await
        }));
    }
    for t in tasks {
        assert_eq!(t.await.unwrap().unwrap(), 20, "every append succeeds");
    }
    let lines = db::flow::log(&h.app.db, job.id, 0, 1_000).await.unwrap();
    assert_eq!(lines.len(), 320);
    let seqs: std::collections::BTreeSet<i64> = lines.iter().map(|l| l.seq).collect();
    assert_eq!(seqs.len(), 320, "sequence numbers are unique");
}

/// A repository's own Flow setting overrides hub.toml either way.
#[tokio::test(flavor = "multi_thread")]
async fn a_repository_setting_turns_flow_on_or_off() {
    let h = hub().await;
    create(&h.app, "other", "plain").await;
    create(&h.app, "andy", "quiet").await;
    // A run elsewhere first, so hub-wide ids and per-repository numbers differ.
    create(&h.app, "andy", "first").await;
    {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/andy/first", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        assert_eq!(runs_of(&h.app, "andy", "first").await.len(), 1);
    }
    let row = db::repos::get(&h.app.db, "other", "plain").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, row.id, "on").await.unwrap();
    let quiet = db::repos::get(&h.app.db, "andy", "quiet").await.unwrap().unwrap();
    db::repos::set_flow(&h.app.db, quiet.id, "off").await.unwrap();
    for (owner, name) in [("other", "plain"), ("andy", "quiet")] {
        let work = tempfile::tempdir().unwrap();
        let c = client(&format!("{}/{owner}/{name}", h.base), work.path());
        write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
        write(work.path(), "test.sh", "echo ok\n");
        c.commit("Add a pipeline", &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    }
    let plain = runs_of(&h.app, "other", "plain").await;
    assert_eq!(plain.len(), 1, "turned on");
    // With andy/first's jobs ahead, plain's run says where it stands.
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["repo"], "andy/first");
    let (_, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(page.contains("Waiting for a runner: 3rd in line") && page.contains("Running now: andy/first #1"), "{page}");
    let (_, body) = call(&h.app, "GET", "/api/v1/repos/other/plain/flow/runs/1", None, None).await;
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["queue_position"], 3);
    let (_, list) = call(&h.app, "GET", "/other/plain/flow", None, None).await;
    assert!(list.contains("queued, 3rd in line"), "{list}");
    // Runs are numbered per repository: this is plain's first, whatever
    // other repositories ran before.
    assert_eq!(plain[0].number, 1);
    let checks = checks_when(&h.app, row.id, &plain[0].state, 3).await;
    assert!(checks.iter().all(|c| c.url.ends_with("/other/plain/flow/1")), "{checks:?}");
    let (st, page) = call(&h.app, "GET", "/other/plain/flow/1", None, None).await;
    assert!(st == StatusCode::OK && page.contains("Run #1"), "{st}");
    assert!(runs_of(&h.app, "andy", "quiet").await.is_empty(), "turned off despite andy/*");
}

/// "Run pipeline" and "Re-run" (hub spec §13.2): runs that move no
/// bookmark, through the API and the pages.
#[tokio::test(flavor = "multi_thread")]
async fn runs_start_without_a_push_and_finished_runs_re_run() {
    let h = hub().await;
    // The site token acts as the first site administrator.
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let first = runs_of(&h.app, "andy", "demo").await;
    assert_eq!((first.len(), first[0].trigger.as_str()), (1, "push"));

    // Run pipeline on the default bookmark: a new run of the same state.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["number"], 2);
    assert!(v["url"].as_str().unwrap().ends_with("/andy/demo/flow/2"));
    let second = db::flow::run(&h.app.db, first[0].repo_id, 2).await.unwrap().unwrap();
    assert_eq!((second.trigger.as_str(), second.actor.as_str(), second.state.as_str()), ("manual", "andy", first[0].state.as_str()));
    assert_eq!(db::flow::jobs(&h.app.db, second.id).await.unwrap().len(), 3);
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/2", None, None).await;
    assert!(page.contains("started by andy"), "the run page says what started it");

    // Run pipeline on one change: its state, with no bookmark.
    let head = first[0].head_list()[0].to_string();
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{head}"}}"#))).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let on_change = db::flow::run(&h.app.db, first[0].repo_id, 3).await.unwrap().unwrap();
    assert_eq!(
        (on_change.bookmark.as_str(), on_change.heads.as_str(), on_change.state.as_str()),
        ("", head.as_str(), first[0].state.as_str())
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/3", None, None).await;
    assert!(page.contains("title=\"Run on one change, not a bookmark\">change</span>"), "run page");
    let (st, body) =
        call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(format!(r#"{{"change":"{}"}}"#, "ab".repeat(32))))
            .await;
    assert!(st == StatusCode::CONFLICT && body.contains("no change"), "{st} {body}");
    // Viewers who cannot start runs are told why, and get no button.
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Sign in with write access to run the pipeline.") && !page.contains("primary-btn"));
    let (_, page) = call(&h.app, "GET", "/andy/demo/bookmarks", None, None).await;
    assert!(!page.contains("/flow/start"), "no run buttons for anonymous viewers");

    // A live run is not re-run; unknown bookmarks and runs are refused.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert!(st == StatusCode::CONFLICT && body.contains("still queued"), "{st} {body}");
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some(r#"{"bookmark":"nope"}"#.into())).await;
    assert_eq!(st, StatusCode::CONFLICT);
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/99/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    // Anonymous callers cannot start runs.
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", None, Some("{}".into())).await;
    assert_ne!(st, StatusCode::CREATED);

    // A finished run re-runs as a new run of the same state.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    for r in [&first[0], &second] {
        for j in db::flow::jobs(&h.app.db, r.id).await.unwrap() {
            sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE id = ?").bind(j.id).execute(&h.app.db).await.unwrap();
        }
        sqlx::query("UPDATE flow_runs SET status = 'passed', finished_ms = 1 WHERE id = ?").bind(r.id).execute(&h.app.db).await.unwrap();
    }
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs/1/rerun", Some("site"), None).await;
    assert_eq!(st, StatusCode::CREATED, "{body}");
    let third = db::flow::run(&h.app.db, row.id, 4).await.unwrap().unwrap();
    assert_eq!(
        (third.trigger_label().as_str(), third.status.as_str(), third.state.as_str()),
        ("rerun:1", "queued", first[0].state.as_str())
    );
    assert_eq!(third.trigger, "rerun:1:push", "a rerun keeps the event it repeats");
    let checks = db::hooks::checks_for_state(&h.app.db, row.id, &third.state).await.unwrap();
    assert!(
        checks.iter().all(|c| c.status == "queued" && c.url.ends_with("/flow/4")),
        "the re-run's checks replace the old ones: {checks:?}"
    );
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow/4", None, None).await;
    assert!(page.contains("re-run of #1 by andy"));
    assert!(!page.contains("RE-RUN"), "anonymous viewers get no button");

    // With Flow off, nothing starts.
    db::repos::set_flow(&h.app.db, row.id, "off").await.unwrap();
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/demo/flow/runs", Some("site"), Some("{}".into())).await;
    assert!(st == StatusCode::CONFLICT && body.contains("not enabled"), "{body}");
    let (_, page) = call(&h.app, "GET", "/andy/demo/flow", None, None).await;
    assert!(page.contains("Flow is off for this repository"), "and the Flow tab says so");
}

/// The Flow API (hub spec §13.2): runs, a run with its jobs, a job's log,
/// checks by the heads a client has, and JSON for unknown API routes.
#[tokio::test(flavor = "multi_thread")]
async fn runs_jobs_and_logs_are_in_the_api() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    let head = c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);

    // A job runs and logs.
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let recs =
        serde_json::json!({ "records": [{ "t": 1, "stream": "stdout", "text": "one" }, { "t": 3, "stream": "stderr", "text": "three" }] });
    let (st, _) = call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(recs.to_string())).await;
    assert!(st.is_success(), "{st}");

    let (st, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs", None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let runs: Vec<hub_api::FlowRun> = serde_json::from_str(&body).unwrap();
    assert_eq!(runs.len(), 1);
    assert_eq!((runs[0].number, runs[0].state.as_str(), runs[0].trigger.as_str()), (1, run.state.as_str(), "push"));
    assert!(runs[0].url.ends_with("/andy/demo/flow/1") && runs[0].jobs.is_empty());

    let (_, body) = call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1", None, None).await;
    let one: hub_api::FlowRun = serde_json::from_str(&body).unwrap();
    let names: Vec<(&str, &str)> = one.jobs.iter().map(|j| (j.name.as_str(), j.status.as_str())).collect();
    assert_eq!(names.len(), 3);
    let running = one.jobs.iter().find(|j| j.status == "running").unwrap();
    assert_eq!(running.name, a["job"]["name"].as_str().unwrap());
    assert!(one.jobs.iter().any(|j| j.name == "test" && j.depends_on == ["build"]));

    let job = running.name.clone();
    let (st, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log"), None, None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let lines: Vec<hub_api::FlowLogLine> = serde_json::from_str(&body).unwrap();
    assert!(lines.iter().any(|l| l.text == "three"), "{body}");
    let last = lines.last().unwrap().seq;
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/repos/andy/demo/flow/runs/1/jobs/{job}/log?after={last}"), None, None).await;
    assert_eq!(body, "[]", "nothing after the last line");
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/flow/runs/1/jobs/nope/log", None, None).await.0, StatusCode::NOT_FOUND);

    // Checks by the state hash, by the head `oxy log` shows, and by bookmark.
    let by = |key: String| {
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "GET", &format!("/api/v1/repos/andy/demo/checks/{key}"), None, None).await;
            assert_eq!(st, StatusCode::OK, "{key}: {body}");
            serde_json::from_str::<Vec<hub_api::Check>>(&body).unwrap().len()
        }
    };
    checks_when(&h.app, run.repo_id, &run.state, 3).await;
    assert_eq!(by(run.state.clone()).await, 3);
    assert_eq!(by(head.to_hex()).await, 3, "a single head names its state");
    assert_eq!(by("main".into()).await, 3);
    let (st, _) = call(&h.app, "GET", "/api/v1/repos/andy/demo/checks/no-such-bookmark", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);

    // An unknown API route answers with a problem document.
    let (st, body) = call(&h.app, "GET", "/api/v1/users/andy/repos", None, None).await;
    assert_eq!(st, StatusCode::NOT_FOUND);
    let v: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(v["status"], 404, "{body}");
}

async fn get_bytes(app: &Shared, path: &str) -> (StatusCode, Vec<u8>) {
    let res = oxygenhub::router(app.clone()).oneshot(Request::get(path).body(Body::empty()).unwrap()).await.unwrap();
    let status = res.status();
    (status, res.into_body().collect().await.unwrap().to_bytes().to_vec())
}

#[tokio::test(flavor = "multi_thread")]
async fn signed_receipts_are_checked_kept_and_shown() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectRef, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let heads: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads.clone()).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let sign = |r: &serde_json::Value, state: ObjectRef, sk: &SecretKey, id: &oxy_core::identity::Identity| {
        let r: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
        let run_url = format!("{url}/flow/{}", run.number);
        hex::encode(SignedReceipt::sign(r.to_oxy(), repo_id, state, &run_url, id, sk).encode())
    };

    let a = claim(&h.app).await.unwrap();
    let b = claim(&h.app).await.unwrap();
    let ph = a["pipeline_hash"].as_str().unwrap().to_string();
    let (a_id, b_id) = (a["job_id"].as_i64().unwrap(), b["job_id"].as_i64().unwrap());
    // The first signed receipt binds the runner token to its key, and is kept.
    let rb = receipt("build", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": rb, "signed_receipt": sign(&rb, state, &sk, &runner) });
    assert_eq!(finish(&h.app, a_id, body).await, StatusCode::NO_CONTENT);
    // Another key under the same token: the result counts, the receipt is not kept.
    let other = SecretKey::generate();
    let rl = receipt("lint", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 300, "receipt": rl,
        "signed_receipt": sign(&rl, state, &other, &other.identity("mallory", "", 1)) });
    assert_eq!(finish(&h.app, b_id, body).await, StatusCode::NO_CONTENT);
    // The right key, but for another state.
    let t = claim(&h.app).await.unwrap();
    let rt = receipt("test", &ph, 0);
    let body = serde_json::json!({ "exit_code": 0, "duration_ms": 500, "receipt": rt,
        "signed_receipt": sign(&rt, ObjectRef([9; 32]), &sk, &runner) });
    assert_eq!(finish(&h.app, t["job_id"].as_i64().unwrap(), body).await, StatusCode::NO_CONTENT);

    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    let job = |n: &str| jobs.iter().find(|j| j.name == n).unwrap().clone();
    assert_eq!(job("build").status, "passed");
    assert_eq!((job("build").receipt_attest.len(), job("build").receipt_signer.as_str()), (64, "test-1 (OxygenFlow runner)"));
    assert_eq!(job("lint").status, "passed", "the job's result stands");
    assert!(job("lint").receipt_attest.is_empty());
    assert!(job("lint").receipt_note.contains("not the key this runner signs with"), "{}", job("lint").receipt_note);
    assert!(job("test").receipt_note.contains("another state"), "{}", job("test").receipt_note);

    // The repository keeps it, and anyone who can read the repository can
    // download it and check it on their own.
    let kept = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().receipts(&heads).unwrap();
    assert_eq!(kept.len(), 1);
    let attest = job("build").receipt_attest;
    let (st, bytes) = get_bytes(&h.app, &format!("/andy/demo/receipts/{attest}")).await;
    assert_eq!(st, StatusCode::OK);
    let sr = SignedReceipt::decode(&bytes).unwrap();
    sr.verify().unwrap();
    assert_eq!((sr.state(), sr.receipt.job_name.as_str()), (state, "build"));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/receipts/{}", "0".repeat(64))).await.0, StatusCode::NOT_FOUND);

    // Pages: the run names the signer or why a receipt was not kept; the
    // change whose state it is lists it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert!(page.contains("signed receipt</a> by test-1"), "run page");
    assert!(page.contains("receipt not kept"), "run page");
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/change/{}", heads[0].to_hex()), None, None).await;
    assert!(page.contains("<dt>Receipts</dt>") && page.contains(&format!("/andy/demo/receipts/{attest}")), "change page");
}

async fn upload(app: &Shared, job: i64, name: &str, bytes: &[u8]) -> (StatusCode, String) {
    let req = Request::post(format!("/api/v1/flow/jobs/{job}/artifacts/{name}"))
        .header(header::AUTHORIZATION, "Bearer runner-secret")
        .header(header::CONTENT_TYPE, "application/octet-stream")
        .body(Body::from(bytes.to_vec()))
        .unwrap();
    let res = oxygenhub::router(app.clone()).oneshot(req).await.unwrap();
    let status = res.status();
    (status, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
}

#[tokio::test(flavor = "multi_thread")]
async fn artifacts_are_kept_named_by_the_receipt_and_served() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    assert_eq!(a["job"]["outputs"], serde_json::json!([{ "name": "tool", "path": "/src/out/tool" }]), "the runner is told");
    let id = a["job_id"].as_i64().unwrap();

    let body: Vec<u8> = (0..300_000u32).map(|i| (i % 253) as u8).collect();
    let (st, text) = upload(&h.app, id, "other", &body).await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "an undeclared artifact: {text}");
    let (st, text) = upload(&h.app, id, "tool", &body).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    assert_eq!(blob, oxy_core::repo::Repo::blob_ref(&body).to_hex(), "the same blob the runner computes");

    // The receipt must name exactly the uploaded artifacts.
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r.clone() })).await,
        StatusCode::BAD_REQUEST
    );
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 0, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);

    // The run page lists it; anyone who can read the repository downloads it.
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    let link = format!("/andy/demo/flow/{}/artifacts/build/tool", run.number);
    assert!(page.contains(&link), "run page lists the artifact");
    let (st, got) = get_bytes(&h.app, &link).await;
    assert_eq!((st, got == body), (StatusCode::OK, true));
    assert_eq!(get_bytes(&h.app, &format!("/andy/demo/flow/{}/artifacts/build/none", run.number)).await.0, StatusCode::NOT_FOUND);

    // A writer attaches the artifact to a release from the run page; the
    // release then names the run and job that built it, as does the API.
    let form = |f: String| {
        Request::post("/register").header(header::CONTENT_TYPE, "application/x-www-form-urlencoded").body(Body::from(f)).unwrap()
    };
    let res = oxygenhub::router(h.app.clone())
        .oneshot(form("username=andy&email=a%40example.com&password=correct-horse-battery".into()))
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let with_cookie = |method: &str, path: &str, body: Option<String>| {
        let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie.clone());
        if body.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()
    };
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    let res = oxygenhub::router(h.app.clone()).oneshot(with_cookie("GET", &format!("/andy/demo/flow/{}", run.number), None)).await.unwrap();
    let page = String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned();
    let attach = format!("/andy/demo/flow/{}/artifacts/build/tool/attach", run.number);
    assert!(page.contains(&attach), "a writer is offered the release");
    let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
    let csrf = page[i..].split('"').next().unwrap().to_string();
    let release_id = db::releases::list(&h.app.db, row.id).await.unwrap()[0].id;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(with_cookie("POST", &attach, Some(format!("_csrf={csrf}&release={release_id}"))))
        .await
        .unwrap();
    assert_eq!(res.status(), StatusCode::SEE_OTHER);
    let (_, page) = call(&h.app, "GET", "/andy/demo/releases/tag/v1", None, None).await;
    assert!(
        page.contains(">tool</a>") && page.contains(&format!("built by <a href=\"/andy/demo/flow/{0}\">run #{0}</a> build", run.number)),
        "{page}"
    );
    let (_, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases", None, None).await;
    let api: serde_json::Value = serde_json::from_str(&api).unwrap();
    let asset = &api[0]["assets"][0];
    assert_eq!(
        (asset["name"].as_str(), asset["built_by"]["run"].as_i64(), asset["built_by"]["job"].as_str()),
        (Some("tool"), Some(run.number), Some("build"))
    );
    assert!(asset["built_by"]["receipt"].is_null(), "that receipt was not signed");
}

/// oxygenflow#37: a failing job keeps a `when "always"` artifact; its
/// receipt names none, and the run page says the file is outside it.
#[tokio::test(flavor = "multi_thread")]
async fn a_failing_job_keeps_an_artifact_outside_its_receipt() {
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"test\"\n    output {{\n        artifact \"junit\" {{\n            path \"/src/out/junit.xml\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    // A JUnit report (#114): parsed on upload, counted, failures listed.
    let xml = br#"<testsuites><testsuite name="pytest"><testcase classname="tests.test_io" name="test_read" time="0.5"/><testcase classname="tests.test_io" name="test_write" time="1.5"><failure message="disk full">E   OSError: disk full</failure></testcase></testsuite></testsuites>"#;
    let (st, text) = upload(&h.app, id, "junit", xml).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    let mut r = receipt("test", a["pipeline_hash"].as_str().unwrap(), 1);
    r["outputs"] = serde_json::json!([blob]);
    assert_eq!(
        finish(&h.app, id, serde_json::json!({ "exit_code": 1, "duration_ms": 700, "receipt": r.clone() })).await,
        StatusCode::BAD_REQUEST,
        "a failing receipt names no outputs"
    );
    r["outputs"] = serde_json::json!([]);
    assert_eq!(finish(&h.app, id, serde_json::json!({ "exit_code": 1, "duration_ms": 700, "receipt": r })).await, StatusCode::NO_CONTENT);
    let (_, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.number), None, None).await;
    assert!(page.contains(&format!("/andy/demo/flow/{}/artifacts/test/junit", run.number)), "listed");
    assert!(page.contains("not in its receipt"), "marked");
    assert!(page.contains("2 tests: 1 failed"), "counts beside the job");
    assert!(page.contains("tests.test_io::test_write") && page.contains("disk full") && page.contains("OSError"), "the failure");
    assert!(page.contains("tests: 2 tests: 1 failed in 2.0 s"), "said in the job's log");
}

/// A release's trust report (hub spec §9.1): a signed tag, an asset built
/// by a job with a signed receipt from the tagged state, and one uploaded
/// by hand; the page, its JSON export and the API.
#[tokio::test(flavor = "multi_thread")]
async fn a_release_is_audited_from_authors_to_receipts() {
    use oxy_core::hash::{hash_object, ChangeHash, ObjectType};
    use oxy_core::identity::SecretKey;
    use oxy_core::objects::StateObject;
    use oxy_core::receipt::SignedReceipt;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let mut c = client(&url, work.path());
    c.create_identity("Andy", "andy@example.com", oxy_core::repo::IdentityScope::Repo).unwrap();
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"build\" {{\n    sandbox \"alpine\"\n    exec \"make\"\n    output {{\n        artifact \"tool\" {{\n            path \"/src/out/tool\"\n        }}\n    }}\n}}\n"
        ),
    );
    c.snapshot("test").unwrap();
    c.commit("Build a tool", &[]).unwrap();
    let heads = c.bookmark_heads().unwrap();
    c.tag_create("v1", &heads, "first").unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));

    // The runner builds, uploads the artifact and signs its receipt.
    let run = runs_of(&h.app, "andy", "demo").await.remove(0);
    let a = claim(&h.app).await.unwrap();
    let id = a["job_id"].as_i64().unwrap();
    let body: Vec<u8> = b"#!/bin/sh\necho tool\n".to_vec();
    let (_, text) = upload(&h.app, id, "tool", &body).await;
    let blob = serde_json::from_str::<serde_json::Value>(&text).unwrap()["blob"].as_str().unwrap().to_string();
    let mut r = receipt("build", a["pipeline_hash"].as_str().unwrap(), 0);
    r["outputs"] = serde_json::json!([blob]);
    let heads_run: Vec<ChangeHash> = run.head_list().into_iter().filter_map(ChangeHash::from_hex).collect();
    let state = hash_object(ObjectType::State, &StateObject::new(heads_run).encode());
    let repo_id = h.app.repos.open(&RepoName::parse("andy/demo").unwrap()).unwrap().id().unwrap();
    let sk = SecretKey::generate();
    let runner = sk.identity("test-1 (OxygenFlow runner)", "", 1);
    let fr: flow_core::Receipt = serde_json::from_value(r.clone()).unwrap();
    let signed = hex::encode(SignedReceipt::sign(fr.to_oxy(), repo_id, state, &format!("{url}/flow/1"), &runner, &sk).encode());
    let done = serde_json::json!({ "exit_code": 0, "duration_ms": 900, "receipt": r, "signed_receipt": signed });
    assert_eq!(finish(&h.app, id, done).await, StatusCode::NO_CONTENT);

    // A release with the built asset and a file uploaded by hand.
    let row = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let rel = db::releases::create(&h.app.db, row.id, "v1", "One", "", false, None).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "tool", "", &body).await.unwrap();
    oxygenhub::web::releases::store_asset(&h.app, rel.id, "notes.txt", "", b"by hand\n").await.unwrap();

    let (st, page) = call(&h.app, "GET", "/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    for want in [
        "Tag v1 is signed",
        "The one change in the release is authenticated",
        "tool: signed receipt from run #1 build",
        "a runner key of this hub",
        "tool: built from the tagged state",
        "tool: pinned environment",
        "notes.txt: no build found",
        "<svg class=\"trust-graph\"",
        // oxygenhub#79: a summary of badges, badges in the graph, and the
        // graph drawn at its own size rather than stretched to the page.
        "Tag signed by",
        "1 of 1 changes signed or sealed",
        "1 of 2 assets built and attested by OxygenFlow",
        "attested build",
        "uploaded by hand",
        "receipt verified",
        "class=\"shackle\"",
        "<title>notes.txt:",
    ] {
        assert!(page.contains(want), "missing {want:?}");
    }
    let svg = &page[page.find("<svg class=\"trust-graph\"").unwrap()..];
    assert!(svg[..200].contains(" width=\"") && svg[..200].contains(" height=\""), "the graph has its own size: {}", &svg[..200]);
    let (_, releases) = call(&h.app, "GET", "/andy/demo/releases", None, None).await;
    assert!(releases.contains("/andy/demo/releases/trust/v1"), "the release links its audit");

    // The export carries the signed receipt, which checks on its own.
    let (st, json) = call(&h.app, "GET", "/andy/demo/releases/trust/v1?format=json", None, None).await;
    assert_eq!(st, StatusCode::OK);
    let t: serde_json::Value = serde_json::from_str(&json).unwrap();
    assert_eq!((t["tag"].as_str(), t["tag_signature"].as_str(), t["changes"].as_i64()), (Some("v1"), Some("valid"), Some(1)));
    assert_eq!(t["authors"][0]["who"]["name"], "Andy");
    let tool = t["assets"].as_array().unwrap().iter().find(|a| a["name"] == "tool").unwrap();
    let hex_receipt = tool["build"]["signed_receipt"].as_str().unwrap();
    let sr = SignedReceipt::decode(&hex::decode(hex_receipt).unwrap()).unwrap();
    sr.verify().unwrap();
    assert_eq!(sr.receipt.outputs[0].to_hex(), blob);
    let audit = t["audit"].as_array().unwrap();
    assert!(audit.iter().any(|c| c["verdict"] == "warn" && c["title"].as_str().unwrap().starts_with("notes.txt")));
    assert!(!audit.iter().any(|c| c["verdict"] == "fail"), "{audit:?}");
    // The API gives the same report; an unknown release is a 404.
    let (st, api) = call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v1", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert_eq!(serde_json::from_str::<serde_json::Value>(&api).unwrap()["assets"].as_array().unwrap().len(), 2);
    assert_eq!(call(&h.app, "GET", "/api/v1/repos/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
    assert_eq!(call(&h.app, "GET", "/andy/demo/releases/trust/v9", None, None).await.0, StatusCode::NOT_FOUND);
}

/// Runners (hub spec §13.2): an administrator makes a registration token,
/// a runner registers with it (binding its key), claims, shows on Admin →
/// Runners with its history, and is drained, disabled and removed.
#[tokio::test(flavor = "multi_thread")]
async fn runners_register_show_their_history_and_are_drained_and_removed() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    // A signed-in site administrator.
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=carol&email=carol%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let admin = db::accounts::account_by_name(&h.app.db, "carol").await.unwrap().unwrap();
    db::accounts::set_admin(&h.app.db, admin.id, true).await.unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/admin/runners".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("No runners yet"));
    assert_eq!(call(&h.app, "GET", "/admin/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // A one-time registration token, shown once with the command.
    let (_, page) = web("POST", "/admin/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    assert!(token.starts_with("flowreg_"), "{token}");

    // The runner registers, binding its key; the token then is spent.
    let sk = SecretKey::generate();
    let identity = sk.identity("plex (OxygenFlow runner)", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": ["big-disk"], "max_jobs": 2, "memory_mib": 16384, "cpus": 4 });
    let reg = serde_json::json!({ "token": token, "name": "plex", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    assert!(credential.starts_with("flowrunner_"));
    assert_eq!(
        call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0,
        StatusCode::FORBIDDEN,
        "used once"
    );

    // It claims with its credential; the hub.toml runner shows too.
    let claim_as = |who: String| {
        let app = h.app.clone();
        let info = info.clone();
        async move {
            let body = serde_json::json!({ "runner": "ignored", "runtimes": ["microvm"], "info": info }).to_string();
            call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await
        }
    };
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "nothing queued");
    assert_eq!(claim(&h.app).await, None);
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(
        page.contains(">plex</strong>") && page.contains("st-online") && page.contains("x86_64 · linux") && page.contains("big-disk"),
        "{page}"
    );
    assert!(page.contains(">test-1</strong>") && page.contains("hub.toml"), "the configured runner, by its claimed name");
    let plex = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex").unwrap();
    assert_eq!(plex.key, identity.hash().to_hex(), "bound at registration");

    // A run: plex takes a job, and its history shows it.
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    runs_of(&h.app, "andy", "demo").await;
    let (st, body) = claim_as(credential.clone()).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let job: serde_json::Value = serde_json::from_str(&body).unwrap();
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    assert!(
        page.contains("andy/demo") && page.contains(job["job"]["name"].as_str().unwrap()) && page.contains("Running now</dt><dd>1 job"),
        "{page}"
    );
    // The queue on Admin → Runners and through the API (#41, part 4).
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    assert!(page.contains("<h2>Queue</h2>") && page.contains("<td>plex</td>"), "{page}");
    let (_, tokens) = web("GET", "/settings/tokens".into(), None).await;
    let (_, made) = web("POST", "/settings/tokens".into(), Some(format!("_csrf={}&name=admin&scopes=user%3Aread", csrf_of(&tokens)))).await;
    let i = made.find("oxyhub_").unwrap();
    let admin_token: String = made[i..].chars().take_while(|c| c.is_ascii_alphanumeric() || *c == '_').collect();
    let (st, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();
    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();
    assert_eq!((plex_row["running"].as_i64(), plex_row["arch"].as_str()), (Some(1), Some("x86_64")));
    // A runner's own labels and time limit from its page (#98).
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    let (st, _) = web(
        "POST",
        format!("/admin/runners/{}/limits", plex.id),
        Some(format!("_csrf={}&labels=gpu%2C+plex&timeout_minutes=5", csrf_of(&page))),
    )
    .await;
    assert_eq!(st, StatusCode::SEE_OTHER);
    let (_, page) = web("GET", format!("/admin/runners/{}", plex.id), None).await;
    assert!(page.contains(">gpu<") && page.contains("5 min for its jobs"), "{page}");
    let (_, body) = call(&h.app, "GET", "/api/v1/runners", Some(&admin_token), None).await;
    let listed: serde_json::Value = serde_json::from_str(&body).unwrap();
    let plex_row = listed.as_array().unwrap().iter().find(|r| r["name"] == "plex").unwrap();
    assert!(plex_row["labels"].as_array().unwrap().iter().any(|l| l == "gpu"), "{plex_row}");
    let (st, _) = web(
        "POST",
        format!("/admin/runners/{}/limits", plex.id),
        Some(format!("_csrf={}&labels=bad%20label%21&timeout_minutes=", csrf_of(&page))),
    )
    .await;
    assert_eq!(st, StatusCode::BAD_REQUEST, "a label with a space or punctuation is refused");
    let (_, body) = call(&h.app, "GET", "/api/v1/runners/queue", Some(&admin_token), None).await;
    let queue: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert!(queue.as_array().unwrap().iter().any(|q| q["status"] == "running" && q["runner"] == "plex"), "{queue}");
    let (_, body) = call(&h.app, "GET", &format!("/api/v1/runners/{}/jobs", plex.id), Some(&admin_token), None).await;
    let jobs: serde_json::Value = serde_json::from_str(&body).unwrap();
    assert_eq!(jobs[0]["repo"], "andy/demo", "{jobs}");
    assert_eq!(call(&h.app, "GET", "/api/v1/runners", None, None).await.0, StatusCode::NOT_FOUND, "administrators only");

    // Draining: no new work; removing while busy is refused.
    let (_, page) = web("GET", "/admin/runners".into(), None).await;
    let csrf = csrf_of(&page);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=draining"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::NO_CONTENT, "a draining runner takes nothing");
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("is running jobs: drain it first"), "{page}");
    // Its job finishes; disabled, it is refused; removed, its credential is dead.
    let id = job["job_id"].as_i64().unwrap();
    let done = call(
        &h.app,
        "POST",
        &format!("/api/v1/flow/jobs/{id}/finish"),
        Some(&credential),
        Some(r#"{"exit_code":0,"duration_ms":5}"#.into()),
    )
    .await;
    assert_eq!(done.0, StatusCode::NO_CONTENT);
    web("POST", format!("/admin/runners/{}/state", plex.id), Some(format!("_csrf={csrf}&state=disabled"))).await;
    assert_eq!(claim_as(credential.clone()).await.0, StatusCode::FORBIDDEN);
    let (_, page) = web("POST", format!("/admin/runners/{}/remove", plex.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("Removed plex."), "{page}");
    assert_eq!(claim_as(credential).await.0, StatusCode::UNAUTHORIZED);
}

/// Runner scopes (oxygenhub#41): a runner registered for one repository, or
/// one owner, takes only those jobs; a site runner takes any.
#[tokio::test(flavor = "multi_thread")]
async fn scoped_runners_take_only_their_scope() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    create(&h.app, "andy", "other").await;
    let register = |scope: db::runners::Scope, name: &'static str| {
        let app = h.app.clone();
        async move {
            let token = format!("flowreg_{name}");
            db::runners::add_registration(&app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &scope).await.unwrap();
            let identity = SecretKey::generate().identity(name, "", 1);
            let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
            let reg = serde_json::json!({ "token": token, "name": name, "identity": hex::encode(identity.encode()), "info": info });
            let (st, body) = call(&app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
            assert_eq!(st, StatusCode::OK, "{body}");
            serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string()
        }
    };
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let only_demo = register(db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) }, "demo-runner").await;
    let carols = register(db::runners::Scope { owner: "carol".into(), repo_id: None }, "carol-runner").await;
    let andys = register(db::runners::Scope { owner: "andy".into(), repo_id: None }, "andy-runner").await;
    let claim_as = |who: String| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"] }).to_string();
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(&who), Some(body)).await;
            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["repo"].as_str().unwrap().to_string())
        }
    };
    // Jobs queued in andy/other only.
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/other", h.base), work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "other").await.is_empty());
    assert_eq!(claim_as(only_demo.clone()).await, None, "nothing of andy/demo is queued");
    assert_eq!(claim_as(carols).await, None, "nothing of carol's");
    assert_eq!(claim_as(andys).await.as_deref(), Some("andy/other"), "the owner's runner takes its repositories' jobs");
    // The site runner (hub.toml) takes any.
    assert_eq!(claim(&h.app).await.unwrap()["repo"], "andy/other");
    assert_eq!(claim_as(only_demo).await, None, "still nothing of its own");
}

/// Settings → Runners (oxygenhub#41, part 2b): a repository's admin adds a
/// runner for that repository alone, sees which runners serve it, and
/// removes only its own.
#[tokio::test(flavor = "multi_thread")]
async fn repository_admins_add_and_remove_their_own_runners() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    let res = oxygenhub::router(h.app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=andy%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    create(&h.app, "andy", "demo").await;
    let demo = db::repos::get(&h.app.db, "andy", "demo").await.unwrap().unwrap();
    let web = |method: &'static str, path: String, body: Option<String>| {
        let (app, cookie) = (h.app.clone(), cookie.clone());
        async move {
            let mut req = Request::builder().method(method).uri(path).header(header::COOKIE, cookie);
            if body.is_some() {
                req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
            }
            let res = oxygenhub::router(app).oneshot(req.body(body.map(Body::from).unwrap_or_else(Body::empty)).unwrap()).await.unwrap();
            let st = res.status();
            (st, String::from_utf8_lossy(&res.into_body().collect().await.unwrap().to_bytes()).into_owned())
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    let (st, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(page.contains("<h2>Runners</h2>"), "the owner sees the section");
    let (_, page) = web("POST", "/andy/demo/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("mine", "", 1);
    let info = serde_json::json!({ "arch": "x86_64", "os": "linux", "version": "0.1.0", "labels": [], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "mine", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let mine = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "mine").unwrap();
    assert_eq!(mine.scope(), db::runners::Scope { owner: "andy".into(), repo_id: Some(demo.id) });
    // The site's runner (hub.toml, from its first claim) serves it too, but is not the owner's to remove.
    claim(&h.app).await;
    let site = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "test-1").unwrap();
    let (_, page) = web("GET", "/andy/demo/settings".into(), None).await;
    assert!(page.contains("<strong>mine</strong>") && page.contains("this repository"), "{page}");
    assert!(page.contains("<strong>test-1</strong>") && page.contains("the whole site"), "{page}");
    let csrf = csrf_of(&page);
    let (st, _) = web("POST", format!("/andy/demo/settings/runners/{}/remove", site.id), Some(format!("_csrf={csrf}"))).await;
    assert_eq!(st, StatusCode::NOT_FOUND, "not the repository's own");
    let (_, page) = web("POST", format!("/andy/demo/settings/runners/{}/remove", mine.id), Some(format!("_csrf={csrf}"))).await;
    assert!(page.contains("mine removed"), "{page}");
    assert!(db::runners::get(&h.app.db, mine.id).await.unwrap().is_none());

    // The same for an organisation's owner.
    let (_, page) = web("GET", "/orgs/new".into(), None).await;
    web("POST", "/orgs/new".into(), Some(format!("_csrf={}&name=acme&display_name=Acme", csrf_of(&page)))).await;
    let (st, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert_eq!(st, StatusCode::OK);
    let (_, page) = web("POST", "/orgs/acme/settings/runners/token".into(), Some(format!("_csrf={}", csrf_of(&page)))).await;
    let i = page.find("--token ").unwrap() + "--token ".len();
    let token: String = page[i..].chars().take_while(|c| !c.is_whitespace()).collect();
    let identity = SecretKey::generate().identity("acme-runner", "", 1);
    let reg = serde_json::json!({ "token": token, "name": "acme-runner", "identity": hex::encode(identity.encode()), "info": info });
    assert_eq!(call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await.0, StatusCode::OK);
    let acme = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "acme-runner").unwrap();
    assert_eq!(acme.scope(), db::runners::Scope { owner: "acme".into(), repo_id: None });
    let (_, page) = web("GET", "/orgs/acme/settings".into(), None).await;
    assert!(page.contains("<strong>acme-runner</strong>") && page.contains("the organisation's repositories"), "{page}");
    let (_, page) = web("POST", format!("/orgs/acme/settings/runners/{}/remove", acme.id), Some(format!("_csrf={}", csrf_of(&page)))).await;
    assert!(page.contains("acme-runner removed"), "{page}");
}

/// runs-on (oxygenhub#41, part 3): a job that asks for a label goes only to
/// a runner that has it, and waits for one otherwise.
#[tokio::test(flavor = "multi_thread")]
async fn jobs_that_ask_for_labels_wait_for_a_runner_with_them() {
    use oxy_core::identity::SecretKey;
    let h = hub().await;
    create(&h.app, "andy", "demo").await;
    let work = tempfile::tempdir().unwrap();
    let c = client(&format!("{}/andy/demo", h.base), work.path());
    let cross = "job \"cross\" {\n    sandbox \"alpine\"\n    runs-on \"riscv64\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(cross));
    c.snapshot("t").unwrap();
    c.commit("pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert!(!runs_of(&h.app, "andy", "demo").await.is_empty());
    // The site runner has no labels: it takes the others, never cross.
    let mut taken = Vec::new();
    while let Some(j) = claim(&h.app).await {
        taken.push(j["job"]["name"].as_str().unwrap().to_string());
    }
    assert!(!taken.contains(&"cross".to_string()) && taken.contains(&"lint".to_string()), "{taken:?}");
    let queue = db::flow::queue(&h.app.db).await.unwrap();
    let waiting = queue.iter().find(|q| q.job == "cross").unwrap();
    assert_eq!((waiting.status.as_str(), waiting.needs()), ("queued", vec!["riscv64".to_string()]));
    // A runner registered with the label takes it.
    let token = "flowreg_riscv";
    db::runners::add_registration(&h.app.db, &blake3::hash(token.as_bytes()).to_hex(), "admin", &Default::default()).await.unwrap();
    let identity = SecretKey::generate().identity("visionfive", "", 1);
    let info = serde_json::json!({ "arch": "riscv64", "os": "linux", "version": "0.1.0", "labels": ["riscv64"], "max_jobs": 1 });
    let reg = serde_json::json!({ "token": token, "name": "visionfive", "identity": hex::encode(identity.encode()), "info": info });
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/runners/register", None, Some(reg.to_string())).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let credential = serde_json::from_str::<serde_json::Value>(&body).unwrap()["credential"].as_str().unwrap().to_string();
    let claim_body = serde_json::json!({ "runner": "x", "runtimes": ["microvm"], "info": info }).to_string();
    let (st, body) = call(&h.app, "POST", "/api/v1/flow/claim", Some(&credential), Some(claim_body)).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    assert_eq!(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"], "cross");
}

/// oxygenhub#71: a job its runner stopped reporting (restarted, or lost
/// the hub as the job ended) is ended once it has run past the time limit
/// and a grace: an error with the reason, its dependents skipped and its
/// run settled, as if the runner had reported the error.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_no_runner_reports_is_ended_after_the_time_limit() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "lost").await;
    let url = format!("{}/andy/lost", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "lost").await.len(), 1);
    let job = claim(&h.app).await.expect("a job to claim");
    let id = job["job_id"].as_i64().unwrap();

    // Running, but well within the limit: left alone.
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 0);
    // Started long ago and never reported: ended.
    let old = db::now_ms() - 24 * 3600 * 1000;
    sqlx::query("UPDATE flow_jobs SET started_ms = ? WHERE id = ?").bind(old).bind(id).execute(&h.app.db).await.unwrap();
    assert_eq!(oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap(), 1);
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "lost").await[0];
    let jobs = db::flow::jobs(&h.app.db, run.id).await.unwrap();
    assert!(jobs.iter().all(|j| j.status != "running"), "nothing left running");
    // A late report, should one come, does not overwrite the ending.
    assert_ne!(finish(&h.app, id, serde_json::json!({ "exit_code": 0 })).await, StatusCode::NO_CONTENT);
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "error");
}

/// oxygenhub#73: a runner that lists the images it holds is offered only
/// the jobs whose rootfs is among them; one that does not say gets any.
#[tokio::test(flavor = "multi_thread")]
async fn a_runner_is_offered_only_jobs_whose_image_it_holds() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "images").await;
    let url = format!("{}/andy/images", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "images").await.len(), 1);
    let claim_with = |images: Vec<String>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "images": images }).to_string();
        let app = h.app.clone();
        async move {
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await;
            (st, body)
        }
    };
    let (st, _) = claim_with(vec!["ab".repeat(32)]).await;
    assert_eq!(st, StatusCode::NO_CONTENT, "no job for a runner without the image");
    let jobs = db::flow::jobs(&h.app.db, runs_of(&h.app, "andy", "images").await[0].id).await.unwrap();
    let rootfs: serde_json::Value = serde_json::from_str(&jobs[0].sandbox).unwrap();
    let rootfs = rootfs["rootfs"].as_str().unwrap().to_string();
    let (st, body) = claim_with(vec!["ab".repeat(32), rootfs]).await;
    assert_eq!(st, StatusCode::OK, "{body}");
}

/// oxygenhub#71: a runner that lists the jobs it holds has the others it
/// was given ended at once (it restarted, or never received them), not at
/// the time limit; a held job is left alone.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_its_runner_no_longer_holds_is_ended_at_its_next_claim() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "held").await;
    let url = format!("{}/andy/held", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "held").await.len(), 1);
    let id = claim(&h.app).await.expect("a job to claim")["job_id"].as_i64().unwrap();
    let claim_holding = |holding: Vec<i64>| {
        let body = serde_json::json!({ "runner": "test-1", "runtimes": ["microvm"], "holding": holding }).to_string();
        let app = h.app.clone();
        async move { call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body)).await.0 }
    };
    claim_holding(vec![id]).await;
    assert_eq!(db::flow::job(&h.app.db, id).await.unwrap().unwrap().status, "running", "a held job stays");
    claim_holding(vec![]).await;
    let j = db::flow::job(&h.app.db, id).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert!(j.error.starts_with("lost:") && j.error.contains("no longer holds"), "{}", j.error);
    let run = &runs_of(&h.app, "andy", "held").await[0];
    assert!(db::flow::jobs(&h.app.db, run.id).await.unwrap().iter().all(|j| j.status != "running"), "the run settles");
}

/// oxygen#59: a job with a `scope` runs on a push only when the push's
/// changes touch a path it names (a release job, when Cargo.toml's version
/// moves); a bookmark's first run and manual runs have every job.
#[tokio::test(flavor = "multi_thread")]
async fn a_scoped_job_runs_only_when_a_push_touches_its_paths() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "scoped").await;
    let url = format!("{}/andy/scoped", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\" \"packaging/**\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    write(work.path(), "Cargo.toml", "version = \"0.1.0\"\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let names = |app: Shared, n: usize| async move {
        let mut runs = Vec::new();
        for _ in 0..100 {
            runs = runs_of(&app, "andy", "scoped").await;
            if runs.len() >= n {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert_eq!(runs.len(), n, "runs");
        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
        j.sort();
        // Passed, so the next push is scoped against it, not superseding it.
        sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = ?").bind(runs[0].id).execute(&app.db).await.unwrap();
        j
    };
    assert_eq!(names(h.app.clone(), 1).await, ["build", "dist", "lint", "test"], "the first run has every job");
    write(work.path(), "test.sh", "echo still ok\n");
    c.commit("Edit the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(names(h.app.clone(), 2).await, ["build", "lint", "test"], "dist is out of scope");
    write(work.path(), "Cargo.toml", "version = \"0.1.1\"\n");
    c.commit("Release 0.1.1", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(names(h.app.clone(), 3).await, ["build", "dist", "lint", "test"], "a version bump runs dist");
}

/// oxygenflow#32: a run still going is cancelled from the API, the web or
/// by a newer push to its bookmark; its runner is refused from then on,
/// and the next push is scoped past the cancelled run.
#[tokio::test(flavor = "multi_thread")]
async fn runs_are_cancelled_by_hand_or_by_a_newer_push() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "stop").await;
    let url = format!("{}/andy/stop", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"dist\" {\n    sandbox \"alpine\"\n    scope \"Cargo.toml\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    write(work.path(), "Cargo.toml", "version = \"0.1.0\"\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let wait_runs = |app: Shared, n: usize| async move {
        for _ in 0..100 {
            let runs = runs_of(&app, "andy", "stop").await;
            if runs.len() >= n {
                return runs;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        panic!("no run {n}");
    };
    let first = wait_runs(h.app.clone(), 1).await;
    sqlx::query("UPDATE flow_runs SET status = 'passed' WHERE id = ?").bind(first[0].id).execute(&h.app.db).await.unwrap();
    sqlx::query("UPDATE flow_jobs SET status = 'passed' WHERE run_id = ?").bind(first[0].id).execute(&h.app.db).await.unwrap();

    // Run 2 (a version bump) has a job running when run 3 is pushed.
    write(work.path(), "Cargo.toml", "version = \"0.1.1\"\n");
    c.commit("Release 0.1.1", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    wait_runs(h.app.clone(), 2).await;
    let job = claim(&h.app).await.expect("a job of run 2");
    let id = job["job_id"].as_i64().unwrap();
    write(work.path(), "test.sh", "echo still ok\n");
    c.commit("Edit the test", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = wait_runs(h.app.clone(), 3).await;
    let third = &runs[0];
    let mut second = runs[1].clone();
    for _ in 0..100 {
        second = db::flow::run_by_id(&h.app.db, runs[1].id).await.unwrap().unwrap();
        if !second.live() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert_eq!(second.status, "cancelled", "a newer push supersedes a run still going");
    let jobs = db::flow::jobs(&h.app.db, second.id).await.unwrap();
    assert!(jobs.iter().all(|j| j.status == "cancelled" && j.error == "superseded by run #3"), "{jobs:?}");
    let mut names: Vec<String> = db::flow::jobs(&h.app.db, third.id).await.unwrap().into_iter().map(|j| j.name).collect();
    names.sort();
    assert_eq!(names, ["build", "dist", "lint", "test"], "run 3 also checks what run 2 would have: the version bump");
    let mut checks = vec![];
    for _ in 0..100 {
        checks = db::hooks::checks_for_state(&h.app.db, second.repo_id, &second.state).await.unwrap();
        if checks.iter().all(|c| c.status == "cancelled") {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(50)).await;
    }
    assert!(checks.iter().all(|c| c.status == "cancelled"), "{checks:?}");
    // The runner of the cancelled job is refused, so it stops the job.
    let (st, _) =
        call(&h.app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(r#"{"records":[]}"#.into())).await;
    assert_eq!(st, StatusCode::CONFLICT);

    // By hand: the API, then a run already ended is refused.
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", Some("site"), None).await;
    assert_eq!(st, StatusCode::OK, "{body}");
    let third = db::flow::run_by_id(&h.app.db, third.id).await.unwrap().unwrap();
    assert_eq!(third.status, "cancelled");
    assert!(db::flow::jobs(&h.app.db, third.id).await.unwrap().iter().all(|j| j.error == "cancelled by andy"));
    let (st, body) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", Some("site"), None).await;
    assert!(st == StatusCode::CONFLICT && body.contains("already cancelled"), "{st} {body}");
    let (st, _) = call(&h.app, "POST", "/api/v1/repos/andy/stop/flow/runs/3/cancel", None, None).await;
    assert_ne!(st, StatusCode::OK, "anonymous callers cannot cancel");
    let (_, page) = call(&h.app, "GET", "/andy/stop/flow/3", None, None).await;
    assert!(page.contains("⊘") && !page.contains("CANCEL</button>"), "a cancelled run shows so, with no button");
}

/// oxygenflow#23 and oxygenhub#89: a job runs under its pipeline's
/// timeout, capped at the site's maximum, else the site's default; and the
/// hub gives a job up only past its own limit.
#[tokio::test(flavor = "multi_thread")]
async fn a_job_runs_under_its_own_time_limit_within_the_sites() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_DEFAULT, "30").await.unwrap();
    db::accounts::set_setting(&h.app.db, oxygenhub::flow::SETTING_MAX, "120").await.unwrap();
    create(&h.app, "andy", "limits").await;
    let url = format!("{}/andy/limits", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let extra = "job \"slow\" {\n    sandbox \"alpine\"\n    timeout \"90m\"\n    exec \"true\"\n}\njob \"huge\" {\n    sandbox \"alpine\"\n    timeout \"5h\"\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "limits").await.len(), 1);
    let mut limits = std::collections::HashMap::new();
    let mut ids = std::collections::HashMap::new();
    while let Some(j) = claim(&h.app).await {
        let name = j["job"]["name"].as_str().unwrap().to_string();
        limits.insert(name.clone(), j["timeout_s"].as_u64().unwrap());
        ids.insert(name, j["job_id"].as_i64().unwrap());
    }
    assert_eq!(limits["slow"], 5400, "its own");
    assert_eq!(limits["huge"], 7200, "cut to the site's maximum");
    assert_eq!(limits["build"], 1800, "the site's default");
    // 40 minutes in: past the default and its grace, within slow's limit.
    let started = db::now_ms() - 40 * 60 * 1000;
    sqlx::query("UPDATE flow_jobs SET started_ms = ?").bind(started).execute(&h.app.db).await.unwrap();
    oxygenhub::flow::reap_lost_jobs(&h.app).await.unwrap();
    let status = |id: i64| {
        let app = h.app.clone();
        async move { db::flow::job(&app.db, id).await.unwrap().unwrap().status }
    };
    assert_eq!(status(ids["build"]).await, "error", "past its 30 minutes");
    assert_eq!(status(ids["slow"]).await, "running", "within its 90");
    assert_eq!(status(ids["huge"]).await, "running", "within its 120");
}

/// oxygenflow#24: a job's `on` decides which events start it: a pushed
/// tag runs only the jobs that name it; a push to a bookmark the pipeline
/// does not name starts no run; a rerun repeats its run's event.
#[tokio::test(flavor = "multi_thread")]
async fn on_decides_which_events_start_a_job() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "events").await;
    let url = format!("{}/andy/events", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    let text = format!(
        "on {{\n    push \"main\"\n    manual\n}}\n{}job \"dist\" {{\n    sandbox \"alpine\"\n    on {{\n        tag \"v*\"\n    }}\n    exec \"true\"\n}}\n",
        pipeline("")
    );
    // `on` must come after schema-version; pipeline() starts with it.
    let text = text.replacen(
        "on {\n    push \"main\"\n    manual\n}\nschema-version 1\n",
        "schema-version 1\non {\n    push \"main\"\n    manual\n}\n",
        1,
    );
    write(work.path(), ".flow/pipeline.kdl", &text);
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let jobs_of_latest = |app: Shared, n: usize| async move {
        let mut runs = Vec::new();
        for _ in 0..100 {
            runs = runs_of(&app, "andy", "events").await;
            if runs.len() >= n {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        assert_eq!(runs.len(), n, "runs");
        let mut j: Vec<String> = db::flow::jobs(&app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
        j.sort();
        (runs[0].clone(), j)
    };
    let (_, j) = jobs_of_latest(h.app.clone(), 1).await;
    assert_eq!(j, ["build", "lint", "test"], "a push to main: no dist");
    // A tag: only dist.
    let heads = c.bookmark_heads().unwrap();
    c.tag_create("v1.0.0", &heads, "first").unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let (run, j) = jobs_of_latest(h.app.clone(), 2).await;
    assert_eq!(j, ["dist"], "a tag runs only what names it");
    assert_eq!(run.trigger, "tag:v1.0.0");
    // A rerun repeats the tag's event.
    let row = db::repos::get(&h.app.db, "andy", "events").await.unwrap().unwrap();
    db::flow::finish(
        &h.app.db,
        db::flow::jobs(&h.app.db, run.id).await.unwrap()[0].id,
        db::flow::Outcome { status: "passed", exit_code: Some(0), error: "", receipt: "", receipt_hash: "" },
    )
    .await
    .unwrap();
    let run = db::flow::run_by_id(&h.app.db, run.id).await.unwrap().unwrap();
    oxygenhub::flow::rerun(&h.app, &row, &run, "andy").await.unwrap();
    let (_, j) = jobs_of_latest(h.app.clone(), 3).await;
    assert_eq!(j, ["dist"], "the rerun is the tag's run again");
    // A bookmark the pipeline does not name starts nothing.
    c.bookmark_create("develop", None).unwrap();
    assert!(c.push_bookmark("origin", "develop", false).unwrap().iter().all(|r| r.ok));
    tokio::time::sleep(std::time::Duration::from_millis(300)).await;
    assert_eq!(runs_of(&h.app, "andy", "events").await.len(), 3, "no run for develop");
}

/// oxygenflow#24: a job with a `schedule` line runs on the default
/// bookmark when the line comes due, and only then; other jobs stay out.
#[tokio::test(flavor = "multi_thread")]
async fn scheduled_jobs_run_when_due() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "nightly").await;
    let url = format!("{}/andy/nightly", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    // An odd minute, so the hub's own ticker cannot fire it during the test.
    let extra = "job \"nightly\" {\n    sandbox \"alpine\"\n    on {\n        schedule \"17 4 29 2 *\"\n    }\n    exec \"true\"\n}\n";
    write(work.path(), ".flow/pipeline.kdl", &pipeline(extra));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Start", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    assert_eq!(runs_of(&h.app, "andy", "nightly").await.len(), 1);
    let first = runs_of(&h.app, "andy", "nightly").await;
    let names: Vec<String> = db::flow::jobs(&h.app.db, first[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
    assert!(!names.contains(&"nightly".to_string()), "a push does not run a scheduled job");
    use flow_core::pipeline::When;
    let not_yet = When { minute: 16, hour: 4, day: 29, month: 2, weekday: 4 };
    assert_eq!(oxygenhub::flow::run_due(&h.app, not_yet).await, 0);
    let due = When { minute: 17, ..not_yet };
    assert_eq!(oxygenhub::flow::run_due(&h.app, due).await, 1);
    let runs = runs_of(&h.app, "andy", "nightly").await;
    assert_eq!(runs.len(), 2);
    assert_eq!(runs[0].trigger, "schedule");
    let names: Vec<String> = db::flow::jobs(&h.app.db, runs[0].id).await.unwrap().into_iter().map(|j| j.name).collect();
    assert_eq!(names, ["nightly"]);
    assert_eq!(runs[0].bookmark, "main", "on the default bookmark");
}

#[tokio::test(flavor = "multi_thread")]
async fn variables_from_three_levels_reach_a_job_inner_overriding_outer() {
    use oxygenhub::levels::Scope;
    use oxygenhub::variables;
    let h = hub().await;
    let app = &h.app;
    // Andy, the site's administrator, owns the acme organisation.
    let res = oxygenhub::router(app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=a%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let andy = db::accounts::account_by_name(&app.db, "andy").await.unwrap().unwrap();
    db::accounts::create_org(&app.db, "acme", "Acme", andy.id).await.unwrap();
    create(app, "acme", "app").await;
    let page = |path: &str, form: Option<String>| {
        let mut req =
            Request::builder().method(if form.is_some() { "POST" } else { "GET" }).uri(path).header(header::COOKIE, cookie.clone());
        if form.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        let req = req.body(Body::from(form.unwrap_or_default())).unwrap();
        let app = app.clone();
        async move {
            let res = oxygenhub::router(app).oneshot(req).await.unwrap();
            let status = res.status();
            let body = String::from_utf8(res.into_body().collect().await.unwrap().to_bytes().to_vec()).unwrap();
            (status, body)
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };

    // The site sets CC and LANG, the organisation its own CC, the
    // repository RUST_LOG from its settings page.
    variables::set(&app.db, Scope::Site, "CC", "gcc").await.unwrap();
    variables::set(&app.db, Scope::Site, "LANG", "C").await.unwrap();
    let org = db::accounts::org_by_name(&app.db, "acme").await.unwrap().unwrap();
    variables::set(&app.db, Scope::Org(org.id), "CC", "clang").await.unwrap();
    let (st, settings) = page("/acme/app/settings", None).await;
    assert_eq!(st, StatusCode::OK, "{settings}");
    assert!(settings.contains("Inherited") && settings.contains("the acme organisation") && settings.contains("overridden"), "{settings}");
    let csrf = csrf_of(&settings);
    let (_, body) = page("/acme/app/settings/variables", Some(format!("_csrf={csrf}&action=set&name=RUST_LOG&value=debug"))).await;
    assert!(body.contains("RUST_LOG set.") && body.contains("<code>debug</code>"), "{body}");
    let (_, body) = page("/acme/app/settings/variables", Some(format!("_csrf={csrf}&action=set&name=FLOW_REPO&value=x"))).await;
    assert!(body.contains("not FLOW_"), "{body}");
    let repo = db::repos::get(&app.db, "acme", "app").await.unwrap().unwrap();
    db::repos::set_flow(&app.db, repo.id, "on").await.unwrap();
    let env = variables::env_for_repo(&app.db, &repo).await.unwrap();
    assert_eq!(
        env.iter().map(|(k, v)| (k.as_str(), v.as_str())).collect::<Vec<_>>(),
        [("CC", "clang"), ("LANG", "C"), ("RUST_LOG", "debug")],
        "the organisation's CC wins over the site's"
    );

    // A claim carries them.
    let url = format!("{}/acme/app", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = runs_of(app, "acme", "app").await;
    assert_eq!(runs.len(), 1, "{runs:?}");
    let got = claim(app).await.unwrap_or_else(|| panic!("a job: {:?}", runs[0]));
    assert_eq!(got["env"], serde_json::json!({ "CC": "clang", "LANG": "C", "RUST_LOG": "debug" }), "{got}");

    // Removing a site variable from the admin page takes effect at the next claim.
    let (_, admin) = page("/admin", None).await;
    assert!(admin.contains("Variables") && admin.contains("<code>gcc</code>"), "{admin}");
    let (_, body) = page("/admin/variables", Some(format!("_csrf={}&action=delete&name=LANG", csrf_of(&admin)))).await;
    assert!(body.contains("LANG removed."), "{body}");
    let got = claim(app).await.expect("another job");
    assert_eq!(got["env"], serde_json::json!({ "CC": "clang", "RUST_LOG": "debug" }), "{got}");
}

#[tokio::test(flavor = "multi_thread")]
async fn secrets_a_job_declares_come_with_the_claim_and_never_reach_the_log() {
    let h = hub().await;
    let app = &h.app;
    let res = oxygenhub::router(app.clone())
        .oneshot(
            Request::post("/register")
                .header(header::CONTENT_TYPE, "application/x-www-form-urlencoded")
                .body(Body::from("username=andy&email=a%40example.com&password=correct-horse-battery"))
                .unwrap(),
        )
        .await
        .unwrap();
    let cookie = res.headers().get(header::SET_COOKIE).unwrap().to_str().unwrap().split(';').next().unwrap().to_string();
    let page = |path: &str, form: Option<String>| {
        let mut req =
            Request::builder().method(if form.is_some() { "POST" } else { "GET" }).uri(path).header(header::COOKIE, cookie.clone());
        if form.is_some() {
            req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        }
        let req = req.body(Body::from(form.unwrap_or_default())).unwrap();
        let app = app.clone();
        async move {
            let res = oxygenhub::router(app).oneshot(req).await.unwrap();
            let status = res.status();
            let body = String::from_utf8(res.into_body().collect().await.unwrap().to_bytes().to_vec()).unwrap();
            (status, body)
        }
    };
    let csrf_of = |page: &str| {
        let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
        page[i..].split('"').next().unwrap().to_string()
    };
    create(app, "andy", "demo").await;
    // The site sets deploy-token; the repository overrides it. Neither
    // page shows a value, only that it is set.
    let (_, admin) = page("/admin", None).await;
    let csrf = csrf_of(&admin);
    let (_, body) = page("/admin/secrets", Some(format!("_csrf={csrf}&action=set&name=deploy-token&value=site-level-value"))).await;
    assert!(body.contains("deploy-token set.") && !body.contains("site-level-value"), "{body}");
    let (_, body) = page("/admin/secrets", Some(format!("_csrf={csrf}&action=set&name=bad+name&value=x"))).await;
    assert!(body.contains("name is 1 to 64 of"), "{body}");
    let (_, settings) = page("/andy/demo/settings", None).await;
    assert!(
        settings.contains("Secrets") && settings.contains("<code>deploy-token</code>") && settings.contains("the site</td>"),
        "{settings}"
    );
    let csrf_r = csrf_of(&settings);
    let (_, body) =
        page("/andy/demo/settings/secrets", Some(format!("_csrf={csrf_r}&action=set&name=deploy-token&value=hunter22-repo-value"))).await;
    assert!(body.contains("deploy-token set.") && body.contains("overridden") && !body.contains("hunter22"), "{body}");
    let repo = db::repos::get(&app.db, "andy", "demo").await.unwrap().unwrap();
    assert!(
        !String::from_utf8_lossy(&oxygenhub::secrets::list(&app.db, oxygenhub::levels::Scope::Repo(repo.id)).await.unwrap()[0].value)
            .contains("hunter22"),
        "sealed at rest"
    );

    // A job that declares the secret, and one the levels do not hold.
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &pipeline("job \"deploy\" {\n    sandbox \"alpine\"\n    secret \"deploy-token\" {\n        mount-path \"/run/secrets/token\"\n    }\n    secret \"missing\" {\n        mount-path \"/run/secrets/missing\"\n    }\n    exec \"deploy\"\n}\n"),
    );
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let runs = runs_of(app, "andy", "demo").await;
    let jobs = match runs.first() {
        Some(r) => db::flow::jobs(&app.db, r.id).await.unwrap().iter().map(|j| format!("{}:{}", j.name, j.status)).collect::<Vec<_>>(),
        None => vec![],
    };
    let mut deploy = None;
    for _ in 0..50 {
        let Some(got) = claim(app).await else {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            continue;
        };
        if got["job"]["name"] == "deploy" {
            deploy = Some(got);
            break;
        }
        assert_eq!(got["secrets"], serde_json::Value::Null, "{got}");
    }
    let got = deploy.unwrap_or_else(|| panic!("the deploy job: runs {runs:?} jobs {jobs:?}"));
    assert_eq!(
        got["secrets"],
        serde_json::json!([{ "name": "deploy-token", "path": "/run/secrets/token", "value": "hunter22-repo-value" }]),
        "the repository's value, the missing one left out: {got}"
    );
    let id = got["job_id"].as_i64().unwrap();
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    assert!(log.iter().any(|l| l.stream == "flow" && l.text.contains("secret \"missing\" is not set")), "{log:?}");

    // The value never reaches the stored log, plain or base64.
    let records = serde_json::json!({ "records": [
        { "t": 1, "stream": "out", "text": "token=hunter22-repo-value" },
        { "t": 2, "stream": "err", "text": "b64 aHVudGVyMjItcmVwby12YWx1ZQ== end" },
    ] });
    let (st, _) = call(app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(records.to_string())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    let texts: Vec<&str> = log.iter().map(|l| l.text.as_str()).collect();
    assert!(texts.contains(&"token=***") && texts.contains(&"b64 *** end"), "{texts:?}");
    assert!(!texts.iter().any(|t| t.contains("hunter22")), "{texts:?}");
}

#[tokio::test(flavor = "multi_thread")]
async fn a_job_s_identity_is_a_signed_token_the_discovery_document_lets_anyone_verify() {
    let h = hub().await;
    let app = &h.app;
    let (st, disc) = call(app, "GET", "/.well-known/openid-configuration", None, None).await;
    assert_eq!(st, StatusCode::OK, "{disc}");
    let disc: serde_json::Value = serde_json::from_str(&disc).unwrap();
    assert_eq!(disc["issuer"], h.base.as_str());
    assert_eq!(disc["jwks_uri"], format!("{}/.well-known/jwks.json", h.base));
    let (st, jwks) = call(app, "GET", "/.well-known/jwks.json", None, None).await;
    assert_eq!(st, StatusCode::OK, "{jwks}");
    let jwks: serde_json::Value = serde_json::from_str(&jwks).unwrap();
    assert_eq!(jwks["keys"][0]["alg"], "RS256");
    let kid = jwks["keys"][0]["kid"].as_str().unwrap().to_string();

    create(app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &pipeline("job \"deploy\" {\n    sandbox \"alpine\"\n    identity \"sts.amazonaws.com\" {\n        mount-path \"/run/secrets/aws-token\"\n    }\n    exec \"deploy\"\n}\n"),
    );
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let mut deploy = None;
    for _ in 0..50 {
        let Some(got) = claim(app).await else {
            tokio::time::sleep(std::time::Duration::from_millis(100)).await;
            continue;
        };
        if got["job"]["name"] == "deploy" {
            deploy = Some(got);
            break;
        }
    }
    let got = deploy.expect("the deploy job");
    let secrets = got["secrets"].as_array().expect("secrets");
    assert_eq!(secrets.len(), 1);
    assert_eq!(
        (secrets[0]["name"].as_str(), secrets[0]["path"].as_str()),
        (Some("identity:sts.amazonaws.com"), Some("/run/secrets/aws-token"))
    );
    let token = secrets[0]["value"].as_str().unwrap().to_string();
    let issuer = app.oidc().await.unwrap();
    assert_eq!(issuer.kid, kid, "the key the JWKS publishes signed it");
    let claims = issuer.verify(&token, "sts.amazonaws.com", db::now_ms()).expect("verifies");
    assert_eq!(claims["sub"], "repo:andy/demo:bookmark:main");
    assert_eq!(claims["repository"], "andy/demo");
    assert_eq!(claims["job"], "deploy");
    assert_eq!(claims["run_number"], 1);
    assert_eq!(claims["runner"], "test-1");
    assert_eq!(claims["iss"], h.base.as_str());
    assert!(claims["state"].as_str().unwrap().len() == 64);
    // A job that prints its token leaves *** in the log.
    let id = got["job_id"].as_i64().unwrap();
    let records = serde_json::json!({ "records": [{ "t": 1, "stream": "out", "text": format!("TOKEN={token} end") }] });
    let (st, _) = call(app, "POST", &format!("/api/v1/flow/jobs/{id}/log"), Some("runner-secret"), Some(records.to_string())).await;
    assert_eq!(st, StatusCode::NO_CONTENT);
    let log = db::flow::log(&app.db, id, 0, 100).await.unwrap();
    assert!(log.iter().any(|l| l.text == "TOKEN=*** end"), "{log:?}");
}

#[tokio::test(flavor = "multi_thread")]
async fn jobs_are_offered_only_to_runners_of_their_architecture() {
    let h = hub().await;
    let app = &h.app;
    create(app, "andy", "demo").await;
    let url = format!("{}/andy/demo", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline("sandbox \"board\" {\n    rootfs \"cas://blob/{ROOTFS}\"\n    arch \"riscv64\"\n}\njob \"flash\" {\n    sandbox \"board\"\n    exec \"flash\"\n}\n").replace("{ROOTFS}", &"ab".repeat(32)));
    write(work.path(), "test.sh", "echo ok\n");
    c.snapshot("test").unwrap();
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let claim_as = |arch: &'static str, archs: serde_json::Value| {
        let app = app.clone();
        async move {
            let body = serde_json::json!({
                "runner": format!("{arch}-1"), "runtimes": ["microvm"],
                "info": { "arch": arch, "os": "linux", "version": "t", "labels": [], "max_jobs": 1 },
                "image_archs": archs,
            });
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some("runner-secret"), Some(body.to_string())).await;
            match st {
                StatusCode::NO_CONTENT => None,
                StatusCode::OK => {
                    Some(serde_json::from_str::<serde_json::Value>(&body).unwrap()["job"]["name"].as_str().unwrap().to_string())
                }
                other => panic!("claim: {other} {body}"),
            }
        }
    };
    // An x86_64 runner that says the alpine image is x86_64: it takes the
    // alpine jobs, never the one that asks for riscv64.
    let rootfs = pipeline("").split("cas://blob/").nth(1).unwrap()[..64].to_string();
    for _ in 0..50 {
        if !runs_of(app, "andy", "demo").await.is_empty() {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
    let runs = runs_of(app, "andy", "demo").await;
    assert_eq!(runs.len(), 1, "{runs:?}");
    let mut taken = vec![];
    while let Some(j) = claim_as("x86_64", serde_json::json!({ rootfs.clone(): "x86_64" })).await {
        taken.push(j);
    }
    assert_eq!(taken, ["build", "lint"], "test waits for build; flash is for riscv64");
    // A riscv64 runner: the alpine image is known to be x86_64, so only flash.
    assert_eq!(claim_as("riscv64", serde_json::json!({})).await.as_deref(), Some("flash"));
    assert_eq!(claim_as("riscv64", serde_json::json!({})).await, None);
    let (_, queue) = call(app, "GET", "/admin/runners", None, None).await;
    let _ = queue;
}

/// OxygenFlow spec §11.5: a runner hands back a job it cannot run. The
/// job is queued again and not offered to that runner again, the decline
/// shows on the run's page and is kept for the runner's, and once three
/// runners have declined it the job ends as unplaceable with their reasons.
#[tokio::test(flavor = "multi_thread")]
async fn a_declined_job_goes_to_another_runner_or_ends_unplaceable() {
    let h = hub().await;
    db::accounts::create_account(&h.app.db, "andy", "a@example.com", "x", "salt", true).await.unwrap();
    create(&h.app, "andy", "declines").await;
    let url = format!("{}/andy/declines", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(work.path(), ".flow/pipeline.kdl", &pipeline(""));
    write(work.path(), "test.sh", "echo ok\n");
    c.commit("Add a pipeline", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "declines").await[0].clone();
    let claim_as = |token: &'static str, name: &'static str| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "runner": name, "runtimes": ["microvm"] }).to_string();
            let (st, body) = call(&app, "POST", "/api/v1/flow/claim", Some(token), Some(body)).await;
            (st == StatusCode::OK).then(|| serde_json::from_str::<serde_json::Value>(&body).unwrap()["job_id"].as_i64().unwrap())
        }
    };
    let decline = |token: &'static str, id: i64, reason: &'static str| {
        let app = h.app.clone();
        async move {
            let body = serde_json::json!({ "reason": reason }).to_string();
            call(&app, "POST", &format!("/api/v1/flow/jobs/{id}/decline"), Some(token), Some(body)).await.0
        }
    };
    let job = claim_as("runner-secret", "earth-1").await.expect("a job");
    // The other runners are known to the hub before they decline anything.
    assert_eq!(decline("runner-two", job, "not mine").await, StatusCode::CONFLICT, "only the runner it runs on may decline it");
    assert_eq!(decline("runner-secret", job, "the MicroVM tier needs /dev/kvm").await, StatusCode::NO_CONTENT);
    let j = db::flow::job(&h.app.db, job).await.unwrap().unwrap();
    assert_eq!((j.status.as_str(), j.runner.as_str()), ("queued", ""));
    assert_eq!(j.waiting, "declined by earth-1: the MicroVM tier needs /dev/kvm; waiting for another runner");
    // earth-1 is offered the other ready job, not this one, while it stays the same.
    let next = claim_as("runner-secret", "earth-1").await.expect("the other ready job");
    assert_ne!(next, job);
    let page = call(&h.app, "GET", &format!("/andy/declines/flow/{}", run.number), None, None).await.1;
    assert!(page.contains("Waiting: declined by earth-1: the MicroVM tier needs /dev/kvm"), "{page}");
    assert_eq!(claim_as("runner-two", "plex-1").await, Some(job), "another runner takes it");
    assert_eq!(decline("runner-two", job, "rootfs 36fc4e323804 is not on this runner").await, StatusCode::NO_CONTENT);
    assert_eq!(claim_as("runner-three", "narvi").await, Some(job));
    assert_eq!(decline("runner-three", job, "rootfs 36fc4e323804 is for x86_64; this runner is riscv64").await, StatusCode::NO_CONTENT);
    let j = db::flow::job(&h.app.db, job).await.unwrap().unwrap();
    assert_eq!(j.status, "error");
    assert_eq!(
        j.error,
        "unplaceable: declined by 3 runners: earth-1: the MicroVM tier needs /dev/kvm; plex-1: rootfs 36fc4e323804 is not on this runner; narvi: rootfs 36fc4e323804 is for x86_64; this runner is riscv64"
    );
    let r = db::runners::list(&h.app.db).await.unwrap().into_iter().find(|r| r.name == "plex-1").unwrap();
    let declined = db::flow::declines_of_runner(&h.app.db, r.id, 10).await.unwrap();
    assert_eq!(declined.len(), 1);
    assert_eq!((declined[0].job_id, declined[0].reason.as_str()), (job, "rootfs 36fc4e323804 is not on this runner"));
    assert_eq!(r.tier_list(), vec!["microvm"]);
}

/// oxygenhub#113: a job's artifact named `coverage` (LCOV) is read when
/// it is uploaded. The job's log says what it found; the Coverage tab,
/// the file view, the badge and the run page show it.
#[tokio::test(flavor = "multi_thread")]
async fn a_coverage_artifact_becomes_a_coverage_report() {
    let h = hub().await;
    create(&h.app, "andy", "cov").await;
    let url = format!("{}/andy/cov", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"coverage\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"coverage\"\n    output {{\n        artifact \"coverage\" {{\n            path \"/src/lcov.info\"\n        }}\n    }}\n}}\n"
        ),
    );
    write(work.path(), "src/lib.rs", "pub fn add(a: u32, b: u32) -> u32 {\n    a + b\n}\n\npub fn unused() {\n    todo!()\n}\n");
    c.snapshot("test").unwrap();
    c.commit("Code and coverage", &[]).unwrap();
    assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
    let run = runs_of(&h.app, "andy", "cov").await.remove(0);
    assert!(!call(&h.app, "GET", "/andy/cov", None, None).await.1.contains(">Coverage</a>"), "no tab before a report");
    let id = claim(&h.app).await.unwrap()["job_id"].as_i64().unwrap();

    let (st, text) = upload(&h.app, id, "coverage", b"this is not a report").await;
    assert_eq!(st, StatusCode::OK, "a bad report never fails the upload: {text}");
    let lcov = "TN:\nSF:/src/src/lib.rs\nFN:1,add\nFN:5,unused\nFNDA:3,add\nFNDA:0,unused\nDA:1,3\nDA:2,3\nDA:5,0\nDA:6,0\nend_of_record\n\
SF:/src/src/gone.rs\nDA:1,1\nend_of_record\nSF:/usr/lib/rustlib/x.rs\nDA:1,1\nend_of_record\n";
    let (st, text) = upload(&h.app, id, "coverage", lcov.as_bytes()).await;
    assert_eq!(st, StatusCode::OK, "{text}");
    let log: Vec<String> = db::flow::log(&h.app.db, id, 0, 50).await.unwrap().into_iter().map(|l| l.text).collect();
    assert!(log.iter().any(|l| l.contains("coverage: this is not an LCOV report")), "{log:?}");
    assert!(
        log.iter().any(|l| l == "coverage: 50.0% of lines (2 of 4) in 1 files (2 named files are not in this state and were left out)"),
        "{log:?}"
    );

    let (st, page) = call(&h.app, "GET", "/andy/cov/coverage", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert!(
        page.contains(">Coverage</a>")
            && page.contains("50.0%")
            && page.contains("<strong>2</strong> of <strong>4</strong> relevant lines covered"),
        "{page}"
    );
    assert!(page.contains("1 of 2") && page.contains("/andy/cov/coverage/") && page.contains("/file/src/lib.rs"), "{page}");
    let report = crate_report(&h.app, run.id).await;
    let (st, file) = call(&h.app, "GET", &format!("/andy/cov/coverage/{report}/file/src/lib.rs"), None, None).await;
    assert_eq!(st, StatusCode::OK, "{file}");
    assert_eq!(file.matches("class=\"cov-hit\"").count(), 2, "{file}");
    assert_eq!(file.matches("class=\"cov-miss\"").count(), 2, "{file}");
    assert!(file.contains("3×"), "{file}");
    assert_eq!(call(&h.app, "GET", &format!("/andy/cov/coverage/{report}/file/src/gone.rs"), None, None).await.0, StatusCode::NOT_FOUND);
    let (_, badge) = call(&h.app, "GET", "/andy/cov/coverage/badge.svg", None, None).await;
    assert!(badge.contains("coverage: 50.0%"), "{badge}");
    let (_, run_page) = call(&h.app, "GET", &format!("/andy/cov/flow/{}", run.number), None, None).await;
    assert!(run_page.contains("coverage 50.0%"), "{run_page}");
}

async fn crate_report(app: &Shared, run_id: i64) -> i64 {
    oxygenhub::coverage::for_run(&app.db, run_id).await.unwrap()[0].id
}

/// The Tests tab (#114): three runs' JUnit reports on a bookmark; a test
/// breaks in the second and is fixed in the third.
#[tokio::test(flavor = "multi_thread")]
async fn junit_reports_make_a_tests_tab() {
    let h = hub().await;
    create(&h.app, "andy", "t").await;
    let url = format!("{}/andy/t", h.base);
    let work = tempfile::tempdir().unwrap();
    let c = client(&url, work.path());
    write(
        work.path(),
        ".flow/pipeline.kdl",
        &format!(
            "schema-version 1\nsandbox \"alpine\" {{\n    rootfs \"cas://blob/{ROOTFS}\"\n}}\njob \"test\" {{\n    sandbox \"alpine\"\n    exec \"make\" \"test\"\n    output {{\n        artifact \"junit\" {{\n            path \"/src/junit.xml\"\n        }}\n    }}\n}}\n"
        ),
    );
    assert!(!call(&h.app, "GET", "/andy/t", None, None).await.1.contains(">Tests</a>"));
    let case = |name: &str, ok: bool, secs: &str| {
        if ok {
            format!(r#"<testcase classname="t.suite" name="{name}" time="{secs}"/>"#)
        } else {
            format!(
                r#"<testcase classname="t.suite" name="{name}" time="{secs}"><failure message="{name} broke">AssertionError</failure></testcase>"#
            )
        }
    };
    let runs = [
        vec![case("a", true, "0.1"), case("b", true, "2.5"), case("c", true, "0.2")],
        vec![case("a", true, "0.1"), case("b", false, "2.6"), case("c", true, "0.2"), case("d", true, "0.3")],
        vec![case("a", true, "0.1"), case("b", true, "2.4"), case("c", true, "0.2"), case("d", true, "0.3")],
    ];
    for (i, cases) in runs.iter().enumerate() {
        write(work.path(), "n.txt", &i.to_string());
        c.snapshot("test").unwrap();
        c.commit(&format!("Run {i}"), &[]).unwrap();
        assert!(c.push("origin", false).unwrap().iter().all(|r| r.ok));
        // The run is made after the push answers.
        let mut a = None;
        for _ in 0..100 {
            a = claim(&h.app).await;
            if a.is_some() {
                break;
            }
            tokio::time::sleep(std::time::Duration::from_millis(50)).await;
        }
        let a = a.expect("a job to claim");
        let id = a["job_id"].as_i64().unwrap();
        let xml = format!("<testsuites><testsuite name=\"t\">{}</testsuite></testsuites>", cases.concat());
        assert_eq!(upload(&h.app, id, "junit", xml.as_bytes()).await.0, StatusCode::OK);
        let failed = cases.iter().any(|c| c.contains("<failure"));
        let mut r = receipt("test", a["pipeline_hash"].as_str().unwrap(), failed as u32);
        if !failed {
            r["outputs"] = serde_json::json!([oxy_core::repo::Repo::blob_ref(xml.as_bytes()).to_hex()]);
        }
        let code = failed as i32;
        assert_eq!(
            finish(&h.app, id, serde_json::json!({ "exit_code": code, "duration_ms": 700, "receipt": r })).await,
            StatusCode::NO_CONTENT
        );
    }
    let (st, page) = call(&h.app, "GET", "/andy/t/tests", None, None).await;
    assert_eq!(st, StatusCode::OK, "{page}");
    if let Some(out) = std::env::var_os("OXYHUB_DUMP_TESTS_PAGE") {
        std::fs::write(out, &page).unwrap();
    }
    assert!(page.contains(">Tests</a>"), "the tab");
    assert!(page.contains("100%") && page.contains("of 4 tests passed"), "the latest run passes");
    assert!(page.contains("1 fixed") && page.contains("t.suite::b"), "b fixed against the run before");
    assert!(page.contains("1 ✕") && page.contains("2 ✓"), "b is flaky: {page}");
    assert!(page.contains("<svg class=\"cov-chart\""), "charts");
    // The slowest first.
    let slow = page.find("Slowest tests").unwrap();
    assert!(page[slow..].find("t.suite::b").unwrap() < page[slow..].find("t.suite::d").unwrap());
    // The second run, picked: b newly failing, its failure listed.
    let row = db::repos::get(&h.app.db, "andy", "t").await.unwrap().unwrap();
    let second = oxygenhub::junit::on_bookmark(&h.app.db, row.id, "main", 10).await.unwrap()[1].id;
    let (_, page) = call(&h.app, "GET", &format!("/andy/t/tests?report={second}"), None, None).await;
    assert!(page.contains("1 failing") && page.contains("Newly failing") && page.contains("b broke"), "{page}");
    assert!(page.contains("1 new tests"), "d is new");
}