//! End-to-end protocol checks against a WASI plugin under Wasmtime.
use sausage_plugin::serde_json::{Value, json};
use std::{
    io::{BufRead, BufReader, Write},
    path::{Path, PathBuf},
    process::{Command, Stdio},
};
struct Process {
    child: std::process::Child,
    input: Option<std::process::ChildStdin>,
    output: BufReader<std::process::ChildStdout>,
}
impl Process {
    fn finishes(&mut self) {
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(4);
        loop {
            if let Some(status) = self.child.try_wait().unwrap() {
                assert!(status.success());
                return;
            }
            if std::time::Instant::now() >= deadline {
                self.child.kill().unwrap();
                let _ = self.child.wait();
                panic!("process survived exit or EOF for four seconds");
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
    }
    fn send(&mut self, v: Value) {
        writeln!(self.input.as_mut().unwrap(), "{v}").unwrap();
    }
    fn read(&mut self) -> Value {
        loop {
            let mut s = String::new();
            if self.output.read_line(&mut s).unwrap() == 0 {
                let mut stderr = String::new();
                let _ =
                    std::io::Read::read_to_string(self.child.stderr.as_mut().unwrap(), &mut stderr);
                panic!("unexpected EOF from wasmtime. Its stderr:\n{stderr}");
            }
            if let Ok(v) = sausage_plugin::serde_json::from_str::<Value>(&s)
                && v["jsonrpc"] == "2.0"
            {
                return v;
            }
        }
    }
    fn request(&mut self, method: &str, params: Value) {
        self.send(json!({"jsonrpc":"2.0","id":"probe-1","method":method,"params":params}));
    }
    fn answer(&mut self, r: &Value, result: Value) {
        self.send(json!({"jsonrpc":"2.0","id":r["id"],"result":result}));
    }
}
fn start(wasm: &Path, dir: &Path) -> Process {
    let mut child = Command::new("wasmtime")
        .args([
            "run",
            // Without a disk cache: the Nix sandbox has no writable home folder.
            "-C",
            "cache=n",
            "-S",
            "inherit-network=y",
            "--env",
            "SAUSAGE_PID=probe-00000000-0000-7000-8000-000000000000",
            "--env",
            "SAUSAGE_PLUGIN=probe",
            "--dir",
        ])
        .arg(format!("{}::/", dir.display()))
        .arg(wasm)
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .unwrap();
    let input = child.stdin.take();
    let output = BufReader::new(child.stdout.take().unwrap());
    Process {
        child,
        input,
        output,
    }
}
#[test]
fn protocol() {
    let root = PathBuf::from(env!("CARGO_MANIFEST_DIR"));
    let target = root.join("../target/sdk/probe");
    assert!(
        Command::new("cargo")
            .env("CARGO_TARGET_DIR", &target)
            .args(["build", "--offline", "--manifest-path"])
            .arg(root.join("tests/plugins/probe/Cargo.toml"))
            .args(["--target", "wasm32-wasip2"])
            .status()
            .unwrap()
            .success()
    );
    let wasm = target.join("wasm32-wasip2/debug/sdk-probe.wasm");
    let dir = std::env::temp_dir().join(format!("sausage-sdk-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    for method in ["sleepy_exit", "sleepy"] {
        let mut p = start(&wasm, &dir);
        p.request(method, json!({}));
        assert_eq!(p.read()["result"], Value::Null);
        if method == "sleepy" {
            drop(p.input.take());
        }
        p.finishes();
    }
    let mut p = start(&wasm, &dir);
    p.request("rpc.discover", json!({}));
    let d = p.read();
    assert_eq!(d["result"]["x-sausage-grace"], 12);
    assert_eq!(
        d["result"]["methods"]
            .as_array()
            .unwrap()
            .iter()
            .find(|m| m["name"] == "add")
            .unwrap()["description"],
        "Add two numbers.\n\nThe result is their sum."
    );
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("add", json!({"a":2,"b":4}));
    assert_eq!(p.read()["result"], 6);
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("echo", json!({"text":"x".repeat(70_000)}));
    assert_eq!(p.read()["result"], "x".repeat(70_000));
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("echo", json!({"text":5}));
    assert_eq!(p.read()["error"]["code"], -32602);
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("serde_error", json!({}));
    assert_eq!(p.read()["error"]["code"], -32603);
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("fail", json!({}));
    assert_eq!(p.read()["error"]["code"], -32010);
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("exercise", json!({}));
    let a = p.read();
    let b = p.read();
    assert_eq!(a["method"], "other.first");
    assert_eq!(b["method"], "core.call");
    assert_eq!(b["params"]["detach"], true);
    p.send(json!({"jsonrpc":"2.0","id":9876,"result":{}}));
    p.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":a["id"],"name":"tick","params":{}}}));
    p.answer(&b, json!({}));
    p.answer(&a, json!({}));
    assert_eq!(p.read()["method"], "core.notify");
    let grant = p.read();
    assert_eq!(grant["method"], "core.lock");
    p.answer(&grant, json!({"held":true}));
    assert_eq!(p.read()["method"], "core.unlock");
    let attempt = p.read();
    assert_eq!(attempt["params"]["wait"], false);
    p.answer(&attempt, json!({"held":false}));
    assert_eq!(p.read()["result"], "done");
    assert!(p.child.wait().unwrap().success());
    let mut err = String::new();
    std::io::Read::read_to_string(p.child.stderr.as_mut().unwrap(), &mut err).unwrap();
    assert!(err.contains("unknown ID 9876"));
    assert!(err.contains("notification: tick"));
    let mut p = start(&wasm, &dir);
    p.request("rpc.retired", json!({"previous":"hash"}));
    assert_eq!(p.read()["result"], "hash");
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("late", json!({}));
    let grant = p.read();
    let barrier = p.read();
    p.answer(&grant, json!({"held":true}));
    assert_eq!(p.read()["method"], "core.unlock");
    p.answer(&barrier, json!({}));
    let kept = p.read();
    p.answer(&kept, json!({"held":true}));
    assert_eq!(p.read()["result"], "late done");
    assert!(p.child.wait().unwrap().success());
    let mut p = start(&wasm, &dir);
    p.request("wait", json!({}));
    assert_eq!(p.read()["method"], "other.wait");
    drop(p.input.take());
    assert_eq!(p.read()["params"]["name"], "stopped");
    assert!(p.child.wait().unwrap().success());
    assert_eq!(std::fs::read(dir.join("stopped")).unwrap(), b"yes");
    let mut p = start(&wasm, &dir);
    p.request("start", json!({}));
    let first = p.read();
    let second = p.read();
    let outgoing = if first["method"] == "other.background" {
        first
    } else {
        second
    };
    assert_eq!(outgoing["method"], "other.background");
    p.answer(&outgoing, json!({}));
    assert!(p.child.wait().unwrap().success());
    // Replacement is a notification, not EOF. Drain keeps both endpoint handlers
    // and background calls alive, while closing idle connections and the listener.
    assert!(
        !d["result"]["methods"]
            .as_array()
            .unwrap()
            .iter()
            .any(|m| m["name"] == "rpc.superseded" || m["name"] == "superseded")
    );
    std::fs::remove_file(dir.join("stopped")).unwrap();
    let mut provider = start(&wasm, &dir);
    provider.request("serve", json!({}));
    assert_eq!(provider.read()["result"], Value::Null);
    provider.request("start", json!({}));
    let first = provider.read();
    let second = provider.read();
    let background = if first["method"] == "other.background" {
        first
    } else {
        second
    };
    assert_eq!(background["method"], "other.background");
    let address = std::fs::read_to_string(dir.join("endpoint"))
        .unwrap()
        .split_whitespace()
        .next()
        .unwrap()
        .trim_start_matches("tcp:")
        .to_owned();
    let mut idle = std::net::TcpStream::connect(&address).unwrap();
    idle.set_read_timeout(Some(std::time::Duration::from_secs(4)))
        .unwrap();
    writeln!(
        idle,
        "{}",
        json!({"jsonrpc":"2.0","id":0,"method":"rpc.endpoint","params":{}})
    )
    .unwrap();
    let mut idle = BufReader::new(idle);
    let mut line = String::new();
    idle.read_line(&mut line).unwrap();
    assert_eq!(
        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["id"],
        0
    );
    let mut active = std::net::TcpStream::connect(&address).unwrap();
    active
        .set_read_timeout(Some(std::time::Duration::from_secs(4)))
        .unwrap();
    writeln!(
        active,
        "{}",
        json!({"jsonrpc":"2.0","id":1,"method":"wait","params":{}})
    )
    .unwrap();
    let inflight = provider.read();
    assert_eq!(inflight["method"], "other.inflight");
    provider
        .send(json!({"jsonrpc":"2.0","method":"rpc.superseded","params":{"current":"new-hash"}}));
    let hook = provider.read();
    assert_eq!(hook["method"], "other.superseded");
    assert_eq!(hook["params"]["current"], "new-hash");
    provider.answer(&hook, json!({}));
    assert_eq!(provider.read()["params"]["name"], "draining");
    line.clear();
    assert_eq!(idle.read_line(&mut line).unwrap(), 0);
    // drain() wakes independent accept/read cancellations. Idle EOF does not
    // order listener destruction: TCP may still complete a handshake, but that
    // connection must close without serving work. Do not retry the connect.
    if let Ok(mut late) = std::net::TcpStream::connect(&address) {
        eprintln!("TCP handshake raced listener cancellation; waiting for close");
        // Reuse the four-second socket bound used by idle/active above.
        late.set_read_timeout(Some(std::time::Duration::from_secs(4)))
            .unwrap();
        let closed = std::io::Read::read(&mut late, &mut [0]);
        assert!(
            matches!(closed, Ok(0))
                || matches!(&closed, Err(e) if e.kind() == std::io::ErrorKind::ConnectionReset),
            "late connection survived drain: {closed:?}"
        );
    }
    assert!(provider.child.try_wait().unwrap().is_none());
    provider.answer(&inflight, json!({}));
    line.clear();
    BufReader::new(active).read_line(&mut line).unwrap();
    assert_eq!(
        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["result"],
        "finished"
    );
    assert!(provider.child.try_wait().unwrap().is_none());
    provider.answer(&background, json!({}));
    provider.finishes();
    assert!(std::net::TcpStream::connect(&address).is_err());
    assert!(
        !dir.join("stopped").exists(),
        "drain is not the EOF stop hook"
    );
    let provider_doc = &d["result"];
    let imports = target.join("imports");
    std::fs::create_dir_all(&imports).unwrap();
    std::fs::write(
        imports.join("probe.json"),
        sausage_plugin::serde_json::to_vec_pretty(provider_doc).unwrap(),
    )
    .unwrap();
    let generated = sausage_interface::client_rs("probe", provider_doc).unwrap();
    std::fs::write(target.join("client.rs"), generated).unwrap();
    assert!(
        Command::new("cargo")
            .env("CARGO_TARGET_DIR", &target)
            .args(["build", "--offline", "--manifest-path"])
            .arg(root.join("tests/plugins/consumer/Cargo.toml"))
            .args(["--target", "wasm32-wasip2"])
            .status()
            .unwrap()
            .success()
    );
    let consumer = target.join("wasm32-wasip2/debug/sdk-consumer.wasm");
    let mut client = start(&consumer, &dir);
    client.request("rpc.discover", json!({}));
    let consumer_doc = client.read()["result"].clone();
    assert!(client.child.wait().unwrap().success());
    let imported = &consumer_doc["x-sausage-imports"]["probe"];
    assert!(sausage_interface::compat::check(imported, provider_doc).is_empty());
    let mut broken = provider_doc.clone();
    broken["methods"]
        .as_array_mut()
        .unwrap()
        .retain(|m| m["name"] != "add");
    assert!(!sausage_interface::compat::check(imported, &broken).is_empty());
    let mut client = start(&consumer, &dir);
    client.request("run", json!({}));
    let outgoing = client.read();
    assert_eq!(outgoing["method"], "core.call");
    assert_eq!(outgoing["params"]["method"], "probe.add");
    let mut provider = start(&wasm, &dir);
    provider.request("add", outgoing["params"]["params"].clone());
    let reply = provider.read();
    assert!(provider.child.wait().unwrap().success());
    client.send(json!({"jsonrpc":"2.0","method":"core.notify","params":{"call":outgoing["id"],"name":"typed","params":{}}}));
    client.answer(&outgoing, reply["result"].clone());
    assert_eq!(client.read()["result"], 5);
    assert!(client.child.wait().unwrap().success());
    let mut provider = start(&wasm, &dir);
    provider.request("serve", json!({}));
    assert_eq!(provider.read()["result"], Value::Null);
    let mut client = start(&consumer, &dir);
    client.request("endpoint", json!({}));
    assert_eq!(client.read()["result"], 7);
    assert!(client.child.wait().unwrap().success());
    let mut stderr = String::new();
    std::io::Read::read_to_string(client.child.stderr.as_mut().unwrap(), &mut stderr).unwrap();
    assert!(stderr.contains("endpoint notification: endpoint-progress"));
    let path = dir.join("endpoint");
    let original = std::fs::read_to_string(&path).unwrap();
    std::fs::write(&path, original.replace("probe-00000000", "probe-11111111")).unwrap();
    let mut client = start(&consumer, &dir);
    client.request("endpoint", json!({}));
    assert!(
        client.read()["error"]["message"]
            .as_str()
            .unwrap()
            .contains("identity mismatch")
    );
    assert!(client.child.wait().unwrap().success());
    std::fs::write(&path, original).unwrap();
    // The endpoint server ignores non-RPC lines before a request.
    let address = std::fs::read_to_string(&path)
        .unwrap()
        .split_whitespace()
        .next()
        .unwrap()
        .trim_start_matches("tcp:")
        .to_owned();
    let mut stream = std::net::TcpStream::connect(address).unwrap();
    stream
        .set_read_timeout(Some(std::time::Duration::from_secs(4)))
        .unwrap();
    stream.write_all(b"\nnot-json\n[]\n{\"jsonrpc\":\"2.0\",\"id\":99,\"method\":\"rpc.endpoint\",\"params\":{}}\n").unwrap();
    let mut reader = BufReader::new(stream);
    let mut line = String::new();
    reader.read_line(&mut line).unwrap();
    assert_eq!(
        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["id"],
        99
    );
    reader
        .get_mut()
        .write_all(b"{\"jsonrpc\":\"2.0\",\"id\":100,\"method\":\"stop\",\"params\":{}}\n")
        .unwrap();
    line.clear();
    reader.read_line(&mut line).unwrap();
    assert_eq!(
        sausage_plugin::serde_json::from_str::<Value>(&line).unwrap()["id"],
        100
    );
    provider.finishes();

    // The endpoint client ignores non-RPC lines during verification and responses.
    let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
    let owner = "probe-00000000-0000-7000-8000-000000000000";
    std::fs::write(
        &path,
        format!("tcp:{} {owner} control\n", listener.local_addr().unwrap()),
    )
    .unwrap();
    let server = std::thread::spawn(move || {
        for _ in 0..2 {
            let (stream, _) = listener.accept().unwrap();
            stream
                .set_read_timeout(Some(std::time::Duration::from_secs(4)))
                .unwrap();
            let mut reader = BufReader::new(stream);
            loop {
                let mut line = String::new();
                if reader.read_line(&mut line).unwrap() == 0 {
                    break;
                }
                let request: Value = sausage_plugin::serde_json::from_str(&line).unwrap();
                let result = if request["method"] == "rpc.endpoint" {
                    json!({"owner":owner,"plugin":"probe","interface":"control"})
                } else {
                    json!(7)
                };
                writeln!(
                    reader.get_mut(),
                    "\nnot-json\n[]\n{}",
                    json!({"jsonrpc":"2.0","id":request["id"],"result":result})
                )
                .unwrap();
            }
        }
    });
    let mut client = start(&consumer, &dir);
    client.request("endpoint", json!({}));
    assert_eq!(client.read()["result"], 7);
    client.finishes();
    server.join().unwrap();
    assert_eq!(
        std::fs::read(root.join("tests/plugins/consumer/imports/probe.json")).unwrap(),
        include_bytes!("plugins/consumer/imports/probe.json")
    );
    std::fs::remove_dir_all(dir).unwrap();
}