//! Sessions at `draft-ietf-moq-transport-18` and `-16` through the relay, beside -22
//! ones: the relay chooses each session's draft by ALPN or `WT-Available-Protocols` and
//! bridges it to its -22 model (`transport::bridge`). WAI objects and their receipts
//! cross every pairing of drafts and verify at each subscriber; a FETCH from the cache
//! returns them at -18 and -16; what one draft cannot say is refused by name; and two
//! sessions written byte by byte from the drafts' own messages, not through this crate's
//! client, check the relay's -18 and -16 wire.

mod common;

use common::*;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::draft::Draft;
use joule_moq::draft22::data::FetchEntry;
use joule_moq::draft22::{Location, Parameter, Parameters};
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{draft16 as m16, draft18 as m18};
use joule_moq_relay::client::ClientOptions;
use joule_moq_relay::transport::{quic, SharedConnection, TransportError};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

const LIMIT: Duration = Duration::from_secs(120);

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_18_over_quic() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D18))).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_16_over_quic() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::QuicDraft(Draft::D16))).await.expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn every_session_at_18_over_webtransport() {
    timeout(LIMIT, two_publishers_three_subscribers(Via::WtDraft(Draft::D18))).await.expect("in time");
}

/// Publishers at -16 and -18, subscribers at -22, -18 (WebTransport) and -16: every
/// object reaches every subscriber and verifies against its publisher's receipts, which
/// crossed two bridges unchanged.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn publishers_and_subscribers_at_different_drafts() {
    let vias = [Via::QuicDraft(Draft::D16), Via::QuicDraft(Draft::D18), Via::Quic, Via::WtDraft(Draft::D18), Via::QuicDraft(Draft::D16)];
    timeout(LIMIT, two_publishers_three_subscribers_on(vias)).await.expect("in time");
}

/// A late joiner at -18 and one at -16 FETCH a track and its receipts from the relay's
/// cache, published at -22: the exclusive End Locations of -18 and -16 and the absolute
/// Group IDs of a -16 fetch stream come back to the same objects, which verify.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_fetch_from_the_cache_at_18_and_16() {
    timeout(LIMIT, async {
        const GROUPS: u64 = 6;
        let server = relay(|_| {}).await;
        let key = ed25519_dalek::SigningKey::from_bytes(&[0x2b; 32]);
        let t = Tracks::new(ns(&["vod", "drafts"]), "ep", 10, 5, 100);
        let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let _held = publish_tracks(&publisher, &t, None).await;
        for g in 0..GROUPS {
            publish_group(&publisher, &key, &t, g, &payload("ep", g, 1500)).await;
        }
        let relay = server.relay().clone();
        let (mk, rk) = ((t.ns.clone(), t.media.clone()), (t.ns.clone(), t.receipts.clone()));
        assert!(
            until(STEP, || relay.largest(&mk) == Some(Location::new(GROUPS - 1, 0)) && relay.cache_covers(&mk, Location::new(0, 0)) && relay.cache_covers(&rk, Location::new(0, 0))).await,
            "the cache holds the track"
        );
        let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
        for d in [Draft::D18, Draft::D16] {
            let (late, _ep) = connect(&server, Via::QuicDraft(d), ClientOptions::default()).await;
            let (ok, media) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.unwrap_or_else(|e| panic!("FETCH at {d}: {e}"));
            assert_eq!(ok.end_location, Location::new(GROUPS - 1, 0), "{d}: the inclusive end, back from the draft's exclusive one");
            let (_, rec) = late.fetch(t.ns.clone(), t.receipts.clone(), Parameters::new()).await.expect("FETCH the receipts");
            let objs = |v: &[FetchEntry]| v.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.clone()) } else { None }).collect::<Vec<_>>();
            let (media, rec) = (objs(&media), objs(&rec));
            assert_eq!((media.len(), rec.len()), (GROUPS as usize, GROUPS as usize), "{d}");
            for (m, r) in media.iter().zip(&rec) {
                let b = ReceiptBundle::from_payload(&r.payload).unwrap();
                let vg = verifier.verify_group(&b.group).expect("a receipt verifies");
                verifier.verify_payload(&m.payload, &b.objects[0], &vg).expect("an object verifies");
                assert_eq!(m.payload, payload("ep", m.group_id, 1500));
            }
            // A range: groups 2 to 3, whole, which -18 and -16 say as End {4, 0}.
            let range = Parameters::new().with(Parameter::LocationFilter(joule_moq::draft22::LocationFilter::AbsoluteStartGroupEnd { start: Location::new(2, 0), end_group_delta: 1 }));
            let (_, part) = late.fetch(t.ns.clone(), t.media.clone(), range).await.expect("FETCH a range");
            assert_eq!(objs(&part).iter().map(|o| o.group_id).collect::<Vec<_>>(), vec![2, 3], "{d}");
            assert!(late.conn.notes().iter().all(|n| !n.starts_with("refused")), "{d}: {:?}", late.conn.notes());
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// A SUBSCRIBE whose Range Filter -18 cannot say is refused at this end, by name, and
/// never reaches the relay; the session goes on.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn what_18_cannot_say_is_refused_here() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (c, _ep) = connect(&server, Via::QuicDraft(Draft::D18), ClientOptions::default()).await;
        let p = Parameters::new().with(Parameter::RangeFilter { ty: joule_moq::draft22::params::param::OBJECTID_FILTER, filter: None });
        let e = c.subscribe(ns(&["a"]), name("b"), p).await.expect_err("refused");
        assert!(e.contains("REQUEST_ERROR NOT_SUPPORTED") && e.contains("Range Filter"), "{e}");
        assert!(c.conn.notes().iter().any(|n| n.starts_with("refused: a Range Filter has no -18 form")), "{:?}", c.conn.notes());
        // The session is whole: a request -18 can say gets the relay's answer.
        let e = c.subscribe(ns(&["nothing", "here"]), name("b"), Parameters::new()).await.expect_err("no publisher");
        assert!(e.contains("DOES_NOT_EXIST"), "{e}");
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

async fn raw_session(server: &joule_moq_relay::RelayServer, d: Draft) -> (SharedConnection, quinn::Endpoint) {
    let ep = quic::client_endpoint(loopback(), vec![server.identity.chain[0].clone()], vec![d.alpn().to_vec()], None).unwrap();
    let (conn, got) = timeout(STEP, quic::connect_any(&ep, server.quic_addr.unwrap(), "localhost")).await.expect("in time").expect("a QUIC session");
    assert_eq!(got, d);
    (Arc::new(conn), ep)
}

/// Read framed messages (a 16-bit Length after the type) of the draft whose integer's
/// length `len` reads from a first byte, until one is whole.
async fn read_framed(r: &mut joule_moq_relay::transport::BoxRecv, buf: &mut Vec<u8>, len: fn(u8) -> usize) -> Vec<u8> {
    loop {
        if let Some(&b0) = buf.first() {
            let tl = len(b0);
            if buf.len() >= tl + 2 {
                let n = tl + 2 + u16::from_be_bytes([buf[tl], buf[tl + 1]]) as usize;
                if buf.len() >= n {
                    return buf.drain(..n).collect();
                }
            }
        }
        match timeout(STEP, r.read_chunk()).await.expect("in time").expect("a chunk") {
            Some(b) => buf.extend_from_slice(&b),
            None => panic!("the stream ended"),
        }
    }
}

/// An -18 session written from -18's messages: SETUP each way on unidirectional streams,
/// a PUBLISH_NAMESPACE answered with REQUEST_OK, a Joining FETCH (no -22 form) refused
/// with NOT_SUPPORTED, a FETCH with Fetch Type 4 closing the session.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_raw_18_session() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (conn, _ep) = raw_session(&server, Draft::D18).await;
        let mut ctl = conn.open_uni().await.unwrap();
        let mut o = m18::kvp::SetupOptions::new();
        o.push(m18::kvp::KeyValuePair::bytes(m18::kvp::opt::MOQT_IMPLEMENTATION, b"raw-18".to_vec()));
        ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: o }).encode().unwrap()).await.unwrap();
        // The relay's SETUP, on its control stream.
        let mut theirs = timeout(STEP, conn.accept_uni()).await.expect("in time").expect("a stream");
        let mut d = m18::UniStreamDecoder::new(joule_moq::draft22::GroupOrder::Ascending, Default::default());
        let setup = loop {
            match d.next().unwrap() {
                Some(m18::UniEvent::Control(m)) => break m,
                Some(_) => continue,
                None => d.push(&timeout(STEP, theirs.read_chunk()).await.unwrap().unwrap().unwrap()),
            }
        };
        let m18::ControlMessage::Setup(s) = setup else { panic!("{setup:?}") };
        assert!(s.options.implementation().is_some());
        assert!(s.options.0.iter().all(|p| p.ty != 0x06 && p.ty != 0x08), "no -22-only option reaches an -18 peer");

        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        let pn = m18::ControlMessage::PublishNamespace(m18::NamespaceRequest { request_id: 0, prefix: ns(&["raw18"]), params: m18::Parameters::new() });
        tx.write_all(&pn.encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
        assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");

        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        let j = m18::ControlMessage::Fetch(m18::Fetch { request_id: 2, kind: m18::FetchKind::RelativeJoining { joining_request_id: 0, joining_start: 1 }, params: m18::Parameters::new() });
        tx.write_all(&j.encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        match m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap() {
            m18::ControlMessage::RequestError(e) => assert_eq!(e.code, m18::codes::RequestErrorCode::NOT_SUPPORTED),
            other => panic!("{other:?}"),
        }

        let (mut tx, _rx) = conn.open_bi().await.unwrap();
        tx.write_all(&[0x16, 0x00, 0x03, 0x04, 0x04, 0x00]).await.unwrap();
        match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
            TransportError::Closed { code, .. } => assert_eq!(code, 0x3, "PROTOCOL_VIOLATION"),
            other => panic!("{other:?}"),
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// A FIN from an -18 requester on PUBLISH_NAMESPACE withdraws the namespace (-18 §6.2
/// withdraws it by cancelling the request; -18 has no FIN that leaves a request
/// standing), while a -22 publisher's FIN does not (-22 §6.4.2.2). A -22 client's FIN
/// sent through an -18 session is withheld, so it does not withdraw either.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_18_publishers_fin_withdraws_its_namespace() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (sub, _e1) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (_stx, mut srx) = sub.subscribe_namespace(ns(&["fin"])).await.expect("SUBSCRIBE_NAMESPACE");
        // An -18 publisher written from -18's messages: PUBLISH_NAMESPACE, then a FIN.
        let (conn, _ep) = raw_session(&server, Draft::D18).await;
        let mut ctl = conn.open_uni().await.unwrap();
        ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: m18::kvp::SetupOptions::new() }).encode().unwrap()).await.unwrap();
        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        let pn = m18::ControlMessage::PublishNamespace(m18::NamespaceRequest { request_id: 0, prefix: ns(&["fin", "raw18"]), params: m18::Parameters::new() });
        tx.write_all(&pn.encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
        assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");
        match timeout(STEP, srx.next()).await {
            Ok(Ok(Some(joule_moq::draft22::ControlMessage::Namespace(s)))) => assert_eq!(s, ns(&["raw18"])),
            other => panic!("no NAMESPACE: {other:?}"),
        }
        tx.finish().unwrap();
        match timeout(STEP, srx.next()).await {
            Ok(Ok(Some(joule_moq::draft22::ControlMessage::NamespaceDone(s)))) => assert_eq!(s, ns(&["raw18"])),
            other => panic!("the -18 FIN did not withdraw the namespace: {other:?}"),
        }

        // A -22 publisher, and a -22 client over -18: each FINs its PUBLISH_NAMESPACE, and
        // each namespace stands.
        for (via, field) in [(Via::Quic, "at22"), (Via::QuicDraft(Draft::D18), "through18")] {
            let (p, _e) = connect(&server, via, ClientOptions::default()).await;
            let (ptx, _prx) = p.publish_namespace(ns(&["fin", field]), Parameters::new()).await.expect("PUBLISH_NAMESPACE");
            match timeout(STEP, srx.next()).await {
                Ok(Ok(Some(joule_moq::draft22::ControlMessage::Namespace(s)))) => assert_eq!(s, ns(&[field])),
                other => panic!("{via:?}: no NAMESPACE: {other:?}"),
            }
            ptx.fin().await;
            let after = timeout(Duration::from_millis(500), srx.next()).await;
            assert!(after.is_err(), "{via:?}: a FIN that is not a withdrawal withdrew it: {after:?}");
            // The namespace goes with the session.
            p.close(0, "done");
            match timeout(STEP, srx.next()).await {
                Ok(Ok(Some(joule_moq::draft22::ControlMessage::NamespaceDone(s)))) => assert_eq!(s, ns(&[field])),
                other => panic!("{via:?}: no NAMESPACE_DONE at the session's end: {other:?}"),
            }
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// An -18 publisher that closes its control stream while a data stream of its is still
/// arriving and before its PUBLISH_DONE: closing the control stream is a protocol
/// violation (-18 §3.3), but the relay reads the stream to its end and takes the
/// PUBLISH_DONE before it closes the session for it, so the subscriber gets every object
/// and the publisher's own status.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_18_publisher_that_closes_its_control_stream_first_loses_nothing() {
    timeout(LIMIT, async {
        use joule_moq::draft22::data::{SubgroupCodec, SubgroupHeader, SubgroupId, SubgroupObject};
        let server = relay(|_| {}).await;
        let (sub, _e) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (conn, _ep) = raw_session(&server, Draft::D18).await;
        let mut ctl = conn.open_uni().await.unwrap();
        ctl.write_all(&m18::ControlMessage::Setup(m18::Setup { options: m18::kvp::SetupOptions::new() }).encode().unwrap()).await.unwrap();
        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        let p = m18::ControlMessage::Publish(m18::Publish { request_id: 0, namespace: ns(&["late"]), name: name("track"), track_alias: 1, params: m18::Parameters::new(), track_properties: Default::default() });
        tx.write_all(&p.encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let m = m18::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, joule_moq::draft22::varint::len_from_first).await).unwrap();
        assert!(matches!(m, m18::ControlMessage::RequestOk(_)), "{m:?}");
        let (_ok, _stx, mut srx) = sub.subscribe(ns(&["late"]), name("track"), Parameters::new()).await.expect("SUBSCRIBE");

        // The data stream's header and first object; the control stream closed; then the
        // second object, the stream's FIN and PUBLISH_DONE.
        let h = SubgroupHeader { track_alias: 1, group_id: 0, subgroup_id: SubgroupId::Zero, publisher_priority: Some(0), properties: false, end_of_group: true, first_object: true };
        let mut codec = SubgroupCodec::new(h);
        let (mut first, mut second) = (Vec::new(), Vec::new());
        h.encode(&mut first);
        codec.encode_object(&SubgroupObject::normal(0, b"before".to_vec()), &mut first).unwrap();
        codec.encode_object(&SubgroupObject::normal(1, b"after".to_vec()), &mut second).unwrap();
        let mut data = conn.open_uni().await.unwrap();
        data.write_all(&first).await.unwrap();
        tokio::time::sleep(Duration::from_millis(100)).await;
        ctl.finish().unwrap();
        tokio::time::sleep(Duration::from_millis(300)).await;
        data.write_all(&second).await.unwrap();
        data.finish().unwrap();
        let done = m18::ControlMessage::PublishDone(m18::PublishDone { status: m18::codes::PublishDoneCode::TRACK_ENDED, stream_count: 1, reason: Vec::new() });
        tx.write_all(&done.encode().unwrap()).await.unwrap();

        let got = collect(&sub, STEP, |g| g.iter().map(|x| x.objects.len()).sum::<usize>() >= 2).await;
        let payloads: Vec<&[u8]> = got.iter().flat_map(|g| &g.objects).map(|o| o.payload.as_slice()).collect();
        assert_eq!(payloads, vec![&b"before"[..], &b"after"[..]]);
        match timeout(STEP, srx.next()).await {
            Ok(Ok(Some(joule_moq::draft22::ControlMessage::PublishDone(d)))) => assert_eq!(d.status, joule_moq::draft22::PublishDoneCode::TRACK_ENDED),
            other => panic!("no PUBLISH_DONE: {other:?}"),
        }
        // And then the session closes, for the closed control stream.
        match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
            TransportError::Closed { code, .. } => assert_eq!(code, 0x3, "PROTOCOL_VIOLATION"),
            other => panic!("{other:?}"),
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}

/// A -16 session written from -16's messages: CLIENT_SETUP on the bidirectional control
/// stream the client opens, SERVER_SETUP back with MAX_REQUEST_ID, a PUBLISH_NAMESPACE
/// answered by Request ID; a request whose ID is ahead of one still in flight on another
/// stream admitted (a SUBSCRIBE at 4 before the SUBSCRIBE_NAMESPACE at 2 on its own
/// stream, which then fills the gap), and a Request ID seen before closing the session
/// with INVALID_REQUEST_ID.
#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_raw_16_session() {
    timeout(LIMIT, async {
        let server = relay(|_| {}).await;
        let (conn, _ep) = raw_session(&server, Draft::D16).await;
        let (mut tx, mut rx) = conn.open_bi().await.unwrap();
        let mut p = m16::kvp::SetupParameters::new();
        p.push(m16::kvp::KeyValuePair::varint(m16::kvp::setup::MAX_REQUEST_ID, 100)).push(m16::kvp::KeyValuePair::bytes(m16::kvp::setup::MOQT_IMPLEMENTATION, b"raw-16".to_vec()));
        tx.write_all(&m16::ControlMessage::ClientSetup(p).encode().unwrap()).await.unwrap();
        let mut buf = Vec::new();
        let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
        let m16::ControlMessage::ServerSetup(sp) = m else { panic!("{m:?}") };
        assert!(sp.max_request_id() > 0, "the relay grants requests");

        let pn = m16::ControlMessage::PublishNamespace(m16::control::PublishNamespace { request_id: 0, namespace: ns(&["raw16"]), params: m16::params::Parameters::new() });
        tx.write_all(&pn.encode().unwrap()).await.unwrap();
        let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
        assert!(matches!(&m, m16::ControlMessage::RequestOk(o) if o.request_id == 0), "{m:?}");

        let s = m16::ControlMessage::Subscribe(m16::control::TrackRequest { request_id: 4, namespace: ns(&["elsewhere"]), name: name("t"), params: m16::params::Parameters::new() });
        tx.write_all(&s.encode().unwrap()).await.unwrap();
        let m = m16::ControlMessage::decode_exact(&read_framed(&mut rx, &mut buf, m16::io::len_from_first).await).unwrap();
        assert!(matches!(&m, m16::ControlMessage::RequestError(e) if e.request_id == 4), "the request ahead is answered: {m:?}");
        let (mut ntx, mut nrx) = conn.open_bi().await.unwrap();
        let sn = m16::ControlMessage::SubscribeNamespace(m16::control::SubscribeNamespace { request_id: 2, prefix: ns(&["somewhere"]), options: m16::control::SubscribeOptions::Namespace, params: m16::params::Parameters::new() });
        ntx.write_all(&sn.encode().unwrap()).await.unwrap();
        let mut nbuf = Vec::new();
        let m = m16::ControlMessage::decode_exact(&read_framed(&mut nrx, &mut nbuf, m16::io::len_from_first).await).unwrap();
        assert!(matches!(&m, m16::ControlMessage::RequestOk(o) if o.request_id == 2), "the request behind it, on its own stream: {m:?}");

        tx.write_all(&s.encode().unwrap()).await.unwrap();
        match timeout(STEP, conn.closed()).await.expect("the relay closes the session") {
            TransportError::Closed { code, .. } => assert_eq!(code, 0x4, "INVALID_REQUEST_ID"),
            other => panic!("{other:?}"),
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}