//! End-to-end integration tests.
//!
//! Each test spins up the server in a background OS thread on an
//! ephemeral port, opens a regular `std::net::TcpStream` against it, and
//! exercises the wire. Single thread for the smarm runtime per test so
//! teardown happens cleanly when the test thread is dropped (the process
//! exits at the end of the test binary).

use std::io::{Read, Write};
use std::net::{SocketAddr, TcpListener, TcpStream};
use std::time::Duration;

use urus::{serve_with, Conn, Config, Next, Pipeline, Router};

// ---------------------------------------------------------------------------
// Helpers
// ---------------------------------------------------------------------------

/// Reserve a free localhost port by binding briefly and reading back the
/// assigned port number. The port is released the instant we drop the
/// listener; there's a tiny race window before the urus server claims it
/// but it's well below the threshold for flakiness in practice.
fn free_port() -> u16 {
    let l = TcpListener::bind("127.0.0.1:0").unwrap();
    l.local_addr().unwrap().port()
}

fn spawn_server(pipeline: Pipeline) -> u16 {
    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:     2,
            scheduler_threads: Some(2),
            ..Config::new(addr)
        };
        serve_with(cfg, pipeline).unwrap();
    });
    // Wait for the server to actually be listening.
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() {
            return port;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("server didn't come up on {addr}");
}

fn send_request(port: u16, req: &[u8]) -> Vec<u8> {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(req).unwrap();
    s.shutdown(std::net::Shutdown::Write).ok();
    let mut buf = Vec::new();
    s.read_to_end(&mut buf).unwrap();
    buf
}

fn http_status(resp: &[u8]) -> u16 {
    let s = std::str::from_utf8(&resp[..resp.len().min(64)]).unwrap();
    let parts: Vec<&str> = s.splitn(3, ' ').collect();
    parts[1].parse().unwrap()
}

fn http_body(resp: &[u8]) -> &[u8] {
    // Split on the first \r\n\r\n.
    for i in 0..resp.len().saturating_sub(3) {
        if &resp[i..i + 4] == b"\r\n\r\n" {
            return &resp[i + 4..];
        }
    }
    &[]
}

// ---------------------------------------------------------------------------
// Tests
// ---------------------------------------------------------------------------

#[test]
fn hello_world() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| {
            c.put_status(200).put_body("hello urus")
        })
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"hello urus");
}

#[test]
fn server_and_listeners_are_registered() {
    // `whereis` must run inside the runtime (the test itself is a foreign
    // OS thread with no runtime in its TLS), so probe from a handler —
    // connection actors live in the runtime by construction.
    let pipe = Pipeline::new().plug(
        Router::new().get("/whereis", |c: Conn, _n: Next| {
            let ok = smarm::whereis("urus.server").is_some()
                && smarm::whereis("urus.listener.0").is_some()
                && smarm::whereis("urus.listener.1").is_some(); // pool of 2
            c.put_status(200).put_body(if ok { "registered" } else { "missing" })
        })
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"GET /whereis HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"registered");
}

#[test]
fn echo_body() {
    let pipe = Pipeline::new().plug(
        Router::new().post("/echo", |c: Conn, _n: Next| {
            let body = c.body.as_bytes().to_vec();
            c.put_status(200).put_body(body)
        })
    );
    let port = spawn_server(pipe);
    let resp = send_request(
        port,
        b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nConnection: close\r\n\r\nhello",
    );
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"hello");
}

#[test]
fn path_params() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/users/:id", |c: Conn, _n: Next| {
            let id = c.params.get("id").unwrap_or("?").to_string();
            c.put_status(200).put_body(format!("user={id}"))
        })
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"GET /users/42 HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"user=42");
}

#[test]
fn method_not_allowed() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/only-get", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"POST /only-get HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 405);
}

#[test]
fn unknown_route_falls_to_404() {
    // Router falls through; connection actor's default emits 404.
    let pipe = Pipeline::new().plug(
        Router::new().get("/known", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"GET /missing HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 404);
}

#[test]
fn keep_alive_two_requests_on_one_connection() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200).put_body("ok"))
    );
    let port = spawn_server(pipe);
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();

    // First request, no Connection header (HTTP/1.1 default = keep-alive).
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();

    // Read the first response (Content-Length: 2 -> body "ok").
    let mut buf = vec![0u8; 1024];
    let mut total = 0;
    let mut first_end = None;
    while first_end.is_none() {
        let n = s.read(&mut buf[total..]).unwrap();
        assert!(n > 0, "server hung up early");
        total += n;
        // First response ends at header-terminator + 2 bytes of body.
        for i in 0..total.saturating_sub(3) {
            if &buf[i..i + 4] == b"\r\n\r\n" {
                let body_start = i + 4;
                if total >= body_start + 2 {
                    first_end = Some(body_start + 2);
                }
                break;
            }
        }
    }
    let first = &buf[..first_end.unwrap()];
    assert_eq!(http_status(first), 200);
    assert_eq!(http_body(first), b"ok");

    // Second request on the same socket — proves keep-alive works.
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap();
    let mut rest = Vec::new();
    s.read_to_end(&mut rest).unwrap();
    assert_eq!(http_status(&rest), 200);
    assert_eq!(http_body(&rest), b"ok");
}

#[test]
fn panicking_handler_yields_500() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/boom", |_c: Conn, _n: Next| -> Conn {
            panic!("handler explosion")
        })
    );
    let port = spawn_server(pipe);
    let resp = send_request(port, b"GET /boom HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 500);
}

#[test]
fn middleware_can_short_circuit() {
    // Middleware that requires `X-Auth: secret`; otherwise returns 401
    // without calling `next.run(conn)`.
    let auth = |c: Conn, n: Next| {
        if c.headers.get("x-auth").map(|s| s == "secret").unwrap_or(false) {
            n.run(c)
        } else {
            c.put_status(401).halt()
        }
    };
    let pipe = Pipeline::new()
        .plug(auth)
        .plug(Router::new().get("/secret", |c: Conn, _n: Next| c.put_status(200).put_body("ok")));

    let port = spawn_server(pipe);

    // Without auth → 401.
    let resp = send_request(port, b"GET /secret HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 401);

    // With auth → 200 + body.
    let resp = send_request(
        port,
        b"GET /secret HTTP/1.1\r\nHost: x\r\nX-Auth: secret\r\nConnection: close\r\n\r\n",
    );
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"ok");
}

// ---------------------------------------------------------------------------
// Supervision (v0.2 chunk 1)
// ---------------------------------------------------------------------------

/// A panicking listener must be restarted by the pool supervisor, and the
/// connection that was pending when it died must still be served.
///
/// Pool size is 1 on purpose: with a sibling listener the test would pass
/// even if restarts were broken (the sibling would pick up the accept).
/// With one listener, the request after the injected panic can only
/// succeed if a fresh listener came up on the same fd.
#[test]
fn panicking_listener_restarts() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/ping", |c: Conn, _n: Next| c.put_status(200).put_body("pong"))
    );

    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:     1,
            scheduler_threads: Some(2),
            ..Config::new(addr)
        };
        serve_with(cfg, pipe).unwrap();
    });
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() {
            break;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    // Sanity: server answers before the fault.
    let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 200);

    // Arm the fault: the listener's next accept-loop iteration panics
    // *before* accepting, so our connection waits in the kernel backlog
    // until the restarted listener picks it up.
    urus::serve::INJECT_LISTENER_PANICS.store(1, std::sync::atomic::Ordering::Relaxed);

    let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
    assert_eq!(http_status(&resp), 200, "request pending across the panic was not served");
    assert_eq!(http_body(&resp), b"pong");
    assert_eq!(
        urus::serve::INJECT_LISTENER_PANICS.load(std::sync::atomic::Ordering::Relaxed),
        0,
        "fault was never consumed — listener didn't wake for the connection"
    );

    // The restarted listener keeps serving.
    for _ in 0..5 {
        let resp = send_request(port, b"GET /ping HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n");
        assert_eq!(http_status(&resp), 200);
    }
}

// ---------------------------------------------------------------------------
// Graceful shutdown (v0.2 chunk 2)
// ---------------------------------------------------------------------------

/// Boot a server with a shutdown handle. Returns (port, handle, done_rx)
/// where done_rx fires when serve_with_shutdown returns.
fn spawn_server_with_handle(
    pipeline: Pipeline,
    drain: Duration,
) -> (u16, urus::Handle, std::sync::mpsc::Receiver<()>) {
    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    let (handle, signal) = urus::shutdown_handle();
    let (done_tx, done_rx) = std::sync::mpsc::channel();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:     2,
            scheduler_threads: Some(2),
            drain_timeout:     drain,
            ..Config::new(addr)
        };
        urus::serve_with_shutdown(cfg, pipeline, signal).unwrap();
        let _ = done_tx.send(());
    });
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() {
            return (port, handle, done_rx);
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("server didn't come up on {addr}");
}

/// Shutdown with nothing in flight returns promptly — well inside the
/// (deliberately huge) drain window, proving idle conns don't run the
/// clock out.
#[test]
fn shutdown_with_no_connections_returns() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
    );
    let (_port, handle, done_rx) = spawn_server_with_handle(pipe, Duration::from_secs(30));
    handle.shutdown();
    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve_with_shutdown did not return after shutdown()");
}

/// An idle keep-alive connection is closed immediately on shutdown (the
/// drain deadline of 30s must NOT be what gates the return), and the
/// server refuses new connections afterwards.
#[test]
fn shutdown_closes_idle_keepalive_promptly() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200).put_body("ok"))
    );
    let (port, handle, done_rx) = spawn_server_with_handle(pipe, Duration::from_secs(30));

    // Complete one request on a keep-alive connection and leave it open.
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
    let mut buf = [0u8; 1024];
    let n = s.read(&mut buf).unwrap();
    assert!(n > 0 && buf.starts_with(b"HTTP/1.1 200"));

    let t0 = std::time::Instant::now();
    handle.shutdown();

    // The parked conn actor is stopped; its socket closes → we see EOF.
    let n = s.read(&mut buf).expect("read after shutdown");
    assert_eq!(n, 0, "expected EOF on the idle keep-alive connection");

    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return");
    assert!(
        t0.elapsed() < Duration::from_secs(5),
        "shutdown waited for the drain deadline instead of stopping the idle conn"
    );

    // Listeners are gone: new connections are refused.
    assert!(
        TcpStream::connect(("127.0.0.1", port)).is_err(),
        "server still accepting after shutdown"
    );
}

/// A request in flight when shutdown is signalled completes with a real
/// response before its connection is closed.
#[test]
fn shutdown_drains_in_flight_request() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/slow", |c: Conn, _n: Next| {
            smarm::sleep(Duration::from_millis(400));
            c.put_status(200).put_body("made it")
        })
    );
    let (port, handle, done_rx) = spawn_server_with_handle(pipe, Duration::from_secs(10));

    // Fire the slow request, then shut down while it is in flight.
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /slow HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap();
    std::thread::sleep(Duration::from_millis(100)); // let the head land
    handle.shutdown();

    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap();
    assert_eq!(http_status(&resp), 200, "in-flight request was cut off by shutdown");
    assert_eq!(http_body(&resp), b"made it");

    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return after drain");
}

/// A request still in flight at the drain deadline is force-stopped: the
/// connection dies without a response, but serve returns near the
/// deadline rather than hanging on the stuck handler's full duration.
#[test]
fn shutdown_force_stops_at_drain_deadline() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/stuck", |c: Conn, _n: Next| {
            smarm::sleep(Duration::from_secs(60));
            c.put_status(200)
        })
    );
    let (port, handle, done_rx) = spawn_server_with_handle(pipe, Duration::from_millis(300));

    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
    s.write_all(b"GET /stuck HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n").unwrap();
    std::thread::sleep(Duration::from_millis(100));
    handle.shutdown();

    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return after force-stop deadline");
}

// ---------------------------------------------------------------------------
// Timeouts (v0.2 chunk 3)
// ---------------------------------------------------------------------------

fn spawn_server_with_timeouts(
    pipeline: Pipeline,
    keep_alive: Duration,
    request: Duration,
) -> u16 {
    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:      2,
            scheduler_threads:  Some(2),
            keep_alive_timeout: keep_alive,
            request_timeout:    request,
            ..Config::new(addr)
        };
        serve_with(cfg, pipeline).unwrap();
    });
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() {
            return port;
        }
        std::thread::sleep(Duration::from_millis(50));
    }
    panic!("server didn't come up on {addr}");
}

/// Read from `s` until the response head is complete (double CRLF). Only
/// suitable for responses with an empty body.
fn read_response_head(s: &mut TcpStream) -> Vec<u8> {
    let mut resp = Vec::new();
    let mut byte = [0u8; 1];
    while !resp.ends_with(b"\r\n\r\n") {
        match s.read(&mut byte) {
            Ok(0) => break,
            Ok(_) => resp.push(byte[0]),
            Err(e) => panic!("read failed mid-response: {e}"),
        }
    }
    resp
}

/// An idle keep-alive connection is reaped once keep_alive_timeout passes
/// with no next request: the server closes the socket (clean EOF on our
/// side) well before the much larger request_timeout.
#[test]
fn idle_keepalive_reaped_at_keep_alive_timeout() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server_with_timeouts(
        pipe,
        Duration::from_millis(300), // keep_alive_timeout under test
        Duration::from_secs(10),    // request_timeout out of the way
    );

    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n\r\n").unwrap(); // keep-alive
    let head = read_response_head(&mut s);
    assert_eq!(http_status(&head), 200);

    // Now go quiet. The server should close us at ~300ms; the 5s read
    // timeout on our side is the failure detector.
    let start = std::time::Instant::now();
    let mut byte = [0u8; 1];
    match s.read(&mut byte) {
        Ok(0) => {} // clean EOF: reaped
        Ok(n) => panic!("expected EOF, got {n} unexpected byte(s)"),
        Err(e) => panic!("expected EOF, read errored: {e}"),
    }
    assert!(
        start.elapsed() < Duration::from_secs(3),
        "reap took {:?}, expected ~300ms", start.elapsed()
    );
}

/// A slowloris client that sends a partial head and then stalls is killed
/// at request_timeout with a best-effort 408, even though the (large)
/// keep-alive budget hasn't expired.
#[test]
fn slowloris_partial_head_killed_at_request_timeout() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server_with_timeouts(
        pipe,
        Duration::from_secs(10),    // keep_alive_timeout out of the way
        Duration::from_millis(300), // request_timeout under test
    );

    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    // Partial head: request clock starts on these bytes, never completes.
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\n").unwrap();

    let start = std::time::Instant::now();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).expect("expected 408+EOF or EOF");
    assert!(
        start.elapsed() < Duration::from_secs(3),
        "kill took {:?}, expected ~300ms", start.elapsed()
    );
    // The 408 is best-effort (single non-parking write), but with our read
    // side live it should land.
    assert!(!resp.is_empty(), "expected a best-effort 408 before close");
    assert_eq!(http_status(&resp), 408);
}

// ---------------------------------------------------------------------------
// Streaming responses (v0.3 chunk 1)
// ---------------------------------------------------------------------------

/// Router with a fixed route plus a 3-chunk streaming route. The producer
/// is a separate actor (the pull design): the handler spawns it, hands the
/// Receiver to urus, and returns. Dropping the Sender ends the stream.
fn streaming_pipeline(chunk_gap: Duration, chunks: usize) -> Pipeline {
    Pipeline::new().plug(
        Router::new()
            .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello"))
            .get("/stream", move |c: Conn, _n: Next| {
                let (tx, rx) = smarm::channel::<Vec<u8>>();
                smarm::spawn(move || {
                    for i in 0..chunks {
                        if tx.send(format!("part{i}").into_bytes()).is_err() {
                            return; // conn died; stop producing
                        }
                        smarm::sleep(chunk_gap);
                    }
                });
                c.put_status(200).put_body(rx)
            }),
    )
}

/// Read from `s` until the chunked-encoding terminator arrives, then
/// return (head, decoded_body).
fn read_chunked_response(s: &mut TcpStream) -> (String, Vec<u8>) {
    let mut raw = Vec::new();
    let mut buf = [0u8; 4096];
    loop {
        match s.read(&mut buf) {
            Ok(0) => panic!("connection closed before chunked terminator"),
            Ok(n) => raw.extend_from_slice(&buf[..n]),
            Err(e) => panic!("read failed mid-stream: {e}"),
        }
        if raw.windows(5).any(|w| w == b"0\r\n\r\n") {
            break;
        }
    }
    let head_end = raw.windows(4).position(|w| w == b"\r\n\r\n").unwrap() + 4;
    let head = String::from_utf8_lossy(&raw[..head_end]).to_string();

    // Decode the chunked body.
    let mut body = Vec::new();
    let mut rest = &raw[head_end..];
    loop {
        let line_end = rest.windows(2).position(|w| w == b"\r\n").unwrap();
        let size = usize::from_str_radix(
            std::str::from_utf8(&rest[..line_end]).unwrap().trim(),
            16,
        )
        .unwrap();
        rest = &rest[line_end + 2..];
        if size == 0 {
            break;
        }
        body.extend_from_slice(&rest[..size]);
        assert_eq!(&rest[size..size + 2], b"\r\n", "chunk not CRLF-terminated");
        rest = &rest[size + 2..];
    }
    (head, body)
}

/// A streaming handler produces a chunked HTTP/1.1 response whose decoded
/// body is the concatenation of the producer's sends, with no
/// content-length on the wire.
#[test]
fn streaming_response_chunked() {
    let port = spawn_server(streaming_pipeline(Duration::from_millis(10), 3));
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();

    let (head, body) = read_chunked_response(&mut s);
    assert!(head.starts_with("HTTP/1.1 200 OK\r\n"), "head: {head}");
    assert!(head.contains("transfer-encoding: chunked"), "head: {head}");
    assert!(!head.contains("content-length"), "head: {head}");
    assert_eq!(body, b"part0part1part2");
}

/// The connection survives a completed chunked response: a second request
/// on the same socket gets a normal fixed response.
#[test]
fn keep_alive_after_chunked_response() {
    let port = spawn_server(streaming_pipeline(Duration::from_millis(5), 3));
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();

    s.write_all(b"GET /stream HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
    let (_head, body) = read_chunked_response(&mut s);
    assert_eq!(body, b"part0part1part2");

    // Same socket, second request.
    s.write_all(b"GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n")
        .unwrap();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap();
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"hello");
}

/// HTTP/1.0 keep-alive must be advertised back (`connection: keep-alive`)
/// when honored — 1.0 defaults to close, so a silent keep-open leaves
/// spec-following clients (ab -k) waiting for EOF. The socket then serves
/// a second request.
#[test]
fn http10_keepalive_advertised_and_socket_reused() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server(pipe);
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();

    let req = b"GET / HTTP/1.0\r\nHost: x\r\nConnection: keep-alive\r\n\r\n";
    s.write_all(req).unwrap();
    let head = String::from_utf8(read_response_head(&mut s)).unwrap();
    assert!(head.starts_with("HTTP/1.0 200 OK\r\n"), "head: {head}");
    assert!(head.contains("connection: keep-alive"), "head: {head}");

    // Same socket, second request — the keep-alive was real.
    s.write_all(req).unwrap();
    let head2 = String::from_utf8(read_response_head(&mut s)).unwrap();
    assert!(head2.starts_with("HTTP/1.0 200 OK\r\n"), "head: {head2}");
}

/// Error statuses on HTTP/1.0 force close even when the client asked for
/// keep-alive: `connection: close` on the wire, then EOF.
#[test]
fn http10_keepalive_forced_off_on_error_status() {
    let pipe = Pipeline::new().plug(
        Router::new().get("/", |c: Conn, _n: Next| c.put_status(200))
    );
    let port = spawn_server(pipe);
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();

    s.write_all(b"GET /missing HTTP/1.0\r\nHost: x\r\nConnection: keep-alive\r\n\r\n")
        .unwrap();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap(); // EOF: server closed
    let head = String::from_utf8_lossy(&resp).to_string();
    assert!(head.starts_with("HTTP/1.0 404 Not Found\r\n"), "head: {head}");
    assert!(head.contains("connection: close"), "head: {head}");
}

/// HTTP/1.0 has no chunked framing: a stream body goes out raw, the
/// response carries `connection: close` and no transfer-encoding, and EOF
/// delimits the body.
#[test]
fn http10_stream_is_eof_delimited() {
    let port = spawn_server(streaming_pipeline(Duration::from_millis(5), 3));
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /stream HTTP/1.0\r\nHost: x\r\n\r\n").unwrap();

    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap(); // EOF = end of body
    let head = String::from_utf8_lossy(&resp[..]).to_string();
    assert!(head.starts_with("HTTP/1.0 200 OK\r\n"), "head: {head}");
    assert!(!head.contains("transfer-encoding"), "head: {head}");
    assert!(head.contains("connection: close"), "head: {head}");
    assert_eq!(http_body(&resp), b"part0part1part2");
}

/// An infinite stream is an in-flight request: graceful shutdown waits for
/// the drain deadline, then force-stops the conn actor parked in the pump's
/// recv(). serve returns; the client sees EOF; the orphaned producer's next
/// send fails (closed channel) and it exits, letting the runtime wind down.
#[test]
fn shutdown_force_stops_active_stream() {
    let pipe = Pipeline::new().plug(Router::new().get(
        "/infinite",
        |c: Conn, _n: Next| {
            let (tx, rx) = smarm::channel::<Vec<u8>>();
            smarm::spawn(move || loop {
                if tx.send(b"tick".to_vec()).is_err() {
                    return;
                }
                smarm::sleep(Duration::from_millis(20));
            });
            c.put_status(200).put_body(rx)
        },
    ));
    let (port, handle, done_rx) =
        spawn_server_with_handle(pipe, Duration::from_millis(300));

    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /infinite HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();

    // Confirm the stream is live before shutting down.
    let mut buf = [0u8; 1024];
    let n = s.read(&mut buf).unwrap();
    assert!(n > 0);

    handle.shutdown();
    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return: streaming conn not force-stopped at drain deadline");

    // Client side drains to EOF.
    let mut rest = Vec::new();
    let _ = s.read_to_end(&mut rest); // EOF or reset; both fine
}

/// A client that stops reading mid-stream is dropped once a chunk write
/// stalls past write_timeout; the producer observes the closed channel.
#[test]
fn stalled_reader_killed_at_write_timeout() {
    let (dead_tx, dead_rx) = std::sync::mpsc::channel::<()>();
    let pipe = Pipeline::new().plug(Router::new().get(
        "/firehose",
        move |c: Conn, _n: Next| {
            let (tx, rx) = smarm::channel::<Vec<u8>>();
            let dead_tx = dead_tx.clone();
            smarm::spawn(move || loop {
                if tx.send(vec![0u8; 64 * 1024]).is_err() {
                    let _ = dead_tx.send(()); // conn actor dropped the Receiver
                    return;
                }
                smarm::sleep(Duration::from_millis(1));
            });
            c.put_status(200).put_body(rx)
        },
    ));

    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:     2,
            scheduler_threads: Some(2),
            write_timeout:     Duration::from_millis(300),
            ..Config::new(addr)
        };
        serve_with(cfg, pipe).unwrap();
    });
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() {
            break;
        }
        std::thread::sleep(Duration::from_millis(50));
    }

    let mut s = TcpStream::connect(addr).unwrap();
    s.write_all(b"GET /firehose HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();
    // Read a little to confirm liveness, then stall: never read again.
    let mut buf = [0u8; 1024];
    let _ = s.read(&mut buf).unwrap();

    // Socket buffers fill, a chunk write stalls, write_timeout (300ms)
    // expires, the conn actor exits, the Receiver drops, the producer's
    // send fails. Generous wall budget for slow CI.
    dead_rx
        .recv_timeout(Duration::from_secs(10))
        .expect("producer never observed conn death: write_timeout not enforced");
}

// ---------------------------------------------------------------------------
// Chunked request bodies (v0.3 chunk 2)
// ---------------------------------------------------------------------------

fn echo_pipeline() -> Pipeline {
    Pipeline::new().plug(
        Router::new()
            .get("/", |c: Conn, _n: Next| c.put_status(200).put_body("hello"))
            .post("/echo", |c: Conn, _n: Next| {
                let body = c.body.as_bytes().to_vec();
                c.put_status(200).put_body(body)
            }),
    )
}

/// A chunked request body (with chunk extensions, which must be ignored)
/// is decoded before the pipeline runs; the handler sees the joined bytes.
#[test]
fn chunked_request_body_is_decoded() {
    let port = spawn_server(echo_pipeline());
    let resp = send_request(
        port,
        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\nConnection: close\r\n\r\n\
          5\r\nhello\r\n6;ext=1\r\n world\r\n0\r\n\r\n",
    );
    assert_eq!(http_status(&resp), 200);
    assert_eq!(http_body(&resp), b"hello world");
}

/// Trailer headers after the 0-chunk are consumed and discarded; the
/// connection stays usable (keep-alive across a chunked request, with the
/// next request pipelined in the same packet — exercises the raw-consumed
/// accounting).
#[test]
fn chunked_request_trailers_and_pipelining() {
    let port = spawn_server(echo_pipeline());
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(
        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
          3\r\nabc\r\n0\r\nx-trailer: ignored\r\n\r\n\
          GET / HTTP/1.1\r\nHost: x\r\nConnection: close\r\n\r\n",
    )
    .unwrap();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap();
    let text = String::from_utf8_lossy(&resp);
    // Two responses back to back: the echo, then hello.
    assert!(text.contains("abc"), "echo body missing: {text}");
    let second = &text[text.find("abc").unwrap()..];
    assert!(second.contains("hello"), "pipelined second response missing: {text}");
}

/// A chunked body whose decoded size exceeds max_body_bytes is rejected
/// with 413 the moment the limit would be crossed.
#[test]
fn chunked_request_over_limit_413() {
    let port = free_port();
    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
    let pipe = echo_pipeline();
    std::thread::spawn(move || {
        let cfg = Config {
            listener_pool:     2,
            scheduler_threads: Some(2),
            max_body_bytes:    8, // tiny
            ..Config::new(addr)
        };
        serve_with(cfg, pipe).unwrap();
    });
    for _ in 0..50 {
        if TcpStream::connect(addr).is_ok() { break; }
        std::thread::sleep(Duration::from_millis(50));
    }
    let mut s = TcpStream::connect(addr).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(
        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
          9\r\n123456789\r\n0\r\n\r\n",
    )
    .unwrap();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap();
    assert_eq!(http_status(&resp), 413);
}

/// Garbage where a chunk-size line should be is 400.
#[test]
fn chunked_request_malformed_size_400() {
    let port = spawn_server(echo_pipeline());
    let resp = send_request(
        port,
        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n\
          zz\r\nhello\r\n0\r\n\r\n",
    );
    assert_eq!(http_status(&resp), 400);
}

/// Content-Length + Transfer-Encoding: chunked together is the smuggling
/// ambiguity: rejected 400 at parse time.
#[test]
fn chunked_plus_content_length_400() {
    let port = spawn_server(echo_pipeline());
    let resp = send_request(
        port,
        b"POST /echo HTTP/1.1\r\nHost: x\r\nContent-Length: 5\r\nTransfer-Encoding: chunked\r\n\r\n\
          5\r\nhello\r\n0\r\n\r\n",
    );
    assert_eq!(http_status(&resp), 400);
}

/// A chunked body that stalls mid-stream is killed by the request
/// deadline: the connection just closes (no response owed mid-body).
#[test]
fn chunked_request_stall_killed_at_request_timeout() {
    let port = spawn_server_with_timeouts(
        echo_pipeline(),
        Duration::from_secs(30),
        Duration::from_millis(400), // request_timeout
    );
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    // Head + first chunk, then stall forever before the 0-chunk.
    s.write_all(
        b"POST /echo HTTP/1.1\r\nHost: x\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n",
    )
    .unwrap();
    let start = std::time::Instant::now();
    let mut resp = Vec::new();
    s.read_to_end(&mut resp).unwrap(); // server closes; EOF
    assert!(resp.is_empty(), "expected silent close, got: {:?}", String::from_utf8_lossy(&resp));
    assert!(start.elapsed() < Duration::from_secs(3), "close took too long");
}

// ---------------------------------------------------------------------------
// SSE (v0.3 chunk 3)
// ---------------------------------------------------------------------------

/// Conn::sse(): the response is a chunked text/event-stream; events sent
/// through the EventSender arrive framed; the stream ends (0-chunk) when
/// the producer drops the sender.
#[test]
fn sse_events_round_trip() {
    let pipe = Pipeline::new().plug(Router::new().get(
        "/events",
        |c: Conn, _n: Next| {
            let (c, events) = c.sse();
            smarm::spawn(move || {
                let _ = events.send("tick", "1");
                let _ = events.data("plain");
                smarm::sleep(Duration::from_millis(10));
                // events drops here: end of stream.
            });
            c
        },
    ));
    let port = spawn_server(pipe);
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();

    let (head, body) = read_chunked_response(&mut s);
    assert!(head.contains("content-type: text/event-stream"), "head: {head}");
    assert!(head.contains("transfer-encoding: chunked"), "head: {head}");
    let body = String::from_utf8(body).unwrap();
    assert!(body.contains("event: tick\ndata: 1\n\n"), "body: {body}");
    assert!(body.contains("data: plain\n\n"), "body: {body}");
}

/// With a silent producer, the conn actor emits keep-alive comment chunks
/// at the configured heartbeat interval.
#[test]
fn sse_heartbeats_on_silence() {
    let pipe = Pipeline::new().plug(Router::new().get(
        "/events",
        |c: Conn, _n: Next| {
            let (c, events) = c.sse_with_heartbeat(Duration::from_millis(100));
            smarm::spawn(move || {
                // Hold the sender open, silently, then end the stream.
                smarm::sleep(Duration::from_millis(450));
                drop(events);
            });
            c
        },
    ));
    let port = spawn_server(pipe);
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(b"GET /events HTTP/1.1\r\nHost: x\r\n\r\n").unwrap();

    let (_head, body) = read_chunked_response(&mut s);
    let body = String::from_utf8(body).unwrap();
    let beats = body.matches(": keep-alive").count();
    assert!(beats >= 2, "expected >=2 heartbeats in 450ms at 100ms interval, got {beats}: {body:?}");
}

// ---------------------------------------------------------------------------
// WebSocket handshake (v0.4 chunk 1). Duplex lands in chunk 3 — for now an
// accepted upgrade writes the 101 and the actor leaves the HTTP loop, which
// closes the socket; the tests assert the wire artefact, not ws traffic.
// ---------------------------------------------------------------------------

const WS_HANDSHAKE: &[u8] =
    b"GET /ws HTTP/1.1\r\nHost: x\r\nUpgrade: websocket\r\nConnection: Upgrade\r\n\
      Sec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\nSec-WebSocket-Version: 13\r\n\r\n";

/// Echo handler: data messages come straight back; the text "bye" makes
/// the SERVER initiate the close handshake (1000/"goodbye").
struct Echo;

impl urus::WsHandler for Echo {
    fn on_message(&mut self, msg: urus::Message, sender: &urus::WsSender) {
        if matches!(&msg, urus::Message::Text(t) if t == "bye") {
            let _ = sender.close(1000, "goodbye");
            return;
        }
        let _ = sender.send(msg);
    }
}

fn ws_pipeline() -> Pipeline {
    Pipeline::new().plug(Router::new().get("/ws", |c: Conn, _n: Next| c.upgrade(Echo)))
}

// ----- client-side helpers (the test IS the ws client) -----

use urus::ws::frame::{self as wsframe, Frame, Opcode};

/// Complete the handshake on a fresh socket; panics on anything but 101.
fn ws_connect(port: u16) -> TcpStream {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();
    s.write_all(WS_HANDSHAKE).unwrap();
    let head = String::from_utf8(read_response_head(&mut s)).unwrap();
    assert!(head.starts_with("HTTP/1.1 101"), "handshake: {head}");
    s
}

fn ws_send(s: &mut TcpStream, frame: &Frame) {
    s.write_all(&wsframe::encode_masked(frame, [0x11, 0x22, 0x33, 0x44]))
        .unwrap();
}

/// Read one server frame (server frames are unmasked). `buf` carries
/// partial bytes across calls.
fn ws_read_frame(s: &mut TcpStream, buf: &mut Vec<u8>) -> Frame {
    loop {
        if let Some((f, consumed)) =
            wsframe::decode(buf, false, 16 * 1024 * 1024).expect("bad server frame")
        {
            buf.drain(..consumed);
            return f;
        }
        let mut tmp = [0u8; 4096];
        let n = s.read(&mut tmp).expect("read mid-frame");
        assert!(n > 0, "EOF mid-frame; buffered: {buf:?}");
        buf.extend_from_slice(&tmp[..n]);
    }
}

/// Assert the socket yields EOF (the server closed).
fn ws_expect_eof(s: &mut TcpStream, buf: &[u8]) {
    assert!(buf.is_empty(), "unconsumed server bytes at EOF check: {buf:?}");
    let mut rest = Vec::new();
    let _ = s.read_to_end(&mut rest); // EOF or reset; both are closed
    assert!(rest.is_empty(), "unexpected bytes before EOF: {rest:?}");
}

#[test]
fn ws_echo_roundtrip_text_and_binary() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    ws_send(&mut s, &Frame::new(Opcode::Text, "hello"));
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Text);
    assert_eq!(f.payload, b"hello");

    ws_send(&mut s, &Frame::new(Opcode::Binary, vec![0u8, 159, 146, 150]));
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Binary);
    assert_eq!(f.payload, vec![0u8, 159, 146, 150]);
}

/// A first frame pipelined in the SAME write as the handshake must be
/// served: bytes read past the 101 request head carry into the duplex
/// loop (the buf-handover requirement).
#[test]
fn ws_pipelined_first_frame_served() {
    let port = spawn_server(ws_pipeline());
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(5))).unwrap();

    let mut wire = WS_HANDSHAKE.to_vec();
    wire.extend_from_slice(&wsframe::encode_masked(
        &Frame::new(Opcode::Text, "early"),
        [9, 9, 9, 9],
    ));
    s.write_all(&wire).unwrap();

    let head = String::from_utf8(read_response_head(&mut s)).unwrap();
    assert!(head.starts_with("HTTP/1.1 101"), "{head}");
    let mut buf = Vec::new();
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.payload, b"early");
}

/// §5.5.2: a ping is answered with a pong echoing the payload, with no
/// handler involvement.
#[test]
fn ws_ping_gets_pong_with_payload() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    ws_send(&mut s, &Frame::new(Opcode::Ping, "abc"));
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Pong);
    assert_eq!(f.payload, b"abc");

    // The connection is still a working websocket afterwards.
    ws_send(&mut s, &Frame::new(Opcode::Text, "still here"));
    assert_eq!(ws_read_frame(&mut s, &mut buf).payload, b"still here");
}

/// Client-initiated close: the server echoes the status code (§5.5.1)
/// and closes the TCP connection (§7.1.1 — server closes first).
#[test]
fn ws_client_close_is_echoed_then_eof() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    ws_send(
        &mut s,
        &Frame::new(Opcode::Close, wsframe::close_payload(1000, "done")),
    );
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Close);
    let (code, _) = wsframe::parse_close_payload(&f.payload).unwrap().unwrap();
    assert_eq!(code, 1000);
    ws_expect_eof(&mut s, &buf);
}

/// Server-initiated close via WsSender::close: the close frame carries
/// our code/reason; after the client echoes, the socket closes.
#[test]
fn ws_server_initiated_close_handshake() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    ws_send(&mut s, &Frame::new(Opcode::Text, "bye"));
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Close);
    let (code, reason) = wsframe::parse_close_payload(&f.payload).unwrap().unwrap();
    assert_eq!(code, 1000);
    assert_eq!(reason, "goodbye");

    // Echo the close; server completes the handshake and drops the fd.
    ws_send(
        &mut s,
        &Frame::new(Opcode::Close, wsframe::close_payload(1000, "")),
    );
    ws_expect_eof(&mut s, &buf);
}

/// A protocol violation (unmasked client frame) starts the close
/// handshake with 1002.
#[test]
fn ws_unmasked_client_frame_closes_1002() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    // Server-style (unmasked) encoding from the client = violation.
    s.write_all(&Frame::new(Opcode::Text, "naughty").encode()).unwrap();
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Close);
    let (code, _) = wsframe::parse_close_payload(&f.payload).unwrap().unwrap();
    assert_eq!(code, 1002);

    // Complete the handshake so the server returns inside the test budget.
    ws_send(
        &mut s,
        &Frame::new(Opcode::Close, wsframe::close_payload(1002, "")),
    );
    ws_expect_eof(&mut s, &buf);
}

/// A frame whose HEADER already exceeds max_frame_payload is rejected
/// before any payload is buffered: 1009 with no payload bytes sent.
#[test]
fn ws_oversize_frame_header_closes_1009() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    // fin+text, masked, 64-bit length claiming 2 MiB (cap is 1 MiB).
    let header: &[u8] = &[
        0x81, 0xFF, 0, 0, 0, 0, 0, 0x20, 0, 0, // len = 0x200000
        9, 9, 9, 9, // mask key
    ];
    s.write_all(header).unwrap();
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Close);
    let (code, _) = wsframe::parse_close_payload(&f.payload).unwrap().unwrap();
    assert_eq!(code, 1009);

    ws_send(
        &mut s,
        &Frame::new(Opcode::Close, wsframe::close_payload(1009, "")),
    );
    ws_expect_eof(&mut s, &buf);
}

/// Fragmented text reassembles into one message; control frames may
/// interleave mid-fragmentation (§5.4) without disturbing assembly.
#[test]
fn ws_fragmented_message_with_interleaved_ping() {
    let port = spawn_server(ws_pipeline());
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    let first = Frame { fin: false, opcode: Opcode::Text, payload: b"hel".to_vec() };
    ws_send(&mut s, &first);
    ws_send(&mut s, &Frame::new(Opcode::Ping, "mid"));
    let cont = Frame { fin: true, opcode: Opcode::Continuation, payload: b"lo".to_vec() };
    ws_send(&mut s, &cont);

    // Pong for the interleaved ping arrives first (written inline on
    // receipt), then the reassembled echo.
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Pong);
    assert_eq!(f.payload, b"mid");
    let f = ws_read_frame(&mut s, &mut buf);
    assert_eq!(f.opcode, Opcode::Text);
    assert_eq!(f.payload, b"hello");
}

/// An open WebSocket is in-flight work: graceful shutdown waits for the
/// drain deadline, then force-stops the conn actor parked in the duplex
/// select. serve returns; the client sees the socket drop.
#[test]
fn shutdown_force_stops_open_ws() {
    let (port, handle, done_rx) =
        spawn_server_with_handle(ws_pipeline(), Duration::from_millis(300));
    let mut s = ws_connect(port);
    let mut buf = Vec::new();

    // Prove the duplex loop is live before shutting down.
    ws_send(&mut s, &Frame::new(Opcode::Text, "alive?"));
    assert_eq!(ws_read_frame(&mut s, &mut buf).payload, b"alive?");

    handle.shutdown();
    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return: open ws not force-stopped at drain deadline");

    let mut rest = Vec::new();
    let _ = s.read_to_end(&mut rest); // EOF or reset; both fine
}

// ---------------------------------------------------------------------------
// v0.5: pubsub wired to ws (the chat shape) — see examples/ws_chat.rs
// ---------------------------------------------------------------------------

/// Minimal chat handler mirroring the example: on_open subscribes (conn
/// actor pid) + spawns a relay holding ONLY the Receiver and a WsSender
/// clone; on_message broadcast_from's, skipping the sender's own relay.
struct Chat {
    bus: std::sync::Arc<std::sync::OnceLock<urus::PubSub<String>>>,
}

impl urus::WsHandler for Chat {
    fn on_open(&mut self, sender: &urus::WsSender) {
        let bus = self.bus.get_or_init(urus::PubSub::new);
        let rx = bus.subscribe("room").unwrap();
        let _ = bus.broadcast_from(smarm::self_pid(), "room", "joined".into());
        let out = sender.clone();
        smarm::spawn(move || {
            while let Ok(msg) = rx.recv() {
                if out.send(urus::Message::Text((*msg).clone())).is_err() {
                    break;
                }
            }
        });
    }

    fn on_message(&mut self, msg: urus::Message, sender: &urus::WsSender) {
        if let urus::Message::Text(t) = msg {
            // "sync" is the join-ack: a direct (non-broadcast) reply
            // proving on_open — and therefore this client's
            // subscription — completed. The tests need it because the
            // 101 reaches the client BEFORE the conn actor runs
            // on_open; without an ack, "subscribed yet?" is a race.
            if t == "sync" {
                let _ = sender.send(urus::Message::Text("synced".into()));
                return;
            }
            let bus = self.bus.get_or_init(urus::PubSub::new);
            let _ = bus.broadcast_from(smarm::self_pid(), "room", t);
        }
    }
}

fn chat_pipeline() -> Pipeline {
    let bus: std::sync::Arc<std::sync::OnceLock<urus::PubSub<String>>> =
        std::sync::Arc::new(std::sync::OnceLock::new());
    Pipeline::new().plug(Router::new().get("/ws", move |c: Conn, _n: Next| {
        let bus = bus.clone();
        c.upgrade(Chat { bus })
    }))
}

/// Two clients in a room: a broadcast reaches the other client and (via
/// broadcast_from) never echoes to the sender. The on_open hook is what
/// makes the listen-only direction work at all — B receives without
/// having sent a single frame when A's first message arrives.
#[test]
fn ws_chat_broadcast_reaches_other_client_not_sender() {
    let port = spawn_server(chat_pipeline());

    let mut a = ws_connect(port);
    let mut a_buf = Vec::new();
    // Join-ack A before B connects: A's sub must predate B's join
    // notice for the read sequence below to be deterministic.
    ws_send(&mut a, &Frame::new(Opcode::Text, "sync"));
    assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"synced");

    let mut b = ws_connect(port);
    let mut b_buf = Vec::new();
    ws_send(&mut b, &Frame::new(Opcode::Text, "sync"));
    assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"synced");

    // A sees B's join notice — proves A's relay is live and B's
    // subscription (made in on_open, before any client frame) is in.
    assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"joined");

    // A speaks; B receives it as its FIRST broadcast frame.
    ws_send(&mut a, &Frame::new(Opcode::Text, "hello"));
    assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"hello");

    // The no-echo proof without timeout reads: B speaks, and A's next
    // frame must be B's message — if A had been echoed its own "hello",
    // that would arrive first.
    ws_send(&mut b, &Frame::new(Opcode::Text, "yo"));
    assert_eq!(
        ws_read_frame(&mut a, &mut a_buf).payload,
        b"yo",
        "sender was echoed its own broadcast_from message"
    );
}

/// THE structural test for the v0.5 shutdown chain: with two open chat
/// sockets, live relays, and a live pubsub table, graceful shutdown must
/// still terminate. Chain under test: drain force-stops conn actors →
/// monitors prune subscriptions (relay senders drop, relays' recv errs)
/// → listeners stop → last Arc<Pipeline> drops the non-static bus cell
/// in-runtime → table inbox closes → table exits → AllDone → serve
/// returns. A static bus, or a relay holding a PubSub clone, hangs here.
#[test]
fn shutdown_with_open_chat_terminates() {
    let (port, handle, done_rx) =
        spawn_server_with_handle(chat_pipeline(), Duration::from_millis(300));

    let mut a = ws_connect(port);
    let mut a_buf = Vec::new();
    ws_send(&mut a, &Frame::new(Opcode::Text, "sync"));
    assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"synced");

    let mut b = ws_connect(port);
    let mut b_buf = Vec::new();
    ws_send(&mut b, &Frame::new(Opcode::Text, "sync"));
    assert_eq!(ws_read_frame(&mut b, &mut b_buf).payload, b"synced");

    // Prove the full pubsub path is live before shutting down.
    assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"joined");
    ws_send(&mut b, &Frame::new(Opcode::Text, "pre-shutdown"));
    assert_eq!(ws_read_frame(&mut a, &mut a_buf).payload, b"pre-shutdown");

    handle.shutdown();
    done_rx
        .recv_timeout(Duration::from_secs(5))
        .expect("serve did not return: pubsub table or a relay outlived the drain");

    let mut rest = Vec::new();
    let _ = a.read_to_end(&mut rest);
    let _ = b.read_to_end(&mut rest);
}