//! A WHIP publisher (str0m as the client) sends synthetic H.264 and Opus into the
//! gateway. A subscriber at the relay receives every group of both tracks, verifies every
//! receipt under the gateway's key (and the relay's door verifies them too), and the
//! frames' bytes equal what was sent. DELETE ends the publish, and the relay ends the
//! tracks with PUBLISH_DONE.

mod common;

use common::*;
use joule_moq_ingest::frame::Codec;
use std::time::Duration;
use tokio::time::timeout;

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

async fn run() {
    let w = World::start(|_| {}).await;
    let auth = w.auth(&w.publisher, 0);
    let p = publish(&w, &auth).await;
    let watcher = watch(&w).await;

    let mut sent = Sent::default();
    send(&p, &mut sent, 3, 30).await;
    assert!(received(&w, &p, sent.video.len() + sent.audio.len()).await, "every frame reached the gateway");

    // DELETE ends the publish, and the relay sees each track end.
    let r = delete(&w, &p.location, &auth).await;
    assert_eq!(r.status, 200);
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
    settle(&watcher, Duration::from_millis(500)).await;
    let r = delete(&w, &p.location, &auth).await;
    assert_eq!(r.status, 404, "the session is gone");

    let video = watcher.tracks.iter().find(|t| t.kind == 2).unwrap();
    let audio = watcher.tracks.iter().find(|t| t.kind == 1).unwrap();
    let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
    let (af, ag) = verified(&watcher, audio, &w.gateway_key, Codec::Opus);
    assert_eq!(vf.len(), sent.video.len(), "every video frame");
    same(&vf, &sent.video, "video frames byte for byte");
    same(&af, &sent.audio, "audio packets byte for byte");
    assert_eq!(vg, 3, "one video group per keyframe");
    assert!(ag >= 2, "audio groups: {ag}");
    // Each audio group spans less than its 200 ms window on the room clock, and video
    // groups begin at keyframes.
    let mut spans: std::collections::BTreeMap<u64, (u64, u64)> = Default::default();
    for ((alias, g, _), p) in watcher.objects.lock().unwrap().iter() {
        if *alias == audio.media_alias {
            let t = joule_moq_ingest::frame::open(p).unwrap().t;
            let e = spans.entry(*g).or_insert((t, t));
            e.0 = e.0.min(t);
            e.1 = e.1.max(t);
        }
    }
    assert!(spans.values().all(|(a, b)| b - a < 200_000), "every audio group within one window");
    let firsts: Vec<bool> = envelopes(&watcher, video).iter().map(|e| e.key).collect();
    assert_eq!(firsts.iter().filter(|k| **k).count(), 3);
    assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");

    // The session's record, signed by the gateway.
    assert!(until(STEP, || !w.gateway.gateway.records().is_empty()).await);
    let rec = w.gateway.gateway.records().remove(0);
    rec.verify().expect("the record verifies under the gateway's key");
    assert_eq!((rec.record.protocol.as_str(), rec.record.role, rec.record.ended.as_str()), ("whip", 3, "deleted"));
    assert_eq!(rec.record.energy.acquisition, "unmetered");
    let out: u64 = rec.record.tracks.iter().filter(|t| t.dir == 2).map(|t| t.objects).sum();
    assert_eq!(out, (sent.video.len() + sent.audio.len()) as u64);
    assert_eq!(rec.record.ledger.len(), 1);

    p.close();
    w.stop().await;
}