//! The telephony bridge on loopback (`extensions/telephony-bridge.md` §8 to §11), with the
//! tests' own SIP peer and RTP callers (`common::telephony`), a relay whose door checks every
//! receipt of the room's tracks, and the gateway: two calls that hear each other through the
//! room, their tracks verified at a relay subscriber sample for sample; credentials and a PIN;
//! the planted faults over the wire; SIP over TCP and TLS; a call the bridge places and its
//! registration.

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

mod common;

use common::telephony::{heard, heard_to_the_end, holding_pin, offer, payload, Caller, Peer, PIN};
use common::*;
use ed25519_dalek::SigningKey;
use joule_moq::draft22::{Parameters, StreamHeader, UniEvent};
use joule_moq::TrackName;
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::{InboundAuth, NumberConfig, OutboundAuth, PinConfig, TelephonyConfig};
use joule_moq_ingest::telephony::digest::{self, Alg, Challenge, Nonces, Policy};
use joule_moq_ingest::telephony::g711::Law;
use joule_moq_ingest::telephony::{flac, sdp, sip};
use serde_json::Value;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::time::timeout;

fn key(n: u8) -> SigningKey {
    SigningKey::from_bytes(&[n; 32])
}

/// A bridge configuration for `peer`, with `numbers`, on loopback.
fn config(peer: std::net::SocketAddr, numbers: Vec<NumberConfig>) -> TelephonyConfig {
    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: numbers, source: Some("digitalCapture".into()), media_timeout_ms: 5_000, ..TelephonyConfig::default() }
}

fn number(n: &str, grants: Vec<String>) -> NumberConfig {
    NumberConfig { number: n.into(), grants, pin: false, max_calls: None }
}

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()
}

async fn until_records(w: &World, n: usize) -> Vec<Value> {
    assert!(until(STEP, || records(w).len() >= n).await, "{n} call records");
    records(w)
}

/// Each object of a track with the capture instant its properties state.
type Stamps = Arc<Mutex<Vec<Option<u64>>>>;

async fn watch_stamps(w: &World, owner: &str) -> (Stamps, Vec<joule_moq_relay::wire::ReqTx>) {
    let (c, ep) = relay_client(w).await;
    let c = Arc::new(c);
    let ns = catalog::media_namespace(&w.room, owner, media::AUDIO);
    let (_, tx, _rx) = c.subscribe(ns, TrackName::from_str("0"), Parameters::new()).await.expect("SUBSCRIBE");
    let out: Stamps = Arc::default();
    let o2 = out.clone();
    tokio::spawn(async move {
        let _ep = ep;
        while let Some(mut r) = c.next_stream(Duration::from_secs(30)).await {
            let o2 = o2.clone();
            tokio::spawn(async move {
                let Ok(Ok(h)) = timeout(STEP, r.start()).await else { return };
                while let Ok(Ok(Some(e))) = timeout(STEP, r.next()).await {
                    if let (StreamHeader::Subgroup(_), UniEvent::SubgroupObject(o)) = (&h, e) {
                        let inner = o.properties.immutable().ok().flatten().unwrap_or_default();
                        let c = inner.0.iter().find(|k| k.ty == joule_moq::tempo::CAPTURE_TIMESTAMP).and_then(|k| k.as_bytes()).and_then(|b| joule_moq::draft22::varint::get(b, "a timestamp").ok()).map(|v| v.0);
                        o2.lock().unwrap().push(c);
                    }
                }
            });
        }
    });
    (out, vec![tx])
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn two_calls_hear_each_other_through_the_room() {
    timeout(Duration::from_secs(180), async {
        let w = World::start(|_| {}).await;
        let (ka, kb) = (key(0x61), key(0x62));
        let (oa, ob) = (hex::encode(ka.verifying_key().to_bytes()), hex::encode(kb.verifying_key().to_bytes()));
        let peer = Peer::bind("127.0.0.1").await;
        let cfg = config(peer.addr, vec![number("+15550100", vec![World::token(&w.grant(&ka, 0))]), number("+15550101", vec![World::token(&w.grant(&kb, 0))])]);
        let bridge = Bridge::start(w.gateway.gateway.clone(), cfg).await.expect("the bridge starts");
        let sip_addr = bridge.udp_addr.unwrap();
        let mut a = Caller::new("127.0.0.1", "call-a").await;
        let mut b = Caller::new("127.0.0.1", "call-b").await;
        assert_eq!(a.invite(&peer, sip_addr, "+15550100", &[], &offer(a.port(), "0 8 101", "")).await.code(), Some(200));
        // The call took its number's grant: its track is published as A's. Watch it before
        // sound, once the relay has it.
        assert!(until(STEP, || bridge.stats().calls_published.load(std::sync::atomic::Ordering::Relaxed) == 1).await, "A's tracks published");
        let wt = watch_tracks(&w, vec![(media::AUDIO, catalog::media_namespace(&w.room, &oa, media::AUDIO), "0".into())]).await;
        let (stamps, _held) = watch_stamps(&w, &oa).await;
        let cat = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &oa), catalog::CATALOG_NAME.into())]).await;
        assert_eq!(b.invite(&peer, sip_addr, "+15550101", &[], &offer(b.port(), "0 101", "")).await.code(), Some(200));
        assert!(until(STEP, || bridge.stats().calls_published.load(std::sync::atomic::Ordering::Relaxed) == 2).await, "B's tracks published");
        // Three seconds of sound each way, a packet every 20 ms. A sender its host stalls goes on
        // at that pace from where it was, as a caller does, and does not send what it missed in
        // one burst: a burst of the host's making, past what a playout buffer holds (§10.2: a
        // second), would be dropped by the bridge's rule.
        const N: u32 = 150;
        let start_us = now_us();
        let mut sent_a = Vec::new();
        let mut sent_b = Vec::new();
        let mut tick = tokio::time::interval(Duration::from_millis(20));
        tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        for k in 0..N {
            tick.tick().await;
            sent_a.push(Instant::now());
            a.send(&payload(1, k)).await;
            sent_b.push(Instant::now());
            b.send(&payload(2, k)).await;
        }
        // Each hears the other's last packet and the silence after it, however long the host
        // takes to carry them: the wait is on that, not on the clock. A call whose caller sends
        // nothing for its media timeout ends on its own and refuses its BYE, so what was heard
        // is told first.
        let ended = until(STEP, || heard_to_the_end(&b.received, 1, N - 1) && heard_to_the_end(&a.received, 2, N - 1)).await;
        let byes = (a.in_dialog(&peer, sip_addr, "BYE", &[], "").await.code(), b.in_dialog(&peer, sip_addr, "BYE", &[], "").await.code());
        let recs = until_records(&w, 2).await;
        // Each heard the other's every packet after the first, in order, exactly: read by
        // sequence, with nothing between them but silence, and each run of silence a stop of
        // the playout buffer that the hearer's record states (§10.2, §11), the host's and not
        // the mix's: the buffer had run dry, and the sound that ended the stop reached the mix
        // after it. No underrun at all is the real-time test's claim, on a quiet host
        // (tests/telephony_rt.rs).
        let ha = heard(&b.received.lock().unwrap(), 1, &sent_a);
        let hb = heard(&a.received.lock().unwrap(), 2, &sent_b);
        let traces = [common::telephony::trace(&b.received.lock().unwrap(), 1), common::telephony::trace(&a.received.lock().unwrap(), 2)];
        // The sender's own stalls (its host's), for the messages: a packet sent more than 10 ms
        // after its 20 ms.
        let late = |sent: &[Instant]| {
            let l: Vec<f64> = sent.windows(2).map(|w| w[1].duration_since(w[0]).as_secs_f64() * 1000.0 - 20.0).collect();
            format!("the sender stalled more than 10 ms before {} of {N} packets, {:.0} ms at most", l.iter().filter(|&&x| x > 10.0).count(), l.iter().cloned().fold(0.0, f64::max))
        };
        let ticks = |s: &joule_moq_ingest::telephony::bridge::Stats| format!("the mix's ticks {}, {} of them more than 10 ms late", s.ticks.load(std::sync::atomic::Ordering::Relaxed), s.ticks_late.load(std::sync::atomic::Ordering::Relaxed));
        for (i, (who, h, hearer, track, sent)) in [("B hearing A", &ha, &ob, &oa, &sent_a), ("A hearing B", &hb, &oa, &ob, &sent_b)].into_iter().enumerate() {
            let rec = recs.iter().find(|r| r["principal"] == hearer.as_str()).unwrap_or_else(|| panic!("{who}: the hearer's record"));
            let counts = &rec["detail"]["heard"]["tracks"][track.as_str()];
            let (underruns, overflowed) = (counts["underruns"].as_u64().unwrap_or(u64::MAX), counts["overflowed"].as_u64());
            let first = h.first.unwrap_or_else(|| panic!("{who}: nothing heard"));
            // Every run of silence inside the run begins where the buffer ran dry while playing,
            // and the buffer ran dry once more after the last: with the hearer in the mix from
            // before the buffer first played, those are all its underruns, each stated in turn.
            // A stop is the host's where the buffer held less than a frame and the sound that
            // ended it reached the mix after it (from the caller, the relay or the mix's own
            // task late, its tick then late too); as many ticks gave nothing as the hearer heard
            // silence, or more where the host dropped a packet of it. The last, after the call's
            // last sound, has no sound after it. A stop with a frame held, or whose sound had
            // reached the mix before it, is the mix's.
            let stops = counts["stops"].as_array().cloned().unwrap_or_default();
            let explained: Vec<String> = stops
                .iter()
                .enumerate()
                .filter_map(|(i, s)| {
                    let (held, refill, ticks) = (s["held"].as_u64().unwrap_or(u64::MAX), s["refill_us"].as_i64(), s["ticks"].as_u64().unwrap_or(0));
                    let why = if held >= 160 {
                        Some(format!("it held {held} samples"))
                    } else if i < h.silences.len() {
                        if refill.is_none_or(|r| r < 0) {
                            Some(format!("the sound after it reached the mix {refill:?} µs after it"))
                        } else if ticks < h.silences[i] as u64 {
                            Some(format!("{} packets of silence heard, {ticks} ticks stopped", h.silences[i]))
                        } else {
                            None
                        }
                    } else {
                        refill.map(|r| format!("the last, yet sound came {r} µs after it"))
                    };
                    why.map(|w| format!("stop {i} ({s}): {w}"))
                })
                .collect();
            let accounted = underruns == h.gaps as u64 + 1 && stops.len() as u64 == underruns && explained.is_empty();
            // Each packet heard within a second of its sending, less the time the playout buffer
            // stood stopped before it, as far as its delay still carries it: a stop plays the rest
            // of the call that much later (§10.2), and how long a stop lasts is the host's (the
            // sender's own lateness first). A stop counts for no more than the delay grew across
            // it (`common::telephony::heard`).
            let max = h.delays_ms.iter().cloned().fold(0.0, f64::max);
            let net = h.net_delays_ms.iter().cloned().fold(0.0, f64::max);
            if !ended || h.foreign > 0 || first + h.run != N || !accounted || overflowed != Some(0) || net >= 1_000.0 {
                panic!(
                    "{who}: from packet {first}, {} heard in sequence of {N}, {} other packets between them, {} runs of silence ({:?} packets) and {underruns} underruns counted, {} stops stated, those not the host's: {explained:?}; {overflowed:?} samples dropped; the end heard: {ended}; delay {max:.1} ms at most, {net:.1} ms less the buffer's stops (each packet's: {:?}); {}; {}: {}\nrecords' heard: {}",
                    h.run,
                    h.foreign,
                    h.gaps,
                    h.silences,
                    stops.len(),
                    h.net_delays_ms.iter().map(|d| d.round() as i64).collect::<Vec<_>>(),
                    late(sent.as_slice()),
                    ticks(bridge.stats()),
                    traces[i],
                    recs.iter().map(|r| r["detail"]["heard"].to_string()).collect::<Vec<_>>().join("\n")
                );
            }
            let refills: Vec<i64> = stops.iter().filter_map(|s| s["refill_us"].as_i64()).collect();
            let came = match (refills.iter().min(), refills.iter().max()) {
                (Some(a), Some(b)) => format!(", each a stop of a dry buffer whose sound reached the mix {:.1} to {:.1} ms after it", *a as f64 / 1000.0, *b as f64 / 1000.0),
                _ => String::new(),
            };
            eprintln!(
                "{who}: from packet {first}, {} heard, {} runs of silence between{came}; delay {:.1} ms mean, {max:.1} ms at most, {net:.1} ms at most less the buffer's stops; {}; {}",
                h.run,
                h.gaps,
                h.delays_ms.iter().sum::<f64>() / h.delays_ms.len() as f64,
                late(sent.as_slice()),
                ticks(bridge.stats())
            );
        }
        // A call the bridge ended itself (its media timeout, an ACK not seen, its relay lost)
        // answers the BYE 481: the record says which, and when.
        assert_eq!(byes, (Some(200), Some(200)), "each call ended by its BYE; the records' ends: {:?}", recs.iter().map(|r| (r["principal"].as_str().map(|p| &p[..p.len().min(6)]), r["ended"].as_str(), (r["end"].as_u64().unwrap_or(0).saturating_sub(r["start"].as_u64().unwrap_or(0))) / 1000)).collect::<Vec<_>>());
        settle(&wt, Duration::from_millis(500)).await;
        // A's track at the relay: every object verified against the bridge's receipts, and
        // its samples the characters A sent, character for character.
        let (frames, groups) = verified(&wt, &wt.tracks[0], &w.gateway_key, Codec::Flac);
        assert!(groups >= 10, "groups of 100 ms over three seconds: {groups}");
        let mut chars = Vec::new();
        for f in &frames {
            let d = flac::read(f).expect("a FLAC stream");
            assert_eq!((d.channels, d.bits, d.rate), (1, 16, 8_000));
            chars.extend(d.samples.iter().map(|&s| Law::Mu.encode(s as i16)));
        }
        let want: Vec<u8> = (0..N).flat_map(|k| payload(1, k)).collect();
        assert_eq!(chars.len(), want.len(), "every sample of A's call, none made");
        assert!(chars == want, "A's track is the characters A sent");
        // Each object states a capture instant within the call.
        let st = stamps.lock().unwrap().clone();
        assert!(!st.is_empty());
        for c in &st {
            let c = c.expect("a CaptureTimestamp");
            assert!(c + 2_000 >= start_us / 1000 && c <= now_us() / 1000, "a capture instant within the call: {c}");
        }
        // The catalog states the track's playout timing.
        settle(&cat, Duration::from_millis(200)).await;
        let last = cat.objects.lock().unwrap().iter().rfind(|(k, _)| k.0 == cat.tracks[0].media_alias).map(|(_, b)| serde_json::from_slice::<Value>(b).unwrap()).expect("a catalog");
        let t = last["tracks"].as_array().unwrap().iter().find(|t| t["packaging"] == "wai").unwrap().clone();
        assert_eq!((t["wai_capabilities"].clone(), t["wai_tempo"]["playout_delay_ms"].as_u64()), (serde_json::json!(["wai.audio.flac"]), Some(160)));
        // The records: one a call, its statement and origin claim, nothing lost, no caller.
        for r in &recs {
            let d = &r["detail"];
            assert_eq!(r["ended"], "bye");
            assert_eq!(d["rtp"]["lost"], 0);
            assert_eq!(d["rtp"]["refused"], serde_json::json!({}));
            assert_eq!(d["statement"]["kind"], "wai-telephony-statement");
            assert_eq!(d["statement"]["codec"], "PCMU/8000");
            assert!(d.get("caller").is_none(), "no record says who called");
            let c = &d["origin_claims"][0];
            assert_eq!((c["class"].as_str(), c["step"].as_str(), c["operation"].as_str(), c["record"].as_str()), (Some("captured"), Some("capture"), Some("telephony:received"), d["statement_digest"].as_str()));
        }
        let principals: Vec<&str> = recs.iter().map(|r| r["principal"].as_str().unwrap()).collect();
        assert!(principals.contains(&oa.as_str()) && principals.contains(&ob.as_str()));
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

/// What the test above reads of a hearing (`common::telephony::heard`), on one made up: a run
/// of silence counts as a stop for its length and no more than the delay grew across it,
/// silence a mix late to tick sends at once counts for none, and a fall in the delay plays the
/// stops' delay away first. Each stop counted 20 ms a packet whatever the delay did, the burst
/// below would leave packet 4 heard 290 ms before it was sent.
#[test]
fn a_hearing_is_read_by_sequence_and_each_stop_for_the_delay_it_added() {
    let t0 = Instant::now();
    let ms = |x: u64| t0 + Duration::from_millis(x);
    let sent: Vec<Instant> = [0, 20, 40, 60, 80, 300, 320, 340].into_iter().map(ms).collect();
    let silence = vec![0xffu8; 160];
    let mut r: Vec<(Instant, Vec<u8>)> = vec![(ms(200), payload(1, 0)), (ms(220), payload(1, 1))];
    // A stop of five ticks at the mix's pace: packet 2 heard 100 ms later than its own.
    r.extend((0..5).map(|i| (ms(240 + 20 * i), silence.clone())));
    r.push((ms(340), payload(1, 2)));
    // Two ticks of silence, the delay grown 150 ms across them: 40 of it a stop.
    r.extend([(ms(360), silence.clone()), (ms(380), silence.clone())]);
    r.push((ms(510), payload(1, 3)));
    // Thirty ticks of silence at once, the delay not grown: no stop.
    r.extend((0..30).map(|_| (ms(525), silence.clone())));
    r.push((ms(530), payload(1, 4)));
    // The sender stalled 200 ms and the buffer played on: the delay falls by that much.
    r.extend([(ms(550), payload(1, 5)), (ms(570), payload(1, 6)), (ms(590), payload(1, 7)), (ms(610), silence.clone())]);
    let h = heard(&r, 1, &sent);
    assert_eq!((h.first, h.run, h.breaks, h.foreign, h.gaps, h.silences.as_slice()), (Some(0), 8, 37, 0, 3, [5, 2, 30].as_slice()));
    let round = |d: &[f64]| d.iter().map(|x| x.round() as i64).collect::<Vec<_>>();
    assert_eq!(round(&h.delays_ms), [200, 200, 300, 450, 450, 250, 250, 250]);
    assert_eq!(round(&h.net_delays_ms), [200, 200, 200, 310, 310, 250, 250, 250]);
    // A packet heard out of its place breaks the run, and is not silence.
    let r = vec![(ms(200), payload(1, 0)), (ms(220), payload(1, 1)), (ms(240), payload(1, 3)), (ms(260), payload(1, 2))];
    let h = heard(&r, 1, &sent[..4]);
    assert_eq!((h.first, h.run, h.breaks, h.foreign, h.gaps, h.silences.len()), (Some(0), 3, 1, 1, 0, 0));
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn credentials_and_a_pin() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|_| {}).await;
        let kp = key(0x63);
        let peer = Peer::bind("127.0.0.1").await;
        let mut cfg = config(peer.addr, vec![NumberConfig { number: "+15550200".into(), grants: vec![], pin: true, max_calls: Some(4) }]);
        cfg.auth = Some(InboundAuth { realm: "bridge.test".into(), user: "trunk".into(), password_path: "/nonexistent".into(), algorithms: vec![Alg::Sha256, Alg::Md5], nonce_ttl_s: 300, password: Some("correct horse".into()) });
        cfg.pin = vec![PinConfig { pin_path: "/nonexistent".into(), grants: vec![World::token(&w.grant(&kp, 0))], pin: Some(PIN.into()) }];
        let bridge = Bridge::start(w.gateway.gateway.clone(), cfg).await.expect("the bridge starts");
        let sip_addr = bridge.udp_addr.unwrap();
        let mut c = Caller::new("127.0.0.1", "call-pin").await;
        let body = offer(c.port(), "0 101", "");
        // Challenged, then answered with SHA-256.
        let r = c.invite(&peer, sip_addr, "+15550200", &[], &body).await;
        assert_eq!(r.code(), Some(401));
        assert_eq!(r.rows("WWW-Authenticate").len(), 2);
        let ch = Challenge::read(r.rows("WWW-Authenticate")[0]).unwrap();
        assert_eq!(ch.alg, Alg::Sha256);
        let uri = format!("sip:+15550200@{sip_addr}");
        let good = ch.answer("trunk", "correct horse", "INVITE", &uri, "c0ffee", 1);
        // A response for a nonce the bridge never issued first.
        let other = Challenge { nonce: "00112233445566778899aabbccddeeff".into(), ..ch.clone() }.answer("trunk", "correct horse", "INVITE", &uri, "c0ffee", 1);
        c.cseq += 1;
        let r = c.invite(&peer, sip_addr, "+15550200", &[("Authorization", &other)], &body).await;
        assert_eq!((r.code(), r.get("Warning")), (Some(401), Some("399 127.0.0.1 \"digest_nonce_unknown\"")));
        c.cseq += 1;
        assert_eq!(c.invite(&peer, sip_addr, "+15550200", &[("Authorization", &good)], &body).await.code(), Some(200));
        assert!(until(STEP, || bridge.calls_open() == 1).await);
        // The same credentials replayed by another call (a new Call-ID and branch, the same
        // nonce, count, cnonce and response): its count is not above the one accepted (§5.3
        // rule 10), and no call opens.
        let mut replay = Caller::new("127.0.0.1", "call-replay").await;
        let r = replay.invite(&peer, sip_addr, "+15550200", &[("Authorization", &good)], &offer(replay.port(), "0 101", "")).await;
        assert_eq!((r.code(), r.get("Warning")), (Some(401), Some("399 127.0.0.1 \"digest_nc_replayed\"")));
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(bridge.calls_open(), 1, "the replayed credentials opened no call");
        // A wrong PIN, then the right one: the call joins under its grant.
        for _ in 0..PIN.len() {
            c.digit(9).await;
        }
        for d in PIN.bytes() {
            c.digit(d - b'0').await;
        }
        let owner = hex::encode(kp.verifying_key().to_bytes());
        let wt = watch_tracks(&w, vec![(media::AUDIO, catalog::media_namespace(&w.room, &owner, media::AUDIO), "0".into())]).await;
        let mut tick = tokio::time::interval(Duration::from_millis(20));
        for k in 0..40 {
            tick.tick().await;
            c.send(&payload(3, k)).await;
        }
        tokio::time::sleep(Duration::from_millis(400)).await;
        assert_eq!(c.in_dialog(&peer, sip_addr, "BYE", &[], "").await.code(), Some(200), "inside the dialog, not challenged");
        let rec = until_records(&w, 1).await.remove(0);
        settle(&wt, Duration::from_millis(400)).await;
        let (frames, _) = verified(&wt, &wt.tracks[0], &w.gateway_key, Codec::Flac);
        assert!(!frames.is_empty(), "the call published once its PIN was entered");
        assert_eq!(rec["principal"], owner);
        assert_eq!(rec["detail"]["pin"], serde_json::json!({"asked": true, "tries": 2, "wrong": 1}));
        // No record holds a PIN, in any form the search looks for; and the search finds one
        // planted in this record in each of those forms.
        assert_eq!(holding_pin(&rec, PIN), Vec::<String>::new(), "no record holds a PIN: {rec}");
        let digits: Vec<u8> = PIN.bytes().map(|b| b - b'0').collect();
        for planted in [serde_json::json!(PIN), serde_json::json!(PIN.parse::<u64>().unwrap()), serde_json::json!(digits), serde_json::json!(PIN.as_bytes())] {
            let mut r = rec.clone();
            r["detail"]["pin"]["planted"] = planted.clone();
            assert_eq!(holding_pin(&r, PIN), vec!["/detail/pin/planted".to_string()], "the search finds a PIN held as {planted}");
        }
        assert!(rec["detail"]["counts"]["digits"].as_u64().unwrap() >= 2 * PIN.len() as u64);
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn faults_on_the_wire_are_refused_each_with_its_reason() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|_| {}).await;
        let kf = key(0x64);
        let peer = Peer::bind("127.0.0.1").await;
        let cfg = config(peer.addr, vec![number("+15550300", vec![World::token(&w.grant(&kf, 0))])]);
        let bridge = Bridge::start(w.gateway.gateway.clone(), cfg).await.expect("the bridge starts");
        let sip_addr = bridge.udp_addr.unwrap();
        let warning = |m: &sip::Message| m.get("Warning").map(str::to_string).unwrap_or_default();
        // An offer with no codec the bridge both receives and sends.
        let mut x = Caller::new("127.0.0.1", "call-nocodec").await;
        let r = x.invite(&peer, sip_addr, "+15550300", &[], &format!("v=0\r\no=c 1 1 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\nm=audio {} RTP/AVP 111\r\na=rtpmap:111 opus/48000/2\r\n", x.port())).await;
        assert_eq!((r.code(), warning(&r)), (Some(488), "305 127.0.0.1 \"sdp_no_common_codec\"".to_string()));
        // A call, then a BYE for another dialog, and a re-INVITE that changes the codec; the
        // call goes on after each.
        let mut c = Caller::new("127.0.0.1", "call-faults").await;
        assert_eq!(c.invite(&peer, sip_addr, "+15550300", &[], &offer(c.port(), "0 101", "")).await.code(), Some(200));
        assert!(until(STEP, || bridge.calls_open() == 1).await);
        let bye = peer.request("BYE", &format!("sip:bridge@{sip_addr}"), "call-faults", &c.from_tag, &format!("<sip:x@{sip_addr}>;tag=ffffffffffffffff"), 9, &[], "");
        peer.send(sip_addr, &bye).await;
        let r = peer.final_to("call-faults", 9).await;
        assert_eq!((r.code(), warning(&r)), (Some(481), "399 127.0.0.1 \"dialog_unknown\"".to_string()));
        tokio::time::sleep(Duration::from_millis(200)).await;
        assert_eq!(bridge.calls_open(), 1, "a BYE for another dialog ends no call");
        let pcma = offer(c.port(), "8 101", "").replace("o=caller 1 1", "o=caller 1 2");
        let r = c.in_dialog(&peer, sip_addr, "INVITE", &[], &pcma).await;
        assert_eq!((r.code(), warning(&r)), (Some(488), "305 127.0.0.1 \"reinvite_codec_change\"".to_string()));
        // The call goes on in its law (§8.5): the bridge still sends it µ-law, here its silence.
        let n0 = c.received.lock().unwrap().len();
        assert!(until(STEP, || c.received.lock().unwrap().len() >= n0 + 10).await, "the bridge still sends the call");
        assert!(c.received.lock().unwrap()[n0..].iter().all(|(_, p)| p.len() == 160 && p.iter().all(|&x| x == Law::Mu.encode(0))), "every packet after the refused re-INVITE in µ-law");
        // RTP: good packets, a G.711 payload of the wrong length for its packet time, telephone
        // events split wrongly, and (where the host routes 127.0.0.2) a packet from an address
        // the description does not state, mid-call, then good packets again.
        for k in 0..10 {
            c.send(&payload(4, k)).await;
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        let short = joule_moq_ingest::telephony::rtp::write(false, 0, c.seq, c.ts, c.ssrc, &payload(4, 10)[..159]);
        c.rtp.send_to(&short, c.bridge_rtp).await.unwrap();
        c.seq = c.seq.wrapping_add(1);
        let ev = |seq: u16, ts: u32, m: bool, e: u8, d: u16| joule_moq_ingest::telephony::rtp::write(m, 101, seq, ts, c.ssrc, &[e, 10, (d >> 8) as u8, d as u8]);
        let t0 = c.ts;
        for p in [ev(c.seq, t0, true, 5, 0xffff), ev(c.seq.wrapping_add(1), t0.wrapping_add(0xffff + 160), false, 5, 160)] {
            c.rtp.send_to(&p, c.bridge_rtp).await.unwrap();
        }
        c.seq = c.seq.wrapping_add(2);
        let rogue = if cfg!(target_os = "linux") { tokio::net::UdpSocket::bind("127.0.0.2:0").await.ok() } else { None };
        if let Some(s) = &rogue {
            s.send_to(&joule_moq_ingest::telephony::rtp::write(false, 0, c.seq, c.ts, c.ssrc, &payload(4, 11)), c.bridge_rtp).await.unwrap();
            // A call from a peer the bridge is not configured with: an INVITE that would be
            // answered from the peer's address (a configured number, an offer that reads), its
            // Via naming the sender, so that any answer would come back to it. It is dropped,
            // unanswered, and counted (§8.1); no call opens.
            let me = s.local_addr().unwrap();
            let body = offer(me.port(), "0 101", "");
            let invite = format!("INVITE sip:+15550300@{sip_addr} SIP/2.0\r\nVia: SIP/2.0/UDP {me};branch=z9hG4bKrogue1\r\nMax-Forwards: 70\r\nFrom: <sip:pbx@{me}>;tag=rogue\r\nTo: <sip:+15550300@{sip_addr}>\r\nCall-ID: rogue-call\r\nCSeq: 1 INVITE\r\nContact: <sip:pbx@{me}>\r\nContent-Type: application/sdp\r\nContent-Length: {}\r\n\r\n{body}", body.len());
            s.send_to(invite.as_bytes(), sip_addr).await.unwrap();
        }
        tokio::time::sleep(Duration::from_millis(300)).await;
        for k in 12..20 {
            c.send(&payload(4, k)).await;
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        if let Some(s) = &rogue {
            let mut buf = [0u8; 2048];
            assert!(timeout(Duration::from_millis(700), s.recv_from(&mut buf)).await.is_err(), "nothing answers a peer the bridge is not configured with");
            assert_eq!(bridge.stats().foreign_dropped.load(std::sync::atomic::Ordering::Relaxed), 1, "the call from an unconfigured peer, dropped and counted");
            assert_eq!(bridge.calls_open(), 1, "it opened no call");
        }
        assert_eq!(c.in_dialog(&peer, sip_addr, "BYE", &[], "").await.code(), Some(200));
        let rec = until_records(&w, 1).await.remove(0);
        let refused = &rec["detail"]["rtp"]["refused"];
        let mut want = serde_json::json!({"g711_ptime_length": 1, "dtmf_segment_gap": 1});
        if rogue.is_some() {
            want["rtp_source_not_stated"] = 1.into();
        }
        assert_eq!(refused, &want, "each fault refused by its own reason, and nothing else refused");
        // The call went on, its law kept: the one number lost is the wrong-length packet's,
        // which is refused before the sequence counts it (§7.1); none after the faults.
        assert_eq!(rec["detail"]["rtp"]["lost"], 1);
        assert_eq!(rec["detail"]["statement"]["law"], "PCMU");
        tokio::time::sleep(Duration::from_millis(300)).await;
        assert_eq!(records(&w).len(), 1, "the call from an unconfigured peer left no record");
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn opus_received_is_carried_as_it_came() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|_| {}).await;
        let ko = key(0x66);
        let owner = hex::encode(ko.verifying_key().to_bytes());
        let peer = Peer::bind("127.0.0.1").await;
        let bridge = Bridge::start(w.gateway.gateway.clone(), config(peer.addr, vec![number("+15550500", vec![World::token(&w.grant(&ko, 0))])])).await.expect("the bridge starts");
        let sip_addr = bridge.udp_addr.unwrap();
        let mut c = Caller::new("127.0.0.1", "call-opus").await;
        // An offer that lists Opus first: the answer names the law first, which the bridge
        // sends, then Opus, which it receives (§6.2).
        let r = c.invite(&peer, sip_addr, "+15550500", &[], &offer(c.port(), "111 0 101", "a=rtpmap:111 opus/48000/2\r\n")).await;
        assert_eq!(r.code(), Some(200));
        let a = sdp::read(&r.body).expect("the answer");
        assert_eq!(a.media[0].fmts, vec!["0", "111", "101"]);
        assert!(until(STEP, || bridge.calls_open() == 1).await);
        tokio::time::sleep(Duration::from_millis(300)).await;
        let wt = watch_tracks(&w, vec![(media::AUDIO, catalog::media_namespace(&w.room, &owner, media::AUDIO), "0".into())]).await;
        let cat = watch_tracks(&w, vec![(0, catalog::catalog_namespace(&w.room, &owner), catalog::CATALOG_NAME.into())]).await;
        // Fifty Opus packets of 20 ms, each of its own length and bytes: the bridge decodes
        // none of them.
        let packets: Vec<Vec<u8>> = (0..50u32).map(|k| (0..20 + k % 40).map(|i| (k * 7 + i * 13) as u8).collect()).collect();
        let mut tick = tokio::time::interval(Duration::from_millis(20));
        for p in &packets {
            tick.tick().await;
            let pk = joule_moq_ingest::telephony::rtp::write(false, 111, c.seq, c.ts, c.ssrc, p);
            c.rtp.send_to(&pk, c.bridge_rtp).await.unwrap();
            c.seq = c.seq.wrapping_add(1);
            c.ts = c.ts.wrapping_add(960);
        }
        tokio::time::sleep(Duration::from_millis(500)).await;
        assert_eq!(c.in_dialog(&peer, sip_addr, "BYE", &[], "").await.code(), Some(200));
        let rec = until_records(&w, 1).await.remove(0);
        settle(&wt, Duration::from_millis(500)).await;
        // Each object one RTP payload, as it came, every one verified against its receipt.
        let (frames, _) = verified(&wt, &wt.tracks[0], &w.gateway_key, Codec::Opus);
        assert!(frames == packets, "the track is the Opus packets the caller sent, {} of {}", frames.len(), packets.len());
        settle(&cat, Duration::from_millis(200)).await;
        let last = cat.objects.lock().unwrap().iter().rfind(|(k, _)| k.0 == cat.tracks[0].media_alias).map(|(_, b)| serde_json::from_slice::<Value>(b).unwrap()).expect("a catalog");
        let t = last["tracks"].as_array().unwrap().iter().find(|t| t["packaging"] == "wai").unwrap().clone();
        assert_eq!(t["wai_capabilities"], serde_json::json!(["wai.audio.opus.packet"]));
        let d = &rec["detail"];
        assert_eq!((d["statement"]["codec"].as_str(), d["statement"]["law"].as_str(), d["statement"]["transport"].as_str()), (Some("opus/48000/2"), Some("PCMU"), Some("udp")));
        assert_eq!((d["rtp"]["refused"].clone(), d["rtp"]["lost"].as_u64()), (serde_json::json!({}), Some(0)));
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn sip_over_tcp_and_tls() {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    timeout(Duration::from_secs(60), async {
        let w = World::start(|_| {}).await;
        let id = joule_moq_relay::cert::RelayIdentity::dev_self_signed(&["localhost".to_string()]).unwrap();
        let chain: Vec<Vec<u8>> = id.chain.iter().map(|c| c.to_vec()).collect();
        let root = chain[0].clone();
        let (kt, ks) = (key(0x67), key(0x68));
        let mut cfg = config("127.0.0.1:5060".parse().unwrap(), vec![number("+15550600", vec![World::token(&w.grant(&kt, 0)), World::token(&w.grant(&ks, 0))])]);
        cfg.sip_udp = None;
        cfg.sip_tcp = Some("127.0.0.1:0".parse().unwrap());
        cfg.sip_tls = Some("127.0.0.1:0".parse().unwrap());
        cfg.tls_der = Some((chain, id.key.secret_der().to_vec()));
        let bridge = Bridge::start(w.gateway.gateway.clone(), cfg).await.expect("the bridge starts");
        let options = |port: u16, transport: &str| format!("OPTIONS sip:bridge@127.0.0.1 SIP/2.0\r\nVia: SIP/2.0/{transport} 127.0.0.1:{port};branch=z9hG4bKo{port}\r\nMax-Forwards: 70\r\nFrom: <sip:trunk@127.0.0.1>;tag=t\r\nTo: <sip:bridge@127.0.0.1>\r\nCall-ID: o{port}\r\nCSeq: 1 OPTIONS\r\nContent-Length: 0\r\n\r\n");
        /// The next message on a stream, `buf` holding what was read past the last.
        async fn answer<S: tokio::io::AsyncRead + Unpin>(s: &mut S, buf: &mut Vec<u8>) -> sip::Message {
            let mut chunk = [0u8; 4096];
            loop {
                if let Ok(Some((a, b))) = sip::frame(&buf[..]) {
                    let m = sip::read(&buf[a..b], true).expect("a response that reads");
                    buf.drain(..b);
                    return m;
                }
                let n = s.read(&mut chunk).await.expect("a read");
                assert!(n > 0, "the bridge closed the stream");
                buf.extend_from_slice(&chunk[..n]);
            }
        }
        /// A call over the stream `s`: INVITE, its final response, whose Contact must be
        /// `contact` (the listener of the stream's transport, §8.3), ACK, then BYE and its 200.
        async fn call_over<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin>(s: &mut S, buf: &mut Vec<u8>, port: u16, transport: &str, call_id: &str, contact: &str) {
            let rtp = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
            let body = offer(rtp.local_addr().unwrap().port(), "0 101", "");
            let head = |method: &str, cseq: u32, to: &str, branch: &str| format!("{method} sip:+15550600@127.0.0.1 SIP/2.0\r\nVia: SIP/2.0/{transport} 127.0.0.1:{port};branch=z9hG4bK{branch}\r\nMax-Forwards: 70\r\nFrom: <sip:trunk@127.0.0.1>;tag=ft-{call_id}\r\nTo: {to}\r\nCall-ID: {call_id}\r\nCSeq: {cseq} {method}\r\nContact: <sip:trunk@127.0.0.1:{port};transport={}>\r\n", transport.to_lowercase());
            s.write_all(format!("{}Content-Type: application/sdp\r\nContent-Length: {}\r\n\r\n{body}", head("INVITE", 1, "<sip:+15550600@127.0.0.1>", &format!("i{call_id}")), body.len()).as_bytes()).await.unwrap();
            let ok = loop {
                let r = answer(s, buf).await;
                if r.code().is_some_and(|c| c >= 200) {
                    break r;
                }
            };
            assert_eq!(ok.code(), Some(200), "the call over {transport} is answered");
            assert_eq!(ok.get("Contact"), Some(contact), "a dialog over {transport} is reached over {transport}");
            let to = ok.get("To").unwrap().to_string();
            s.write_all(format!("{}Content-Length: 0\r\n\r\n", head("ACK", 1, &to, &format!("a{call_id}"))).as_bytes()).await.unwrap();
            s.write_all(format!("{}Content-Length: 0\r\n\r\n", head("BYE", 2, &to, &format!("b{call_id}"))).as_bytes()).await.unwrap();
            let bye = answer(s, buf).await;
            assert_eq!((bye.code(), bye.cseq().map(|c| c.1.to_string())), (Some(200), Some("BYE".into())));
        }
        let mut tcp = tokio::net::TcpStream::connect(bridge.tcp_addr.unwrap()).await.unwrap();
        let port = tcp.local_addr().unwrap().port();
        // CRLFs before a start line are passed over (§4.1).
        tcp.write_all(format!("\r\n\r\n{}", options(port, "TCP")).as_bytes()).await.unwrap();
        let mut tbuf = Vec::new();
        let r = answer(&mut tcp, &mut tbuf).await;
        assert_eq!((r.code(), r.get("Allow")), (Some(200), Some("INVITE, ACK, BYE, CANCEL, OPTIONS")));
        call_over(&mut tcp, &mut tbuf, port, "TCP", "call-tcp", &format!("<sip:bridge@127.0.0.1:{};transport=tcp>", bridge.tcp_addr.unwrap().port())).await;
        let mut roots = rustls::RootCertStore::empty();
        roots.add(rustls::pki_types::CertificateDer::from(root)).unwrap();
        let tc = rustls::ClientConfig::builder_with_provider(Arc::new(rustls::crypto::ring::default_provider())).with_safe_default_protocol_versions().unwrap().with_root_certificates(roots).with_no_client_auth();
        let raw = tokio::net::TcpStream::connect(bridge.tls_addr.unwrap()).await.unwrap();
        let port = raw.local_addr().unwrap().port();
        let mut tls = tokio_rustls::TlsConnector::from(Arc::new(tc)).connect(rustls::pki_types::ServerName::try_from("localhost").unwrap(), raw).await.expect("the TLS handshake");
        tls.write_all(options(port, "TLS").as_bytes()).await.unwrap();
        let mut sbuf = Vec::new();
        let r = answer(&mut tls, &mut sbuf).await;
        assert_eq!(r.code(), Some(200));
        call_over(&mut tls, &mut sbuf, port, "TLS", "call-tls", &format!("<sips:bridge@127.0.0.1:{}>", bridge.tls_addr.unwrap().port())).await;
        // Each call's statement names the transport its dialog was made over (§11).
        let recs = until_records(&w, 2).await;
        let mut transports: Vec<(String, String)> = recs.iter().map(|r| (r["detail"]["statement"]["transport"].as_str().unwrap_or_default().to_string(), r["ended"].as_str().unwrap_or_default().to_string())).collect();
        transports.sort();
        assert_eq!(transports, vec![("tcp".to_string(), "bye".to_string()), ("tls".to_string(), "bye".to_string())]);
        bridge.shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn a_call_the_bridge_places_and_its_registration() {
    timeout(Duration::from_secs(120), async {
        let w = World::start(|_| {}).await;
        let ko = key(0x65);
        let peer = Peer::bind("127.0.0.1").await;
        let mut cfg = config(peer.addr, vec![]);
        cfg.outbound = Some(OutboundAuth { user: "bridge".into(), password_path: "/nonexistent".into(), register: true, register_expires_s: 60, aor: String::new(), password: Some("s3cret".into()) });
        let bridge = Arc::new(Bridge::start(w.gateway.gateway.clone(), cfg).await.expect("the bridge starts"));
        // The peer as a registrar: a challenge, then the REGISTER with credentials, granted.
        let policy = Policy { realm: "pbx.test".into(), accounts: vec![("bridge".into(), "s3cret".into())], algorithms: vec![Alg::Md5], nonce_ttl_s: 60 };
        let mut nonces = Nonces::new([3; 32]);
        let (reg, from) = peer.next(|m| m.method() == Some("REGISTER")).await;
        let n = nonces.issue(now_us() / 1_000_000);
        peer.send(from, &peer.respond(&reg, 401, "Unauthorized", Some("r1"), &[("WWW-Authenticate", digest::challenge("pbx.test", &n, Alg::Md5, false))], "")).await;
        let (reg2, from) = peer.next(|m| m.method() == Some("REGISTER") && m.get("Authorization").is_some()).await;
        let rows = reg2.rows("Authorization");
        assert_eq!(digest::check(&rows, "REGISTER", reg2.uri().unwrap(), &policy, &mut nonces, now_us() / 1_000_000), Ok("bridge".into()));
        peer.send(from, &peer.respond(&reg2, 200, "OK", Some("r1"), &[("Expires", "60".into())], "")).await;
        assert!(until(STEP, || bridge.stats().registrations.load(std::sync::atomic::Ordering::Relaxed) == 1).await);
        // A call the bridge places: challenged once by the PBX, then answered.
        let grant = World::token(&w.grant(&ko, 0));
        let b2 = bridge.clone();
        let placed = tokio::spawn(async move { b2.place_call("+15550400", &grant, 10_000).await });
        let (inv, from) = peer.next(|m| m.method() == Some("INVITE")).await;
        let o = sdp::read(&inv.body).expect("the bridge's offer");
        assert_eq!(o.media[0].fmts, vec!["0", "8", "101"]);
        let n = nonces.issue(now_us() / 1_000_000);
        peer.send(from, &peer.respond(&inv, 407, "Proxy Authentication Required", Some("p1"), &[("Proxy-Authenticate", digest::challenge("pbx.test", &n, Alg::Md5, false))], "")).await;
        let _ = peer.next(|m| m.method() == Some("ACK") && m.cseq().is_some_and(|c| c.0 == 1)).await;
        let (inv2, from) = peer.next(|m| m.method() == Some("INVITE") && m.get("Proxy-Authorization").is_some()).await;
        assert_eq!(digest::check(&inv2.rows("Proxy-Authorization"), "INVITE", inv2.uri().unwrap(), &policy, &mut nonces, now_us() / 1_000_000), Ok("bridge".into()));
        let callee = Caller::new("127.0.0.1", "callee").await;
        let answer = format!("v=0\r\no=pbx 1 1 IN IP4 127.0.0.1\r\ns=-\r\nc=IN IP4 127.0.0.1\r\nt=0 0\r\nm=audio {} RTP/AVP 8 101\r\na=rtpmap:101 telephone-event/8000\r\n", callee.port());
        peer.send(from, &peer.respond(&inv2, 200, "OK", Some("callee-tag"), &[("Contact", format!("<sip:pbx@{}>", peer.addr))], &answer)).await;
        let (ack, _) = peer.next(|m| m.method() == Some("ACK") && m.cseq().is_some_and(|c| c.0 == 2)).await;
        assert_eq!(ack.to_tag().as_deref(), Some("callee-tag"));
        let call_id = placed.await.unwrap().expect("the call is placed");
        // It sends the room's mix in the law the answer named (PCMA), silence here.
        assert!(until(STEP, || callee.received.lock().unwrap().len() >= 10).await, "the callee hears the bridge");
        assert!(callee.received.lock().unwrap().iter().all(|(_, p)| p.len() == 160 && p.iter().all(|&c| c == Law::A.encode(0))));
        // The PBX ends it.
        let bye = peer.request("BYE", &format!("sip:bridge@{}", bridge.udp_addr.unwrap()), &call_id, "callee-tag", &format!("{};tag={}", inv2.get("From").unwrap().split(";tag=").next().unwrap(), inv2.from_tag().unwrap()), 1, &[], "");
        peer.send(bridge.udp_addr.unwrap(), &bye).await;
        assert_eq!(peer.final_to(&call_id, 1).await.code(), Some(200));
        let rec = until_records(&w, 1).await.remove(0);
        assert_eq!((rec["ended"].as_str(), rec["detail"]["statement"]["direction"].as_str(), rec["detail"]["statement"]["law"].as_str()), (Some("bye"), Some("outbound"), Some("PCMA")));
        Arc::try_unwrap(bridge).ok().expect("one owner").shutdown().await;
        w.stop().await;
    })
    .await
    .expect("in time");
}