open-standards/joule-code/joule-code-rs/crates/joule-moq-room/tests/cli_dev_room.rs

//! The development room's doors on the command line, end to end. `wai dev --control` runs WAI's
//! one local stack in a project directory (the relay and this crate's room service as processes
//! of their own, and this crate's synthetic participants), and in that directory
//!
//! - `wai room join --principal NAME --yes` has the stack's room service grant the test principal
//!   the publisher role, and its participants enter it and publish with receipts, which the relay
//!   serves; the stack takes the line on its origin while its standard input is the test's;
//! - `wai session show` gives the room as a verifier is given it, which wai-rs's live-rooms
//!   verifier accepts with both principals present; after `wai room leave` it accepts with one;
//! - `wai room show --target local` reads the room from the stack's room service, through its
//!   origin;
//! - `wai log tail` reads the relay's log, and `wai session list` lists the signed summary of the
//!   session the relay closed when one left;
//! - `wai room invite` signs an invitation the room's authority holds;
//! - and once the stack has stopped, a door finds no stack.
//!
//! `wai` is built from `wai/wai-rs` with the feature `mcp` into the target directory
//! tests/mcp_dev_room.rs builds it in, and the relay `wai-relay` from the joule-code workspace.

use serde_json::Value;
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
use std::time::{Duration, Instant};

fn cargo() -> String {
    std::env::var("CARGO").ok().or(option_env!("CARGO").map(String::from)).unwrap_or_else(|| "cargo".into())
}

fn exe(name: &str) -> String {
    if cfg!(windows) { format!("{name}.exe") } else { name.to_string() }
}

fn wai_bin() -> PathBuf {
    let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../../wai/wai-rs/Cargo.toml");
    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("wai-mcp");
    let status = Command::new(cargo())
        .args(["build", "--locked", "--no-default-features", "--features", "mcp", "--bin", "wai", "--manifest-path"])
        .arg(&manifest)
        .env("CARGO_TARGET_DIR", &target)
        .status()
        .expect("cargo runs");
    assert!(status.success(), "the wai binary builds with the feature mcp");
    target.join("debug").join(exe("wai"))
}

fn relay_bin() -> PathBuf {
    let manifest = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../Cargo.toml");
    let target = Path::new(env!("CARGO_TARGET_TMPDIR")).join("wai-relay");
    let status = Command::new(cargo())
        .args(["build", "--locked", "-p", "joule-moq-relay", "--bin", "wai-relay", "--manifest-path"])
        .arg(&manifest)
        .env("CARGO_TARGET_DIR", &target)
        .status()
        .expect("cargo runs");
    assert!(status.success(), "the relay builds");
    target.join("debug").join(exe("wai-relay"))
}

fn text(o: &Output) -> String {
    format!("{}{}", String::from_utf8_lossy(&o.stdout), String::from_utf8_lossy(&o.stderr))
}

/// `wai` in the project, as an agent runs it: `--json`, no terminal; its exit and its envelope.
fn wai(bin: &Path, project: &Path, args: &[&str]) -> (Option<i32>, Value) {
    let o = Command::new(bin).current_dir(project).args(args).arg("--json").stdin(Stdio::null()).output().expect("wai runs");
    let v = serde_json::from_slice(&o.stdout).unwrap_or_else(|e| panic!("{args:?}: not an envelope ({e}): {}", text(&o)));
    (o.status.code(), v)
}

/// How long [`Running`] waits for `wai dev`, once killed, to be reaped. `SIGKILL` ends a process
/// at once unless it is stuck in the kernel; a drop that waited on such a one without a bound
/// would hang the test binary instead of failing it.
const REAPED_WITHIN: Duration = Duration::from_secs(10);

/// `wai dev`, spawned as the leader of a process group of its own, which the relay, the room
/// service and the participants it starts join: killed with its whole group, and reaped, when
/// it is dropped, at the end of the test or while a failed assertion unwinds through it. A
/// [`Child`] is not killed when it is dropped, and `wai dev` killed alone would leave its
/// services running. It reads as the child it holds.
struct Running(Child);

impl Running {
    /// Kills the group and reaps its leader, waiting up to [`REAPED_WITHIN`]: whether it was
    /// reaped. The group first, while its leader is not yet reaped: until then the leader's pid,
    /// which is the group's id, is no other process's. A leader that has already ended is reaped
    /// and not signalled, and neither is its group: `wai dev` that ends of itself has stopped its
    /// services.
    fn stop(&mut self) -> bool {
        #[cfg(unix)]
        if matches!(self.0.try_wait(), Ok(None)) {
            let group = format!("-{}", self.0.id());
            let _ = Command::new("kill").args(["-KILL", "--", &group]).stdout(Stdio::null()).stderr(Stdio::null()).status();
        }
        let _ = self.0.kill();
        let t = Instant::now();
        loop {
            match self.0.try_wait() {
                Ok(Some(_)) => return true,
                Ok(None) if t.elapsed() < REAPED_WITHIN => std::thread::sleep(Duration::from_millis(10)),
                _ => return false,
            }
        }
    }
}

impl std::ops::Deref for Running {
    type Target = Child;

    fn deref(&self) -> &Child {
        &self.0
    }
}

impl std::ops::DerefMut for Running {
    fn deref_mut(&mut self) -> &mut Child {
        &mut self.0
    }
}

impl Drop for Running {
    fn drop(&mut self) {
        if !self.stop() {
            eprintln!("process {} was killed and not reaped within {} s", self.0.id(), REAPED_WITHIN.as_secs());
        }
    }
}

fn unhex(s: &str) -> Vec<u8> {
    (0..s.len()).step_by(2).map(|i| u8::from_str_radix(&s[i..i + 2], 16).unwrap()).collect()
}

/// The session `wai session show` gave, verified as live-rooms §9 verifies a room: the outcome
/// and the principals present.
fn verified(s: &Value) -> (String, Vec<String>) {
    let list = |v: &Value| v.as_array().unwrap().iter().map(|x| unhex(x.as_str().unwrap())).collect::<Vec<_>>();
    let chains: Vec<Vec<Vec<u8>>> = s["chains"].as_array().unwrap().iter().map(list).collect();
    let run = wai::live_room::verify_room(&unhex(s["descriptor"].as_str().unwrap()), &list(&s["grants"]), s["key_log_head"].as_str(), &chains);
    let outcome = match run.outcome {
        wai::live_room::RoomOutcome::Accept => "accept".to_string(),
        _ => "not accepted".to_string(),
    };
    let mut present: Vec<String> = run.state.present.keys().map(|p| p.iter().map(|b| format!("{b:02x}")).collect()).collect();
    present.sort();
    (outcome, present)
}

#[test]
fn the_command_line_joins_reads_and_leaves_the_stack_wai_dev_runs() {
    let wai_path = wai_bin();
    let relay = relay_bin();
    let project = std::env::temp_dir().join(format!("joule-moq-room-cli-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&project);
    std::fs::create_dir_all(&project).unwrap();
    let mut cmd = Command::new(&wai_path);
    cmd.current_dir(&project)
        .args(["dev", "--control", "--json", "--port", "0", "--publishers", "4", "--relay"])
        .arg(&relay)
        .arg("--room")
        .arg(env!("CARGO_BIN_EXE_wai-room"))
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::null());
    #[cfg(unix)]
    std::os::unix::process::CommandExt::process_group(&mut cmd, 0);
    // Held as it starts: a test that fails before the stack is asked to stop kills it, with the
    // relay, the room service and the participants.
    let mut dev = Running(cmd.spawn().expect("wai dev starts"));
    let mut stdin = dev.stdin.take().unwrap();
    let mut out = BufReader::new(dev.stdout.take().unwrap());
    let mut first = String::new();
    out.read_line(&mut first).unwrap();
    let ready: Value = serde_json::from_str(&first).unwrap_or_else(|e| panic!("the ready line ({e}): {first:?}"));
    assert_eq!(ready["ready"], true, "{ready}");
    let room = ready["room"].as_str().unwrap().to_string();
    assert!(project.join(".wai/dev/stack.json").is_file(), "the stack is recorded where the doors look");

    let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
        let w = |args: &[&str]| wai(&wai_path, &project, args);
        // Two test principals join, through the stack's origin.
        let mut keys = Vec::new();
        for name in ["alice", "bob"] {
            let (code, v) = w(&["room", "join", "--principal", name, "--objects", "2", "--yes"]);
            assert_eq!(code, Some(0), "{name}: {v}");
            let d = &v["data"];
            assert_eq!((d["events"].clone(), d["room"].as_str()), (serde_json::json!(["join", "publish", "publish"]), Some(room.as_str())), "{name}: {v}");
            assert!(d["objects_served"].as_u64().unwrap() >= 2, "the relay serves what it published with its receipts: {v}");
            keys.push(d["key"].as_str().unwrap().to_string());
        }
        let (alice, bob) = (keys[0].clone(), keys[1].clone());
        keys.sort();
        // The session, verified: both present.
        let (code, s) = w(&["session", "show"]);
        assert_eq!(code, Some(0), "{s}");
        assert_eq!(s["data"]["room"].as_str(), Some(room.as_str()), "{s}");
        assert_eq!(verified(&s["data"]), ("accept".to_string(), keys.clone()));
        // The room, from the stack's room service.
        let (code, r) = w(&["room", "show", "--target", "local"]);
        assert_eq!((code, r["data"]["status"].as_u64(), r["data"]["target"]["name"].as_str()), (Some(0), Some(200), Some("local")), "{r}");
        // Alice leaves; the session verifies with Bob alone, and the relay has a summary of hers.
        let (code, l) = w(&["room", "leave", "--principal", "alice", "--yes"]);
        assert_eq!(code, Some(0), "{l}");
        assert!(l["data"]["events"].as_array().unwrap().iter().any(|e| e == "leave"), "{l}");
        let (_, s) = w(&["session", "show"]);
        assert_eq!(verified(&s["data"]), ("accept".to_string(), vec![bob.clone()]), "bob alone, after {alice} left");
        let until = Instant::now() + Duration::from_secs(20);
        let summaries = loop {
            let (code, v) = w(&["session", "list"]);
            assert_eq!(code, Some(0), "{v}");
            let n = v["data"]["sessions"].as_array().unwrap().len();
            if n > 0 || Instant::now() > until {
                break v;
            }
            std::thread::sleep(Duration::from_millis(250));
        };
        let first = &summaries["data"]["sessions"][0];
        assert!(first["summary"]["session"].is_u64() && first["signature"].as_str().is_some_and(|s| s.len() == 128), "{summaries}");
        // The relay's log, read from the stack's directory.
        let (code, t) = w(&["log", "tail", "--name", "relay"]);
        assert_eq!(code, Some(0), "{t}");
        assert!(!t["data"]["text"].as_str().unwrap().is_empty() && t["data"]["next"].as_u64().unwrap() > 0, "{t}");
        // An invitation the room's authority signed.
        let (code, i) = w(&["room", "invite", "--seconds", "600"]);
        assert_eq!(code, Some(0), "{i}");
        let authority: [u8; 32] = unhex(ready["authority"].as_str().unwrap()).try_into().unwrap();
        let vk = ed25519_dalek::VerifyingKey::from_bytes(&authority).unwrap();
        let now = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_secs();
        let (_, end) = joule_moq_room::remote::read_invitation(i["data"]["invitation"].as_str().unwrap(), &vk, now).expect("the room's authority signed it");
        assert!(end > now + 500 && end <= now + 601, "good for the seconds asked: {end} {now}");
    }));

    // The stack stops at the end of its standard input, and its record goes with it.
    drop(stdin.write_all(b"{\"op\":\"stop\"}\n"));
    drop(stdin);
    let until = Instant::now() + Duration::from_secs(30);
    while Instant::now() < until && matches!(dev.try_wait(), Ok(None)) {
        std::thread::sleep(Duration::from_millis(100));
    }
    // A stack that has not stopped by now is killed, with its group.
    dev.stop();
    if let Err(e) = result {
        let _ = std::fs::remove_dir_all(&project);
        std::panic::resume_unwind(e);
    }
    assert!(!project.join(".wai/dev/stack.json").exists(), "the record is removed when the stack stops");
    let (code, v) = wai(&wai_path, &project, &["session", "show"]);
    assert_eq!((code, v["error"]["code"].as_str()), (Some(5), Some("stack-not-found")), "{v}");
    let _ = std::fs::remove_dir_all(&project);
}