//! Milestone 0.5 end to end: the JSON API with scoped tokens, webhooks
//! delivered to a local receiver with signatures and retries, checks on a
//! projected state that gate landing, and the proposal event stream.

use axum::body::Body;
use axum::http::{header, Request, StatusCode};
use axum::routing::post;
use axum::Router;
use http_body_util::BodyExt;
use oxy_core::repo::Repo;
use oxygenhub::config::Config;
use oxygenhub::db;
use oxygenhub::repos::RepoName;
use oxygenhub::Shared;
use std::path::Path;
use std::sync::{Arc, Mutex};
use tower::ServiceExt;

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: &[u8]) {
    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.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 }
}

struct Res {
    status: StatusCode,
    body: String,
    cookie: Option<String>,
}

async fn call(
    app: &Shared,
    method: &str,
    path: &str,
    bearer: Option<&str>,
    cookie: Option<&str>,
    json: Option<&str>,
    form: Option<&str>,
) -> Res {
    let mut req = Request::builder().method(method).uri(path);
    if let Some(b) = bearer {
        req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));
    }
    if let Some(c) = cookie {
        req = req.header(header::COOKIE, format!("oxyhub={c}"));
    }
    let body = if let Some(j) = json {
        req = req.header(header::CONTENT_TYPE, "application/json");
        Body::from(j.to_string())
    } else if let Some(f) = form {
        req = req.header(header::CONTENT_TYPE, "application/x-www-form-urlencoded");
        Body::from(f.to_string())
    } else {
        Body::empty()
    };
    let res = oxygenhub::router(app.clone()).oneshot(req.body(body).unwrap()).await.unwrap();
    let status = res.status();
    let cookie = res
        .headers()
        .get(header::SET_COOKIE)
        .and_then(|v| v.to_str().ok())
        .and_then(|v| v.split(';').next())
        .and_then(|kv| kv.split_once('='))
        .map(|(_, v)| v.to_string());
    let bytes = res.into_body().collect().await.unwrap().to_bytes();
    Res { status, body: String::from_utf8_lossy(&bytes).into_owned(), cookie }
}

async fn api(app: &Shared, method: &str, path: &str, bearer: Option<&str>, json: Option<&str>) -> (StatusCode, serde_json::Value) {
    let r = call(app, method, path, bearer, None, json, None).await;
    let v = serde_json::from_str(&r.body).unwrap_or(serde_json::Value::String(r.body.clone()));
    (r.status, v)
}

fn csrf_of(page: &str) -> String {
    let i = page.find("name=\"_csrf\" value=\"").expect("csrf field") + "name=\"_csrf\" value=\"".len();
    page[i..].split('"').next().unwrap().to_string()
}

/// A webhook receiver that records what it gets and fails the first time.
struct Receiver {
    url: String,
    seen: Arc<Mutex<Vec<(String, String, String)>>>,
    fail_first: Arc<Mutex<bool>>,
}

async fn receiver() -> Receiver {
    let seen: Arc<Mutex<Vec<(String, String, String)>>> = Arc::new(Mutex::new(vec![]));
    let fail_first = Arc::new(Mutex::new(true));
    let (s2, f2) = (seen.clone(), fail_first.clone());
    let router = Router::new().route(
        "/hook",
        post(move |headers: header::HeaderMap, body: String| {
            let (s2, f2) = (s2.clone(), f2.clone());
            async move {
                let event = headers.get("x-oxyhub-event").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
                let sig = headers.get("x-oxyhub-signature").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
                let mut f = f2.lock().unwrap();
                if *f {
                    *f = false;
                    return (StatusCode::INTERNAL_SERVER_ERROR, "flaky");
                }
                s2.lock().unwrap().push((event, sig, body));
                (StatusCode::OK, "ok")
            }
        }),
    );
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });
    Receiver { url: format!("http://127.0.0.1:{port}/hook"), seen, fail_first }
}

#[tokio::test(flavor = "multi_thread")]
async fn api_webhooks_and_checks() {
    let h = hub().await;
    let app = &h.app;
    // Accounts and tokens through the pages.
    let r =
        call(app, "POST", "/register", None, None, None, Some("username=andy&email=a%40example.com&password=correct-horse-battery")).await;
    let andy = r.cookie.unwrap();
    let r =
        call(app, "POST", "/register", None, None, None, Some("username=bob&email=b%40example.com&password=correct-horse-battery")).await;
    let bob = r.cookie.unwrap();
    let mint = |cookie: String, scopes: &'static str| {
        let app = app.clone();
        async move {
            let page = call(&app, "GET", "/settings/tokens", None, Some(&cookie), None, None).await;
            let csrf = csrf_of(&page.body);
            let r = call(&app, "POST", "/settings/tokens", None, Some(&cookie), None, Some(&format!("_csrf={csrf}&name=t&{scopes}"))).await;
            r.body.split("oxyhub_").nth(1).map(|s| format!("oxyhub_{}", s.split('<').next().unwrap().trim())).unwrap()
        }
    };
    let andy_admin = mint(andy.clone(), "scopes=repo%3Aadmin").await;
    let andy_ro = mint(andy.clone(), "scopes=repo%3Aread").await;
    let bob_rw = mint(bob.clone(), "scopes=repo%3Aread&scopes=repo%3Awrite").await;

    // Create a repository through the API; the read-only token cannot.
    let (st, _) = api(app, "POST", "/api/v1/repos", Some(&andy_ro), Some(r#"{"owner":"andy","name":"x"}"#)).await;
    assert_eq!(st, StatusCode::FORBIDDEN);
    let (st, v) = api(
        app,
        "POST",
        "/api/v1/repos",
        Some(&andy_admin),
        Some(r#"{"owner":"andy","name":"demo","description":"d","visibility":"public"}"#),
    )
    .await;
    assert_eq!(st, StatusCode::CREATED, "{v}");
    assert_eq!(v["clone_url"], format!("{}/andy/demo", h.base));
    let (st, _) = api(app, "POST", "/api/v1/repos", Some(&andy_admin), Some(r#"{"owner":"andy","name":"demo"}"#)).await;
    assert_eq!(st, StatusCode::CONFLICT);
    let (st, _) = api(app, "GET", "/api/v1/user", Some(&andy_ro), None).await;
    assert_eq!(st, StatusCode::FORBIDDEN, "user:read is a scope of its own");
    let (st, v) = api(app, "GET", "/api/v1/users/andy", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert_eq!(v["username"], "andy");

    // A webhook on the repository; push, then see the delivery with a valid signature after a retry.
    let rx = receiver().await;
    let (st, v) = api(
        app,
        "POST",
        "/api/v1/repos/andy/demo/hooks",
        Some(&andy_admin),
        Some(&format!(r#"{{"url":"{}","secret":"s3","events":"push,proposal,check"}}"#, rx.url)),
    )
    .await;
    assert_eq!(st, StatusCode::CREATED, "{v}");
    let work = tempfile::tempdir().unwrap();
    let mut client = Repo::init(work.path()).unwrap();
    client.config.set("remote.origin.url", &format!("{}/andy/demo", h.base)).unwrap();
    client.config.set("remote.origin.token", "site").unwrap();
    client.save_config().unwrap();
    write(work.path(), "a.txt", b"a\n");
    client.snapshot("t").unwrap();
    let base = client.commit("base", &[]).unwrap();
    assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));
    // The worker retries with backoff; drive it directly for the retry.
    tokio::time::sleep(std::time::Duration::from_millis(400)).await;
    assert!(rx.seen.lock().unwrap().is_empty(), "the first attempt failed");
    sqlx::query("UPDATE deliveries SET next_ms = 0").execute(&app.db).await.unwrap();
    oxygenhub::hooks::deliver_due(app).await;
    let seen = rx.seen.lock().unwrap().clone();
    assert_eq!(seen.len(), 1, "{seen:?}");
    assert_eq!(seen[0].0, "push");
    assert_eq!(seen[0].1, oxygenhub::hooks::signature("s3", seen[0].2.as_bytes()));
    let payload: serde_json::Value = serde_json::from_str(&seen[0].2).unwrap();
    assert_eq!(payload["repository"]["name"], "demo");
    assert_eq!(payload["data"]["updates"][0]["reference"], "main");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/hooks", Some(&andy_admin), None).await;
    assert_eq!(v.as_array().unwrap().len(), 1);

    // History, tree, file, blame, bookmarks, trust, pushes through the API.
    let (st, v) = api(app, "GET", "/api/v1/repos/andy/demo/history/main", None, None).await;
    assert_eq!(st, StatusCode::OK);
    assert_eq!(v["changes"][0]["hash"], base.to_hex());
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/tree/main", None, None).await;
    assert_eq!(v[0]["name"], "a.txt");
    let r = call(app, "GET", "/api/v1/repos/andy/demo/file/main/a.txt", None, None, None, None).await;
    assert_eq!(r.body, "a\n");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/blame/main/a.txt", None, None).await;
    assert_eq!(v[0]["change"], base.to_hex());
    let (_, v) = api(app, "GET", &format!("/api/v1/repos/andy/demo/changes/{}", base.to_hex()), None, None).await;
    assert_eq!(v["files"][0]["path"], "a.txt");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/trust", None, None).await;
    assert!(v["policy"].is_null());
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/pushes", None, None).await;
    assert_eq!(v[0]["reference"], "main");

    // Issues through the API: Bob opens one, comments, and closes it.
    let (st, v) = api(app, "POST", "/api/v1/repos/andy/demo/issues", Some(&bob_rw), Some(r#"{"title":"Bug","body":"it broke"}"#)).await;
    assert_eq!(st, StatusCode::CREATED, "{v}");
    assert_eq!(v["number"], 1);
    let (st, _) = api(app, "POST", "/api/v1/repos/andy/demo/issues/1/comments", Some(&bob_rw), Some(r#"{"body":"more"}"#)).await;
    assert_eq!(st, StatusCode::CREATED);
    let (st, v) = api(app, "POST", "/api/v1/repos/andy/demo/issues/1", Some(&bob_rw), Some(r#"{"state":"closed"}"#)).await;
    assert_eq!(st, StatusCode::OK, "{v}");
    assert_eq!(v["state"], "closed");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/issues?state=all", None, None).await;
    assert_eq!(v[0]["comments"], 1);
    let (_, v) = api(app, "GET", "/api/v1/search?q=broke", None, None).await;
    assert_eq!(v[0]["number"], 1);

    // A proposal through the API, then a required check gates its landing.
    client.bookmark_create("feat", Some(&[base])).unwrap();
    client.bookmark_switch("feat", false).unwrap();
    write(work.path(), "b.txt", b"b\n");
    client.snapshot("t").unwrap();
    let feat = client.commit("feat", &[]).unwrap();
    assert!(client.push_bookmark("origin", "feat", false).unwrap().iter().all(|r| r.ok));
    let (st, v) =
        api(app, "POST", "/api/v1/repos/andy/demo/proposals", Some(&bob_rw), Some(r#"{"title":"Feat","target":"main","source":"feat"}"#))
            .await;
    assert_eq!(st, StatusCode::CREATED, "{v}");
    assert_eq!(v["number"], 2);
    let state = v["projected_state"].as_str().unwrap().to_string();
    assert_eq!(v["landable"], true);
    // Require the "build" check.
    let settings = call(app, "GET", "/andy/demo/settings", None, Some(&andy), None, None).await;
    let csrf = csrf_of(&settings.body);
    assert_eq!(
        call(app, "POST", "/andy/demo/settings/checks", None, Some(&andy), None, Some(&format!("_csrf={csrf}&required_checks=build")))
            .await
            .status,
        StatusCode::OK
    );
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2", None, None).await;
    assert_eq!(v["landable"], false);
    assert!(v["blockers"][0].as_str().unwrap().contains("build"), "{v}");
    // A checker reports on the projected state: first running, then passed; the page streams it.
    let (st, v) = api(
        app,
        "POST",
        "/api/v1/repos/andy/demo/checks",
        Some(&andy_ro),
        Some(&format!(r#"{{"state":"{state}","name":"build","status":"running"}}"#)),
    )
    .await;
    assert_eq!(st, StatusCode::FORBIDDEN, "{v}");
    let mut events = app.events.proposals.subscribe();
    let (st, v) = api(
        app,
        "POST",
        "/api/v1/repos/andy/demo/checks",
        Some(&bob_rw),
        Some(&format!(r#"{{"state":"{state}","name":"build","status":"passed","summary":"42 tests","url":"https://ci.example/1"}}"#)),
    )
    .await;
    assert_eq!(st, StatusCode::CREATED, "{v}");
    let (_, ev) = events.recv().await.unwrap();
    assert_eq!(ev.kind, "check");
    assert_eq!(ev.check.unwrap().name, "build");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2/checks", None, None).await;
    assert_eq!(v[0]["status"], "passed");
    let (_, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2", None, None).await;
    assert_eq!(v["landable"], true, "{v}");
    let page = call(app, "GET", "/andy/demo/proposal/2", None, Some(&andy), None, None).await;
    assert!(page.body.contains("42 tests"), "{}", page.body);
    // The check and the proposal reached the webhook too.
    sqlx::query("UPDATE deliveries SET next_ms = 0").execute(&app.db).await.unwrap();
    oxygenhub::hooks::deliver_due(app).await;
    let kinds: Vec<String> = rx.seen.lock().unwrap().iter().map(|s| s.0.clone()).collect();
    assert!(kinds.contains(&"proposal.opened".to_string()) && kinds.contains(&"check".to_string()), "{kinds:?}");
    let _ = feat;
    let _ = rx.fail_first;
}