//! # nodelay — TCP_NODELAY latency assertion (deploy slice A §5.1)
//!
//! The S1–S8 sweep regresses tick-side latency; this test regresses the
//! LISTENER: server-originated WS frames at the trunk's 20 ms cadence,
//! through `rutster::serve::serve_with_nodelay` (the production serve
//! path) over a real loopback TCP socket, measured send→recv per frame.
//!
//! Runs ONLY under `--features=sim-bench` (the CI-regressed job,
//! `--test-threads=1`). A regression here means Nagle is back on the
//! accepted sockets (axum #2521 class) and every proxied trunk call
//! would jitter by the peer's delayed-ACK timer.
//!
//! ## Why this test needs socket2 + TCP_QUICKACK suppression (Option D)
//!
//! On Linux loopback the kernel's TCP_QUICKACK heuristic often acks each
//! small segment immediately, so even with `tcp_nodelay(false)` on the
//! server the Nagle/delayed-ACK stall does not appear at a 20 ms cadence.
//! That made the original Plan A Step 5 load-bearing check pass with
//! Nagle on, which the plan says is an invalid tripwire. We therefore
//! force delayed-ACK mode on the *client* socket with
//! `sock_ref.set_tcp_quickack(false)` and then **re-assert it after every
//! `ws.next()`**, because `TCP_QUICKACK` is not persistent — the kernel
//! re-engages quickack between calls. Once quickack is suppressed, the
//! loopback client delays its ACKs and `tcp_nodelay(false)` on the server
//! exhibits the ~40 ms Nagle stall the assertion is meant to catch.
//!
//! This is Linux-only (`set_tcp_quickack` is cfg-gated in socket2), so
//! the helper is cfg-gated with a no-op on other platforms. The CI
//! sim-bench job runs on Linux, so the tripwire is effective there.

use std::time::{Duration, Instant};

use futures_util::{SinkExt, StreamExt};
use rutster_media::PcmFrame;
use rutster_trunk::twilio_media_streams::{RegisterTrunkInboundChannel, TwilioMediaStreamsServer};
use tokio::sync::mpsc;
use tokio_tungstenite::MaybeTlsStream;
use tokio_tungstenite::tungstenite::Message;

use crate::thresholds::WS_FRAME_SEND_TO_RECV_P99_MS;

/// 200 frames × 20 ms ≈ 4 s of wall clock — enough samples for a
/// meaningful p99 (index 197) without dominating the sim-bench job.
const FRAMES: usize = 200;
const CADENCE: Duration = Duration::from_millis(20);

/// Disable TCP_QUICKACK on a tokio TcpStream so the peer enters delayed-ACK
/// mode. `TCP_QUICKACK` resets after protocol events, so callers must
/// re-apply this after each receive when determinism matters.
#[cfg(target_os = "linux")]
fn suppress_tcp_quickack(stream: &tokio::net::TcpStream) {
    let sock_ref = socket2::SockRef::from(stream);
    // Ignore errors: if the socket disappeared the recv loop will fail
    // soon anyway; setting a socket option must not abort the test.
    let _ = sock_ref.set_tcp_quickack(false);
}

/// On non-Linux platforms `set_tcp_quickack` is unavailable; the assertion
/// remains a general latency tripwire but cannot force the Nagle stall on
/// loopback. The CI sim-bench job is Linux-bound, so the Linux path covers
/// the regression case.
#[cfg(not(target_os = "linux"))]
fn suppress_tcp_quickack(_stream: &tokio::net::TcpStream) {}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn ws_frame_send_to_recv_p99_stays_under_nodelay_threshold() {
    // 1. Real listener + the PRODUCTION serve path (Task 1's helper).
    let (register_tx, mut register_rx) = mpsc::channel::<RegisterTrunkInboundChannel>(4);
    let app = TwilioMediaStreamsServer::router(register_tx, Duration::from_secs(20));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let (_stop_tx, stop_rx) = tokio::sync::oneshot::channel::<()>();
    tokio::spawn(rutster::serve::serve_with_nodelay(listener, app, async {
        let _ = stop_rx.await;
    }));

    // 2. Stub media thread: ack the registration; keep the channel ends.
    let register_task = tokio::spawn(async move {
        let register = register_rx.recv().await.expect("pump registers");
        // Destructure BEFORE reply.send() consumes `reply` — channel
        // ends captured by value, intentionally kept alive for the
        // whole run (mirrors tests/ws_ping.rs's fix for the same E0382).
        let RegisterTrunkInboundChannel {
            reply,
            inbound_from_twilio_rx,
            outbound_to_twilio_tx,
            ..
        } = register;
        let _ = reply.send(rutster_call_model::ChannelId::new());
        (inbound_from_twilio_rx, outbound_to_twilio_tx)
    });

    // 3. Twilio-side client: connected + start handshake.
    let (mut ws, _resp) =
        tokio_tungstenite::connect_async(format!("ws://{addr}/twilio/media-stream"))
            .await
            .expect("WS connect");
    // Force delayed-ACK mode on the client socket. Linux loopback
    // otherwise quick-acks every small segment and hides the Nagle stall.
    if let MaybeTlsStream::Plain(stream) = ws.get_ref() {
        suppress_tcp_quickack(stream);
    }
    ws.send(Message::Text(
        r#"{"event":"connected","protocol":"twilio-media-stream","version":"1.0.0"}"#.into(),
    ))
    .await
    .unwrap();
    ws.send(Message::Text(
        r#"{"event":"start","start":{"streamSid":"MZsim","callSid":"CAsim"}}"#.into(),
    ))
    .await
    .unwrap();
    let (_inbound_from_twilio_rx, outbound_to_twilio_tx) =
        register_task.await.expect("register task");

    // 4. Server-originated frames at the trunk cadence. mpsc, the pump
    //    loop, and the WS all preserve order, so sent frame k IS received
    //    frame k — instants pair by index; no payload tagging needed
    //    (µ-law would quantize a tag anyway).
    let outbound = outbound_to_twilio_tx.clone();
    let sender = tokio::spawn(async move {
        let mut send_instants = Vec::with_capacity(FRAMES);
        let mut cadence = tokio::time::interval(CADENCE);
        cadence.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay);
        for _ in 0..FRAMES {
            cadence.tick().await;
            send_instants.push(Instant::now());
            outbound
                .send(PcmFrame::zeroed())
                .await
                .expect("pump alive for the whole run");
        }
        send_instants
    });

    let mut recv_instants = Vec::with_capacity(FRAMES);
    while recv_instants.len() < FRAMES {
        let msg = tokio::time::timeout(Duration::from_secs(10), ws.next())
            .await
            .expect("frame within 10s")
            .expect("WS stream open")
            .expect("WS frame ok");
        // TCP_QUICKACK is not persistent: the kernel re-enables it as
        // data arrives, so we suppress it again after every receive to
        // keep delayed-ACK mode active for the next send/receive pair.
        if let MaybeTlsStream::Plain(stream) = ws.get_ref() {
            suppress_tcp_quickack(stream);
        }
        // Only Text frames are media envelopes; skip the §5.2 keepalive
        // Ping (and anything else) so pairing stays index-aligned.
        if matches!(msg, Message::Text(_)) {
            recv_instants.push(Instant::now());
        }
    }
    let send_instants = sender.await.expect("sender task");

    let mut latencies_ms: Vec<f64> = send_instants
        .iter()
        .zip(&recv_instants)
        .map(|(s, r)| r.duration_since(*s).as_secs_f64() * 1000.0)
        .collect();
    latencies_ms.sort_by(|a, b| a.partial_cmp(b).expect("no NaN latencies"));
    let p99 = latencies_ms[(latencies_ms.len() * 99).div_ceil(100) - 1];
    assert!(
        p99 <= WS_FRAME_SEND_TO_RECV_P99_MS,
        "p99 WS frame send→recv {p99:.2}ms > {WS_FRAME_SEND_TO_RECV_P99_MS}ms \
         (TCP_NODELAY regression: Nagle/delayed-ACK is stalling the 20ms \
         trunk cadence; axum #2521 class; deploy spec §5.1)"
    );
}
