//! Slice-2 integration tests: smoke + reconnect-path + DELETE handshake.
//!
//! Drives `rutster_tap_echo::start_echo_server` + `rutster::tap_engine::
//! spawn_tap_engine` directly (no WebRTC peer is wired — slice-1's e2e is
//! manual per spec §8.4; the peer-WebRTC integration lives in README's
//! manual test plan).
//!
//! # Why `tokio_tungstenite::tungstenite::Message` (and not
//! `tokio_tungstenite::Message`)
//!
//! tokio-tungstenite 0.24 re-exports the `tungstenite` crate
//! (`pub use tungstenite;`) but does NOT re-export `Message` at its own
//! crate root. The fully-qualified path through the re-exported
//! `tungstenite` module is the version-stable import — same deviation as
//! Task 4 (see `crates/rutster-tap-echo/src/lib.rs`'s line-118 comment,
//! `crates/rutster-tap/src/tap_client.rs`).
//!
//! # Why `IntoClientRequest` is imported explicitly
//!
//! `connect_async` accepts anything implementing `IntoClientRequest`, but
//! `.into_client_request()` is a *trait method* — so the trait must be in
//! scope to call it. Same pattern the binary uses in
//! `tap_engine.rs::connect_brain`.

use std::sync::atomic::Ordering;
use std::time::Duration;

use futures_util::{SinkExt, StreamExt};
use rutster::session_map::AppState;
use rutster::tap_engine::spawn_tap_engine;
use rutster_call_model::ChannelId;
use rutster_media::{AudioPipe, AudioSink, AudioSource};
use rutster_tap::PcmFrame;
use rutster_tap_echo::start_echo_server;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;

#[tokio::test]
async fn echo_server_starts_and_accepts_connections() {
    // Smoke test: the EchoServer binds + accepts a WS connection.
    let handle = start_echo_server("127.0.0.1:0".parse().unwrap())
        .await
        .unwrap();
    let url = url::Url::parse(&format!("ws://{}/echo", handle.addr)).unwrap();
    // Try a WS connect.
    let req = url.as_str().into_client_request().unwrap();
    let (ws, _resp) = tokio_tungstenite::connect_async(req).await.unwrap();
    // Send hello, expect ack.
    let mut ws = ws;
    let hello = rutster_tap::encode_hello("test-session", 0, 0).unwrap();
    ws.send(tokio_tungstenite::tungstenite::Message::Text(hello))
        .await
        .unwrap();
    let ack = ws.next().await.unwrap().unwrap().into_text().unwrap();
    assert!(ack.contains("\"type\":\"hello\""));
    let _ = handle.shutdown.send(());
}

/// slice-2 spec §8.4 mandated reconnect-path test. Asserts the
/// resilience posture from spec §5.2 + §5.3: kill the brain mid-call →
/// `Channel` stays `Connected` (we don't observe it transition), the
/// `reconnect_attempts` counter increments, playout goes silent during
/// the outage, and after the brain restarts the tap reconnects + audio
/// resumes with *fresh* frames (no stale bleed-through — the §5.3 step 4
/// playout-ring flush contract).
#[tokio::test]
async fn reconnect_after_brain_kill_resumes_audio_and_flushes_playout() {
    // 1. Start EchoServer on ephemeral port.
    let handle = start_echo_server("127.0.0.1:0".parse().unwrap())
        .await
        .unwrap();
    let url = url::Url::parse(&format!("ws://{}/echo", handle.addr)).unwrap();

    // 2. Spawn TapEngine directly against that URL. No `RtcSession`, no
    //    `AppState` — we're exercising the engine + pipe + EchoServer
    //    closed loop, which is what §8.4's reconnect assertion targets.
    //    The binary's poll task (`drive_all_sessions`) is bypassed; the
    //    flush side-channel gets drained manually below to exercise the
    //    Fix-3 playout-ring-flush contract end-to-end.
    let session_id = ChannelId::new();
    let app_state = AppState::default();
    let (advisory_tx, _advisory_rx) =
        tokio::sync::mpsc::channel::<rutster_media::AdvisoryEvent>(16);
    let (mut pipe, mut conn) = spawn_tap_engine(session_id, url, app_state, advisory_tx);

    // 3. Push TWO frames with the same marker `samples[0] = 7` back-to-back
    //    before the kill so the playout ring has buffered content that the
    //    read loop has NOT yet drained. This makes the step-7 "silence after
    //    kill + flush" assertion NON-VACUOUS — there's actually something in
    //    the ring to flush. Spec §5.3 step 4 contract: the flush side-channel
    //    must wipe these stale frames so they don't bleed through post-restart.
    //    Wait long enough first for the engine dial → handshake → pump cycle;
    //    ~200 ms covers a loopback dial + the WS handshake.
    tokio::time::sleep(Duration::from_millis(200)).await;
    let mut frame = PcmFrame::zeroed();
    frame.samples[0] = 7;
    pipe.on_pcm_frame(frame.clone());
    pipe.on_pcm_frame(frame.clone());

    // Drain the engine's audio_out through `next_pcm_frame` until SOME frame
    // arrives (echo round-trip) — but only drain ONE; we want at least one
    // echo frame still buffered in the playout ring at kill time so the
    // step-7 flush has something to wipe. Bounded to ~400 ms to fail fast.
    let mut got_echo = false;
    for _ in 0..20 {
        tokio::time::sleep(Duration::from_millis(20)).await;
        if let Some(f) = pipe.next_pcm_frame() {
            assert_eq!(f.samples.len(), 480);
            got_echo = true;
            break;
        }
    }
    assert!(
        got_echo,
        "brain should echo back our PCM within ~400 ms of pump start"
    );
    // Push one more stale frame so the ring definitely has buffered content
    // (the read above may have drained both); this guarantees the §5.3 step-4
    // flush has something to wipe.
    pipe.on_pcm_frame(frame);

    // 4. Kill the EchoServer. `start_echo_server`'s accept_loop aborts
    //    each spawned per-connection task on shutdown — this forces the
    //    WS stream on the engine side to end (ws.next() → None) so the
    //    pump loop actually returns and the engine enters reconnect.
    //    A listener-only shutdown (without aborting connection tasks)
    //    would leave the live WS half-open and the reconnect assertion
    //    below would never fire (see `rutster-tap-echo::accept_loop`).
    let _ = handle.shutdown.send(());
    handle.join.await.unwrap();

    // 5. Verify reconnect_attempts > 0 within ~2 s. The engine burns its
    //    first backoff slot (250 ms) on a clean Ok(()) pump exit, then
    //    starts dialing the dead port (ECONNREFUSED, fast), increments
    //    again, etc. ~2 s is comfortably above the first two cycles.
    let mut attempts = 0u64;
    for _ in 0..40 {
        attempts = conn.metrics.reconnect_attempts.load(Ordering::Relaxed);
        if attempts > 0 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert!(
        attempts > 0,
        "reconnect_attempts should increment after brain kill"
    );

    // 6. Drain the Fix-3 flush side-channel — simulates what the binary's
    //    `drive_all_sessions` poll task does after each `run_poll_once`.
    //    The signal was emitted by `run_engine_loop` when the pump cycle
    //    exited (Ok(()) on WS-endl). Without draining + applying the flush
    //    here, stale brain-proposed frames would keep playing via
    //    `next_pcm_frame` until the ring drains naturally — that's the
    //    bug fix-3 prevents; this drain makes the test exercise the
    //    actual contract.
    let mut should_flush = false;
    if let Some(rx) = conn.flush_rx.as_mut() {
        while let Ok(()) = rx.try_recv() {
            should_flush = true;
        }
    }
    if should_flush {
        pipe.clear_playout_ring();
    }

    // 7. Verify playout has gone silent. After the flush + during the
    //    outage (no new audio_out frames queued), `next_pcm_frame` returns
    //    None (underflow / silence per spec §4.1).
    let mut silence_count = 0;
    for _ in 0..5 {
        if pipe.next_pcm_frame().is_none() {
            silence_count += 1;
        }
    }
    assert!(
        silence_count >= 3,
        "playout should be silent after brain kill + flush; got {silence_count}/5 None"
    );

    // 8. Restart the EchoServer on the SAME addr:port. tokio's
    //    `TcpListener::bind` sets `SO_REUSEADDR` on Unix (tokio's
    //    internal `socket2`-based bind), so the rebinding succeeds even
    //    though the previous accept_loop just dropped the listener.
    let addr = handle.addr;
    let handle2 = start_echo_server(addr)
        .await
        .expect("restart EchoServer on the same port (SO_REUSEADDR)");

    // 9. Verify reconnect succeeds + audio resumes within the engine's
    //    next backoff window. The backoff has been climbing each failed
    //    attempt during the outage; in the worst case it's at 1 s or 2 s
    //    by the time we restart. 6 s total budget > 5 s cap, so we'll
    //    definitely observe the resume within the loop.
    //
    //    __STRENGTHENED ASSERTION__ (final-fixes re-review Important #2):
    //    the original test only checked `next_pcm_frame().is_some()` — a
    //    future refactor that silently broke the §5.3 playout-ring flush
    //    would still pass, because the stale-frame bleed-through would
    //    still surface as Some. We now explicitly assert the returned frame
    //    carries the FRESH marker (`samples[0] == 9`) and NOT the stale
    //    marker (`samples[0] == 7`) — actively witnessing the §5.3 step 4
    //    "no stale bleed-through" contract.
    let mut resumed = false;
    let restart_at = std::time::Instant::now();
    while restart_at.elapsed() < Duration::from_secs(6) {
        // Push a fresh PCM marker `samples[0] = 9`; if the engine has
        // reconnected, it'll echo through to the (flushed) playout ring.
        let mut f = PcmFrame::zeroed();
        f.samples[0] = 9;
        pipe.on_pcm_frame(f);
        tokio::time::sleep(Duration::from_millis(50)).await;
        if let Some(returned) = pipe.next_pcm_frame() {
            // Fresh-marker: witnesses reconnect + audio resumption.
            assert_eq!(
                returned.samples[0], 9,
                "resumed frame should carry the FRESH marker (samples[0] == 9), \
                 not the stale samples[0] == 7 — catches a broken §5.3 flush \
                 side-channel that would bleed stale frames through."
            );
            resumed = true;
            break;
        }
    }
    assert!(resumed, "audio should resume after EchoServer restart");

    // 9b. Belt-and-suspenders: walk a few more frames and assert NONE of
    //     them carry the stale marker `samples[0] == 7`. The fresh-marker
    //     check above is sufficient to catch the failure mode, but walking
    //     the next few frames actively witnesses that no stale frame is
    //     queued behind the fresh one in the playout ring.
    for _ in 0..5 {
        if let Some(returned) = pipe.next_pcm_frame() {
            assert_ne!(
                returned.samples[0], 7,
                "no stale bleed-through (samples[0] == 7) permitted after the \
                 §5.3 step-4 flush + reconnect"
            );
        }
    }

    // 10. Tear down the engine task to avoid leaking.
    let _ = conn.close_tx.send(());
    conn.join.abort();
    let _ = handle2.shutdown.send(());
}
