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

//! The local stack's ports, held in one place. Every default port of every program, and every
//! port one program's example configuration names to reach another, agrees with the program that
//! serves it, and with the port table of wai/quickstart/README.md:
//!
//! - wai-relay: relay.example.toml's endpoints are `wai-relay --dev`'s (`RelayConfig::dev`);
//! - wai-room: room.example.toml's `relay_url` is the relay example's WebSocket endpoint and path;
//! - wai-ingest: gateway.example.toml, read by the gateway's own reader (`GatewayConfig`): its
//!   `[relay] addr` is the relay example's QUIC endpoint, and each listener it names is a row,
//!   those it leaves commented out (off until set) said to be, each of those read as the gateway
//!   reads it once its line is uncommented;
//! - `wai dev`'s origin, which the remote desktop template's host.sh reaches by default;
//! - every other daemon whose default is in its source: the audit collector, the ledger, the
//!   fleet service, the quantum job service, the worlds' relay, the stitched presentation
//!   verifier and `wai http origin`;
//!
//! no two defaults are one port; the room service started from its example file opens its
//! clock's session on the relay started from the relay's, each moved to a free port alike; and
//! the gateway started from its example, its ports moved to free ones and its files made, opens
//! the listeners the table says are on and none of those it says are off.

use joule_moq_ingest::{GatewayConfig, GatewayServer};
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::keys::Keys;
use joule_moq_room::{Config, Service};
use std::collections::BTreeMap;
use std::net::SocketAddr;
use std::path::{Path, PathBuf};
use std::sync::atomic::Ordering;
use std::time::Duration;

fn repo() -> PathBuf {
    Path::new(env!("CARGO_MANIFEST_DIR")).join("../../../..")
}

fn read(rel: &str) -> String {
    std::fs::read_to_string(repo().join(rel)).unwrap_or_else(|e| panic!("{rel}: {e}"))
}

const ROOM_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-room/room.example.toml";
const RELAY_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-relay/relay.example.toml";
const GATEWAY_EXAMPLE: &str = "joule-code/joule-code-rs/crates/joule-moq-ingest/gateway.example.toml";
const TABLE: &str = "wai/quickstart/README.md";

/// `ws://host:port/path` as (scheme, authority, path).
fn split_url(u: &str) -> (&str, &str, &str) {
    let (scheme, rest) = u.split_once("://").unwrap_or_else(|| panic!("{u}: a URL"));
    let (authority, path) = rest.find('/').map_or((rest, ""), |i| (&rest[..i], &rest[i..]));
    (scheme, authority, path)
}

/// A row of the port table: its program, what listens there, and the setting that moves it.
struct Row {
    program: String,
    what: String,
    setting: String,
}

/// What a row says of a listener that is off until its setting is given.
const OFF: &str = "off until set";

/// The port table: each row's port (with `/udp` for a UDP one) and the row.
fn table() -> BTreeMap<String, Row> {
    let text = read(TABLE);
    let heading = "\n## Ports\n";
    let at = text.find(heading).expect("the README has a port table") + heading.len();
    let mut rows = BTreeMap::new();
    for line in text[at..].lines().take_while(|l| !l.starts_with("## ")) {
        let cells: Vec<&str> = line.split('|').map(str::trim).collect();
        if cells.len() < 6 || !cells[1].starts_with(|c: char| c.is_ascii_digit()) {
            continue;
        }
        let unquote = |s: &str| s.trim_matches('`').to_string();
        let row = Row { program: unquote(cells[2]), what: cells[3].to_string(), setting: unquote(cells[4]) };
        assert!(rows.insert(cells[1].to_string(), row).is_none(), "{}: twice in the table", cells[1]);
    }
    rows
}

/// `text` with the line that gives `key` in its table `[section]` uncommented, when the example
/// leaves it commented out (a listener the program opens only once it is set); `None` when the
/// line is not commented out.
fn uncommented(text: &str, section: &str, key: &str) -> Option<String> {
    let header = format!("[{section}]");
    let lines: Vec<&str> = text.lines().collect();
    let at = lines.iter().position(|l| l.trim() == header).unwrap_or_else(|| panic!("no {header}"));
    for (i, line) in lines.iter().enumerate().skip(at + 1).take_while(|(_, l)| !l.trim_start().starts_with('[')) {
        let (off, body) = match line.trim_start().strip_prefix('#') {
            Some(rest) => (true, rest.trim_start()),
            None => (false, line.trim_start()),
        };
        if body.strip_prefix(key).is_some_and(|r| r.trim_start().starts_with('=')) {
            if !off {
                return None;
            }
            let mut out: Vec<String> = lines.iter().map(|l| l.to_string()).collect();
            out[i] = body.to_string();
            return Some(out.join("\n"));
        }
    }
    panic!("{header} names no {key}")
}

/// The gateway's listeners as its reader reads them, each with its section, its key, whether it
/// is UDP, and its address in a configuration.
type Listener = (&'static str, &'static str, bool, fn(&GatewayConfig) -> Option<SocketAddr>);
const GATEWAY_LISTENERS: &[Listener] = &[
    ("http", "listen", false, |c| Some(c.http.listen)),
    ("rtmp", "listen", false, |c| c.rtmp.listen),
    ("rtmp", "tls_listen", false, |c| c.rtmp.tls_listen),
    ("injector", "listen", false, |c| c.injector.listen),
    ("contrib.srt", "listen", true, |c| c.contrib.srt.listen),
];

/// The numbers of four or five digits on a line of source: the ports it names.
fn ports_on(line: &str) -> Vec<u16> {
    line.split(|c: char| !c.is_ascii_digit()).filter(|n| (4..=5).contains(&n.len())).filter_map(|n| n.parse().ok()).collect()
}

#[test]
fn every_default_port_agrees_with_the_program_that_serves_it() {
    let t = table();
    let mut held = Vec::new();
    let mut row = |port: String, program: &str, setting: &str, off: bool| {
        let r = t.get(&port).unwrap_or_else(|| panic!("the table has no row for {port} ({program})"));
        assert_eq!((r.program.as_str(), r.setting.as_str()), (program, setting), "the table's row for {port}");
        assert_eq!(r.what.contains(OFF), off, "the table's row for {port} says {OFF:?} exactly when the listener is off until set");
        held.push(port);
    };

    // The relay: its example's endpoints are its development defaults, and the table's.
    let relay = RelayConfig::from_toml(&read(RELAY_EXAMPLE)).expect("the relay's example reads");
    let dev = RelayConfig::dev();
    assert_eq!((relay.quic_listen, relay.websocket_listen, relay.path.as_str()), (dev.quic_listen, dev.websocket_listen, dev.path.as_str()), "{RELAY_EXAMPLE} is `wai-relay --dev`");
    let quic = relay.quic_listen.expect("a QUIC endpoint");
    let ws = relay.websocket_listen.expect("a WebSocket endpoint");
    row(format!("{}/udp", quic.port()), "wai-relay", "quic_listen", false);
    row(ws.port().to_string(), "wai-relay", "websocket_listen", false);

    // The room service: its relay is the relay's WebSocket endpoint, at the relay's path.
    let room = Config::from_toml(&read(ROOM_EXAMPLE)).expect("the room's example reads");
    let (scheme, authority, path) = split_url(&room.relay_url);
    let at: SocketAddr = authority.parse().unwrap_or_else(|e| panic!("{authority}: {e}"));
    assert_eq!((scheme, at, path), ("ws", ws, relay.path.as_str()), "{ROOM_EXAMPLE}'s relay_url is the relay's WebSocket endpoint");
    row(room.listen.port().to_string(), "wai-room", "listen", false);

    // The ingest gateway, its example read by its own reader: its relay is the relay's QUIC
    // endpoint, and each listener its example names is a row. One the example leaves commented
    // out is off (the reader reads none) and its row says so; uncommented, the reader reads it at
    // the row's port.
    let text = read(GATEWAY_EXAMPLE);
    let gw = GatewayConfig::from_toml(&text).unwrap_or_else(|e| panic!("{GATEWAY_EXAMPLE} reads as the gateway reads it: {e}"));
    assert_eq!(gw.relay.addr, quic, "{GATEWAY_EXAMPLE}'s [relay] addr is the relay's QUIC endpoint");
    for (section, key, udp, at) in GATEWAY_LISTENERS {
        let (a, off) = match uncommented(&text, section, key) {
            None => (at(&gw).unwrap_or_else(|| panic!("{GATEWAY_EXAMPLE}: [{section}] {key} is set, and the gateway reads none")), false),
            Some(on) => {
                assert_eq!(at(&gw), None, "{GATEWAY_EXAMPLE}: [{section}] {key} is commented out, and the gateway reads one");
                let c = GatewayConfig::from_toml(&on).unwrap_or_else(|e| panic!("{GATEWAY_EXAMPLE} with [{section}] {key} set: {e}"));
                (at(&c).unwrap_or_else(|| panic!("{GATEWAY_EXAMPLE}: [{section}] {key}, uncommented, is not what the gateway reads")), true)
            }
        };
        let port = format!("{}{}", a.port(), if *udp { "/udp" } else { "" });
        row(port, "wai-ingest", &format!("[{section}] {key}"), off);
    }

    // `wai dev`'s origin: its default, and the one the remote desktop template's host reaches.
    let dev_rs = read("wai/wai-rs/src/bin/wai_cli/dx/dev.rs");
    let origin = t.iter().find(|(_, r)| r.program == "wai dev").map(|(port, _)| port.clone()).expect("a row for wai dev");
    assert!(dev_rs.contains(&format!("None => {origin},")), "wai dev's default port is {origin}");
    row(origin.clone(), "wai dev", "--port", false);
    assert!(read("wai/examples/remote-desktop/host.sh").contains(&format!("WAI_DEV_ORIGIN:-http://127.0.0.1:{origin}}}")), "host.sh reaches wai dev's origin");

    // The daemons whose defaults are in their sources: each source, the text on the line that
    // reads the setting, the program and the setting. That line names one port, the table's.
    let sources = [
        ("wai/wai-audit/src/main.rs", "env_or(\"WAI_AUDIT_ADDR\"", "wai-audit", "WAI_AUDIT_ADDR"),
        ("wai/wai-ledger/src/main.rs", "var(\"WAI_LEDGER_ADDR\")", "wai-ledger", "WAI_LEDGER_ADDR"),
        ("wai/wai-fleet/src/main.rs", "var(\"WAI_FLEET_ADDR\")", "wai-fleet", "WAI_FLEET_ADDR"),
        ("wai/wai-quantum-service/src/main.rs", "var(\"WAI_QSVC_ADDR\")", "wai-quantum-service", "WAI_QSVC_ADDR"),
        ("wai/wai-relay/src/main.rs", "var(\"WAI_RELAY_BIND\")", "wai-world-relay", "WAI_RELAY_BIND"),
        ("wai/wai-rs/src/bin/wai_stitched_presentation.rs", "opt(a, \"--listen\")", "wai_stitched_presentation serve", "--listen"),
        ("wai/wai-rs/src/bin/wai_cli/http.rs", "opt(args, \"--port\").map_or(", "wai http origin", "--port"),
    ];
    for (src, reads, program, setting) in sources {
        let text = read(src);
        let lines: Vec<&str> = text.lines().filter(|l| l.contains(reads) && !l.trim_start().starts_with("//")).collect();
        assert_eq!(lines.len(), 1, "{src}: one line reads {setting}: {lines:?}");
        let ports = ports_on(lines[0]);
        assert_eq!(ports.len(), 1, "{src}: the line that reads {setting} names one port: {}", lines[0]);
        row(ports[0].to_string(), program, setting, false);
    }

    // Every row is one held above; the table holds each port once, so no two programs' defaults
    // are one port on one protocol.
    held.sort();
    assert_eq!(held, t.keys().cloned().collect::<Vec<_>>(), "the rows held and the table's");
}

/// The room service, started from room.example.toml, opens its clock's session on the relay
/// started from relay.example.toml. Both move to free ports alike (the room's `relay_url` keeps
/// its scheme, host and path, its port the relay's), and the relay keeps nothing on disk.
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_room_example_reaches_the_relay_example() {
    let mut rc = RelayConfig::from_toml(&read(RELAY_EXAMPLE)).unwrap();
    rc.quic_listen = None;
    rc.websocket_listen = Some("127.0.0.1:0".parse().unwrap());
    rc.dvr = None;
    let relay = RelayServer::start(rc).await.expect("the relay starts from its example");
    let ws = relay.websocket_addr.expect("its WebSocket endpoint");

    let text = read(ROOM_EXAMPLE);
    let example = Config::from_toml(&text).unwrap();
    let (scheme, authority, path) = split_url(&example.relay_url);
    let moved = format!("{scheme}://{}:{}{path}", authority.rsplit_once(':').unwrap().0, ws.port());
    let text = text
        .replace(&format!("relay_url = \"{}\"", example.relay_url), &format!("relay_url = \"{moved}\""))
        .replace(&format!("listen = \"{}\"", example.listen), "listen = \"127.0.0.1:0\"");
    let cfg = Config::from_toml(&text).unwrap();
    assert_eq!(cfg.relay_url, moved, "the example's relay_url, moved");
    let svc = Service::new(Keys::generate(), &cfg);
    tokio::spawn(svc.clone().run_clock());
    let up = tokio::time::timeout(Duration::from_secs(30), async {
        while !svc.connected.load(Ordering::Acquire) {
            tokio::time::sleep(Duration::from_millis(50)).await;
        }
    })
    .await;
    assert!(up.is_ok(), "the clock's session opens on the relay at {moved}");
}

/// The gateway, started from gateway.example.toml with its ports moved to free ones and the
/// files it names made (its signing key, the relay's certificate and token, a room's signed
/// descriptor): it serves HTTP, opens RTMP, which the example sets, and none of the listeners
/// the example leaves off (RTMPS, the SCTE 104 injector, SRT).
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn the_gateway_starts_from_its_example() {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let dir = std::env::temp_dir().join(format!("ports-gateway-{}", std::process::id()));
    std::fs::create_dir_all(&dir).unwrap();
    let keys = Keys::generate();
    let now_us = std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).unwrap().as_micros() as u64;
    let room = joule_moq_room::epoch::descriptor(&keys, joule_moq_room::epoch::roles(false, 0, false, false), 8, 4, now_us);
    let mut text = read(GATEWAY_EXAMPLE);
    let example = GatewayConfig::from_toml(&text).unwrap();
    for (name, bytes) in [("gateway.key", vec![7u8; 32]), ("relay-cert.pem", Vec::new()), ("relay-token", b"token".to_vec()), ("room.wrd1", room.to_bytes())] {
        std::fs::write(dir.join(name), bytes).unwrap();
        let from = format!("\"{name}\"");
        assert_eq!(text.matches(&from).count(), 1, "{GATEWAY_EXAMPLE} names {name} once");
        text = text.replace(&from, &format!("{:?}", dir.join(name).to_str().unwrap()));
    }
    for at in [Some(example.http.listen), example.rtmp.listen] {
        let from = format!("listen = \"{}\"", at.unwrap());
        assert_eq!(text.matches(&from).count(), 1, "{GATEWAY_EXAMPLE}: {from} once");
        text = text.replace(&from, "listen = \"127.0.0.1:0\"");
    }
    let cfg = GatewayConfig::from_toml(&text).unwrap();
    let gw = GatewayServer::start(cfg).await.unwrap_or_else(|e| panic!("the gateway starts from its example: {e}"));
    assert!(gw.addr.ip().is_loopback() && gw.rtmp_addr.is_some_and(|a| a.ip().is_loopback()), "HTTP and RTMP, on loopback");
    assert_eq!((gw.rtmps_addr, gw.injector_addr, gw.srt_addr), (None, None, None), "RTMPS, the injector and SRT, off until set");
    assert!(gw.gateway.rooms.contains_key(&room.room_id()), "the room its example names");
    let mut s = tokio::net::TcpStream::connect(gw.addr).await.unwrap();
    s.write_all(b"GET / HTTP/1.1\r\nHost: localhost\r\nConnection: close\r\n\r\n").await.unwrap();
    let mut head = [0u8; 9];
    s.read_exact(&mut head).await.unwrap();
    assert_eq!(&head, b"HTTP/1.1 ", "the gateway answers HTTP at its listener");
    let _ = std::fs::remove_dir_all(&dir);
}