//! A FETCH that arrives while an earlier group's stream is still arriving: the cache does
//! not yet cover the range (the group is not whole until its FIN), and the publisher, which
//! published by PUBLISH, takes no FETCH. The relay waits a moment for the group to finish
//! (within the fetch's fill budget) and serves the whole range from its cache, rather than
//! asking the publisher.
//!
//! A track of one group per object, each on its own stream, as a page's audio is (one
//! group every 20 ms), always has a stream or two in flight; through an HTTP front the
//! window is wider, and a late joiner's fetch from group 0 met it (seen on the deployed
//! room, 2026-10-19).

mod common;

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

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

fn header(alias: u64, g: u64) -> SubgroupHeader {
    SubgroupHeader { track_alias: alias, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: false, end_of_group: true, first_object: true }
}

async fn run() {
    let server = relay(|_| {}).await;
    let t = Tracks::new(ns(&["room", "p", "audio"]), "0", 10, 5, 64);
    let (publisher, _ep) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let _held = publish_tracks(&publisher, &t, None).await;
    for g in 0..4 {
        publisher.send_subgroup(header(t.media_alias, g), &[SubgroupObject::normal(0, payload("a", g, 200))]).await.unwrap();
    }
    // Group 4's stream opens and carries its object, and its FIN is late; group 5 is whole.
    let mut b = Vec::new();
    let h4 = header(t.media_alias, 4);
    h4.encode(&mut b);
    SubgroupCodec::new(h4).encode_object(&SubgroupObject::normal(0, payload("a", 4, 200)), &mut b).unwrap();
    let mut open = publisher.conn.open_uni().await.unwrap();
    open.write_all(&b).await.unwrap();
    publisher.send_subgroup(header(t.media_alias, 5), &[SubgroupObject::normal(0, payload("a", 5, 200))]).await.unwrap();
    tokio::time::sleep(Duration::from_millis(200)).await;

    let (late, _) = connect(&server, Via::WebSocket, ClientOptions::default()).await;
    let fin = tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(300)).await;
        open.finish().unwrap();
        open
    });
    let (ok, entries) = late.fetch(t.ns.clone(), t.media.clone(), Parameters::new()).await.expect("the fetch is served");
    let _open = fin.await.unwrap();
    let groups: Vec<u64> = entries.iter().filter_map(|e| if let FetchEntry::Object(o) = e { Some(o.group_id) } else { None }).collect();
    assert_eq!(groups, (0..6).collect::<Vec<_>>(), "every group, from the cache");
    assert_eq!(ok.end_location.group, 5);
    let s = server.relay().stats();
    assert_eq!((s.fetch_served_from_cache, s.fetch_served_upstream), (1, 0), "{s:?}");
    late.close(0, "done");
    publisher.close(0, "done");
    server.shutdown().await;
}