//! Playout timing end to end (`draft-nandakumar-moq-tempo-00`; WAI co-watch §15): a publisher
//! stamps a room's source on a session ([`tempo::Stamper`]), the relay forwards it, rewriting
//! HopTime for the subscriber that asked, and the subscriber holds each object it receives to
//! the room ([`tempo::Receiver`]): the instant it states is the one the controlled position
//! reaches it at, its delay is within the track's bound, and its HopTime neither goes back
//! nor jumps.
//!
//! - every object arrives with the instant the publisher stamped, and a HopTime of the relay's
//!   clock, where the publisher's own clock (here a day behind) would say nothing of the hop;
//! - a subscriber that did not ask reads the same instants, with the publisher's HopTime;
//! - planted, on the same session and through the same relay, each refused at the subscriber
//!   with its reason: an instant one millisecond off the controlled position
//!   (`tempo_off_position`), a delay past the track's bound (`tempo_delay_out_of_bound`), a
//!   CaptureTimestamp moved outside Immutable Properties (`tempo_malformed`), and, at the
//!   publisher, an object the controls never reach, which is not stamped (`tempo_not_due`).

use joule_moq::draft22::data::{SubgroupHeader, SubgroupId, SubgroupObject};
use joule_moq::draft22::kvp::{prop, KeyValuePair, Properties};
use joule_moq::draft22::{Parameter, Parameters, StreamHeader, UniEvent};
use joule_moq::tempo::{CAPTURE_TIMESTAMP, PLAYOUT_DELAY};
use joule_moq::{TrackName, TrackNamespace};
use joule_moq_relay::client::{Client, ClientOptions};
use joule_moq_relay::transport::mux::MuxConfig;
use joule_moq_relay::transport::websocket;
use joule_moq_relay::{RelayConfig, RelayServer};
use joule_moq_room::tempo::{self, corpus, CatalogTempo, Reason, Receiver, Stamper, Timing};
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;
use wai::co_watch::action;

const STEP: Duration = Duration::from_secs(20);
const ALIAS: u64 = 3;
const LEN_US: u64 = 40_000;
const MAX_DELAY_MS: u32 = 2000;

async fn session(server: &RelayServer) -> Client {
    let url = format!("ws://{}{}", server.websocket_addr.unwrap(), server.relay().config().path);
    let conn: joule_moq_relay::transport::SharedConnection = Arc::new(websocket::connect(&url, MuxConfig::default()).await.expect("the relay"));
    Client::connect(conn, ClientOptions::default()).await.expect("SETUP")
}

async fn send(p: &Client, g: u64, o: SubgroupObject) {
    let h = SubgroupHeader { track_alias: ALIAS, group_id: g, subgroup_id: SubgroupId::Zero, publisher_priority: Some(64), properties: true, end_of_group: true, first_object: true };
    p.send_subgroup(h, &[o]).await.expect("the object");
}

/// The next `n` groups a subscriber receives, by Group ID: each one's object.
async fn receive(c: &Client, n: usize) -> Vec<(u64, SubgroupObject)> {
    let mut out = Vec::new();
    while out.len() < n {
        let mut r = c.next_data(STEP).await.expect("a data stream in time");
        let StreamHeader::Subgroup(h) = r.header.clone().expect("a header") else { panic!("a subgroup stream") };
        match timeout(STEP, r.next()).await.expect("an object in time") {
            Ok(Some(UniEvent::SubgroupObject(o))) => out.push((h.group_id, o)),
            other => panic!("{other:?}"),
        }
    }
    out.sort_by_key(|(g, _)| *g);
    out
}

fn now_ms() -> u64 {
    joule_moq_relay::relay::now_unix_ms()
}

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

async fn run() {
    let mut cfg = RelayConfig::loopback();
    cfg.quic_listen = None;
    let server = RelayServer::start(cfg).await.expect("the relay starts");
    // Played from 1 s to 3 s, held to 4.5 s, played on.
    let tl = corpus::timeline(&[(action::PLAY, 1_000_000, 0), (action::PAUSE, 3_000_000, 0), (action::PLAY, 4_500_000, 0)]);
    let t0 = corpus::T0_US;
    let catalog = CatalogTempo { max_delay_ms: MAX_DELAY_MS, playout_delay_ms: None, aligned: None };
    let stamper = Stamper { timeline: tl.clone(), t0_us: t0, max_delay_ms: MAX_DELAY_MS };

    let ns = TrackNamespace::from_strs(["room", "source", "video"]);
    let name = TrackName::from_str("0");
    let p = session(&server).await;
    let mut track = Properties::new();
    track.push(KeyValuePair::varint(prop::DEFAULT_PUBLISHER_PRIORITY, 64));
    let _pub = p.publish(ns.clone(), name.clone(), ALIAS, Parameters::new(), track).await.expect("PUBLISH");
    let asked = session(&server).await;
    let plain = session(&server).await;
    let _s1 = asked.subscribe(ns.clone(), name.clone(), Parameters::new().with(Parameter::SyncSupport)).await.expect("SUBSCRIBE with Sync Support");
    let _s2 = plain.subscribe(ns.clone(), name.clone(), Parameters::new()).await.expect("SUBSCRIBE");

    // The publisher's clock is a day behind: its HopTime says nothing of the last hop.
    let day = 86_400_000;
    const N: u64 = 8;
    let q_of = |g: u64| 1_900_000 + g * LEN_US; // across the pause: some objects are due after it
    let mut stamped = Vec::new();
    let before = now_ms();
    for g in 0..N {
        let x = tempo::due(&tl, 0, q_of(g), LEN_US).expect("the position reaches the object");
        let capture_ms = (t0 + x) / 1000 - 150;
        let o = stamper.object(0, q_of(g), LEN_US, capture_ms, now_ms() - day, &Properties::new(), format!("frame {g}").into_bytes()).expect("a stamp");
        stamped.push(x);
        send(&p, g, o).await;
    }

    // Each group rides a stream of its own, and the relay forwards streams beside one another:
    // it may stamp group 5 a millisecond before group 4. HopTime is held to its order along a
    // stream and to the sink's limit across them (`HopCheck`), never to Group ID order.
    for (c, rewritten) in [(&asked, true), (&plain, false)] {
        let mut rx = Receiver::new(tl.clone(), t0, catalog, 60_000);
        let got = receive(c, N as usize).await;
        let after = now_ms();
        let mut hop_before: Option<(u64, u64)> = None;
        for (g, o) in &got {
            assert_eq!(o.payload, format!("frame {g}").into_bytes());
            let r = rx.object((*g, 0), 0, q_of(*g), LEN_US, &o.properties, now_ms()).unwrap_or_else(|e| panic!("group {g}: {e}"));
            assert_eq!(r.due_us, stamped[*g as usize], "the instant the publisher stamped");
            let hop = r.hop_ms.expect("a HopTime");
            // Seen, not required: the relay stamping a later Group's stream first.
            if let Some((pg, ph)) = hop_before.filter(|(_, ph)| hop < *ph) {
                eprintln!("group {g} forwarded before group {pg}: HopTime {hop} < {ph}");
            }
            hop_before = Some((*g, hop));
            if rewritten {
                assert!(hop >= before && hop <= after, "the relay's clock: {before} <= {hop} <= {after}");
                assert!(r.oneway_ms.unwrap() >= 0 && r.oneway_ms.unwrap() < 60_000, "a last hop of the relay's clock");
            } else {
                assert!(hop < before - day + 60_000, "the publisher's HopTime, as sent");
            }
        }
    }

    // ── Planted, each sent through the relay and refused at the subscriber ──
    let x = tempo::due(&tl, 0, q_of(20), LEN_US).unwrap();
    let capture_ms = (t0 + x) / 1000 - 150;
    let good = tempo::stamp(&tl, t0, 0, q_of(20), LEN_US, capture_ms, MAX_DELAY_MS).unwrap();
    let object = |t: Timing| SubgroupObject { object_id: 0, properties: Timing { hop_ms: Some(now_ms()), ..t }.write(&Properties::new()).unwrap(), status: joule_moq::draft22::data::ObjectStatus::Normal, payload: b"planted".to_vec() };
    // 100: an instant one millisecond off the controlled position.
    send(&p, 100, object(Timing { delay_ms: good.delay_ms.map(|d| d + 1), ..good })).await;
    // 101: a delay past the track's bound, the same instant.
    send(&p, 101, object(Timing { capture_ms: Some(capture_ms - 5000), delay_ms: good.delay_ms.map(|d| d + 5000), ..good })).await;
    // 102: the capture time outside Immutable Properties, where a relay could rewrite it.
    let mut inner = Properties::new();
    inner.push(KeyValuePair::bytes(PLAYOUT_DELAY, good.delay_ms.unwrap().to_be_bytes().to_vec()));
    let mut outside = Properties::new();
    let mut v = Vec::new();
    joule_moq::draft22::varint::put(&mut v, capture_ms);
    outside.push(KeyValuePair::bytes(CAPTURE_TIMESTAMP, v));
    outside.set_immutable(&inner).unwrap();
    send(&p, 102, SubgroupObject { object_id: 0, properties: outside, status: joule_moq::draft22::data::ObjectStatus::Normal, payload: b"planted".to_vec() }).await;
    // 103: the claim as stamped, which stands.
    send(&p, 103, object(good)).await;
    let got = receive(&asked, 4).await;
    let mut rx = Receiver::new(tl.clone(), t0, catalog, 60_000);
    let verdicts: Vec<Result<u64, Reason>> = got.iter().map(|(g, o)| rx.object((*g, 0), 0, q_of(20), LEN_US, &o.properties, now_ms()).map(|r| r.due_us)).collect();
    assert_eq!(verdicts, [Err(Reason::OffPosition), Err(Reason::DelayOutOfBound), Err(Reason::Malformed), Ok(x)]);
    // And an object the controls never reach is not stamped at all.
    assert_eq!(stamper.object(0, corpus::DURATION_US + 1_000_000, LEN_US, capture_ms, now_ms(), &Properties::new(), Vec::new()).unwrap_err(), Reason::NotDue);

    for c in [&asked, &plain, &p] {
        c.close(0, "done");
    }
    server.shutdown().await;
}