//! The QUIC listener's UDP sockets: the sessions spread across them, and a viewer whose address
//! changes keeps its connection, because the kernel steers a packet by its connection ID and not by
//! its address (Linux). On the other Unix systems the sockets are one endpoint's, and what it sends
//! goes out through all of them, each viewer's through one; a viewer that moves is still read, since
//! every socket is. On a system with one socket, this holds trivially.

mod common;

use common::*;
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Parameters, UniEvent};
use joule_moq_relay::client::ClientOptions;
use std::time::Duration;
use tokio::time::timeout;

const SOCKETS: usize = 4;
const VIEWERS: usize = 32;

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_viewer_whose_address_changes_keeps_its_connection() {
    timeout(Duration::from_secs(120), run()).await.expect("the test finishes in time");
}

async fn next_payload(c: &joule_moq_relay::client::Client) -> Vec<u8> {
    let mut r = timeout(STEP, c.next_data(STEP)).await.expect("in time").expect("a stream");
    loop {
        match timeout(STEP, r.next()).await.expect("in time").expect("a read") {
            Some(UniEvent::SubgroupObject(o)) => return o.payload,
            Some(_) => continue,
            None => panic!("a stream with no object"),
        }
    }
}

async fn run() {
    let server = relay(|c| c.limits.quic_sockets = SOCKETS).await;
    let (publisher, _pub_ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (space, track) = (ns(&["moved"]), name("video"));
    let _published = publisher.publish(space.clone(), track.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH");

    let mut viewers = Vec::with_capacity(VIEWERS);
    for _ in 0..VIEWERS {
        let (c, ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let held = c.subscribe(space.clone(), track.clone(), Parameters::new()).await.expect("SUBSCRIBE");
        viewers.push((c, ep.expect("a QUIC endpoint"), held));
    }

    let per_socket = server.quic_connections_per_socket();
    assert_eq!(per_socket.iter().sum::<usize>(), VIEWERS + 1, "{per_socket:?}");
    if cfg!(target_os = "linux") {
        assert_eq!(per_socket.len(), SOCKETS);
        assert!(per_socket.iter().filter(|n| **n > 0).count() > 1, "the sessions spread across the sockets: {per_socket:?}");
    }

    let send = |g: u64| {
        let h = SubgroupHeader { track_alias: 1, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(128), properties: false, end_of_group: true, first_object: true };
        (h, [SubgroupObject::normal(0, g.to_le_bytes().to_vec())])
    };
    let (h, o) = send(0);
    timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
    for (c, _, _) in &viewers {
        assert_eq!(next_payload(c).await, 0u64.to_le_bytes());
    }
    // Off Linux, one endpoint's sends went out through more than one of its sockets: before, every
    // datagram to every viewer went through one.
    if cfg!(unix) && !cfg!(target_os = "linux") {
        let sends = server.quic_sends_per_socket();
        assert_eq!(sends.len(), SOCKETS, "{sends:?}");
        assert!(sends.iter().filter(|n| **n > 0).count() > 1, "the sends spread across the sockets: {sends:?}");
    }

    // Every viewer moves to a new address. Steered by address, most of their packets would reach
    // sockets that hold no such connection, and their sessions would end.
    for (_, ep, _) in &viewers {
        ep.rebind(std::net::UdpSocket::bind(loopback()).expect("a new socket")).expect("rebinding");
    }
    for g in 1..=3 {
        let (h, o) = send(g);
        timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
        for (c, _, _) in &viewers {
            assert_eq!(next_payload(c).await, g.to_le_bytes(), "group {g}, after the move");
        }
    }
    assert_eq!(server.quic_connections_per_socket().iter().sum::<usize>(), VIEWERS + 1, "no session ended");
}

/// Two relays never share a UDP port. A second relay asked for the port a first one holds is
/// refused, as on a system with one socket a listener; before, its sockets joined the first
/// relay's group, the kernel steered each relay's packets to the other's sockets, and a client
/// of one met the other's certificate ("invalid peer certificate: BadSignature", both being
/// self-signed under one name) or no answer. The same happened by chance between two relays
/// bound to port 0, in one test binary or two (`transport::quic`'s unit tests hold that case).
#[cfg(unix)]
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_second_relay_is_refused_the_port_a_first_one_holds() {
    use joule_moq_relay::transport::quic;
    let first = relay(|c| c.limits.quic_sockets = SOCKETS).await;
    let mut cfg = joule_moq_relay::RelayConfig::loopback();
    cfg.limits.quic_sockets = SOCKETS;
    cfg.quic_listen = first.quic_addr;
    match joule_moq_relay::RelayServer::start(cfg).await {
        Err(e) => assert!(e.contains("binding"), "{e}"),
        Ok(second) => {
            let ep = quic::client_endpoint(loopback(), vec![second.identity.chain[0].clone()], vec![joule_moq::draft22::ALPN.to_vec()], None).unwrap();
            let met = match timeout(Duration::from_secs(5), quic::connect(&ep, second.quic_addr.unwrap(), "localhost")).await {
                Err(_) => "no answer in five seconds".to_string(),
                Ok(Err(e)) => e.to_string(),
                Ok(Ok(_)) => "a session".to_string(),
            };
            panic!("two relays hold {}: a client of the second met: {met}", first.quic_addr.unwrap());
        }
    }
    // The first relay is untouched by the refusal.
    let (c, _ep) = connect(&first, Via::Quic, ClientOptions::default()).await;
    c.close(0, "done");
    first.shutdown().await;
}