//! 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()];

    config.flow.repos = vec!["andy/*".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())

}

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![]

}

#[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)]);

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

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

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

    // 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.id)) && list.contains("st-failed"), "{list}");

    let (st, page) = call(&h.app, "GET", &format!("/andy/demo/flow/{}", run.id), 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.id), 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.id), None, None).await;

    assert!(page.contains("The pipeline is not valid"));

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

}

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

async fn claims_carry_pinned_subrepos_and_the_job_token_reads_only_those() {