//! A data stream whose header comes late is still read. QUIC opens every stream below the
//! highest one a packet has named (RFC 9000 §3.2): a stream whose first packet was lost is
//! open at the relay, with nothing in it, from the moment a later stream's packet arrives
//! until QUIC sends its own again. The relay gave such a stream ten seconds for its header
//! and then stopped it. On a path that drops packets a publisher's retransmission took
//! longer: its object reached the relay's host after the stream was stopped, the publisher
//! was told nothing it reads (its write had long returned), and no subscriber was sent the
//! object (seen through `joule-moq-room`'s lossy participants: 103 of a publisher's 240
//! streams stopped in one run, 82 of 120 objects delivered to nobody).
//!
//! Here the delay is the publisher's own: it opens two streams, writes the second whole, and
//! writes the first twelve seconds later.

mod common;

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

/// Longer than the ten seconds the relay waited.
const LATE: Duration = Duration::from_secs(12);

fn stream_of(alias: u64, g: u64) -> Vec<u8> {
    let h = SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true };
    let mut b = Vec::new();
    h.encode(&mut b);
    SubgroupCodec::new(h).encode_object(&SubgroupObject::normal(0, format!("group {g}").into_bytes()), &mut b).expect("an object");
    b
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_stream_whose_header_comes_late_is_read() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|_| {}).await;
        let (space, track) = (ns(&["late"]), name("header"));
        let (p, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let _held = p.publish(space.clone(), track.clone(), 4, Parameters::new(), Default::default()).await.expect("PUBLISH");
        let (s, _es) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let (ok, _stx, _srx) = s.subscribe(space.clone(), track.clone(), Parameters::new()).await.expect("SUBSCRIBE");
        // Two streams, in the order of their IDs. Nothing of the first is sent yet; the
        // second's bytes open both at the relay.
        let mut first = p.conn.open_uni().await.expect("a stream");
        let mut second = p.conn.open_uni().await.expect("a stream");
        second.write_all(&stream_of(4, 1)).await.expect("written");
        second.finish().expect("finished");
        let got = collect(&s, STEP, |g| !g.is_empty()).await;
        assert_eq!(got.iter().map(|g| (g.header.track_alias, g.header.group_id)).collect::<Vec<_>>(), [(ok.track_alias, 1)], "the second stream's object");
        tokio::time::sleep(LATE).await;
        // The first stream's bytes, late: the relay has not stopped it, and its object arrives.
        first.write_all(&stream_of(4, 0)).await.expect("the relay has not stopped the stream");
        first.finish().expect("finished");
        let got = collect(&s, STEP, |g| !g.is_empty()).await;
        assert_eq!(got.len(), 1, "the late stream's object reaches the subscriber");
        assert_eq!((got[0].header.group_id, got[0].reset, got[0].objects[0].payload.as_slice()), (0, None, b"group 0".as_slice()));
        assert!(matches!(timeout(Duration::from_millis(200), first.delivered()).await, Ok(Ok(None)) | Err(_)), "the relay did not stop the publisher's stream");
        for c in [&p, &s] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}