//! Real SDK watch callbacks and raw WASI filesystem mutations share the host namespace.
use super::common::{CACHE, probe, sdk_probe};
use sausage::{Host, world::World};
use serde_json::{Value, json};
use std::time::Duration;

// Same call bound as core.rs; no sleeps or polling.
async fn call(h: &Host, method: &str, p: Value) -> Value {
    tokio::time::timeout(Duration::from_secs(20), h.call(method, p))
        .await
        .unwrap()
        .unwrap()
}
async fn event(h: &Host, rx: &mut tokio::sync::broadcast::Receiver<Value>, name: &str) -> Value {
    let result = tokio::time::timeout(Duration::from_secs(20), async {
        loop {
            let e = rx.recv().await.unwrap();
            if e["params"]["name"] == name {
                return e["params"]["params"].clone();
            }
        }
    })
    .await;
    match result {
        Ok(v) => v,
        Err(e) => {
            eprintln!(
                "JOURNAL {}",
                h.call("core.log", json!({"limit":100})).await.unwrap()
            );
            panic!("{e}");
        }
    }
}
async fn start() -> (tempfile::TempDir, Host) {
    let dir = tempfile::tempdir().unwrap();
    let world = World::create(&dir.path().join("world")).unwrap();
    for (name, wasm) in [("probe", probe()), ("sdk", sdk_probe())] {
        std::fs::create_dir_all(world.plugin_dir(name)).unwrap();
        std::fs::copy(wasm, world.plugin_dir(name).join("plugin.wasm")).unwrap();
    }
    let host = Host::start_with_cache(&world.root, std::path::Path::new(CACHE))
        .await
        .unwrap();
    (dir, host)
}
fn watch(h: &Host, path: &str) -> tokio::task::JoinHandle<sausage::RpcResult> {
    let (h, path) = (h.clone(), path.to_owned());
    tokio::spawn(async move { h.call("sdk.watch", json!({"path":path})).await })
}
async fn mutate(h: &Host, op: &str, path: &str) {
    call(h, "probe.mutate", json!({"op":op,"path":path})).await;
}
/// PROTOCOL 4: pre-arm state is in the first scan; each post-scan mutation wakes
/// the SDK without completing the watch. Covers descriptor/stream/truncate paths.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn armed_scan_and_later_changes() {
    let (_dir, h) = start().await;
    mutate(&h, "mkdir", "/state/mailbox").await;
    call(
        &h,
        "probe.write",
        json!({"path":"/state/mailbox/before", "text":"before"}),
    )
    .await;
    let mut rx = h.subscribe();
    let task = watch(&h, "/state/mailbox/../mailbox/");
    let armed = event(&h, &mut rx, "armed").await;
    assert_eq!(armed["entries"], json!(["before"]));
    assert!(!task.is_finished());
    for (op, expected) in [
        ("descriptor", "descriptor"),
        ("stream", "streamptor"),
        ("append", "streamptorappend"),
        ("size", ""),
        ("descriptor", "descriptor"),
        ("truncate", ""),
    ] {
        mutate(&h, op, "/state/mailbox/before").await;
        let mut scan = event(&h, &mut rx, "changed").await;
        while scan["text"] != expected {
            scan = event(&h, &mut rx, "changed").await;
        }
    }
    mutate(&h, "burst", "/state/mailbox").await;
    let mut scanned = event(&h, &mut rx, "changed").await;
    while !scanned["entries"].as_array().unwrap().contains(&json!("4")) {
        scanned = event(&h, &mut rx, "changed").await;
    }
    while rx.try_recv().is_ok() {}
    mutate(&h, "delete", "/state/mailbox/4").await;
    let mut scanned = event(&h, &mut rx, "changed").await;
    while scanned["entries"].as_array().unwrap().contains(&json!("4")) {
        scanned = event(&h, &mut rx, "changed").await;
    }
    call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
    assert!(task.await.unwrap().is_err());
    h.stop().await;
}
/// PROTOCOL 4: atomic publication, rename source/destination, mounted paths,
/// absent paths, and caller-death cleanup. No external host writes are required.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn rename_delete_and_mounts() {
    let (_dir, h) = start().await;
    for base in ["/state/inbox", "/run/inbox"] {
        let mut rx = h.subscribe();
        let task = watch(&h, base);
        let armed = event(&h, &mut rx, "armed").await;
        assert_eq!(armed["entries"], json!([]));
        let staging = format!("{base}-staging");
        mutate(&h, "mkdir", &staging).await;
        call(
            &h,
            "probe.mutate",
            json!({"op":"rename","path":staging,"to":base}),
        )
        .await;
        event(&h, &mut rx, "changed").await;
        mutate(&h, "atomic", &format!("{base}/published")).await;
        let mut scan = event(&h, &mut rx, "changed").await;
        while !scan["entries"]
            .as_array()
            .unwrap()
            .contains(&json!("published"))
        {
            scan = event(&h, &mut rx, "changed").await;
        }
        assert_eq!(
            call(
                &h,
                "probe.read",
                json!({"path":format!("{base}/published")})
            )
            .await,
            "complete"
        );
        while rx.try_recv().is_ok() {}
        call(
            &h,
            "probe.mutate",
            json!({"op":"rename","path":base,"to":format!("{base}-moved")}),
        )
        .await;
        let mut scan = event(&h, &mut rx, "changed").await;
        while scan["entries"] != json!([]) {
            scan = event(&h, &mut rx, "changed").await;
        }
        call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
        assert!(task.await.unwrap().is_err());
    }
    // /self watches bind to this caller, and mutations are visible via /proc/<pid>.
    let mut rx = h.subscribe();
    let task = watch(&h, "/self");
    let armed = event(&h, &mut rx, "armed").await;
    assert!(
        armed["entries"]
            .as_array()
            .unwrap()
            .contains(&json!("owned"))
    );
    event(&h, &mut rx, "changed").await;
    assert_eq!(
        call(
            &h,
            "probe.read",
            json!({"path":format!("/proc/{}/owned", armed["pid"].as_str().unwrap())})
        )
        .await,
        "visible via proc"
    );
    call(&h, "core.kill", json!({"pid":armed["pid"]})).await;
    assert!(task.await.unwrap().is_err());
    h.stop().await;
}

/// PROTOCOL 9: core discovery includes the armed watch and matches its proc projection.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn discovery() {
    let (_dir, h) = start().await;
    let doc = call(&h, "core.rpc.discover", json!({})).await;
    let projected = call(&h, "probe.read", json!({"path":"/proc/core/discover.json"})).await;
    assert_eq!(
        doc,
        serde_json::from_str::<Value>(projected.as_str().unwrap()).unwrap()
    );
    let watch = doc["methods"]
        .as_array()
        .unwrap()
        .iter()
        .find(|m| m["name"] == "watch")
        .unwrap();
    assert_eq!(watch["params"][0]["name"], "path");
    assert!(watch["description"].as_str().unwrap().contains("armed"));
    h.stop().await;
}