//! The event bus (docs/roles.md, #273): a hub hears its own events once;
//! on PostgreSQL, two hubs on one database hear each other's: a push made
//! through a web process reaches live pages and the webhook worker of
//! another process, and a process whose listening connection is ended
//! listens again on a new one.

use oxygenhub::bus::{Event, Topic};
use oxygenhub::config::Config;
use std::path::Path;
use std::time::Duration;

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 config(dir: &Path, roles: &[&str]) -> Config {
    let mut config = Config::load(dir).unwrap();
    config.server.roles = roles.iter().map(|r| r.to_string()).collect();
    config.registration.setup_code = "test-setup".into();
    config
}

/// The next event on `rx` that `want` picks, within `limit`.
async fn heard<T: Clone>(rx: &mut tokio::sync::broadcast::Receiver<T>, limit: Duration, want: impl Fn(&T) -> bool) -> Option<T> {
    tokio::time::timeout(limit, async {
        loop {
            match rx.recv().await {
                Ok(e) if want(&e) => return Some(e),
                Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,
                Err(_) => return None,
            }
        }
    })
    .await
    .ok()
    .flatten()
}

#[tokio::test(flavor = "multi_thread")]
async fn a_hub_hears_its_own_events_once() {
    hermetic();
    let dir = tempfile::tempdir().unwrap();
    let app = oxygenhub::build(config(dir.path(), &[])).await.unwrap();
    let mut live = app.bus.subscribe(Topic::Live);
    let mut deliveries = app.bus.subscribe(Topic::Deliveries);
    app.bus.live("repo:1", "push");
    app.bus.delivery_waiting();
    let second = Duration::from_secs(1);
    assert_eq!(heard(&mut live, second, |_| true).await, Some(Event::Live { topic: "repo:1".into(), kind: "push".into() }));
    assert_eq!(heard(&mut deliveries, second, |_| true).await, Some(Event::DeliveryWaiting));
    // On PostgreSQL the event also went out through the database and came
    // back to this process's listener, which skips its own.
    tokio::time::sleep(Duration::from_millis(500)).await;
    assert!(live.try_recv().is_err(), "heard once");
    assert!(deliveries.try_recv().is_err(), "heard once");
    assert_eq!(app.bus.live.last_id(), 1);
}

#[cfg(feature = "postgres")]
mod postgres {
    use super::*;
    use axum::body::Body;
    use axum::http::{header, Request, StatusCode};
    use axum::routing::post;
    use http_body_util::BodyExt;
    use oxy_core::repo::Repo;
    use oxygenhub::Shared;
    use std::sync::{Arc, Mutex};
    use tower::ServiceExt;

    fn pg(dir: &Path, schema: &str, roles: &[&str]) -> Config {
        let mut c = config(dir, roles);
        c.database.schema = schema.to_string();
        c
    }

    async fn call(
        app: &Shared,
        method: &str,
        path: &str,
        auth: (Option<&str>, Option<&str>),
        body: Option<(&str, String)>,
    ) -> (StatusCode, String, Option<String>) {
        let mut req = Request::builder().method(method).uri(path);
        if let Some(b) = auth.0 {
            req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));
        }
        if let Some(c) = auth.1 {
            req = req.header(header::COOKIE, format!("oxyhub={c}"));
        }
        let body = match body {
            Some((kind, b)) => {
                req = req.header(header::CONTENT_TYPE, kind);
                Body::from(b)
            }
            None => 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();
        (status, String::from_utf8_lossy(&bytes).into_owned(), cookie)
    }

    /// A webhook receiver that records the events it gets.
    async fn receiver() -> (String, Arc<Mutex<Vec<String>>>) {
        let seen: Arc<Mutex<Vec<String>>> = Arc::default();
        let s2 = seen.clone();
        let router = axum::Router::new().route(
            "/hook",
            post(move |headers: header::HeaderMap| {
                let s2 = s2.clone();
                async move {
                    let event = headers.get("x-oxyhub-event").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();
                    s2.lock().unwrap().push(event);
                    StatusCode::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() });
        (format!("http://127.0.0.1:{port}/hook"), seen)
    }

    /// A push through a web-only hub: a live page's subscriber on a
    /// worker-only hub hears it, and that hub's webhook worker, nudged over
    /// the bus, delivers its webhook at once rather than on its
    /// once-a-minute pass. The web hub hears its own push once.
    #[tokio::test(flavor = "multi_thread")]
    async fn a_push_through_one_hub_reaches_the_other_hubs_pages_and_worker() {
        hermetic();
        let (da, dbb) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
        let schema = format!("hub_bus_{}", da.path().file_name().unwrap().to_string_lossy());
        let mut ca = pg(da.path(), &schema, &["web"]);
        ca.auth.push_tokens = vec!["site".into()];
        ca.registration.mode = "open".into();
        ca.outbound.allow = vec!["http://127.0.0.1".into()];
        let tcp = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let base = format!("http://127.0.0.1:{}", tcp.local_addr().unwrap().port());
        ca.server.base_url = base.clone();
        let a = oxygenhub::build(ca).await.unwrap();
        let router = oxygenhub::router(a.clone());
        tokio::spawn(async move { axum::serve(tcp, router).await.unwrap() });
        let mut cb = pg(dbb.path(), &schema, &["worker"]);
        cb.outbound.allow = vec!["http://127.0.0.1".into()];
        let b = oxygenhub::build(cb).await.unwrap();

        // An account, an admin token, a repository and its webhook, on A.
        let form = "application/x-www-form-urlencoded";
        let (_, _, cookie) = call(
            &a,
            "POST",
            "/register",
            (None, None),
            Some((form, "username=andy&email=a%40example.com&password=correct-horse-battery&setup=test-setup".into())),
        )
        .await;
        let andy = cookie.expect("signed in");
        let (_, page, _) = call(&a, "GET", "/settings/tokens", (None, Some(&andy)), None).await;
        let at = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();
        let csrf = page[at..].split('"').next().unwrap().to_string();
        let (_, page, _) =
            call(&a, "POST", "/settings/tokens", (None, Some(&andy)), Some((form, format!("_csrf={csrf}&name=t&scopes=repo%3Aadmin"))))
                .await;
        let token = format!("oxyhub_{}", page.split("oxyhub_").nth(1).unwrap().split('<').next().unwrap().trim());
        let json = "application/json";
        let (st, body, _) = call(
            &a,
            "POST",
            "/api/v1/repos",
            (Some(&token), None),
            Some((json, r#"{"owner":"andy","name":"demo","visibility":"public"}"#.into())),
        )
        .await;
        assert_eq!(st, StatusCode::CREATED, "{body}");
        let (hook, seen) = receiver().await;
        let (st, body, _) = call(
            &a,
            "POST",
            "/api/v1/repos/andy/demo/hooks",
            (Some(&token), None),
            Some((json, format!(r#"{{"url":"{hook}","secret":"s","events":"push"}}"#))),
        )
        .await;
        assert_eq!(st, StatusCode::CREATED, "{body}");
        let repo = oxygenhub::db::repos::get(&a.db, "andy", "demo").await.unwrap().unwrap();
        let topic = oxygenhub::events::topic::repo(repo.id);

        // B's worker leads the webhook worker and has made its first pass.
        let start = std::time::Instant::now();
        while !b.leader.leading().contains(&"hooks.worker") {
            assert!(start.elapsed() < Duration::from_secs(10), "{:?}", b.leader.leading());
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        tokio::time::sleep(Duration::from_millis(500)).await;

        let mut b_live = b.bus.live.subscribe();
        let work = tempfile::tempdir().unwrap();
        let mut client = Repo::init(work.path()).unwrap();
        client.config.set("remote.origin.url", &format!("{base}/andy/demo")).unwrap();
        client.config.set("remote.origin.token", "site").unwrap();
        client.save_config().unwrap();
        std::fs::write(work.path().join("a.txt"), b"a\n").unwrap();
        client.snapshot("t").unwrap();
        client.commit("base", &[]).unwrap();
        assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

        let push = heard(&mut b_live, Duration::from_secs(10), |e| e.topic == topic).await.expect("B hears A's push");
        assert_eq!(push.kind, "push");
        let start = std::time::Instant::now();
        while seen.lock().unwrap().is_empty() {
            assert!(start.elapsed() < Duration::from_secs(10), "B's worker was not nudged");
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        assert_eq!(*seen.lock().unwrap(), ["push"]);
        eprintln!("delivered by the other process in {:.1} s", start.elapsed().as_secs_f64());

        // A heard its own push once, not again from the database.
        tokio::time::sleep(Duration::from_millis(500)).await;
        let oxygenhub::events::Replay::Events(on_a) = a.bus.live.since(0, std::slice::from_ref(&topic)) else { panic!() };
        assert_eq!(on_a.iter().filter(|e| e.kind == "push").count(), 1, "{on_a:?}");
    }

    /// Proposal events and resolver rooms' choices cross over; a process
    /// whose listening connection is ended listens again, looks for
    /// deliveries it may have missed, and hears the other again.
    #[tokio::test(flavor = "multi_thread")]
    async fn two_hubs_hear_each_other_and_listen_again_after_losing_the_connection() {
        hermetic();
        let (da, dbb) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());
        let schema = format!("hub_bus_{}", da.path().file_name().unwrap().to_string_lossy());
        let a = oxygenhub::build(pg(da.path(), &schema, &[])).await.unwrap();
        let b = oxygenhub::build(pg(dbb.path(), &schema, &[])).await.unwrap();
        // A hub in another schema of the same database hears neither.
        let dc = tempfile::tempdir().unwrap();
        let c = oxygenhub::build(pg(dc.path(), &format!("{schema}_other"), &[])).await.unwrap();
        let mut c_live = c.bus.subscribe(Topic::Live);

        let mut b_proposals = b.bus.subscribe(Topic::Proposals);
        let check = hub_api::ProposalEvent { number: 0, kind: "check".into(), check: None };
        a.bus.proposal(7, check.clone());
        let got = heard(&mut b_proposals, Duration::from_secs(5), |_| true).await;
        assert_eq!(got, Some(Event::Proposal { repo: 7, event: check }));

        // A choice made on A reaches the same room's viewers on B.
        let key = oxygenhub::live::RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() };
        let (mut b_room, _) = b.rooms.join(key.clone(), "tab-b", "bea");
        let choice = oxygenhub::live::Choice { choice: "side2".into(), text: String::new(), by: "andy".into() };
        let event = oxygenhub::live::RoomEvent::Choice { region: "1".into(), choice: choice.clone(), client: "tab-a".into() };
        a.bus.publish(Event::Room { room: key.clone(), event });
        let got = heard(&mut b_room, Duration::from_secs(5), |e| matches!(e, oxygenhub::live::LiveEvent::Choice { .. })).await;
        assert_eq!(got, Some(oxygenhub::live::LiveEvent::Choice { region: "1".into(), choice, client: "tab-a".into() }));

        // B's listening connection is ended, as when the network drops.
        let pid = b.bus.listener_pid().expect("b listens");
        let mut b_deliveries = b.bus.subscribe(Topic::Deliveries);
        let ended: bool = sqlx::query_scalar("SELECT pg_terminate_backend($1)").bind(pid).fetch_one(&a.db).await.unwrap();
        assert!(ended);
        let start = std::time::Instant::now();
        while b.bus.listener_pid().is_none_or(|p| p == pid) {
            assert!(start.elapsed() < Duration::from_secs(15), "b did not listen again");
            tokio::time::sleep(Duration::from_millis(100)).await;
        }
        eprintln!("listening again in {:.1} s", start.elapsed().as_secs_f64());
        // It looks now for deliveries a missed nudge would have announced.
        assert_eq!(heard(&mut b_deliveries, Duration::from_secs(5), |_| true).await, Some(Event::DeliveryWaiting));
        let mut b_live = b.bus.subscribe(Topic::Live);
        a.bus.live("repo:9", "push");
        let got = heard(&mut b_live, Duration::from_secs(5), |_| true).await;
        assert_eq!(got, Some(Event::Live { topic: "repo:9".into(), kind: "push".into() }));
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert!(c_live.try_recv().is_err(), "another schema's hub hears nothing");
    }
}