//! A session whose peer grants no flow-control credit is closed (`limits.stalled_session_ms`).
//!
//! A subscriber's connection-level receive window is shared by every stream of its session.
//! One that reads nothing of its data streams holds that whole window in unread bytes, and
//! the relay can then send it nothing on any stream: no object, and not the PUBLISH_DONE of a
//! subscription it ends. Such a peer still acknowledges, so the QUIC idle timeout never ends
//! its session. The relay closes it `CONTROL_MESSAGE_TIMEOUT`, reason `stalled_session`,
//! once a write of its own has waited on the session that long with no credit granted and no
//! stream data sent.
//!
//! Beside it, on the same relay: a subscriber with the same small window that reads late
//! (each read grants credit) is never closed, and neither is a session that is merely quiet.

mod common;

use common::*;
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{Parameters, SessionError};
use joule_moq_relay::client::ClientOptions;
use joule_moq_relay::session::STALLED_SESSION;
use joule_moq_relay::transport::TransportError;
use std::time::{Duration, Instant};
use tokio::time::timeout;

/// Ten times the slow subscriber's delay before it reads a stream.
const STALL: Duration = Duration::from_millis(2_000);
/// The connection receive window of the two subscribers that do not keep up.
const WINDOW: u32 = 64 * 1024;
const BYTES: usize = 24 * 1024;
/// Six windows' worth.
const GROUPS: u64 = 16;

fn group(g: u64) -> (SubgroupHeader, [SubgroupObject; 1]) {
    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, vec![g as u8; BYTES])])
}

fn narrow() -> quinn::TransportConfig {
    let mut t = quinn::TransportConfig::default();
    t.receive_window(quinn::VarInt::from_u32(WINDOW));
    t
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_session_that_grants_no_credit_is_closed() {
    timeout(Duration::from_secs(120), async {
        let server = relay(|c| c.limits.stalled_session_ms = STALL.as_millis() as u64).await;
        let (space, video) = (ns(&["stalled"]), name("video"));
        let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
        let _v = publisher.publish(space.clone(), video.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH");
        let lazy = || ClientOptions { lazy_data: true, ..ClientOptions::default() };
        let (stalled, _e1) = connect_with(&server, Via::Quic, lazy(), Some(narrow())).await;
        let (slow, _e2) = connect_with(&server, Via::Quic, lazy(), Some(narrow())).await;
        let _s1 = stalled.subscribe(space.clone(), video.clone(), Parameters::new()).await.expect("SUBSCRIBE");
        let _s2 = slow.subscribe(space.clone(), video.clone(), Parameters::new()).await.expect("SUBSCRIBE");
        let began = Instant::now();
        for g in 0..GROUPS {
            let (h, o) = group(g);
            timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
        }
        // The slow one starts on each stream 200 ms after the relay opens it, and reads its
        // streams beside one another, none waiting for another's bytes (a subscriber that
        // waits on its streams in order holds the credit the one it waits on needs, and stalls
        // itself: the README, "A subscriber that does not read"). Each read grants the relay
        // credit, and every group reaches it.
        let slow_reads = async {
            let mut readers = tokio::task::JoinSet::new();
            for _ in 0..GROUPS {
                let mut r = slow.next_stream(STEP).await.unwrap_or_else(|| panic!("the slow subscriber's next stream: its end {:?}; the relay's sessions {:?}", slow.conn.diagnostics(), server.relay().session_diagnostics()));
                readers.spawn(async move {
                    tokio::time::sleep(Duration::from_millis(200)).await;
                    timeout(STEP, r.start()).await.expect("a header in time").expect("a header");
                    read_stream(r).await.expect("a subgroup stream")
                });
            }
            let mut got = 0u64;
            while let Some(g) = readers.join_next().await {
                let g = g.expect("a reader");
                assert_eq!((g.reset, g.objects.len()), (None, 1), "group {}", g.header.group_id);
                got += 1;
            }
            got
        };
        // The one that reads nothing is closed, by the relay, with the code and the reason.
        let closed = async {
            match timeout(STEP, stalled.conn.closed()).await.expect("the stalled session is closed") {
                TransportError::Closed { code, reason, by_peer } => (code, reason, by_peer, began.elapsed()),
                other => panic!("closed otherwise: {other}"),
            }
        };
        let (got, (code, reason, by_peer, after)) = tokio::join!(slow_reads, closed);
        assert_eq!((code, reason.as_str(), by_peer), (SessionError::CONTROL_MESSAGE_TIMEOUT.0, STALLED_SESSION, true));
        assert!(after >= STALL, "not before the time allowed: {after:?}");
        assert_eq!(got, GROUPS, "the slow subscriber has every group");
        let relay = server.relay().clone();
        assert_eq!(relay.stats().sessions_stalled, 1);
        // Quiet for twice the time: the publisher and the slow subscriber stand.
        tokio::time::sleep(2 * STALL).await;
        assert_eq!((relay.stats().sessions_stalled, relay.session_diagnostics().len()), (1, 2), "no other session was closed");
        let (h, o) = group(GROUPS);
        publisher.send_subgroup(h, &o).await.expect("a send");
        let r = slow.next_data(STEP).await.expect("the slow subscriber is still served");
        assert_eq!(read_stream(r).await.expect("a stream").header.group_id, GROUPS);
        for c in [&publisher, &slow] {
            c.close(0, "done");
        }
        server.shutdown().await;
    })
    .await
    .expect("in time");
}