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

    write(work.path(), "a.txt", b"a\nA\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);

    // Its files, and each side's bytes, for editors' diff views.

    let (st, v) = api(app, "GET", "/api/v1/repos/andy/demo/proposals/2/files", None, None).await;

    assert_eq!(st, StatusCode::OK, "{v}");

    let files: Vec<(String, String)> =

        v.as_array().unwrap().iter().map(|f| (f["path"].as_str().unwrap().to_string(), f["kind"].as_str().unwrap().to_string())).collect();

    assert_eq!(files, vec![("a.txt".into(), "modified".into()), ("b.txt".into(), "added".into())], "{v}");

    assert_eq!(v[0]["added"], 1);

    let side = |s: &str, p: &str| format!("/api/v1/repos/andy/demo/proposals/2/content/{s}/{p}");

    assert_eq!(call(app, "GET", &side("base", "a.txt"), None, None, None, None).await.body, "a\n");

    assert_eq!(call(app, "GET", &side("head", "a.txt"), None, None, None, None).await.body, "a\nA\n");

    assert_eq!(call(app, "GET", &side("head", "b.txt"), None, None, None, None).await.body, "b\n");

    assert_eq!(call(app, "GET", &side("base", "b.txt"), None, None, None, None).await.status, StatusCode::NOT_FOUND, "b.txt is new");

    assert_eq!(call(app, "GET", &side("middle", "a.txt"), None, None, None, None).await.status, StatusCode::BAD_REQUEST);

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

    // The inbox through the API: count, list, mark one read, mark all read.

    let bob_user = mint(bob.clone(), "scopes=user%3Aread").await;

    let bob_id = oxygenhub::db::accounts::account_by_name(&app.db, "bob").await.unwrap().unwrap().id;

    oxygenhub::db::notify::mark_read(&app.db, bob_id, None).await.unwrap();

    oxygenhub::db::notify::add(

        &app.db,

        oxygenhub::db::notify::New {

            account_id: bob_id,

            repo_id: None,

            reason: oxygenhub::db::notify::Reason::Participating,

            kind: "mention",

            title: "andy/demo#1: Bug",

            url: "/andy/demo/issue/1",

            actor: "andy",

        },

    )

    .await

    .unwrap();

    oxygenhub::db::notify::add(

        &app.db,

        oxygenhub::db::notify::New {

            account_id: bob_id,

            repo_id: None,

            reason: oxygenhub::db::notify::Reason::Participating,

            kind: "review",

            title: "andy/demo#2: Feat",

            url: "/andy/demo/proposal/2",

            actor: "andy",

        },

    )

    .await

    .unwrap();

    let (st, v) = api(app, "GET", "/api/v1/notifications", Some(&bob_user), None).await;

    assert_eq!(st, StatusCode::OK, "{v}");

    assert_eq!(v["unread"], 2);

    assert_eq!(v["items"][0]["read"], false);

    assert!(v["items"][0]["url"].as_str().unwrap().starts_with(&h.base), "absolute urls: {v}");

    let first = v["items"][0]["id"].as_i64().unwrap();

    let (st, _) = api(app, "POST", &format!("/api/v1/notifications/{first}/read"), Some(&bob_user), None).await;

    assert_eq!(st, StatusCode::NO_CONTENT);

    assert_eq!(api(app, "GET", "/api/v1/notifications", Some(&bob_user), None).await.1["unread"], 1);

    assert_eq!(api(app, "POST", "/api/v1/notifications/read", Some(&bob_user), None).await.0, StatusCode::NO_CONTENT);

    assert_eq!(api(app, "GET", "/api/v1/notifications", Some(&bob_user), None).await.1["unread"], 0);

    // Another account's notification is not the caller's to clear, and a token without user:read sees nothing.