//! SRT into a room's tracks, end to end on loopback: a publisher POSTs `/srt` with its grant
//! and gets a one-time session; it calls the gateway's listener with the stream ID and the
//! passphrase, through a path that drops datagrams each way, and sends a transport stream
//! of H.264 and AAC. A subscriber at the relay receives every group of both tracks, verifies
//! every receipt under the gateway's key, and gets every access unit and every ADTS frame
//! as sent. The link record's parts verify as a series, account for every sequence number,
//! show the losses recovered, and name the head of each track's receipt chain, which is the
//! last group receipt the subscriber verified; they are on the owner's `link` track too.
//! Then the refusals, the gateway calling a publisher's listener, and HEVC.

mod common;

use common::*;
use joule_contrib::lossy::{Counts, Impair, Proxy};
use joule_contrib::srt::{self, conn::CallError, packet::{rej, Packet}, Admitted, CallConfig, ConnRequest, ListenerConfig, Role};
use joule_contrib::ts;
use joule_moq::carriage::ReceiptBundle;
use joule_moq::receipt::ReceiptVerifier;
use joule_moq::{GroupId, ObjectId};
use joule_moq_ingest::catalog;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::linkrec::{self, ended, LinkRecord};
use joule_moq_ingest::room::media;
use std::collections::BTreeSet;
use std::sync::Arc;
use std::time::Duration;
use tokio::time::timeout;

/// Every object of `t` verified against its receipts under `key`, as its raw payload, in
/// group order: one object a group.
fn verified_raw(wt: &Watcher, t: &Watched, key: &ed25519_dalek::SigningKey) -> Vec<Vec<u8>> {
    let v = ReceiptVerifier::with_anchors(vec![key.verifying_key()]);
    let objs = wt.objects.lock().unwrap();
    let mut out = Vec::new();
    let mut prev = None;
    for ((alias, g, o), p) in objs.iter() {
        if *alias != t.media_alias {
            continue;
        }
        let b = objs.iter().find(|(k, _)| k.0 == t.receipts_alias && k.1 == *g).map(|(_, b)| ReceiptBundle::from_payload(b).unwrap()).expect("each group's receipts arrived");
        let vg = v.verify_group(&b.group).expect("the group receipt verifies under the gateway's key");
        if let Some(link) = prev {
            assert_eq!(b.group.parent_receipt_hash, Some(link), "each group chains to the one before");
        }
        prev = Some(b.group.link_hash());
        let r = &b.objects[*o as usize];
        r.check_moqt(b.group.track_alias, GroupId(*g), ObjectId(*o)).unwrap();
        v.verify_payload(p, r, &vg).expect("the object verifies against its receipt");
        out.push(p.clone());
    }
    out
}

async fn open_srt(w: &World, auth: &Auth) -> serde_json::Value {
    let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], b"").await;
    assert_eq!(r.status, 201, "POST /srt: {}", r.text());
    assert!(r.header("location").unwrap().starts_with("/srt/s/"));
    serde_json::from_slice(&r.body).unwrap()
}

fn call_cfg(s: &serde_json::Value) -> CallConfig {
    CallConfig {
        stream_id: s["stream_id"].as_str().map(str::to_string),
        passphrase: s["passphrase"].as_str().map(str::to_string),
        keylen: s["pbkeylen"].as_u64().unwrap() as usize,
        latency_ms: 300,
        ..CallConfig::default()
    }
}

/// The link records of the gateway, read as one series.
fn parts(w: &World) -> Vec<LinkRecord> {
    let recs = w.gateway.gateway.link_records();
    linkrec::verify_series(&recs).unwrap_or_else(|e| panic!("the link record's parts verify as a series: {e:?}"))
}


/// The sequence numbers whose first send the proxy dropped on its way to the receiver:
/// each must be found missing and recovered. Control packets and retransmissions it
/// dropped are not among them.
fn first_sends_lost(c: &Counts) -> BTreeSet<u32> {
    c.dropped_log.lock().unwrap().iter().filter_map(|b| match Packet::read(b) {
        Some(Packet::Data(d)) if !d.rexmit => Some(d.seq),
        _ => None,
    }).collect()
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn an_srt_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(|c| {
        c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
        c.contrib.record_interval_ms = 1_000;
        c.contrib.srt.latency_ms = 250;
    })
    .await;
    let auth = w.auth(&w.publisher, 0);
    let s = open_srt(&w, &auth).await;
    let sid = s["stream_id"].as_str().unwrap();
    assert!(sid.starts_with(&format!("#!::r={}/{},m=publish,s=", hex::encode(w.room), w.owner())), "{sid}");
    // 3% of the datagrams dropped each way, and a burst of six toward the gateway, mid-stream.
    // The publisher closes after about 211 datagrams and sends its three SHUTDOWNs back to
    // back; a burst at 200 swallowed all three when a host sent a few fewer before the close,
    // and the gateway ended the session by its idle timeout.
    let up = Impair { loss_ppm: 30_000, drop_index: &[100, 101, 102, 103, 104, 105], ..Impair::default() };
    let px = Proxy::start(w.gateway.srt_addr.unwrap(), up, Impair { loss_ppm: 30_000, ..Impair::default() }, 0xabc).await.unwrap();
    let link = timeout(STEP, srt::call(px.addr, call_cfg(&s))).await.unwrap().expect("the call is admitted");
    assert_eq!(link.info.cipher_bits, 128);
    // The tracks are published once the call is admitted; a subscriber joins before the
    // first frame.
    assert!(contrib_live(&w, s["session"].as_str().unwrap()).await, "the session publishes its tracks");
    let watcher = watch(&w).await;
    // A watcher of the owner's link track and its receipts.
    let links = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &w.owner()), "link".into())]).await;
    let sent = send_ts(&link, 3, 30, ts::ST_H264).await;
    tokio::time::sleep(Duration::from_millis(800)).await;
    // Close once the gateway has acknowledged every packet: a fixed wait alone left a loss
    // unrecovered on a loaded host, and the accounting below counted it outstanding.
    assert!(until(STEP, || link.sent.lock().unwrap().unacknowledged == 0).await, "the gateway acknowledged every packet before the publisher closed");
    // The path's losses, read once every packet has crossed it: each was acknowledged, so the
    // proxy has handled each of the publisher's 108 data packets, and its seed's drops (five at
    // random in its first hundred datagrams, then the burst of six) are behind it. Read as soon
    // as the publisher had queued its last payload, the count raced the proxy's task: a loaded
    // runner read it before the proxy had handled its 106th datagram.
    assert!(px.up.dropped.load(std::sync::atomic::Ordering::Relaxed) > 10, "the path dropped datagrams");
    link.close();
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
    settle(&watcher, Duration::from_millis(500)).await;

    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let audio = watcher.tracks.iter().find(|t| t.kind == media::AUDIO).unwrap();
    let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
    let (af, _) = verified(&watcher, audio, &w.gateway_key, Codec::Aac);
    same(&vf, &sent.video, "every access unit, byte for byte");
    same(&af, &sent.audio, "every ADTS frame, byte for byte");
    assert_eq!(vg, 3, "one video group per keyframe");
    assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the relay's door verified every object");
    let env = envelopes(&watcher, video);
    assert_eq!(env[0].manifest["model_requirement"]["capability"], "wai.video.avc");
    assert_eq!(env[0].manifest["live"]["rate"], 90_000);
    assert_eq!(envelopes(&watcher, audio)[0].manifest["model_requirement"]["capability"], "wai.audio.aac");

    // The link record.
    assert!(until(STEP, || w.gateway.gateway.link_records().iter().any(|b| LinkRecord::read(b).is_ok_and(|r| r.last))).await, "the last part is signed");
    let ps = parts(&w);
    assert!(ps.len() >= 3, "a part a second, and the last: {}", ps.len());
    let last = ps.last().unwrap();
    assert!(last.last && last.ended == ended::PEER_SHUTDOWN, "the last part ends the session as the publisher ended it: last={} ended={}", last.last, last.ended);
    let (mut delivered, mut first, mut dropped, mut count) = (0, 0, 0, 0);
    for p in &ps {
        assert_eq!((p.protocol, p.mode, p.cipher), (1, 1, 128));
        assert_eq!(p.window, 300_000, "the larger of the two ends' latencies");
        delivered += p.account.delivered;
        first += p.account.first_try;
        dropped += p.account.dropped;
        count += p.account.seq_count;
    }
    assert_eq!(dropped, 0, "every loss recovered within the window");
    // The burst of six dropped datagrams holds as many first sends of data as the timing of the
    // control packets between them leaves: count them, rather than assume all six.
    let lost = first_sends_lost(&px.up);
    assert!(lost.len() >= 2, "the path lost first sends of data: {lost:?}");
    assert_eq!(delivered - first, lost.len() as u64, "each first send lost was found and recovered");
    assert_eq!(count, delivered + last.account.outstanding);
    let counts = link.sent.lock().unwrap().clone();
    assert_eq!((delivered, counts.expired), (counts.sent, 0), "every payload the publisher sent, accounted for, none given up: {counts:?}");
    // Each track's head is the link hash of the last group receipt the subscriber verified.
    for t in [video, audio] {
        let objs = watcher.objects.lock().unwrap();
        let bundles: Vec<ReceiptBundle> = objs.iter().filter(|(k, _)| k.0 == t.receipts_alias).map(|(_, p)| ReceiptBundle::from_payload(p).unwrap()).collect();
        let ns = catalog::media_namespace(&w.room, &w.owner(), t.kind);
        let head = last.tracks.iter().find(|h| h.track == catalog::track_digest(&ns, "0")).expect("the record names the track");
        assert_eq!(head.groups, bundles.len() as u64);
        assert_eq!(head.head, bundles.last().unwrap().group.link_hash(), "the record names the head of the track's receipt chain");
    }
    // Every part is on the owner's link track, each object receipted by the gateway.
    settle(&links, Duration::from_millis(300)).await;
    let lt = &links.tracks[0];
    let got = verified_raw(&links, lt, &w.gateway_key);
    assert_eq!(got, w.gateway.gateway.link_records(), "the link track carries every part, as signed");

    // The record's session record names the protocol.
    assert!(until(STEP, || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "srt")).await);
    let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "srt").unwrap();
    rec.verify().unwrap();
    assert_eq!(rec.record.ended, "peer_shutdown");

    // The session's resource is gone.
    let loc = format!("{}/srt/s/{}", w.base(), s["session"].as_str().unwrap());
    assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 404);
    px.stop();
    w.stop().await;
}

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

async fn refusals() {
    let w = World::start(|c| c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap())).await;
    let addr = w.gateway.srt_addr.unwrap();
    let auth = w.auth(&w.publisher, 0);
    let s = open_srt(&w, &auth).await;
    let call = |cfg: CallConfig| srt::call(addr, CallConfig { connect_timeout_ms: 3_000, ..cfg });
    // No stream ID; one without a session; a session the gateway never gave.
    assert_eq!(call(CallConfig { passphrase: s["passphrase"].as_str().map(str::to_string), ..CallConfig::default() }).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
    let fake = format!("#!::r={}/{},m=publish,s={}", hex::encode(w.room), w.owner(), "00".repeat(16));
    assert_eq!(call(CallConfig { stream_id: Some(fake), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
    // Play mode, which this gateway does not serve.
    let req = s["stream_id"].as_str().unwrap().replace("m=publish", "m=request");
    assert_eq!(call(CallConfig { stream_id: Some(req), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::BAD_MODE)));
    // Another room; another owner.
    let other_room = s["stream_id"].as_str().unwrap().replace(&hex::encode(w.room), &"ab".repeat(32));
    assert_eq!(call(CallConfig { stream_id: Some(other_room), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::NOTFOUND)));
    let other_owner = s["stream_id"].as_str().unwrap().replace(&w.owner(), &"cd".repeat(32));
    assert_eq!(call(CallConfig { stream_id: Some(other_owner), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::FORBIDDEN)));
    // The right session and a wrong passphrase: refused, and the session is spent.
    assert_eq!(call(CallConfig { passphrase: Some("not the session's passphrase".into()), ..call_cfg(&s) }).await.err(), Some(CallError::Rejected(rej::BADSECRET)));
    assert_eq!(call(call_cfg(&s)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)), "a session ID is good for one use");
    // A new session, used once.
    let s2 = open_srt(&w, &auth).await;
    let l = call(call_cfg(&s2)).await.expect("admitted");
    assert_eq!(call(call_cfg(&s2)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
    // While its owner publishes, another session of the same owner is refused: 1409.
    let s3 = open_srt(&w, &auth).await;
    assert!(until(STEP, || w.gateway.gateway.contrib.active.load(std::sync::atomic::Ordering::SeqCst) == 1).await);
    assert_eq!(call(call_cfg(&s3)).await.err(), Some(CallError::Rejected(rej::CONFLICT)));
    l.close();
    // A viewer's grant cannot open a contribution session.
    let viewer = w.auth(&w.publisher, 1);
    let r = viewer.send("POST", &format!("{}/srt", w.base()), &[], b"").await;
    assert_eq!(r.status, 403, "{}", r.text());
    // DELETE of a session waiting to connect cancels it.
    let s4 = open_srt(&w, &auth).await;
    let loc = format!("{}/srt/s/{}", w.base(), s4["session"].as_str().unwrap());
    assert_eq!(auth.send("DELETE", &loc, &[], b"").await.status, 200);
    assert_eq!(call(call_cfg(&s4)).await.err(), Some(CallError::Rejected(rej::UNAUTHORIZED)));
    w.stop().await;
}

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

async fn calls() {
    let w = World::start(|c| c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap())).await;
    // The publisher's listener: it sends to whoever calls with its passphrase.
    let adm = |_: &ConnRequest| Ok(Admitted { passphrase: Some("the encoder's passphrase".into()), role: Role::Send, latency_ms: Some(200), tag: 0 });
    let mut enc = srt::Listener::bind(ListenerConfig::default(), Arc::new(adm)).await.unwrap();
    let auth = w.auth(&w.publisher, 0);
    let body = serde_json::json!({"call": enc.addr.to_string(), "passphrase": "the encoder's passphrase"}).to_string();
    // Not in call_allow: refused.
    let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], body.as_bytes()).await;
    assert_eq!(r.status, 403, "{}", r.text());
    w.stop().await;

    let w = World::start(|c| {
        c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
        c.contrib.srt.call_allow = vec!["127.0.0.1".into()];
    })
    .await;
    let auth = w.auth(&w.publisher, 0);
    let r = auth.send("POST", &format!("{}/srt", w.base()), &[("content-type", "application/json")], body.as_bytes()).await;
    assert_eq!(r.status, 201, "{}", r.text());
    let id: serde_json::Value = serde_json::from_slice(&r.body).unwrap();
    let link = timeout(STEP, enc.accept()).await.unwrap().unwrap();
    assert!(contrib_live(&w, id["session"].as_str().unwrap()).await);
    let watcher = watch(&w).await;
    let sent = send_ts(&link, 2, 30, ts::ST_HEVC).await;
    tokio::time::sleep(Duration::from_millis(700)).await;
    link.close();
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await);
    settle(&watcher, Duration::from_millis(500)).await;
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Hevc);
    same(&vf, &sent.video, "every HEVC access unit, byte for byte");
    assert_eq!(vg, 2, "one group per IRAP");
    let ps = parts(&w);
    assert!(ps.iter().all(|p| p.mode == 2 && p.cipher == 128), "the gateway called");
    w.stop().await;
}

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

/// Contribution at 20 Mbit/s for four seconds, a keyframe each second: video groups of about
/// 2.6 MB beside the audio. The relay's door holds each object until its group's receipts
/// follow the group's last object, so it reads each group whole off its stream before it
/// releases any of it, which the relay's stream window (`limits.quic_stream_window_bytes`)
/// must hold: under QUIC's default of about 1.25 MB the stream stops and the door refuses.
/// The link's window is two seconds: the door is the subject, and on a loaded runner a stall
/// past 300 ms had the gateway drop access units as too late (§4.6) before they reached it.
async fn big_groups() {
    let w = World::start(|c| {
        c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
        c.contrib.max_bitrate_kbps = 1_000_000;
        c.limits.max_queue_bytes = 256 << 20;
    })
    .await;
    let auth = w.auth(&w.publisher, 0);
    let s = open_srt(&w, &auth).await;
    let cfg = CallConfig { latency_ms: 2_000, ..call_cfg(&s) };
    let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), cfg)).await.unwrap().unwrap();
    assert!(contrib_live(&w, s["session"].as_str().unwrap()).await);
    let watcher = watch(&w).await;
    let chunks = joule_contrib::cli::synthetic(4, 20_000);
    joule_contrib::cli::pace(&link, &chunks, 800).await.unwrap();
    assert!(until(STEP, || link.sent.lock().unwrap().unacknowledged == 0).await, "the gateway acknowledged every packet before the publisher closed");
    link.close();
    assert!(until(STEP, || watcher.ended.lock().unwrap().len() == 2).await, "PUBLISH_DONE on both media tracks");
    settle(&watcher, Duration::from_millis(500)).await;
    assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the door released every group");
    assert!(until(STEP, || w.gateway.gateway.link_records().iter().any(|b| LinkRecord::read(b).is_ok_and(|r| r.last))).await, "the last part is signed");
    let ps = parts(&w);
    assert!(ps.iter().all(|p| p.window == 2_000_000), "the caller's window, the larger of the two ends'");
    assert_eq!(ps.iter().map(|p| p.account.dropped).sum::<u64>(), 0, "no packet dropped as too late");
    let all: Vec<u8> = chunks.iter().flat_map(|c| c.1.clone()).collect();
    let mut d = ts::Demux::new();
    let mut pes = d.push(&all);
    pes.extend(d.flush());
    let want: Vec<Vec<u8>> = pes.iter().filter(|p| p.stream_type == ts::ST_H264).map(|p| p.data.clone()).collect();
    let video = watcher.tracks.iter().find(|t| t.kind == media::VIDEO).unwrap();
    let (vf, vg) = verified(&watcher, video, &w.gateway_key, Codec::Avc);
    assert!(want.iter().map(|f| f.len()).sum::<usize>() > 9_000_000, "groups of megabytes");
    same(&vf, &want, "every access unit");
    assert_eq!(vg, 4);
    w.stop().await;
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn contribution_at_twenty_megabits_passes_the_door_with_no_subscriber() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|c| {
            c.contrib.srt.listen = Some("127.0.0.1:0".parse().unwrap());
            c.contrib.max_bitrate_kbps = 1_000_000;
            c.limits.max_queue_bytes = 256 << 20;
        })
        .await;
        let auth = w.auth(&w.publisher, 0);
        let s = open_srt(&w, &auth).await;
        let link = timeout(STEP, srt::call(w.gateway.srt_addr.unwrap(), call_cfg(&s))).await.unwrap().unwrap();
        let chunks = joule_contrib::cli::synthetic(4, 20_000);
        joule_contrib::cli::pace(&link, &chunks, 800).await.unwrap();
        link.close();
        assert!(until(STEP, || w.gateway.gateway.records().iter().any(|r| r.record.protocol == "srt")).await, "the session ended");
        assert_eq!(w.relay.relay().stats().refused_at_door, 0, "the door released every group");
        let rec = w.gateway.gateway.records().into_iter().find(|r| r.record.protocol == "srt").unwrap();
        assert_eq!(rec.record.ended, "peer_shutdown", "the session ended as the publisher ended it, not by the relay");
        w.stop().await;
    })
    .await
    .expect("in time");
}