//! The telephony bridge held to real time (`extensions/telephony-bridge.md` §13): one call's
//! sound carried both ways through the room for 30 seconds, every packet heard in order with
//! no underrun and its end-to-end delay under the stated bound; and as many calls at once as
//! the configuration opens by default (`max_calls`, 64), each carried into the room without loss
//! and each sent the room's mix without a gap.
//!
//! Each reads the host's load first and refuses to measure on a loaded host, failing and
//! saying so (`TELEPHONY_RT_ANY_LOAD=1` measures anyway, and the figure is then the loaded
//! host's). They are run on request on a build host, never on a shared runner:
//!
//! ```text
//! cargo test --release -p joule-moq-ingest --features telephony --test telephony_rt -- --ignored --nocapture --test-threads=1
//! ```
//!
//! `TELEPHONY_RT_SECONDS` (30), `TELEPHONY_RT_CALLS` (the default `max_calls`, 64) and
//! `TELEPHONY_RT_MANY_SECONDS` (20) change the durations and the count.

#![cfg(feature = "telephony")]

mod common;

use common::telephony::{heard, offer, payload, Caller, Peer};
use common::*;
use ed25519_dalek::SigningKey;
use joule_moq_ingest::catalog;
use joule_moq_ingest::frame::Codec;
use joule_moq_ingest::room::media;
use joule_moq_ingest::telephony::bridge::Bridge;
use joule_moq_ingest::telephony::config::{NumberConfig, TelephonyConfig};
use joule_moq_ingest::telephony::flac;
use joule_moq_ingest::telephony::g711::Law;
use serde_json::Value;
use std::sync::atomic::Ordering;
use std::time::{Duration, Instant};
use tokio::time::timeout;

/// The stated bound on one call's end-to-end delay through the room (§10): an object (20 ms),
/// a group (100 ms) and its receipts, the playout buffer's delay (160 ms), two mix ticks
/// (40 ms), and 80 ms for the relay's path and the host's scheduling.
const BOUND_MS: f64 = 400.0;

fn env(name: &str, default: u64) -> u64 {
    std::env::var(name).ok().and_then(|v| v.parse().ok()).unwrap_or(default)
}

/// The host's one-minute load and its threads; a host whose load is a quarter of its threads
/// or more is not measured on, unless asked.
fn quiet_enough() -> String {
    let threads = std::thread::available_parallelism().map(|n| n.get()).unwrap_or(1) as f64;
    let load = std::fs::read_to_string("/proc/loadavg").ok().and_then(|s| s.split_whitespace().next().and_then(|v| v.parse::<f64>().ok()));
    let said = format!("one-minute load {:?} on {threads} threads", load);
    if let Some(l) = load
        && l >= threads / 4.0
        && std::env::var_os("TELEPHONY_RT_ANY_LOAD").is_none()
    {
        panic!("not measured: the host is loaded ({said}); TELEPHONY_RT_ANY_LOAD=1 measures anyway");
    }
    said
}

fn key(n: u32) -> SigningKey {
    let mut s = [0u8; 32];
    s[..4].copy_from_slice(&n.to_be_bytes());
    s[31] = 0x77;
    SigningKey::from_bytes(&s)
}

/// A number a call, `+1555<i>`, each with its one grant: call `i` is grant `i`'s. The bridge
/// opens its default number of calls at once (64), or as many as asked for where that is more:
/// at the default count the calls fill it.
fn config(peer: std::net::SocketAddr, grants: Vec<String>) -> TelephonyConfig {
    let calls = grants.len();
    let number = grants.into_iter().enumerate().map(|(i, g)| NumberConfig { number: format!("+1555{i}"), grants: vec![g], pin: false, max_calls: None }).collect();
    TelephonyConfig { enabled: true, sip_udp: Some("127.0.0.1:0".parse().unwrap()), peer: Some(peer), media_ip: "127.0.0.1".parse().unwrap(), number, max_calls: calls.max(TelephonyConfig::default().max_calls), ..TelephonyConfig::default() }
}

fn records(w: &World) -> Vec<Value> {
    w.gateway.gateway.records().into_iter().filter(|r| r.record.protocol == "telephony").map(|r| serde_json::to_value(&r.record).unwrap()).collect()
}

fn summary(d: &mut [f64]) -> String {
    d.sort_by(|a, b| a.partial_cmp(b).unwrap());
    let q = |p: f64| d[((d.len() as f64 - 1.0) * p).round() as usize];
    format!("mean {:.1} ms, median {:.1}, 99th {:.1}, most {:.1}", d.iter().sum::<f64>() / d.len() as f64, q(0.5), q(0.99), q(1.0))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "a real-time measurement: on request, on a quiet build host"]
async fn one_call_both_ways_in_real_time() {
    let load = quiet_enough();
    let secs = env("TELEPHONY_RT_SECONDS", 30);
    timeout(Duration::from_secs(secs + 120), async {
        let w = World::start(|_| {}).await;
        let (ka, kb) = (key(1), key(2));
        let peer = Peer::bind("127.0.0.1").await;
        let bridge = Bridge::start(w.gateway.gateway.clone(), config(peer.addr, vec![World::token(&w.grant(&ka, 0)), World::token(&w.grant(&kb, 0))])).await.unwrap();
        let sip = bridge.udp_addr.unwrap();
        let mut a = Caller::new("127.0.0.1", "rt-a").await;
        let mut b = Caller::new("127.0.0.1", "rt-b").await;
        assert_eq!(a.invite(&peer, sip, "+15550", &[], &offer(a.port(), "0 101", "")).await.code(), Some(200));
        assert_eq!(b.invite(&peer, sip, "+15551", &[], &offer(b.port(), "0 101", "")).await.code(), Some(200));
        assert!(until(STEP, || bridge.calls_open() == 2).await);
        tokio::time::sleep(Duration::from_millis(500)).await;
        let n = (secs * 50) as u32;
        let (mut sent_a, mut sent_b) = (Vec::new(), Vec::new());
        let start = Instant::now();
        let mut late_sends = 0u32;
        for k in 0..n {
            // Paced on the clock, not on the last send: a sender that falls behind is counted.
            let due = start + Duration::from_millis(20 * k as u64);
            tokio::time::sleep_until(due.into()).await;
            if Instant::now().duration_since(due) > Duration::from_millis(10) {
                late_sends += 1;
            }
            sent_a.push(Instant::now());
            a.send(&payload(1, k)).await;
            sent_b.push(Instant::now());
            b.send(&payload(2, k)).await;
        }
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let ha = heard(&b.received.lock().unwrap(), 1, &sent_a);
        let hb = heard(&a.received.lock().unwrap(), 2, &sent_b);
        let ticks = bridge.stats().ticks.load(Ordering::Relaxed);
        let ticks_late = bridge.stats().ticks_late.load(Ordering::Relaxed);
        eprintln!("telephony real time: {secs} s each way, {load}; the test's sender late (over 10 ms) {late_sends} of {n}; the mix's ticks {ticks}, late (over 10 ms) {ticks_late}");
        for (who, h) in [("B hearing A", ha), ("A hearing B", hb)] {
            let mut d = h.delays_ms.clone();
            let first = h.first.unwrap_or_else(|| panic!("{who}: nothing heard"));
            eprintln!("telephony real time: {who}: from packet {first}, {} of {n} heard in order, {} breaks; delay {}", h.run, h.breaks, summary(&mut d));
            assert_eq!(h.breaks, 0, "{who}: an underrun or a gap");
            assert_eq!(first + h.run, n, "{who}: every packet from the first heard to the last");
            assert!(first <= 25, "{who}: the first packet heard is within half a second: {first}");
            assert!(d.last().copied().unwrap_or(f64::MAX) <= BOUND_MS, "{who}: a delay past the bound of {BOUND_MS} ms");
        }
        assert_eq!(late_sends, 0, "the test's own sender kept time");
        assert_eq!(a.in_dialog(&peer, sip, "BYE", &[], "").await.code(), Some(200));
        assert_eq!(b.in_dialog(&peer, sip, "BYE", &[], "").await.code(), Some(200));
        assert!(until(STEP, || records(&w).len() == 2).await);
        for r in records(&w) {
            assert_eq!((r["detail"]["rtp"]["lost"].as_u64(), r["detail"]["rtp"]["jitter"]["late"].as_u64()), (Some(0), Some(0)));
        }
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 8)]
#[ignore = "a real-time measurement: on request, on a quiet build host"]
async fn the_default_limit_of_calls_at_once() {
    let load = quiet_enough();
    let calls = env("TELEPHONY_RT_CALLS", TelephonyConfig::default().max_calls as u64) as u32;
    let secs = env("TELEPHONY_RT_MANY_SECONDS", 20);
    timeout(Duration::from_secs(secs + 300), async {
        let w = World::start(|_| {}).await;
        let keys: Vec<SigningKey> = (0..calls).map(|i| key(100 + i)).collect();
        let peer = Peer::bind("127.0.0.1").await;
        let bridge = Bridge::start(w.gateway.gateway.clone(), config(peer.addr, keys.iter().map(|k| World::token(&w.grant(k, 0))).collect())).await.unwrap();
        let sip = bridge.udp_addr.unwrap();
        let mut callers = Vec::new();
        for i in 0..calls {
            let mut c = Caller::new("127.0.0.1", &format!("many-{i}")).await;
            assert_eq!(c.invite(&peer, sip, &format!("+1555{i}"), &[], &offer(c.port(), "0 101", "")).await.code(), Some(200), "call {i}");
            callers.push(c);
        }
        assert!(until(STEP, || bridge.calls_open() == calls as usize).await);
        tokio::time::sleep(Duration::from_millis(1000)).await;
        // A relay subscriber on the first five calls' tracks, verifying every object.
        let owners: Vec<String> = keys.iter().map(|k| hex::encode(k.verifying_key().to_bytes())).collect();
        let wt = watch_tracks(&w, owners.iter().take(5).map(|o| (media::AUDIO, catalog::media_namespace(&w.room, o, media::AUDIO), "0".to_string())).collect()).await;
        tokio::time::sleep(Duration::from_millis(300)).await;
        let n = (secs * 50) as u32;
        let start = Instant::now();
        let mut late_sends = 0u32;
        for k in 0..n {
            let due = start + Duration::from_millis(20 * k as u64);
            tokio::time::sleep_until(due.into()).await;
            if Instant::now().duration_since(due) > Duration::from_millis(10) {
                late_sends += 1;
            }
            for (i, c) in callers.iter_mut().enumerate() {
                c.send(&payload(i as u8, k)).await;
            }
        }
        let spent = start.elapsed();
        tokio::time::sleep(Duration::from_millis(1500)).await;
        let ticks_late = bridge.stats().ticks_late.load(Ordering::Relaxed);
        // The mix sent each call a packet every 20 ms, with no gap of more than one tick.
        let mut gaps = 0u32;
        let mut fewest = u32::MAX;
        for c in &callers {
            let r = c.received.lock().unwrap();
            let window: Vec<Instant> = r.iter().map(|(t, _)| *t).filter(|t| *t >= start && *t <= start + spent).collect();
            fewest = fewest.min(window.len() as u32);
            gaps += window.windows(2).filter(|p| p[1].duration_since(p[0]) > Duration::from_millis(60)).count() as u32;
        }
        for c in callers.iter_mut() {
            assert_eq!(c.in_dialog(&peer, sip, "BYE", &[], "").await.code(), Some(200));
        }
        assert!(until(Duration::from_secs(60), || records(&w).len() == calls as usize).await, "every call's record");
        let recs = records(&w);
        let lost: u64 = recs.iter().map(|r| r["detail"]["rtp"]["lost"].as_u64().unwrap_or(u64::MAX)).sum();
        let late: u64 = recs.iter().map(|r| r["detail"]["rtp"]["jitter"]["late"].as_u64().unwrap_or(u64::MAX)).sum();
        let packets: u64 = recs.iter().map(|r| r["detail"]["rtp"]["packets"].as_u64().unwrap_or(0)).sum();
        let into_room: u64 = recs.iter().map(|r| r["detail"]["audio"]["frames_in"].as_u64().unwrap_or(0)).sum();
        let out_of_room: u64 = recs.iter().map(|r| r["detail"]["audio"]["objects_out"].as_u64().unwrap_or(0)).sum();
        eprintln!("telephony real time: {calls} calls for {secs} s, {load}; RTP packets read {packets} of {} sent, lost {lost}, late {late}; objects into the room {into_room}, published {out_of_room}; the mix: fewest packets a call received in the window {fewest} (of about {n}), gaps over 60 ms {gaps}, ticks late {ticks_late}; the test's sender late {late_sends} of {n}", calls as u64 * n as u64);
        assert_eq!(late_sends, 0, "the test's own sender kept time");
        assert_eq!((lost, late), (0, 0));
        assert_eq!(packets, calls as u64 * n as u64, "every packet read");
        assert_eq!((into_room, out_of_room), (packets, packets), "every packet into the room and published, one object each");
        assert_eq!(gaps, 0, "the mix sent every call a packet every tick");
        assert!(fewest + 10 >= n, "each call got the mix's packets: the fewest {fewest} of {n}");
        // Five of the tracks verified at the relay, sample for sample.
        settle(&wt, Duration::from_millis(500)).await;
        for (i, t) in wt.tracks.iter().enumerate() {
            let (frames, _) = verified(&wt, t, &w.gateway_key, Codec::Flac);
            let chars: Vec<u8> = frames.iter().flat_map(|f| flac::read(f).unwrap().samples.into_iter().map(|s| Law::Mu.encode(s as i16))).collect();
            let want: Vec<u8> = (0..n).flat_map(|k| payload(i as u8, k)).collect();
            assert!(chars == want, "call {i}'s track is what it sent");
        }
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}