//! SLM (SLAY Media) protocol sender.
//!
//! Implements the wire protocol used by the OBS plugin receiver:
//!   - Multicast SLM_PKT_ANNOUNCE at ~1 Hz on 239.255.77.43:9878
//!   - Unicast SLM_PKT_CAPS to any receiver that sends SLM_PKT_HELLO
//!   - SLM_PKT_VIDEO chunks + SLM_PKT_AUDIO to all registered receivers
//!   - POSIX SHM for same-machine delivery (Unix only)
//!
//! # Usage
//!
//! ```rust,no_run
//! let sender = slay_slm::SlmSender::new(slay_slm::SlmConfig {
//!     name:       "My Source".into(),
//!     width:      1920,
//!     height:     1080,
//!     fps_num:    30,
//!     fps_den:    1,
//!     compress:   true,
//!     audio_only: false,
//! }).unwrap();
//!
//! // In your frame loop (synchronous; wrap in block_in_place from async):
//! sender.send_video(timestamp_ns, &nv12_bytes);
//! sender.send_audio(timestamp_ns, &f32le_samples);
//! ```

mod proto;
mod shm;

use proto::*;
use shm::SlmShm;

use socket2::SockRef;
use std::collections::HashMap;
use std::net::{IpAddr, Ipv4Addr, SocketAddr, UdpSocket};
use std::sync::{Arc, Mutex};
use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
use std::time::{Duration, Instant};

pub struct SlmConfig {
    pub name:       String,
    pub width:      u32,
    pub height:     u32,
    pub fps_num:    u32,
    pub fps_den:    u32,
    pub compress:   bool,
    pub audio_only: bool,
    /// Video codec advertised in SLM_PKT_CAPS. Use `CODEC_RAW` (0), `CODEC_H264` (1),
    /// or `CODEC_H265` (2) from the `proto` constants re-exported below.
    pub codec:      u8,
    /// FEC group size for UDP video. 1 = disabled; 2/4/8 = 1/N overhead.
    pub fec_n:      u8,
}

pub use proto::{CODEC_RAW, CODEC_H264, CODEC_H265, FRAME_TYPE_P, FRAME_TYPE_IDR};
pub use proto::{build_tally, build_msg, TALLY_OFF, TALLY_PREVIEW, TALLY_PROGRAM, MSG_MAX_BYTES};

#[derive(Debug)]
pub struct SlmError(String);

impl std::fmt::Display for SlmError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { self.0.fmt(f) }
}

impl std::error::Error for SlmError {}

/// Cap on simultaneous UDP receivers. Bounds the map against a spoofed-HELLO
/// flood (finding #8c); 16 is far more than any real studio needs.
const MAX_RECEIVERS: usize = 16;

#[derive(Clone)]
struct Receiver {
    addr:       SocketAddr,
    last_hello: Instant,
    /// EMA of measured one-way latency in microseconds (0 = no STAT yet).
    /// For UDP receivers this is RTT/2; for SHM receivers it is the full elapsed.
    latency_us: u32,
    /// v4 address validation: media is streamed only after the receiver echoes
    /// `nonce` (sent in a PKT_CHALLENGE) back in its HELLO. Persists for the
    /// life of the entry, so clip-start/seek/loop never re-validate.
    validated:  bool,
    /// Challenge nonce issued to this address (0 once consumed is not used; a
    /// fresh nonce is generated per new address).
    nonce:      u64,
}

/// Tracking entry for a same-machine SHM receiver.
struct ShmRx {
    last_hello: Instant,
    latency_us: u32,
}

/// Per-receiver detail snapshot for the UI's receiver list. One entry per
/// connected receiver across both transports.
#[derive(Clone)]
pub struct ReceiverInfo {
    /// True for a same-machine SHM receiver; false for a UDP receiver.
    pub shm:        bool,
    /// Receiver IP for UDP receivers; `None` for SHM (same machine).
    pub addr:       Option<String>,
    /// Measured latency in ms, or `None` until the receiver has reported.
    pub latency_ms: Option<u32>,
}

/// Exponential moving average with alpha = 1/8. Seeds with the first sample.
fn ema(old: u32, sample: u32) -> u32 {
    if old == 0 { sample } else { ((old as u64 * 7 + sample as u64) / 8) as u32 }
}

fn unix_now_ns() -> u64 {
    use std::time::{SystemTime, UNIX_EPOCH};
    SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_nanos() as u64
}

/// Round microseconds to whole milliseconds for display. 0 us (no measurement)
/// maps to `None`; any real sub-millisecond measurement floors at 1 ms so the
/// pill never shows "0ms" while a transport is actively reporting.
fn lat_us_to_ms(us: u32) -> Option<u32> {
    if us == 0 { return None; }
    Some(((us + 500) / 1000).max(1))
}

struct SlmInner {
    data_sock:          UdpSocket,
    receivers:          Mutex<HashMap<SocketAddr, Receiver>>,
    stream_id:          u32,
    /// Pre-built CAPS packet, resent immediately after BYE to avoid a
    /// caps_valid=false gap at the receiver while it waits for the next
    /// keepalive (~250ms).
    caps_pkt:           Vec<u8>,
    compress:           bool,
    fec_n:              u8,
    frame_seq:          AtomicU32,
    audio_seq:          AtomicU32,
    receiver_count:     AtomicU32,
    shm_available:      AtomicBool,
    shm_receiver_count: AtomicU32,
    /// Worst-case measured latency across receivers, in microseconds
    /// (0 = unknown). UDP is RTT/2; SHM is one-way. Fed by PKT_STAT echoes.
    udp_latency_us:     AtomicU32,
    shm_latency_us:     AtomicU32,
    /// Per-receiver detail rebuilt each control-loop iteration for the UI.
    receivers_snapshot: Mutex<Vec<ReceiverInfo>>,
    shm:                Mutex<Option<SlmShm>>,
    running:            AtomicBool,
    /// Operator messages (PKT_MSG) received from receivers, awaiting drain by the
    /// host via `take_messages`. Bounded in the control loop to resist flooding.
    messages:           Mutex<Vec<String>>,
}

/// Cloneable handle to an SLM sender. All methods are synchronous.
///
/// When calling from an async tokio task, wrap `send_video` / `send_audio`
/// in `tokio::task::block_in_place` as these perform blocking UDP sends.
#[derive(Clone)]
pub struct SlmSender(Arc<SlmInner>);

/// Live, routable IPv4 addresses of the host (excludes loopback and
/// link-local 169.254/16). Re-read each announce cycle so a USB NIC or
/// Wi-Fi link coming up or down is picked up without a restart.
fn local_ipv4_ifaces() -> Vec<Ipv4Addr> {
    let Ok(ifaces) = if_addrs::get_if_addrs() else { return Vec::new(); };
    ifaces.into_iter().filter_map(|i| match i.ip() {
        IpAddr::V4(v4) if !v4.is_loopback() && !v4.is_link_local() => Some(v4),
        _ => None,
    }).collect()
}

/// Send the announce out every live IPv4 interface by setting IP_MULTICAST_IF
/// per send. A single send_to leaves egress to the OS, which on a multi-homed
/// host (wired + Wi-Fi on different segments) or a Windows box that defaults
/// multicast to the wrong adapter can miss the segment a receiver is on.
/// Falls back to one OS-routed send if no interface can be enumerated.
fn send_announce_all_ifaces(sock: &UdpSocket, announce: &[u8], mcast_addr: SocketAddr) {
    let ifaces = local_ipv4_ifaces();
    if ifaces.is_empty() {
        let _ = sock.send_to(announce, mcast_addr);
        return;
    }
    let sref = SockRef::from(sock);
    for ip in ifaces {
        if sref.set_multicast_if_v4(&ip).is_ok() {
            let _ = sock.send_to(announce, mcast_addr);
        }
    }
}

impl SlmSender {
    /// Create a new sender and start the background control thread
    /// (multicast announce + HELLO/CAPS handshake loop).
    pub fn new(config: SlmConfig) -> Result<Self, SlmError> {
        let stream_id = rand_stream_id();

        // SHM is NV12-only: skip it for H264/H265 so OBS can't attach to a segment
        // the sender never writes NV12 to and then fail (see 945605c)
        let shm_opt = if config.codec == CODEC_RAW {
            SlmShm::create(
                &config.name,
                config.width, config.height,
                config.fps_num, config.fps_den,
                FMT_NV12,
                if config.audio_only { SOURCE_AUDIO_ONLY } else { 0 },
                SAMPLE_RATE, CHANNELS,
            )
        } else {
            None
        };
        let shm_available = shm_opt.is_some();
        if shm_available {
            log::info!("[SLM] SHM transport ready for '{}'", config.name);
        }

        let data_sock = UdpSocket::bind(("0.0.0.0", 0u16))
            .map_err(|e| SlmError(format!("bind data socket: {e}")))?;
        data_sock.set_nonblocking(false).ok();
        let data_port = data_sock.local_addr()
            .map_err(|e| SlmError(format!("data local_addr: {e}")))?.port();

        let ctrl_sock = UdpSocket::bind(("0.0.0.0", 0u16))
            .map_err(|e| SlmError(format!("bind ctrl socket: {e}")))?;
        let ctrl_port = ctrl_sock.local_addr()
            .map_err(|e| SlmError(format!("ctrl local_addr: {e}")))?.port();
        ctrl_sock.set_read_timeout(Some(Duration::from_millis(250))).ok();

        let caps_pkt = build_caps(
            stream_id,
            config.width as u16, config.height as u16,
            config.fps_num, config.fps_den,
            config.codec,
        );

        let inner = Arc::new(SlmInner {
            data_sock,
            receivers:          Mutex::new(HashMap::new()),
            stream_id,
            caps_pkt:           caps_pkt.clone(),
            compress:           config.compress,
            fec_n:              config.fec_n,
            frame_seq:          AtomicU32::new(0),
            audio_seq:          AtomicU32::new(0),
            receiver_count:     AtomicU32::new(0),
            shm_available:      AtomicBool::new(shm_available),
            shm_receiver_count: AtomicU32::new(0),
            udp_latency_us:     AtomicU32::new(0),
            shm_latency_us:     AtomicU32::new(0),
            receivers_snapshot: Mutex::new(Vec::new()),
            shm:                Mutex::new(shm_opt),
            running:            AtomicBool::new(true),
            messages:           Mutex::new(Vec::new()),
        });

        // Spawn the control thread: send multicast announces, receive HELLOs,
        // reply with CAPS, track SHM receivers, expire stale receivers.
        {
            let inner2   = Arc::clone(&inner);
            let announce  = build_announce(stream_id, &config.name, ctrl_port, config.audio_only);
            let mcast_addr = SocketAddr::new(
                IpAddr::V4(Ipv4Addr::new(MCAST_ADDR[0], MCAST_ADDR[1], MCAST_ADDR[2], MCAST_ADDR[3])),
                MCAST_PORT,
            );

            std::thread::spawn(move || {
                // Must hold a full PKT_MSG: 12-byte header + 1 len + up to
                // MSG_MAX_BYTES (255) of UTF-8. A smaller buffer truncates the
                // datagram at recv_from and corrupts long operator messages.
                let mut buf    = [0u8; 512];
                let mut shm_rx: HashMap<SocketAddr, ShmRx> = HashMap::new();
                loop {
                    send_announce_all_ifaces(&ctrl_sock, &announce, mcast_addr);
                    if !inner2.running.load(Ordering::Relaxed) { break; }

                    {
                        let mut map = inner2.receivers.lock().unwrap();
                        map.retain(|_, r| r.last_hello.elapsed() < Duration::from_secs(10));
                    }
                    shm_rx.retain(|_, r| r.last_hello.elapsed() < Duration::from_secs(10));

                    while let Ok((n, addr)) = ctrl_sock.recv_from(&mut buf) {
                        if n < 12 { continue; }
                        if &buf[..4] != &MAGIC { continue; }
                        let pkt_type = buf[4];
                        if pkt_type == PKT_HELLO || pkt_type == PKT_KEEPALIVE {
                            let flags = u16::from_le_bytes(buf[6..8].try_into().unwrap_or([0; 2]));
                            if (flags & HELLO_FLAG_SHM) != 0 {
                                // A receiver speaks one transport at a time over a
                                // single socket; drop any stale UDP registration so a
                                // live UDP->SHM switch reclassifies within one cycle.
                                inner2.receivers.lock().unwrap().remove(&addr);
                                let is_new = !shm_rx.contains_key(&addr);
                                shm_rx.entry(addr)
                                    .and_modify(|r| r.last_hello = Instant::now())
                                    .or_insert_with(|| ShmRx { last_hello: Instant::now(), latency_us: 0 });
                                if is_new { log::info!("[SLM] SHM receiver: {addr}"); }
                            } else {
                                shm_rx.remove(&addr);
                                // v4: echoed validation nonce sits after the 12-byte
                                // header; 0/absent means not echoed yet.
                                let echoed = if n >= 20 {
                                    u64::from_le_bytes(buf[12..20].try_into().unwrap_or([0; 8]))
                                } else { 0 };

                                let mut challenge: Option<u64> = None;
                                let mut now_validated = false;
                                {
                                    let mut map = inner2.receivers.lock().unwrap();
                                    match map.get_mut(&addr) {
                                        Some(r) => {
                                            r.last_hello = Instant::now();
                                            if r.validated {
                                                now_validated = true;
                                            } else if echoed != 0 && echoed == r.nonce {
                                                r.validated = true;
                                                now_validated = true;
                                                log::info!("[SLM] receiver validated: {addr}");
                                            } else {
                                                challenge = Some(r.nonce); // resend until echoed
                                            }
                                        }
                                        None => {
                                            if map.len() < MAX_RECEIVERS {
                                                let nonce = rand_nonce();
                                                map.insert(addr, Receiver {
                                                    addr, last_hello: Instant::now(), latency_us: 0,
                                                    validated: false, nonce,
                                                });
                                                challenge = Some(nonce);
                                                log::info!("[SLM] new receiver (pending validation): {addr}");
                                            } else {
                                                log::warn!("[SLM] receiver cap ({MAX_RECEIVERS}) reached; ignoring {addr}");
                                            }
                                        }
                                    }
                                }
                                // Challenge the unvalidated address; it must echo the
                                // nonce before media flows. CAPS is withheld until
                                // validated, so an unvalidated/spoofed address gets
                                // neither media nor a useful reflection.
                                if let Some(nonce) = challenge {
                                    let _ = ctrl_sock.send_to(&build_challenge(inner2.stream_id, nonce), addr);
                                }
                                if pkt_type == PKT_HELLO && now_validated {
                                    let _ = ctrl_sock.send_to(&caps_pkt, addr);
                                }
                            }
                        } else if pkt_type == PKT_STAT {
                            // Receiver echoed back a sender_ts_ns; elapsed is measured
                            // entirely on our clock, so no clock sync is needed.
                            if let Some(echo_ts) = parse_stat(&buf[..n]) {
                                let elapsed_us = (unix_now_ns().saturating_sub(echo_ts) / 1000) as u32;
                                if let Some(r) = shm_rx.get_mut(&addr) {
                                    // Same machine: elapsed is the one-way SHM staleness.
                                    r.latency_us = ema(r.latency_us, elapsed_us);
                                } else if let Some(r) = inner2.receivers.lock().unwrap().get_mut(&addr) {
                                    // Network round trip: halve for one-way estimate.
                                    r.latency_us = ema(r.latency_us, elapsed_us / 2);
                                }
                            }
                        } else if pkt_type == PKT_MSG {
                            // Operator message. Accept only from an address we already
                            // track (mirrors PKT_STAT) and cap the queue so a flood
                            // can't grow it without bound before the host drains it.
                            let known = shm_rx.contains_key(&addr)
                                || inner2.receivers.lock().unwrap().contains_key(&addr);
                            if known {
                                if let Some(text) = parse_msg(&buf[..n]) {
                                    let mut msgs = inner2.messages.lock().unwrap();
                                    if msgs.len() < 256 { msgs.push(text); }
                                }
                            }
                        }
                    }

                    let mut snapshot: Vec<ReceiverInfo> = Vec::new();
                    {
                        let map = inner2.receivers.lock().unwrap();
                        // Count/report only validated receivers; a pending-validation
                        // entry is not actually receiving media.
                        inner2.receiver_count.store(
                            map.values().filter(|r| r.validated).count() as u32, Ordering::Relaxed);
                        let udp_lat = map.values().filter(|r| r.validated)
                            .map(|r| r.latency_us).filter(|&l| l > 0).max().unwrap_or(0);
                        inner2.udp_latency_us.store(udp_lat, Ordering::Relaxed);
                        for r in map.values().filter(|r| r.validated) {
                            snapshot.push(ReceiverInfo {
                                shm:        false,
                                addr:       Some(r.addr.ip().to_string()),
                                latency_ms: lat_us_to_ms(r.latency_us),
                            });
                        }
                    }
                    inner2.shm_receiver_count.store(shm_rx.len() as u32, Ordering::Relaxed);
                    let shm_lat = shm_rx.values().map(|r| r.latency_us).filter(|&l| l > 0).max().unwrap_or(0);
                    inner2.shm_latency_us.store(shm_lat, Ordering::Relaxed);
                    for r in shm_rx.values() {
                        snapshot.push(ReceiverInfo {
                            shm:        true,
                            addr:       None,
                            latency_ms: lat_us_to_ms(r.latency_us),
                        });
                    }
                    *inner2.receivers_snapshot.lock().unwrap() = snapshot;
                }
            });
        }

        log::info!("[SLM] sender '{}' ready at {}x{}, data_port={data_port}, ctrl_port={ctrl_port}, stream_id={stream_id:#010x}",
                   config.name, config.width, config.height);

        Ok(SlmSender(inner))
    }

    // -- Frame sending ----------------------------------------------------------

    /// Send a video frame to both SHM and all UDP receivers.
    /// Uses the configured LZ4 compression for UDP.
    pub fn send_video(&self, timestamp_ns: u64, nv12: &[u8]) {
        let seq = self.0.frame_seq.fetch_add(1, Ordering::Relaxed);
        if let Ok(mut shm) = self.0.shm.lock() {
            if let Some(ref mut s) = *shm { s.write_video(nv12, timestamp_ns); }
        }
        if self.0.receiver_count.load(Ordering::Relaxed) == 0 { return; }
        let chunks = build_video_chunks(self.0.stream_id, seq, timestamp_ns, nv12, self.0.compress, FRAME_TYPE_P);
        let fec    = append_fec_packets(&chunks, self.0.fec_n, self.0.stream_id, seq, timestamp_ns);
        let mut pkts = chunks;
        pkts.extend(fec);
        self.udp_send_all(&pkts);
    }

    /// Write a video frame to SHM only (no UDP send).
    /// Used in tee mode where a separate encoded subprocess handles UDP.
    pub fn send_video_shm(&self, timestamp_ns: u64, nv12: &[u8]) {
        if let Ok(mut shm) = self.0.shm.lock() {
            if let Some(ref mut s) = *shm { s.write_video(nv12, timestamp_ns); }
        }
    }

    /// Send a video frame to UDP receivers only (no SHM write).
    /// Always uncompressed -- intended for idle/keepalive black frames.
    pub fn send_video_udp(&self, timestamp_ns: u64, nv12: &[u8]) {
        let seq    = self.0.frame_seq.fetch_add(1, Ordering::Relaxed);
        let chunks = build_video_chunks(self.0.stream_id, seq, timestamp_ns, nv12, false, FRAME_TYPE_P);
        let fec    = append_fec_packets(&chunks, self.0.fec_n, self.0.stream_id, seq, timestamp_ns);
        let mut pkts = chunks;
        pkts.extend(fec);
        self.udp_send_all(&pkts);
    }

    /// Send an encoded (H.264/H.265) access unit to UDP receivers only.
    /// No LZ4 compression, no SHM write. `frame_type` is `FRAME_TYPE_IDR` or
    /// `FRAME_TYPE_P`.
    pub fn send_video_encoded(&self, timestamp_ns: u64, au: &[u8], frame_type: u8) {
        let seq    = self.0.frame_seq.fetch_add(1, Ordering::Relaxed);
        let chunks = build_video_chunks(self.0.stream_id, seq, timestamp_ns, au, false, frame_type);
        let fec    = append_fec_packets(&chunks, self.0.fec_n, self.0.stream_id, seq, timestamp_ns);
        let mut pkts = chunks;
        pkts.extend(fec);
        self.udp_send_all(&pkts);
    }

    /// Send audio samples to both SHM and all UDP receivers.
    pub fn send_audio(&self, timestamp_ns: u64, samples: &[f32]) {
        let seq = self.0.audio_seq.fetch_add(1, Ordering::Relaxed);
        if let Ok(mut shm) = self.0.shm.lock() {
            if let Some(ref mut s) = *shm { s.write_audio(samples, timestamp_ns); }
        }
        if self.0.receiver_count.load(Ordering::Relaxed) == 0 { return; }
        let pkts = build_audio_packets(self.0.stream_id, seq, timestamp_ns, samples);
        self.udp_send_all(&pkts);
    }

    // -- SHM control ------------------------------------------------------------

    /// Update the SHM liveness heartbeat without writing a frame.
    /// Call this from idle/pause loops so OBS doesn't declare the producer dead.
    pub fn update_heartbeat(&self) {
        if let Ok(mut shm) = self.0.shm.lock() {
            if let Some(ref mut s) = *shm { s.update_heartbeat(); }
        }
    }

    /// Set the SHM disconnect flag so the consumer knows the stream ended.
    pub fn send_disconnect(&self) {
        if let Ok(shm) = self.0.shm.lock() {
            if let Some(ref s) = *shm { s.send_disconnect(); }
        }
    }

    /// Clear the SHM disconnect flag before starting a new clip.
    pub fn clear_disconnect(&self) {
        if let Ok(shm) = self.0.shm.lock() {
            if let Some(ref s) = *shm { s.clear_disconnect(); }
        }
    }

    // -- Protocol control -------------------------------------------------------

    /// Broadcast SLM_PKT_BYE to all UDP receivers, then immediately resend
    /// CAPS and set the SHM disconnect flag.
    ///
    /// BYE causes the OBS receiver to call `slm_source_on_disconnect` ->
    /// `obs_source_output_video2(NULL)`, flushing its async video queue and
    /// resetting the A/V sync reference. Without the immediate CAPS that
    /// follows, the receiver's `caps_valid` stays false for up to one
    /// keepalive cycle (~250ms), dropping every audio/video packet during
    /// that window. Sending CAPS atomically after BYE restores `caps_valid`
    /// within the same network round-trip.
    pub fn send_bye(&self) {
        let bye = build_bye(self.0.stream_id);
        self.udp_send_all(&[bye, self.0.caps_pkt.clone()]);
        if let Ok(shm) = self.0.shm.lock() {
            if let Some(ref s) = *shm { s.send_disconnect(); }
        }
    }

    /// Signal the control thread to exit. The thread checks this flag after
    /// each 250 ms recv_from timeout, so it exits within ~250 ms.
    pub fn shutdown(&self) {
        self.0.running.store(false, Ordering::Relaxed);
    }

    // -- Status -----------------------------------------------------------------

    pub fn receiver_count(&self)     -> u32  { self.0.receiver_count.load(Ordering::Relaxed) }
    pub fn shm_available(&self)      -> bool { self.0.shm_available.load(Ordering::Relaxed) }
    pub fn shm_receiver_count(&self) -> u32  { self.0.shm_receiver_count.load(Ordering::Relaxed) }

    /// Measured UDP latency (RTT/2) in whole milliseconds, or `None` until a
    /// receiver has echoed at least one PKT_STAT. Floored at 1 ms.
    pub fn udp_latency_ms(&self) -> Option<u32> { lat_us_to_ms(self.0.udp_latency_us.load(Ordering::Relaxed)) }
    /// Measured SHM one-way latency in whole milliseconds, or `None` until a
    /// receiver has echoed at least one PKT_STAT. Floored at 1 ms.
    pub fn shm_latency_ms(&self) -> Option<u32> { lat_us_to_ms(self.0.shm_latency_us.load(Ordering::Relaxed)) }

    /// Per-receiver detail (transport, address, latency) for the UI list.
    pub fn receivers_detail(&self) -> Vec<ReceiverInfo> {
        self.0.receivers_snapshot.lock().unwrap().clone()
    }

    /// Drain operator messages (PKT_MSG) received from receivers since the last
    /// call, oldest first. Empty when none have arrived.
    pub fn take_messages(&self) -> Vec<String> {
        std::mem::take(&mut *self.0.messages.lock().unwrap())
    }

    // -- Internals --------------------------------------------------------------

    fn udp_send_all(&self, packets: &[Vec<u8>]) {
        // Per-packet spin-wait pacing at ~900 Mbps wire rate; NS_PER_PKT is derived
        // from the full on-wire byte count incl. framing (see 0799016). Reset timing
        // is per-packet, not from a frame t0: after a mid-frame preemption we resume
        // at the normal rate instead of bursting the rest at line rate and overflowing
        // the receiver's NIC RX ring.
        let addrs: Vec<SocketAddr> = {
            let map = self.0.receivers.lock().unwrap();
            // Stream only to validated receivers (v4 address validation). This
            // is the gate that prevents the sender being used as a reflector.
            map.values().filter(|r| r.validated).map(|r| r.addr).collect()
        };
        if addrs.is_empty() { return; }
        const NS_PER_PKT: u64 = 13_400; // ~900 Mbps wire rate for 1440-byte SLM packets
        for pkt in packets.iter() {
            let send_start = Instant::now();
            for &addr in &addrs {
                let _ = self.0.data_sock.send_to(pkt, addr);
            }
            let deadline = send_start + Duration::from_nanos(NS_PER_PKT);
            while Instant::now() < deadline {
                std::hint::spin_loop();
            }
        }
    }
}

fn rand_stream_id() -> u32 {
    use std::time::{SystemTime, UNIX_EPOCH};
    let t = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().subsec_nanos();
    let p = std::process::id();
    t ^ (p << 16) ^ (p >> 8)
}

/// Per-address challenge nonce. Derived from `RandomState` (OS-seeded per
/// process for HashDoS resistance), so the value is unguessable to an off-path
/// attacker that cannot see the PKT_CHALLENGE - exactly the threat this defeats.
/// No external RNG dependency. Never returns 0 (0 means "not echoed yet").
fn rand_nonce() -> u64 {
    use std::collections::hash_map::RandomState;
    use std::hash::{BuildHasher, Hasher};
    use std::sync::atomic::AtomicU64;
    use std::time::{SystemTime, UNIX_EPOCH};
    static COUNTER: AtomicU64 = AtomicU64::new(0);
    let mut h = RandomState::new().build_hasher();
    h.write_u64(COUNTER.fetch_add(1, Ordering::Relaxed));
    h.write_u64(SystemTime::now().duration_since(UNIX_EPOCH)
        .map(|d| d.as_nanos() as u64).unwrap_or(0));
    h.write_u32(std::process::id());
    let n = h.finish();
    if n == 0 { 1 } else { n }
}
