//! A slow subscriber: its low-priority objects time out at the relay, while every receipt
//! still reaches it and its high-priority track is unaffected (`draft-ietf-moq-transport-22`
//! §5.1, §5.2; WAI jwp-receipts §4.1, §4.4).
//!
//! One publisher sends a staged pair over native QUIC: a base track at publisher priority
//! 0 with no delivery timeout, and a layer track at priority 200 whose
//! OBJECT_DELIVERY_TIMEOUT is 100 ms, each with its receipts track. A fast subscriber reads
//! as it goes. A slow one gives the relay small receive windows and reads nothing until the
//! publisher has sent everything and a second and a half more has passed; the relay is
//! given a small send window, so what cannot be sent waits in its scheduler, where the
//! layer's objects pass their deadline. Then it reads all its streams concurrently.

mod common;

use common::*;
use joule_moq::draft22::{Parameter, Parameters, StreamResetCode};
use joule_moq::receipt::ReceiptVerifier;
use joule_moq_relay::client::ClientOptions;
use std::time::Duration;

const GROUPS: u64 = 40;
const STALL: Duration = Duration::from_millis(1500);

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

async fn run() {
    let server = relay(|c| {
        c.limits.quic_send_window_bytes = 64 * 1024;
    })
    .await;
    let key = ed25519_dalek::SigningKey::from_bytes(&[0x51; 32]);
    let space = ns(&["stage", "cam"]);
    let base = Tracks::new(space.clone(), "base", 10, 1, 0);
    let layer = Tracks::new(space.clone(), "layer", 20, 2, 200);

    let (publisher, _ep) = connect(&server, Via::Quic, ClientOptions::default()).await;
    let _b = publish_tracks(&publisher, &base, None).await;
    let _l = publish_tracks(&publisher, &layer, Some(100)).await;

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

    let next = || Parameters::new().with(Parameter::LocationFilter(joule_moq::draft22::LocationFilter::NextObject));
    let fast_base = subscribe_both(&fast, &base, next()).await;
    let fast_layer = subscribe_both(&fast, &layer, next()).await;
    // The slow subscriber also asks for a 50 ms delivery timeout on its receipts
    // subscriptions, which it should not (moq-streaming-format §4.1): a receipts track has
    // none, and the relay applies none to it.
    let receipts_timeout = || Parameters::new().with(Parameter::ObjectDeliveryTimeout(50));
    let slow_base = subscribe_both_with(&slow, &base, next(), receipts_timeout()).await;
    let slow_layer = subscribe_both_with(&slow, &layer, next(), receipts_timeout()).await;

    let (published_tx, published_rx) = tokio::sync::oneshot::channel::<()>();
    let publish = async {
        for g in 0..GROUPS {
            publish_group(&publisher, &key, &base, g, &payload("base", g, 1024)).await;
            publish_group(&publisher, &key, &layer, g, &payload("layer", g, 24 * 1024)).await;
            tokio::time::sleep(Duration::from_millis(15)).await;
        }
        let _ = published_tx.send(());
    };
    // Both subscribers read their streams concurrently, as a subscriber does: one that read
    // them one at a time could wait on a stream whose bytes wait for connection credit
    // held by the unread streams behind it (see the README, "A subscriber that does not
    // read"). Each reads until the base track and every receipt are whole, then until a
    // second passes with nothing new, so every layer stream is read or seen reset.
    let whole = |media: u64, receipts: u64, layer_receipts: u64| {
        move |got: &[Got]| {
            let n = |a: u64| got.iter().filter(|x| x.header.track_alias == a && x.reset.is_none()).count() as u64;
            n(media) == GROUPS && n(receipts) == GROUPS && n(layer_receipts) == GROUPS
        }
    };
    let fast_reads = collect_concurrent(&fast, STEP * 2, Duration::from_secs(1), whole(fast_base.media_alias, fast_base.receipts_alias, fast_layer.receipts_alias));
    let slow_reads = async {
        // Nothing is read while the load arrives, however long a loaded host takes to send
        // it, nor for the stall after: every layer object waiting then is past its deadline.
        let _ = tokio::time::timeout(STEP * 2, published_rx).await;
        tokio::time::sleep(STALL).await;
        collect_concurrent(&slow, STEP * 2, Duration::from_secs(1), whole(slow_base.media_alias, slow_base.receipts_alias, slow_layer.receipts_alias)).await
    };
    let ((), fast_got, slow_got) = tokio::join!(publish, fast_reads, slow_reads);

    // The fast subscriber: every base object and every receipt.
    let (fo, fb, _, _) = verify(&fast_got, &fast_base, &key);
    assert_eq!((fo, fb), (GROUPS as usize, GROUPS as usize));
    let (_, flb, _, _) = verify(&fast_got, &fast_layer, &key);
    assert_eq!(flb, GROUPS as usize, "every layer receipt reached the fast subscriber");

    // The slow subscriber's base track is whole: each group once, none reset. (Objects on
    // different streams may arrive in any order; the order the relay hands them to the
    // transport is checked in priority_fan_out.rs.)
    let (bo, bb, _, _) = verify(&slow_got, &slow_base, &key);
    if (bo, bb) != (GROUPS as usize, GROUPS as usize) {
        // Where the missing base objects were: what the subscriber saw, what its transport
        // counted, and what the relay says it handed to the transport.
        let seen: Vec<(u64, Option<u64>)> = slow_got.iter().filter(|x| x.header.track_alias == slow_base.media_alias).map(|x| (x.header.group_id, x.reset)).collect();
        let missing: Vec<u64> = (0..GROUPS).filter(|g| !seen.iter().any(|(b, r)| b == g && r.is_none())).collect();
        let waited = slow.waited_out.lock().unwrap().clone();
        let client_quic = slow.conn.diagnostics();
        let relay_sessions = server.relay().session_diagnostics();
        // Then read every stream still queued, all at once, as a subscriber reading its
        // streams concurrently would: base objects held only by flow control now arrive.
        let mut reads = Vec::new();
        for _ in 0..8 * GROUPS {
            let Some(mut r) = slow.next_stream(Duration::from_millis(500)).await else { break };
            reads.push(tokio::spawn(async move {
                tokio::time::timeout(Duration::from_secs(10), r.start()).await.ok()?.ok()?;
                read_stream(r).await
            }));
        }
        let mut late_base = Vec::new();
        for t in reads {
            if let Ok(Some(g)) = t.await
                && g.header.track_alias == slow_base.media_alias
                && g.reset.is_none()
            {
                late_base.push(g.header.group_id);
            }
        }
        late_base.sort();
        slow.close(0, "diagnosis");
        let relay = server.relay().clone();
        until(STEP, || relay.summaries().iter().any(|s| s.summary.tally.subscriber.tracks.len() >= 4)).await;
        let base_name = joule_moq::draft22::names::render(&space, Some(&base.media));
        let tallies: Vec<String> = relay
            .summaries()
            .iter()
            .filter_map(|s| s.summary.tally.subscriber.tracks.get(&base_name).map(|t| format!("session {}: {:?}, undelivered at close {}", s.summary.session, t, s.summary.tally.subscriber.undelivered_at_close)))
            .collect();
        panic!(
            "the high-priority track is unaffected: {bo} base objects and {bb} base receipts of {GROUPS}\n\
             missing base groups {missing:?}; base streams seen (group, reset) {seen:?}\n\
             next_data gave up waiting on a header {} time(s), streams queued behind each: {waited:?}\n\
             base groups that arrived once the queued streams were read concurrently: {late_base:?}\n\
             subscriber QUIC: {client_quic:?}\nrelay sessions: {relay_sessions:?}\nrelay's base tallies: {tallies:?}",
            waited.len()
        );
    }
    let mut base_groups: Vec<u64> = slow_got.iter().filter(|x| x.header.track_alias == slow_base.media_alias).map(|x| x.header.group_id).collect();
    base_groups.sort();
    assert_eq!(base_groups, (0..GROUPS).collect::<Vec<_>>());

    // Its layer track lost objects to the delivery timeout; every layer receipt arrived
    // anyway, and each dropped object is accounted for by inclusion (jwp-receipts §4.4).
    let (lo, lb, groups, receipts) = verify(&slow_got, &slow_layer, &key);
    assert_eq!(lb, GROUPS as usize, "every layer receipt reached the slow subscriber");
    assert!(lo < GROUPS as usize, "some layer objects passed their deadline ({lo} of {GROUPS} delivered)");
    let delivered: std::collections::HashSet<u64> =
        slow_got.iter().filter(|x| x.header.track_alias == slow_layer.media_alias && x.reset.is_none()).map(|x| x.header.group_id).collect();
    let verifier = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    let mut inclusion = 0;
    for g in 0..GROUPS {
        if !delivered.contains(&g) {
            verifier.verify_object(&receipts[&g], &groups[&g]).expect("a dropped object's receipt verifies by inclusion");
            inclusion += 1;
        }
    }
    assert_eq!(inclusion as usize, GROUPS as usize - lo);
    for x in slow_got.iter().filter(|x| x.header.track_alias == slow_layer.media_alias) {
        if let Some(code) = x.reset {
            assert_eq!(code, StreamResetCode::DELIVERY_TIMEOUT.0, "a layer stream reset for its deadline");
        }
    }

    // The relay's summary of the slow session says the same.
    slow.close(0, "done");
    let relay = server.relay().clone();
    assert!(until(STEP, || relay.summaries().iter().any(|s| s.summary.tally.subscriber.dropped_delivery_timeout > 0)).await);
    let s = relay.summaries().into_iter().find(|s| s.summary.tally.subscriber.dropped_delivery_timeout > 0).unwrap();
    s.verify().unwrap();
    let t = &s.summary.tally.subscriber.tracks;
    let layer_name = joule_moq::draft22::names::render(&space, Some(&layer.media));
    let base_name = joule_moq::draft22::names::render(&space, Some(&base.media));
    let lt = &t[&layer_name];
    assert_eq!(lt.objects_forwarded as usize, lo, "the relay forwarded exactly what arrived whole");
    assert_eq!(lt.objects_forwarded + lt.dropped_delivery_timeout, GROUPS, "each layer object forwarded or dropped");
    assert_eq!(lt.dropped_receipts_track_subscribed, lt.dropped_delivery_timeout, "every drop on a track whose receipts track the session holds");
    assert_eq!(t[&base_name].dropped_delivery_timeout, 0);
    assert_eq!(t[&base_name].objects_forwarded, GROUPS);

    fast.close(0, "done");
    publisher.close(0, "done");
    server.shutdown().await;
}