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