//! A subscriber that does not read is ended TOO_FAR_BEHIND (`draft-ietf-moq-transport-23`
//! §5.2, §12.4, §12.5; `draft-englishm-moq-relay-dos-02` §5.1).
//!
//! One publisher sends a track with no delivery timeout, a group a stream, each object larger
//! than the stream window a slow subscriber gives the relay. A fast subscriber reads as it
//! goes, and the publisher keeps in step with it. The slow one reads nothing, so each of the
//! relay's writes to it stops on its stream's flow control and what arrives for its
//! subscription waits in its session's scheduler. The relay is given a bound of 128 items a
//! subscription (`limits.max_queued_per_subscription`). Past it the relay drops what the
//! subscription had queued, resets its streams TOO_FAR_BEHIND, those its writes were stopped
//! on, and sends PUBLISH_DONE TOO_FAR_BEHIND; it counts the subscription and the objects
//! dropped. The fast subscriber loses nothing, the slow session and its other subscription
//! stand, and the slow subscriber subscribes again and is served.
//!
//! The slow subscriber's connection window is left wide: one that held the whole connection's
//! credit in streams it had not read could be sent nothing on any stream, its PUBLISH_DONE
//! included, until it read (the README, "A subscriber that does not read"); what bounds that
//! one is `limits.stalled_session_ms` (`tests/stalled_session.rs`).

mod common;

use common::*;
use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::{ControlMessage, Parameters, PublishDoneCode, StreamResetCode};
use joule_moq_relay::client::ClientOptions;
use joule_moq_relay::transport::TransportError;
use joule_moq_relay::wire::StreamError;
use std::time::Duration;
use tokio::time::timeout;

const BOUND: u64 = 128;
const GROUPS: u64 = 200;
/// Larger than the slow subscriber's stream window.
const BYTES: usize = 48 * 1024;
const STREAM_WINDOW: u32 = 32 * 1024;
/// Smaller than it: what the slow subscriber is sent once it reads again.
const SMALL: usize = 4 * 1024;

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

fn group(alias: u64, g: u64, bytes: usize) -> (SubgroupHeader, [SubgroupObject; 1]) {
    let h = SubgroupHeader { track_alias: alias, 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])])
}

async fn run() {
    let server = relay(|c| c.limits.max_queued_per_subscription = BOUND).await;
    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let (space, video, audio) = (ns(&["behind"]), name("video"), name("audio"));
    let _v = publisher.publish(space.clone(), video.clone(), 1, Parameters::new(), Default::default()).await.expect("PUBLISH video");
    let _a = publisher.publish(space.clone(), audio.clone(), 2, Parameters::new(), Default::default()).await.expect("PUBLISH audio");

    let (fast, _ef) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let mut small = quinn::TransportConfig::default();
    small.stream_receive_window(quinn::VarInt::from_u32(STREAM_WINDOW));
    let (slow, _es) = connect_with(&server, Via::Quic, ClientOptions { lazy_data: true, ..ClientOptions::default() }, Some(small)).await;

    let (fast_ok, _ftx, mut frx) = fast.subscribe(space.clone(), video.clone(), Parameters::new()).await.expect("SUBSCRIBE");
    let (slow_ok, _stx, mut srx) = slow.subscribe(space.clone(), video.clone(), Parameters::new()).await.expect("SUBSCRIBE");
    let (slow_audio, _satx, mut sarx) = slow.subscribe(space.clone(), audio.clone(), Parameters::new()).await.expect("SUBSCRIBE audio");

    // The publisher keeps within eight groups of the subscriber that reads, so that one's
    // queue stays far inside the bound however loaded the host is.
    let (seen_tx, mut seen) = tokio::sync::watch::channel(0u64);
    let publish = async {
        for g in 0..GROUPS {
            timeout(STEP, seen.wait_for(|n| n + 8 > g)).await.expect("the fast subscriber keeps up").expect("it is reading");
            let (h, o) = group(1, g, BYTES);
            timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
        }
    };
    let fast_reads = async {
        let mut got = Vec::new();
        while (got.len() as u64) < GROUPS {
            let r = fast.next_data(STEP).await.expect("the fast subscriber's next stream");
            got.push(read_stream(r).await.expect("a subgroup stream"));
            let _ = seen_tx.send(got.len() as u64);
        }
        got
    };
    let ((), fast_got) = tokio::join!(publish, fast_reads);

    // The slow subscription is told, by name, with the streams it was opened.
    let done = match timeout(STEP, srx.next()).await {
        Ok(Ok(Some(ControlMessage::PublishDone(d)))) => d,
        other => panic!("no PUBLISH_DONE for the subscription that fell behind: {other:?}"),
    };
    assert_eq!(done.status, PublishDoneCode::TOO_FAR_BEHIND);
    assert_eq!(done.reason, b"queue_bound");
    assert!(done.stream_count >= 1 && done.stream_count < GROUPS, "streams opened for it: {}", done.stream_count);
    assert!(matches!(timeout(STEP, srx.next()).await, Ok(Ok(None))), "and its request stream ends");

    // The fast subscriber has every group whole, and its subscription stands.
    assert_eq!(fast_got.iter().filter(|x| x.header.track_alias == fast_ok.track_alias && x.reset.is_none() && x.objects.len() == 1).count() as u64, GROUPS);
    assert!(timeout(Duration::from_millis(300), frx.next()).await.is_err(), "the fast subscription was not ended");

    // Counted, once, with what was dropped; nothing of it is left queued.
    let relay = server.relay().clone();
    let s = relay.stats();
    assert_eq!(s.too_far_behind, 1, "{s:?}");
    // The bound's items are objects and their streams' ends, two a group, but for the ends of
    // the streams being written (`write_concurrency`, 8), whose objects were with the transport.
    assert!(s.dropped_too_far_behind >= (BOUND - 8) / 2 && s.dropped_too_far_behind <= BOUND / 2 + 1, "{s:?}");
    assert_eq!(server.relay().session_diagnostics().len(), 3, "no session ended");
    assert!(until(STEP, || relay.stats().queued == 0).await);

    // What the slow subscriber finds when it reads at last: the streams the relay's writes were
    // stopped on, each reset TOO_FAR_BEHIND, and no more streams than the PUBLISH_DONE counted.
    // No object of the track was ever whole on a stream: each is larger than the window.
    let (mut whole, mut reset) = (0u64, 0u64);
    while let Some(mut r) = slow.next_stream(Duration::from_millis(500)).await {
        match timeout(STEP, r.start()).await.expect("in time") {
            Ok(_) => match read_stream(r).await {
                Some(g) if g.reset.is_none() => {
                    assert_eq!(g.header.track_alias, slow_ok.track_alias);
                    whole += 1;
                }
                Some(g) => {
                    assert_eq!(g.reset, Some(StreamResetCode::TOO_FAR_BEHIND.0), "group {}", g.header.group_id);
                    reset += 1;
                }
                None => panic!("not a subgroup stream"),
            },
            Err(StreamError::Transport(TransportError::Reset(code))) => {
                assert_eq!(code, StreamResetCode::TOO_FAR_BEHIND.0);
                reset += 1;
            }
            Err(e) => panic!("a data stream: {e}"),
        }
    }
    assert!(reset >= 1 && whole == 0, "{reset} streams reset TOO_FAR_BEHIND, {whole} whole");
    assert!(reset <= done.stream_count, "{reset} reset of {} opened", done.stream_count);

    // The session stands, and its other subscription: a group of the audio track arrives.
    assert!(timeout(Duration::from_millis(100), sarx.next()).await.is_err(), "the session's other subscription was not ended");
    let (h, o) = group(2, 0, SMALL);
    timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
    let got = collect(&slow, STEP, |g| !g.is_empty()).await;
    assert_eq!((got[0].header.track_alias, got[0].objects[0].payload.len(), got[0].reset), (slow_audio.track_alias, SMALL, None));

    // And the subscriber may subscribe again: it is served from the next group.
    let (again, _atx, _arx) = slow.subscribe(space.clone(), video.clone(), Parameters::new()).await.expect("SUBSCRIBE again");
    assert_ne!(again.track_alias, slow_ok.track_alias);
    let (h, o) = group(1, GROUPS, SMALL);
    timeout(STEP, publisher.send_subgroup(h, &o)).await.expect("in time").expect("a send");
    let got = collect(&slow, STEP, |g| !g.is_empty()).await;
    assert_eq!((got[0].header.track_alias, got[0].header.group_id, got[0].reset), (again.track_alias, GROUPS, None));
    assert_eq!(relay.stats().too_far_behind, 1);

    for c in [&publisher, &fast, &slow] {
        c.close(0, "done");
    }
    server.shutdown().await;
}