//! Interoperation with the reference tools of each transport, run locally as test peers:
//! their SRT caller and listener, their RIST sender and receiver of both profiles, against
//! this crate's listener, caller, receiver and sender, byte for byte, in the clear and
//! encrypted, and through a path that drops datagrams each way, so each side's
//! retransmission answers the other's requests.
//!
//! The tools are named by environment variables, and these tests are ignored unless asked
//! for: run them with `--ignored`, and then a tool that is not named is a failure, never a
//! skip.
//!
//! ```text
//! WAI_SRT_LIVE_TRANSMIT=…/srt-live-transmit WAI_RISTSENDER=…/ristsender \
//! WAI_RISTRECEIVER=…/ristreceiver cargo test -p joule-contrib --features rist --test interop -- --ignored
//! ```

use joule_contrib::link::Link;
use joule_contrib::lossy::{Impair, Proxy};
use joule_contrib::srt::{self, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
use std::net::{SocketAddr, UdpSocket};
use std::process::{Command, Stdio};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

const STEP: Duration = Duration::from_secs(30);

fn tool(var: &str) -> String {
    std::env::var(var).unwrap_or_else(|_| panic!("{var} names the reference tool this test runs; it is not set"))
}

/// A child process killed when dropped.
struct Child(std::process::Child);

impl Drop for Child {
    fn drop(&mut self) {
        let _ = self.0.kill();
        let _ = self.0.wait();
    }
}

fn spawn(bin: &str, args: &[&str], stdout: bool) -> Child {
    let mut c = Command::new(bin);
    c.args(args).stdin(Stdio::null()).stderr(Stdio::null());
    c.stdout(if stdout { Stdio::piped() } else { Stdio::null() });
    Child(c.spawn().unwrap_or_else(|e| panic!("{bin}: {e}")))
}

/// A free UDP port on loopback (even, when `even`).
fn free_port(even: bool) -> u16 {
    loop {
        let s = UdpSocket::bind("127.0.0.1:0").unwrap();
        let p = s.local_addr().unwrap().port();
        drop(s);
        if (!even || p.is_multiple_of(2)) && p < 65_000 && UdpSocket::bind(("127.0.0.1", p + 1)).is_ok() {
            return p;
        }
    }
}

/// The program every test sends: four seconds of a synthetic transport stream, timed.
fn program() -> (Vec<(u64, Vec<u8>)>, Vec<u8>) {
    let c = joule_contrib::cli::synthetic(4, 2500);
    let all = c.iter().flat_map(|x| x.1.clone()).collect();
    (c, all)
}

/// Send the chunks as UDP datagrams to `to`, each at its time.
async fn feed_udp(chunks: Vec<(u64, Vec<u8>)>, to: SocketAddr) {
    let s = tokio::net::UdpSocket::bind("127.0.0.1:0").await.unwrap();
    let start = tokio::time::Instant::now();
    for (at, c) in chunks {
        tokio::time::sleep_until(start + Duration::from_micros(at)).await;
        s.send_to(&c, to).await.unwrap();
    }
}

/// Every UDP datagram that reaches `port` until none has come for `idle`.
#[cfg_attr(not(feature = "rist"), allow(dead_code))]
async fn capture_udp(port: u16, idle: Duration) -> Vec<u8> {
    let s = tokio::net::UdpSocket::bind(("127.0.0.1", port)).await.unwrap();
    let mut out = Vec::new();
    let mut b = vec![0u8; 65_536];
    let mut wait = STEP;
    while let Ok(Ok(n)) = timeout(wait, s.recv(&mut b)).await {
        out.extend_from_slice(&b[..n]);
        wait = idle;
    }
    out
}

async fn drain(link: &mut Link, quiet: Duration) -> Vec<u8> {
    let mut out = Vec::new();
    let mut wait = STEP;
    while let Ok(Some(p)) = timeout(wait, link.data.recv()).await {
        out.extend_from_slice(&p);
        wait = quiet;
    }
    out
}

fn lossy() -> (Impair, Impair) {
    (Impair { loss_ppm: 20_000, spare_first: 10, ..Impair::default() }, Impair { loss_ppm: 20_000, ..Impair::default() })
}

// ── SRT ────────────────────────────────────────────────────────────────────────────

async fn srt_reference_caller(passphrase: Option<&str>, pbkeylen: usize) {
    let slt = tool("WAI_SRT_LIVE_TRANSMIT");
    let pass = passphrase.map(str::to_string);
    let adm = move |_: &ConnRequest| Ok(Admitted { passphrase: pass.clone(), role: Role::Receive, latency_ms: Some(300), tag: 0 });
    let mut l = srt::Listener::bind(ListenerConfig { latency_ms: 300, ..ListenerConfig::default() }, Arc::new(adm)).await.unwrap();
    let (up, down) = lossy();
    let px = Proxy::start(l.addr, up, down, 21).await.unwrap();
    let udp_in = free_port(false);
    let mut url = format!("srt://127.0.0.1:{}?latency=300&streamid=%23!::r=room/owner,m=publish", px.addr.port());
    if let Some(p) = passphrase {
        url.push_str(&format!("&passphrase={p}&pbkeylen={pbkeylen}"));
    }
    let _c = spawn(&slt, &["-q", &format!("udp://127.0.0.1:{udp_in}"), &url], false);
    let mut link = timeout(STEP, l.accept()).await.expect("the reference caller connects").unwrap();
    assert_eq!(link.info.stream_id.as_deref(), Some("#!::r=room/owner,m=publish"));
    assert_eq!(link.info.cipher_bits as usize, passphrase.map_or(0, |_| pbkeylen * 8));
    let (chunks, all) = program();
    tokio::spawn(feed_udp(chunks, format!("127.0.0.1:{udp_in}").parse().unwrap()));
    let got = drain(&mut link, Duration::from_millis(1500)).await;
    assert_eq!(got.len(), all.len(), "every byte");
    assert!(got == all, "byte for byte");
    let part = link.account.lock().unwrap().cut(1, false);
    part.check().unwrap();
    assert_eq!(part.dropped, 0, "{part:?}");
    assert!(part.delivered > part.first_try && part.requests > 0, "losses found, requested and recovered: {part:?}");
    assert!(px.up.dropped.load(std::sync::atomic::Ordering::Relaxed) > 0);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "runs the reference SRT tool: set WAI_SRT_LIVE_TRANSMIT"]
async fn srt_reference_caller_into_this_listener() {
    srt_reference_caller(None, 16).await;
    srt_reference_caller(Some("0123456789abcdef"), 16).await;
    srt_reference_caller(Some("0123456789abcdef"), 32).await;
}

async fn srt_this_caller(passphrase: Option<&str>, pbkeylen: usize) {
    let slt = tool("WAI_SRT_LIVE_TRANSMIT");
    let port = free_port(false);
    let mut url = format!("srt://:{port}?mode=listener&latency=300");
    if let Some(p) = passphrase {
        url.push_str(&format!("&passphrase={p}"));
    }
    // `-a no`: the reference listener exits when the caller ends the connection, and so
    // flushes what it wrote to its standard output (killed, it would lose its buffer).
    let mut c = spawn(&slt, &["-q", "-a", "no", &url, "file://con"], true);
    let mut out = c.0.stdout.take().unwrap();
    let reader = std::thread::spawn(move || {
        let mut b = Vec::new();
        std::io::Read::read_to_end(&mut out, &mut b).unwrap();
        b
    });
    tokio::time::sleep(Duration::from_millis(500)).await;
    let (up, down) = lossy();
    let px = Proxy::start(format!("127.0.0.1:{port}").parse().unwrap(), up, down, 22).await.unwrap();
    let cfg = CallConfig { latency_ms: 300, passphrase: passphrase.map(str::to_string), keylen: pbkeylen, ..CallConfig::default() };
    let mut link = timeout(STEP, srt::call(px.addr, cfg)).await.unwrap().expect("the reference listener admits the call");
    let (chunks, all) = program();
    joule_contrib::cli::pace(&link, &chunks, 1500).await.unwrap();
    let sent = link.sent.lock().unwrap().clone();
    link.close();
    let _ = timeout(Duration::from_secs(3), link.wait_closed()).await;
    let got = tokio::task::spawn_blocking(move || reader.join().unwrap()).await.unwrap();
    drop(c);
    if got != all {
        // Which of the chunks did not arrive, by content.
        let missing: Vec<usize> = chunks.iter().enumerate().filter(|(_, c)| !got.windows(c.1.len()).step_by(4).any(|w| w == &c.1[..])).map(|(i, _)| i).collect();
        eprintln!("missing chunks of {}: {missing:?}", chunks.len());
    }
    assert_eq!(got.len(), all.len(), "every byte reached the reference listener: {sent:?}");
    assert!(got == all, "byte for byte");
    assert!(sent.retransmitted > 0 && sent.requests > 0, "the reference listener's requests were answered: {sent:?}");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "runs the reference SRT tool: set WAI_SRT_LIVE_TRANSMIT"]
async fn this_caller_into_the_srt_reference_listener() {
    srt_this_caller(None, 16).await;
    srt_this_caller(Some("0123456789abcdef"), 24).await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
#[ignore = "runs the reference SRT tool: set WAI_SRT_LIVE_TRANSMIT"]
async fn the_srt_reference_caller_with_a_wrong_passphrase_is_refused() {
    let slt = tool("WAI_SRT_LIVE_TRANSMIT");
    let adm = |_: &ConnRequest| Ok(Admitted { passphrase: Some("the listener's passphrase".into()), role: Role::Receive, latency_ms: None, tag: 0 });
    let mut l = srt::Listener::bind(ListenerConfig::default(), Arc::new(adm)).await.unwrap();
    let url = format!("srt://127.0.0.1:{}?passphrase=another-passphrase", l.addr.port());
    let _c = spawn(&slt, &["-q", "-a", "no", &format!("udp://127.0.0.1:{}", free_port(false)), &url], false);
    assert!(timeout(Duration::from_secs(4), l.accept()).await.is_err(), "no link");
    assert!(l.refused.load(std::sync::atomic::Ordering::Relaxed) > 0, "the call was refused");
}

// ── RIST ───────────────────────────────────────────────────────────────────────────

#[cfg(feature = "rist")]
mod rist_tools {
    use super::*;
    use joule_contrib::rist::{self, recv::RecvConfig, Profile, RistListener, RistListenerConfig, SendConfig};

    async fn reference_sender(profile: Profile, secret: Option<&str>, aes: u16, loss: bool) {
        let rs = tool("WAI_RISTSENDER");
        let mut l = RistListener::bind(RistListenerConfig {
            profile,
            bind_ip: "127.0.0.1".parse().unwrap(),
            port: 0,
            recv: RecvConfig { buffer_us: 1_000_000, ..RecvConfig::default() },
            secret: secret.map(str::to_string),
            expect_cname: Some("tok".into()),
            tag: 0,
        })
        .await
        .unwrap();
        let (up, down) = lossy();
        let px = if loss { Some(Proxy::start(l.addr, up, down, 31).await.unwrap()) } else { None };
        let target = px.as_ref().map_or(l.addr, |p| p.addr);
        let udp_in = free_port(false);
        let mut url = format!("rist://127.0.0.1:{}?cname=tok&buffer=1000", target.port());
        if let Some(s) = secret {
            url.push_str(&format!("&secret={s}&aes-type={aes}"));
        }
        let p = if profile == Profile::Main { "1" } else { "0" };
        let _c = spawn(&rs, &["-i", &format!("udp://@127.0.0.1:{udp_in}"), "-o", &url, "-p", p, "-v", "2"], false);
        tokio::time::sleep(Duration::from_millis(1000)).await;
        let (chunks, all) = program();
        tokio::spawn(feed_udp(chunks, format!("127.0.0.1:{udp_in}").parse().unwrap()));
        let mut link = timeout(STEP, l.accept()).await.expect("the reference sender's flow").unwrap();
        assert_eq!(link.info.cname.as_deref(), Some("tok"));
        let got = drain(&mut link, Duration::from_millis(2500)).await;
        assert_eq!(got.len(), all.len());
        assert!(got == all, "byte for byte");
        let part = link.account.lock().unwrap().cut(1, false);
        part.check().unwrap();
        if loss {
            assert_eq!(part.dropped, 0, "{part:?}");
            assert!(part.delivered > part.first_try, "losses recovered by the reference sender's retransmissions: {part:?}");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore = "runs the reference RIST tools: set WAI_RISTSENDER"]
    async fn rist_reference_sender_into_this_receiver() {
        reference_sender(Profile::Simple, None, 0, false).await;
        reference_sender(Profile::Main, None, 0, true).await;
        reference_sender(Profile::Main, Some("0123456789abcdef"), 128, true).await;
        reference_sender(Profile::Main, Some("0123456789abcdef"), 256, false).await;
    }

    async fn this_sender(profile: Profile, secret: Option<&str>, aes: u16, loss: bool) {
        let rr = tool("WAI_RISTRECEIVER");
        let port = free_port(true);
        let out_port = free_port(false);
        let mut url = format!("rist://@127.0.0.1:{port}?buffer=1000");
        if let Some(s) = secret {
            url.push_str(&format!("&secret={s}&aes-type={aes}"));
        }
        let p = if profile == Profile::Main { "1" } else { "0" };
        let cap = tokio::spawn(capture_udp(out_port, Duration::from_millis(3000)));
        let _c = spawn(&rr, &["-i", &url, "-o", &format!("udp://127.0.0.1:{out_port}"), "-p", p, "-v", "2"], false);
        tokio::time::sleep(Duration::from_millis(700)).await;
        let cfg = SendConfig { profile, cname: "tok".into(), secret: secret.map(str::to_string), aes_256: aes == 256, buffer_ms: 1000, ..SendConfig::default() };
        // The Main Profile's one port can be reached through a lossy path.
        let (up, down) = lossy();
        let px = if loss { Some(Proxy::start(format!("127.0.0.1:{port}").parse().unwrap(), up, down, 32).await.unwrap()) } else { None };
        let target = px.as_ref().map_or_else(|| format!("127.0.0.1:{port}").parse().unwrap(), |p| p.addr);
        let link = rist::send(target, cfg).await.unwrap();
        let (chunks, all) = program();
        joule_contrib::cli::pace(&link, &chunks, 2000).await.unwrap();
        let sent = link.sent.lock().unwrap().clone();
        link.close();
        let got = cap.await.unwrap();
        assert_eq!(got.len(), all.len(), "every byte reached the reference receiver: {sent:?}");
        assert!(got == all, "byte for byte");
        if loss {
            assert!(sent.requests > 0 && sent.retransmitted > 0, "the reference receiver's requests were answered: {sent:?}");
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    #[ignore = "runs the reference RIST tools: set WAI_RISTRECEIVER"]
    async fn this_sender_into_the_rist_reference_receiver() {
        this_sender(Profile::Simple, None, 0, false).await;
        this_sender(Profile::Main, None, 0, true).await;
        this_sender(Profile::Main, Some("0123456789abcdef"), 256, true).await;
    }
}