//! A session's connection-level receive window (`limits.quic_connection_window_bytes`) bounds
//! what the session can make the relay hold unread. A publisher opens many streams and
//! finishes none of their groups: each group's first object waits at the door for receipts
//! that never come, and its stream is not read meanwhile. The publisher stalls once its unread
//! bytes reach the connection window, and the streams it opens after that carry nothing.
//! The stream windows alone would let it leave `STREAMS` stream windows unread.

mod common;

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

const STREAM_WINDOW: u64 = 1 << 20;
const CONNECTION_WINDOW: u64 = 4 << 20;
const STREAMS: u64 = 32;
/// The most the relay reads of a stream before its door holds the stream: one read of
/// the QUIC transport, its first object and what came with it.
const READ_CHUNK: u64 = 64 << 10;
/// Each write's size. A write cut off by a stall may have gone in part, so it is not counted.
const WRITE: usize = 4 << 10;
/// A write that makes no progress for this long has stalled on flow control.
const STALL: Duration = Duration::from_millis(500);

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

async fn stalls() {
    let server = relay(|c| {
        c.limits.quic_stream_window_bytes = STREAM_WINDOW;
        c.limits.quic_connection_window_bytes = CONNECTION_WINDOW;
        c.receipts.door = vec![DoorRule { prefix: vec!["held".into()], trust: vec![], trust_field: None, hold_ms: 100_000 }];
    })
    .await;
    let t = Tracks::new(ns(&["held", "room"]), "video", 10, 3, 128);
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _published = publish_tracks(&publisher, &t, None).await;

    // Each stream: one group's header, a small first object, then an object twice a stream
    // window long, and no FIN. A stream is written until a write stalls, then the next.
    let mut sent = 0u64;
    let mut streams = Vec::new();
    for g in 0..STREAMS {
        let h = SubgroupHeader { track_alias: t.media_alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(t.priority), properties: false, end_of_group: true, first_object: true };
        let mut b = Vec::new();
        h.encode(&mut b);
        let mut codec = SubgroupCodec::new(h);
        codec.encode_object(&SubgroupObject::normal(0, vec![1; 1000]), &mut b).unwrap();
        codec.encode_object(&SubgroupObject::normal(1, vec![2; 2 * STREAM_WINDOW as usize]), &mut b).unwrap();
        let mut st = timeout(STEP, publisher.conn.open_uni()).await.expect("a stream opens in time").expect("a stream opens");
        for c in b.chunks(WRITE) {
            match timeout(STALL, st.write_all(c)).await {
                Ok(Ok(())) => sent += c.len() as u64,
                Ok(Err(e)) => panic!("a write failed: {e}"),
                Err(_) => break,
            }
        }
        streams.push(st);
    }
    let relay = server.relay().clone();
    // The streams opened once the window was full got nothing through, not even a header:
    // the window is the session's, whichever stream the bytes are on.
    assert!(until(STEP, || relay.stats().objects_in >= 1).await, "a group's first object reached the door");
    let held = relay.stats().objects_in;
    eprintln!("sent {sent} bytes on {STREAMS} streams: the connection window {CONNECTION_WINDOW}, a stream window {STREAM_WINDOW}; {held} groups' first objects held at the door");
    assert!(held < STREAMS, "the streams past the window delivered nothing: {held} groups reached the door");
    // What the relay holds of the session is at most what it was sent. That is at most the
    // connection window plus what the relay read itself, one read a stream, and less than
    // the stream windows would allow, which is STREAMS times STREAM_WINDOW.
    let bound = CONNECTION_WINDOW + STREAMS * (READ_CHUNK + WRITE as u64);
    assert!(sent <= bound, "the session stalled at its connection window: {sent} bytes sent, the bound {bound}");
    assert!(sent < STREAMS * STREAM_WINDOW / 2, "the stream windows did not bound it: {sent} bytes sent");
    // It stalled at the window, not before: the window was filled.
    assert!(sent + STREAMS * WRITE as u64 >= CONNECTION_WINDOW, "the session filled its connection window: {sent} bytes sent");
    assert_eq!(relay.stats().sessions_open, 1, "the publisher's session is still open");
    drop(streams);
    publisher.close(0, "done");
}