//! The telephony bridge's tests' own SIP peer and RTP callers (no outside softphone or PBX):
//! a UDP SIP endpoint at the address the bridge is configured with as its peer, requests
//! written here, the bridge's messages read with the crate's reader, and callers that send
//! and receive G.711 over RTP on loopback.

use joule_moq_ingest::telephony::rtp;
use joule_moq_ingest::telephony::sdp;
use joule_moq_ingest::telephony::sip::{self, Message};
use std::net::SocketAddr;
use std::sync::{Arc, Mutex};
use std::time::{Duration, Instant};
use tokio::net::UdpSocket;
use tokio::sync::mpsc;

pub const WAIT: Duration = Duration::from_secs(10);

/// Every packet a caller received, with its arrival: on Linux as the kernel stamped it on
/// reaching the caller's socket, elsewhere when the caller's task read it.
pub type Received = Arc<Mutex<Vec<(Instant, Vec<u8>)>>>;

/// A packet's arrival as the kernel stamped it (`SO_TIMESTAMPNS`, Linux): a test's own task,
/// late to read under load, would otherwise date every packet it finds waiting by its own wake.
#[cfg(target_os = "linux")]
mod arrival {
    use std::os::fd::AsRawFd;
    use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH};

    pub fn stamp(s: &tokio::net::UdpSocket) {
        let on: libc::c_int = 1;
        // SAFETY: a valid socket and an int option of its size.
        let r = unsafe { libc::setsockopt(s.as_raw_fd(), libc::SOL_SOCKET, libc::SO_TIMESTAMPNS, (&on as *const libc::c_int).cast(), std::mem::size_of::<libc::c_int>() as libc::socklen_t) };
        assert_eq!(r, 0, "SO_TIMESTAMPNS: {}", std::io::Error::last_os_error());
    }

    /// One datagram into `buf`, and when it arrived.
    pub async fn recv(s: &tokio::net::UdpSocket, buf: &mut [u8]) -> std::io::Result<(usize, Instant)> {
        let (n, at) = s
            .async_io(tokio::io::Interest::READABLE, || {
                let mut iov = libc::iovec { iov_base: buf.as_mut_ptr().cast(), iov_len: buf.len() };
                let mut control = [0u64; 8];
                // SAFETY: a zeroed msghdr is valid; its buffers outlive the call.
                let mut msg: libc::msghdr = unsafe { std::mem::zeroed() };
                msg.msg_iov = &mut iov;
                msg.msg_iovlen = 1;
                msg.msg_control = control.as_mut_ptr().cast();
                msg.msg_controllen = std::mem::size_of_val(&control) as _;
                // SAFETY: as above.
                let n = unsafe { libc::recvmsg(s.as_raw_fd(), &mut msg, libc::MSG_DONTWAIT) };
                if n < 0 {
                    return Err(std::io::Error::last_os_error());
                }
                let mut at = None;
                // SAFETY: the control messages the kernel wrote into `control`, walked by its macros.
                unsafe {
                    let mut c = libc::CMSG_FIRSTHDR(&msg);
                    while !c.is_null() {
                        if (*c).cmsg_level == libc::SOL_SOCKET && (*c).cmsg_type == libc::SCM_TIMESTAMPNS {
                            let ts = std::ptr::read_unaligned(libc::CMSG_DATA(c) as *const libc::timespec);
                            at = Some(UNIX_EPOCH + Duration::new(ts.tv_sec as u64, ts.tv_nsec as u32));
                        }
                        c = libc::CMSG_NXTHDR(&msg, c);
                    }
                }
                Ok((n as usize, at))
            })
            .await?;
        let now = Instant::now();
        let ago = at.and_then(|k| SystemTime::now().duration_since(k).ok()).unwrap_or_default();
        Ok((n, now.checked_sub(ago).unwrap_or(now)))
    }
}

/// The tests' SIP peer: one UDP socket, every message the bridge sends it kept in arrival
/// order until a test takes it.
pub struct Peer {
    pub sock: Arc<UdpSocket>,
    pub addr: SocketAddr,
    rx: tokio::sync::Mutex<mpsc::UnboundedReceiver<(Message, SocketAddr)>>,
    backlog: Mutex<Vec<(Message, SocketAddr)>>,
    n: std::sync::atomic::AtomicU64,
}

impl Peer {
    pub async fn bind(ip: &str) -> Peer {
        let sock = Arc::new(UdpSocket::bind(format!("{ip}:0")).await.expect("the peer binds"));
        let addr = sock.local_addr().unwrap();
        let (tx, rx) = mpsc::unbounded_channel();
        let s = sock.clone();
        tokio::spawn(async move {
            let mut buf = vec![0u8; 65_536];
            loop {
                let Ok((n, from)) = s.recv_from(&mut buf).await else { return };
                if let Ok(m) = sip::read(&buf[..n], false)
                    && tx.send((m, from)).is_err()
                {
                    return;
                }
            }
        });
        Peer { sock, addr, rx: tokio::sync::Mutex::new(rx), backlog: Mutex::new(Vec::new()), n: 0.into() }
    }

    pub fn fresh(&self, what: &str) -> String {
        format!("{what}{}", self.n.fetch_add(1, std::sync::atomic::Ordering::SeqCst))
    }

    pub async fn send(&self, to: SocketAddr, text: &str) {
        self.sock.send_to(text.as_bytes(), to).await.expect("a datagram");
    }

    /// The first message, kept or to come, that `f` takes.
    pub async fn next(&self, f: impl Fn(&Message) -> bool) -> (Message, SocketAddr) {
        if let Some(i) = self.backlog.lock().unwrap().iter().position(|(m, _)| f(m)) {
            return self.backlog.lock().unwrap().remove(i);
        }
        let deadline = tokio::time::Instant::now() + WAIT;
        let mut rx = self.rx.lock().await;
        loop {
            let got = tokio::time::timeout_at(deadline, rx.recv()).await.expect("a message from the bridge in time").expect("the peer's socket");
            if f(&got.0) {
                return got;
            }
            self.backlog.lock().unwrap().push(got);
        }
    }

    /// The final response to the request of `call_id` and CSeq `cseq`.
    pub async fn final_to(&self, call_id: &str, cseq: u32) -> Message {
        let id = call_id.to_string();
        self.next(move |m| m.code().is_some_and(|c| c >= 200) && m.call_id() == id && m.cseq().is_some_and(|c| c.0 == cseq)).await.0
    }

    /// A request from the peer.
    #[allow(clippy::too_many_arguments)]
    pub fn request(&self, method: &str, uri: &str, call_id: &str, from_tag: &str, to: &str, cseq: u32, extra: &[(&str, &str)], body: &str) -> String {
        let mut s = format!(
            "{method} {uri} SIP/2.0\r\nVia: SIP/2.0/UDP {};branch=z9hG4bK{}\r\nMax-Forwards: 70\r\nFrom: <sip:trunk@{}>;tag={from_tag}\r\nTo: {to}\r\nCall-ID: {call_id}\r\nCSeq: {cseq} {method}\r\nContact: <sip:trunk@{}>\r\n",
            self.addr,
            self.fresh("b"),
            self.addr,
            self.addr
        );
        for (n, v) in extra {
            s.push_str(&format!("{n}: {v}\r\n"));
        }
        if !body.is_empty() {
            s.push_str("Content-Type: application/sdp\r\n");
        }
        s.push_str(&format!("Content-Length: {}\r\n\r\n{body}", body.len()));
        s
    }

    /// The response to a request the bridge sent.
    pub fn respond(&self, req: &Message, code: u16, phrase: &str, to_tag: Option<&str>, extra: &[(&str, String)], body: &str) -> String {
        let mut s = format!("SIP/2.0 {code} {phrase}\r\n");
        for v in req.rows("Via") {
            s.push_str(&format!("Via: {v}\r\n"));
        }
        s.push_str(&format!("From: {}\r\n", req.get("From").unwrap_or_default()));
        let to = req.get("To").unwrap_or_default();
        match to_tag {
            Some(t) if req.to_tag().is_none() => s.push_str(&format!("To: {to};tag={t}\r\n")),
            _ => s.push_str(&format!("To: {to}\r\n")),
        }
        s.push_str(&format!("Call-ID: {}\r\nCSeq: {}\r\n", req.call_id(), req.get("CSeq").unwrap_or_default()));
        for (n, v) in extra {
            s.push_str(&format!("{n}: {v}\r\n"));
        }
        if !body.is_empty() {
            s.push_str("Content-Type: application/sdp\r\n");
        }
        s.push_str(&format!("Content-Length: {}\r\n\r\n{body}", body.len()));
        s
    }
}

/// An offer of PCMU (and the formats `fmts` name) from a caller receiving at `port` on loopback.
pub fn offer(port: u16, fmts: &str, attrs: &str) -> String {
    format!("v=0\r\no=caller 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 {port} RTP/AVP {fmts}\r\na=rtpmap:101 telephone-event/8000\r\na=fmtp:101 0-15\r\na=ptime:20\r\n{attrs}")
}

/// One caller's media: its RTP socket, the bridge's media address once answered, what it
/// sends and every packet it receives with its arrival.
pub struct Caller {
    pub call_id: String,
    pub from_tag: String,
    pub to_tag: String,
    pub cseq: u32,
    pub rtp: Arc<UdpSocket>,
    pub bridge_rtp: SocketAddr,
    pub seq: u16,
    pub ts: u32,
    pub ssrc: u32,
    pub received: Received,
}

impl Caller {
    pub async fn new(ip: &str, call_id: &str) -> Caller {
        let rtp = Arc::new(UdpSocket::bind(format!("{ip}:0")).await.unwrap());
        #[cfg(target_os = "linux")]
        arrival::stamp(&rtp);
        let received: Received = Arc::default();
        let (r, s) = (received.clone(), rtp.clone());
        tokio::spawn(async move {
            let mut buf = vec![0u8; 2048];
            loop {
                #[cfg(target_os = "linux")]
                let got = arrival::recv(&s, &mut buf).await;
                #[cfg(not(target_os = "linux"))]
                let got = s.recv_from(&mut buf).await.map(|(n, _)| (n, Instant::now()));
                let Ok((n, at)) = got else { return };
                if let Ok(p) = rtp::header(&buf[..n]) {
                    r.lock().unwrap().push((at, p.payload.to_vec()));
                }
            }
        });
        Caller { call_id: call_id.into(), from_tag: format!("{call_id}-ft"), to_tag: String::new(), cseq: 1, rtp, bridge_rtp: "0.0.0.0:0".parse().unwrap(), seq: 1000, ts: 160_000, ssrc: 0x1234_0000 | (call_id.len() as u32), received }
    }

    pub fn port(&self) -> u16 {
        self.rtp.local_addr().unwrap().port()
    }

    /// INVITE the bridge at `number` with `body`; the final response, and on a 2xx the ACK sent.
    pub async fn invite(&mut self, peer: &Peer, bridge: SocketAddr, number: &str, extra: &[(&str, &str)], body: &str) -> Message {
        let uri = format!("sip:{number}@{bridge}");
        let req = peer.request("INVITE", &uri, &self.call_id, &self.from_tag, &format!("<{uri}>"), self.cseq, extra, body);
        peer.send(bridge, &req).await;
        let r = peer.final_to(&self.call_id, self.cseq).await;
        if r.code().is_some_and(|c| (200..300).contains(&c)) {
            self.to_tag = r.to_tag().unwrap_or_default();
            if let Ok(a) = sdp::read(&r.body)
                && let Some(m) = a.media.iter().find(|m| m.kind == "audio")
            {
                self.bridge_rtp = SocketAddr::new(m.addr.or(a.addr).unwrap(), m.port);
            }
            let ack = peer.request("ACK", &uri, &self.call_id, &self.from_tag, &format!("<{uri}>;tag={}", self.to_tag), self.cseq, &[], "");
            peer.send(bridge, &ack).await;
        } else {
            // A non-2xx final response is acknowledged too (RFC 3261 §17.1.1.3).
            let ack = peer.request("ACK", &uri, &self.call_id, &self.from_tag, r.get("To").unwrap_or_default(), self.cseq, &[], "");
            peer.send(bridge, &ack).await;
        }
        r
    }

    /// A request in the call's dialog.
    pub async fn in_dialog(&mut self, peer: &Peer, bridge: SocketAddr, method: &str, extra: &[(&str, &str)], body: &str) -> Message {
        self.cseq += 1;
        let uri = format!("sip:bridge@{bridge}");
        let req = peer.request(method, &uri, &self.call_id, &self.from_tag, &format!("<sip:x@{bridge}>;tag={}", self.to_tag), self.cseq, extra, body);
        peer.send(bridge, &req).await;
        peer.final_to(&self.call_id, self.cseq).await
    }

    /// One 20 ms PCMU packet of `payload` to the bridge.
    pub async fn send(&mut self, payload: &[u8]) {
        let p = rtp::write(false, 0, self.seq, self.ts, self.ssrc, payload);
        self.rtp.send_to(&p, self.bridge_rtp).await.unwrap();
        self.seq = self.seq.wrapping_add(1);
        self.ts = self.ts.wrapping_add(payload.len() as u32);
    }

    /// A telephone event (RFC 4733): its reports, then its end three times.
    pub async fn digit(&mut self, event: u8) {
        let ts = self.ts;
        for (j, d) in [160u16, 320, 480].iter().enumerate() {
            let p = rtp::write(j == 0, 101, self.seq, ts, self.ssrc, &[event, 10, (d >> 8) as u8, *d as u8]);
            self.rtp.send_to(&p, self.bridge_rtp).await.unwrap();
            self.seq = self.seq.wrapping_add(1);
            tokio::time::sleep(Duration::from_millis(20)).await;
        }
        for _ in 0..3 {
            let p = rtp::write(false, 101, self.seq, ts, self.ssrc, &[event, 0x80 | 10, 0x02, 0x80]);
            self.rtp.send_to(&p, self.bridge_rtp).await.unwrap();
            self.seq = self.seq.wrapping_add(1);
        }
        self.ts = self.ts.wrapping_add(640);
    }
}

/// A caller's `k`th payload: 160 µ-law characters naming `k` in their first three (each from
/// 0x80 to 0xEF, so never µ-law's negative zero), then a pattern of `salt`.
pub fn payload(salt: u8, k: u32) -> Vec<u8> {
    let mut p = vec![0x80 + (k / (112 * 112) % 112) as u8, 0x80 + (k / 112 % 112) as u8, 0x80 + (k % 112) as u8];
    p.extend((3..160u32).map(|i| 0x80 + ((k * 31 + i * 7 + salt as u32 * 13) % 0x70) as u8));
    p
}

/// The `k` a payload names, where it is one of `salt`'s.
pub fn which(salt: u8, p: &[u8]) -> Option<u32> {
    if p.len() != 160 || p[..3].iter().any(|&c| !(0x80..0xf0).contains(&c)) {
        return None;
    }
    let k = (p[0] - 0x80) as u32 * 112 * 112 + (p[1] - 0x80) as u32 * 112 + (p[2] - 0x80) as u32;
    (payload(salt, k) == p).then_some(k)
}

/// What a caller heard of another's payloads, in order: the first `k` heard, how many
/// consecutive ones followed it, the packets that broke the run (anything before the last
/// that is not the next payload, silence included), of those the ones that are not silence,
/// the runs of silence between two consecutive payloads, and the delay of each heard one from
/// when it was sent; and that delay less the time the playout buffer stood stopped before it,
/// as far as the delay still carries it (§10.2: a buffer that stops plays its later sound that
/// much later, and no sample is dropped to catch up). A run of silence counts for no more than
/// its length, 20 ms a packet, and no more than the delay grew across it: a mix late to tick
/// sends the ticks it missed at once (§10.3), and silence sent so stands for no stop. A fall in
/// the delay plays that much of what the stops added away, before any other delay.
///
/// Read by sequence, a run with no `foreign` packet and `first + run` the count sent is every
/// payload from the first, in order, none missing, none twice and nothing else between them
/// but silence (`gaps` runs of it, each where the playout buffer stopped, their lengths in
/// packets in order in `silences`). Read in real time (`tests/telephony_rt.rs`), `breaks` is
/// zero: no gap at all.
pub struct Heard {
    pub first: Option<u32>,
    pub run: u32,
    pub breaks: u32,
    pub foreign: u32,
    pub gaps: u32,
    pub silences: Vec<u32>,
    pub delays_ms: Vec<f64>,
    pub net_delays_ms: Vec<f64>,
}

/// µ-law's silence, the bridge's frame when nothing plays (§10.3).
pub fn silent(p: &[u8]) -> bool {
    p.len() == 160 && p.iter().all(|&c| c == 0xff)
}

/// What each received packet was, in arrival order: `k` for a payload of `salt`, `s` for
/// silence, `o` for anything else; for a failure's message.
pub fn trace(received: &[(Instant, Vec<u8>)], salt: u8) -> String {
    received.iter().map(|(_, p)| match which(salt, p) {
        Some(k) => format!("{k} "),
        None if silent(p) => "s ".into(),
        None => "o ".into(),
    }).collect()
}

/// A PIN for the tests that enter one, ten digits long. A call's record holds some 670 hex
/// digits and 60 decimal ones that change from run to run (digests, keys, signatures, times,
/// counts), and a given four digits turn up among them by chance in about one record in a
/// hundred: "4321", the PIN these tests entered before, failed "no record holds a PIN" so, with
/// no PIN held. A given ten turn up in about one record in 10^9, so a record that holds these
/// holds the PIN.
pub const PIN: &str = "8406517293";

/// Where `record` holds `pin`, each place by its JSON pointer: a key or a string with the
/// PIN's characters in it, a number whose decimal form has them, and an array of numbers with
/// a run of them as digit values, as RFC 4733 event codes (`*` 10, `#` 11, `A` to `D` 12 to
/// 15) or as ASCII bytes. Empty when the record holds the PIN in none of these forms.
pub fn holding_pin(record: &serde_json::Value, pin: &str) -> Vec<String> {
    use serde_json::Value;
    fn code(c: char) -> u64 {
        match c {
            '0'..='9' => c as u64 - '0' as u64,
            '*' => 10,
            '#' => 11,
            'A'..='D' => c as u64 - 'A' as u64 + 12,
            _ => u64::MAX,
        }
    }
    fn walk(v: &Value, at: &str, pin: &str, runs: &[Vec<u64>], out: &mut Vec<String>) {
        match v {
            Value::String(s) if s.contains(pin) => out.push(at.into()),
            Value::Number(n) if n.to_string().contains(pin) => out.push(at.into()),
            Value::Array(a) => {
                let ns: Vec<Option<u64>> = a.iter().map(Value::as_u64).collect();
                if runs.iter().any(|r| ns.windows(r.len()).any(|w| w.iter().zip(r).all(|(x, y)| *x == Some(*y)))) {
                    out.push(at.into());
                }
                for (i, x) in a.iter().enumerate() {
                    walk(x, &format!("{at}/{i}"), pin, runs, out);
                }
            }
            Value::Object(m) => {
                for (k, x) in m {
                    let here = format!("{at}/{}", k.replace('~', "~0").replace('/', "~1"));
                    if k.contains(pin) {
                        out.push(here.clone());
                    }
                    walk(x, &here, pin, runs, out);
                }
            }
            _ => {}
        }
    }
    let runs = [pin.chars().map(code).collect::<Vec<_>>(), pin.bytes().map(u64::from).collect()];
    let mut out = Vec::new();
    walk(record, "", pin, &runs, &mut out);
    out
}

pub fn heard(received: &[(Instant, Vec<u8>)], salt: u8, sent: &[Instant]) -> Heard {
    let mut h = Heard { first: None, run: 0, breaks: 0, foreign: 0, gaps: 0, silences: Vec::new(), delays_ms: Vec::new(), net_delays_ms: Vec::new() };
    let mut expect: Option<u32> = None;
    // Packets of silence since the last payload, and the delay the stops still add (ms).
    let mut silence = 0u32;
    let mut stopped = 0.0f64;
    let last = sent.len() as u32;
    let delay = |h: &mut Heard, k: u32, at: &Instant, silence: u32, stopped: &mut f64| {
        let d = at.duration_since(sent[k as usize]).as_secs_f64() * 1000.0;
        if let Some(&before) = h.delays_ms.last() {
            let grew = d - before;
            if grew < 0.0 {
                *stopped = (*stopped + grew).max(0.0);
            } else if silence > 0 {
                *stopped += grew.min(20.0 * silence as f64);
            }
        }
        h.delays_ms.push(d);
        h.net_delays_ms.push(d - *stopped);
    };
    for (at, p) in received {
        match (which(salt, p), expect) {
            (Some(k), None) => {
                h.first = Some(k);
                h.run = 1;
                expect = Some(k + 1);
                delay(&mut h, k, at, 0, &mut stopped);
            }
            (Some(k), Some(e)) if k == e => {
                h.run += 1;
                expect = Some(k + 1);
                if silence > 0 {
                    h.gaps += 1;
                    h.silences.push(silence);
                }
                delay(&mut h, k, at, silence, &mut stopped);
                silence = 0;
            }
            (_, Some(e)) if e < last => {
                h.breaks += 1;
                if silent(p) {
                    silence += 1;
                } else {
                    h.foreign += 1;
                }
            }
            _ => {}
        }
    }
    h
}

/// Whether a caller has heard the `last` payload of `salt` and a packet after it: the run is
/// over, and the buffer that played it has run dry (§10.2).
pub fn heard_to_the_end(received: &Received, salt: u8, last: u32) -> bool {
    let r = received.lock().unwrap();
    r.iter().rposition(|(_, p)| which(salt, p) == Some(last)).is_some_and(|i| i + 1 < r.len())
}