qrusty 0.25.2

// tests/ws_tests.rs
//
// Integration tests for the WebSocket API (WS-0001 through WS-0016, WS-0020 through WS-0026).
// Each test spins up an in-process server bound to a random OS-assigned port,
// then connects via tokio-tungstenite so we exercise the full Axum WS stack.

use futures_util::{SinkExt, StreamExt};
use qrusty::{api::ApiServer, storage::Storage};
use reqwest::Client;
use serde_json::{json, Value};
use std::sync::Arc;
use tempfile::TempDir;
use tokio::net::TcpListener;
use tokio::time::{timeout, Duration};
use tokio_tungstenite::{connect_async, tungstenite::Message as TungsteniteMessage};

// ---------------------------------------------------------------------------
// Test helpers
// ---------------------------------------------------------------------------

/// Start a throwaway server; returns (ws_base, http_base, TempDir).
/// The TempDir must be kept alive for the duration of the test.
async fn start_test_server() -> (String, String, TempDir) {
    let temp_dir = TempDir::new().unwrap();
    let storage = Arc::new(Storage::new(temp_dir.path().to_str().unwrap()).unwrap());
    let api = ApiServer::new(storage);
    let router = api
        .router()
        .layer(tower_http::cors::CorsLayer::permissive());
    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });
    (
        format!("ws://127.0.0.1:{}", port),
        format!("http://127.0.0.1:{}", port),
        temp_dir,
    )
}

type WsStream =
    tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>;

async fn ws_connect(ws_base: &str) -> WsStream {
    let url = format!("{}/ws", ws_base);
    let (ws, _) = connect_async(&url).await.unwrap();
    ws
}

async fn send(ws: &mut WsStream, frame: Value) {
    ws.send(TungsteniteMessage::Text(frame.to_string().into()))
        .await
        .unwrap();
}

/// Receive the next text frame and parse as JSON, skipping Ping/Pong frames.
/// Panics if nothing arrives within 5 seconds.
async fn recv(ws: &mut WsStream) -> Value {
    timeout(Duration::from_secs(5), async {
        loop {
            let msg = ws.next().await.expect("stream ended").expect("ws error");
            match msg {
                TungsteniteMessage::Text(t) => {
                    return serde_json::from_str(&t).expect("bad JSON");
                }
                TungsteniteMessage::Ping(_) | TungsteniteMessage::Pong(_) => continue,
                other => panic!("expected text frame, got {:?}", other),
            }
        }
    })
    .await
    .expect("recv timeout")
}

/// Create a queue via the HTTP API.
async fn http_create_queue(http_base: &str, name: &str) {
    Client::new()
        .post(format!("{}/create-queue", http_base))
        .json(&json!({"name": name, "config": {"ordering": "MaxFirst"}}))
        .send()
        .await
        .unwrap()
        .error_for_status()
        .unwrap();
}

/// Publish a message via the HTTP API; returns the assigned id.
async fn http_publish(http_base: &str, queue: &str, payload: &str) -> String {
    let resp = Client::new()
        .post(format!("{}/publish", http_base))
        .json(&json!({"queue": queue, "priority": 0, "payload": payload}))
        .send()
        .await
        .unwrap()
        .error_for_status()
        .unwrap();
    let body: Value = resp.json().await.unwrap();
    body["id"].as_str().unwrap().to_string()
}

// ---------------------------------------------------------------------------
// WS-0001 – connection upgrade
// ---------------------------------------------------------------------------

// Verifies: WS-0001, SYS-0010
#[tokio::test]
async fn test_ws_connection_upgrade() {
    let (ws_base, _http, _dir) = start_test_server().await;
    // Should connect without error.
    let mut ws = ws_connect(&ws_base).await;
    ws.close(None).await.unwrap();
}

// ---------------------------------------------------------------------------
// WS-0002 – framing: binary frames rejected with 1003
// ---------------------------------------------------------------------------

// Verifies: WS-0002
#[tokio::test]
async fn test_ws_binary_frame_rejected_with_1003() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    ws.send(TungsteniteMessage::Binary(b"raw bytes".to_vec().into()))
        .await
        .unwrap();

    let msg = timeout(Duration::from_secs(2), ws.next())
        .await
        .expect("timeout waiting for close")
        .expect("stream ended")
        .expect("ws error");

    match msg {
        TungsteniteMessage::Close(Some(frame)) => {
            assert_eq!(
                frame.code,
                tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Unsupported
            );
        }
        other => panic!("expected close frame, got {:?}", other),
    }
}

// ---------------------------------------------------------------------------
// WS-0002 – framing: invalid JSON rejected with 1007
// ---------------------------------------------------------------------------

// Verifies: WS-0002
#[tokio::test]
async fn test_ws_invalid_json_rejected_with_1007() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    ws.send(TungsteniteMessage::Text("{not valid json".into()))
        .await
        .unwrap();

    let msg = timeout(Duration::from_secs(2), ws.next())
        .await
        .expect("timeout waiting for close")
        .expect("stream ended")
        .expect("ws error");

    match msg {
        TungsteniteMessage::Close(Some(frame)) => {
            assert_eq!(
                frame.code,
                tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Invalid
            );
        }
        other => panic!("expected close frame, got {:?}", other),
    }
}

// ---------------------------------------------------------------------------
// WS-0003 / WS-0004 – req_id is echoed in ok response
// ---------------------------------------------------------------------------

// Verifies: WS-0003, WS-0004
#[tokio::test]
async fn test_ws_req_id_echoed_in_ok() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "echo-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({
            "type": "publish",
            "queue": "echo-q",
            "payload": "hello",
            "req_id": "my-request-42"
        }),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert_eq!(resp["req_id"], "my-request-42");
    assert!(resp["id"].is_string());
}

// ---------------------------------------------------------------------------
// WS-0003 / WS-0004 – response omits req_id when not provided
// ---------------------------------------------------------------------------

// Verifies: WS-0003, WS-0004
#[tokio::test]
async fn test_ws_no_req_id_omitted_in_response() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "no-rid-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "publish", "queue": "no-rid-q", "payload": "x"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert!(resp.get("req_id").is_none(), "req_id should be absent");
}

// ---------------------------------------------------------------------------
// WS-0005 – error frame on unknown type keeps connection open
// ---------------------------------------------------------------------------

// Verifies: WS-0005
#[tokio::test]
async fn test_ws_unknown_type_returns_error_frame_and_stays_open() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "totally-unknown", "req_id": "r1"})).await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "r1");
    assert!(resp["code"].is_string());
    assert!(resp["message"].is_string());

    // Connection must still be usable.
    send(&mut ws, json!({"type": "totally-unknown"})).await;
    let resp2 = recv(&mut ws).await;
    assert_eq!(resp2["type"], "error");
}

// ---------------------------------------------------------------------------
// WS-0006 – publish success
// ---------------------------------------------------------------------------

// Verifies: WS-0006
#[tokio::test]
async fn test_ws_publish_success() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "pub-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "publish", "queue": "pub-q", "payload": "world", "priority": 5}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert!(resp["id"].is_string(), "response should include message id");
}

// ---------------------------------------------------------------------------
// WS-0006 – publish to non-existent queue → error frame, no close
// ---------------------------------------------------------------------------

// Verifies: WS-0006
#[tokio::test]
async fn test_ws_publish_nonexistent_queue_error() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "publish", "queue": "ghost-q", "payload": "x", "req_id": "r99"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "r99");
    assert!(resp["code"].is_string());

    // Connection must still be alive.
    send(&mut ws, json!({"type": "totally-unknown"})).await;
    let still_open = recv(&mut ws).await;
    assert_eq!(still_open["type"], "error");
}

// ---------------------------------------------------------------------------
// WS-0007 / WS-0008 – subscribe then receive delivery frame
// ---------------------------------------------------------------------------

// Verifies: WS-0007, WS-0008
#[tokio::test]
async fn test_ws_subscribe_and_deliver() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "sub-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Subscribe.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "sub-q", "req_id": "sub1"}),
    )
    .await;
    let sub_resp = recv(&mut ws).await;
    assert_eq!(sub_resp["type"], "ok");
    assert_eq!(sub_resp["req_id"], "sub1");

    // Publish a message so the poller picks it up.
    http_publish(&http_base, "sub-q", "hello-ws").await;

    // Wait for the delivery frame (poller may take up to ~100ms).
    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let frame = recv(&mut ws).await;
            if frame["type"] == "deliver" {
                return frame;
            }
        }
    })
    .await
    .expect("did not receive delivery frame in time");

    assert_eq!(delivery["type"], "deliver");
    assert_eq!(delivery["queue"], "sub-q");
    assert_eq!(delivery["payload"], "hello-ws");
    assert!(delivery["id"].is_string());
    assert!(delivery["priority"].is_number());
    assert!(delivery["created_at"].is_string());
    // Delivery frames must NOT carry req_id (WS-0008).
    assert!(
        delivery.get("req_id").is_none(),
        "delivery frame must not include req_id"
    );
}

// ---------------------------------------------------------------------------
// WS-0007 – subscribe to missing queue → error frame
// ---------------------------------------------------------------------------

// Verifies: WS-0007
#[tokio::test]
async fn test_ws_subscribe_nonexistent_queue_error() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "no-such-q", "req_id": "s1"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "s1");
}

// ---------------------------------------------------------------------------
// WS-0009 – unsubscribe stops delivery
// ---------------------------------------------------------------------------

// Verifies: WS-0009
#[tokio::test]
async fn test_ws_unsubscribe_stops_delivery() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "unsub-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "queue": "unsub-q"})).await;
    let _ = recv(&mut ws).await; // ok

    send(
        &mut ws,
        json!({"type": "unsubscribe", "queue": "unsub-q", "req_id": "u1"}),
    )
    .await;
    let unsub_resp = recv(&mut ws).await;
    assert_eq!(unsub_resp["type"], "ok");
    assert_eq!(unsub_resp["req_id"], "u1");

    // Publish; should NOT receive a delivery frame since we unsubscribed.
    http_publish(&http_base, "unsub-q", "should-not-arrive").await;

    let no_delivery = timeout(Duration::from_millis(400), ws.next()).await;
    // Expect a timeout (None), meaning nothing arrived.
    assert!(
        no_delivery.is_err(),
        "should not have received a frame after unsubscribe"
    );
}

// ---------------------------------------------------------------------------
// WS-0010 – multi-queue subscriptions on one connection
// ---------------------------------------------------------------------------

// Verifies: WS-0010
#[tokio::test]
async fn test_ws_multi_queue_subscribe() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "mq-alpha").await;
    http_create_queue(&http_base, "mq-beta").await;
    let mut ws = ws_connect(&ws_base).await;

    // Subscribe to both queues.
    send(&mut ws, json!({"type": "subscribe", "queue": "mq-alpha"})).await;
    let _ = recv(&mut ws).await; // ok
    send(&mut ws, json!({"type": "subscribe", "queue": "mq-beta"})).await;
    let _ = recv(&mut ws).await; // ok

    // Publish to each queue.
    http_publish(&http_base, "mq-alpha", "from-alpha").await;
    http_publish(&http_base, "mq-beta", "from-beta").await;

    // Collect two delivery frames (order may vary).
    let mut queues_seen = std::collections::HashSet::new();
    for _ in 0..2 {
        let frame = timeout(Duration::from_secs(3), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .expect("timeout waiting for delivery");
        queues_seen.insert(frame["queue"].as_str().unwrap().to_string());
    }

    assert!(
        queues_seen.contains("mq-alpha"),
        "missing mq-alpha delivery"
    );
    assert!(queues_seen.contains("mq-beta"), "missing mq-beta delivery");
}

// ---------------------------------------------------------------------------
// WS-0011 – ack removes message
// ---------------------------------------------------------------------------

// Verifies: WS-0011
#[tokio::test]
async fn test_ws_ack_removes_message() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "ack-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "queue": "ack-q"})).await;
    let _ = recv(&mut ws).await; // ok

    http_publish(&http_base, "ack-q", "ack-me").await;

    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .unwrap();

    let msg_id = delivery["id"].as_str().unwrap().to_string();

    send(
        &mut ws,
        json!({"type": "ack", "queue": "ack-q", "id": msg_id, "req_id": "a1"}),
    )
    .await;

    let ack_resp = recv(&mut ws).await;
    assert_eq!(ack_resp["type"], "ok");
    assert_eq!(ack_resp["req_id"], "a1");

    // Queue should now be empty.
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/ack-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(stats["total"], 0, "message should be gone after ack");
}

// ---------------------------------------------------------------------------
// WS-0012 – nack unlocks message
// ---------------------------------------------------------------------------

// Verifies: WS-0012
#[tokio::test]
async fn test_ws_nack_unlocks_message() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "nack-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "queue": "nack-q"})).await;
    let _ = recv(&mut ws).await; // ok

    http_publish(&http_base, "nack-q", "nack-me").await;

    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .unwrap();

    let msg_id = delivery["id"].as_str().unwrap().to_string();

    send(
        &mut ws,
        json!({"type": "nack", "queue": "nack-q", "id": msg_id, "req_id": "n1"}),
    )
    .await;

    // The subscriber is still active, so the server may re-deliver the
    // nacked message before (or instead of) the nack `ok` response.
    // Loop until we see the `ok` for req_id "n1", discarding any
    // interleaved `deliver` frames.
    let nack_resp = loop {
        let f = recv(&mut ws).await;
        if f["type"] == "ok" {
            break f;
        }
        assert_eq!(f["type"], "deliver", "unexpected frame: {f}");
    };
    assert_eq!(nack_resp["req_id"], "n1");

    // Unsubscribe so the server won't immediately re-deliver the nacked
    // message (which would lock it again and make "available" == 0).
    send(
        &mut ws,
        json!({"type": "unsubscribe", "queue": "nack-q", "req_id": "u1"}),
    )
    .await;
    // Drain until we see the unsubscribe ok (discarding any deliver frames).
    loop {
        let f = recv(&mut ws).await;
        if f["type"] == "ok" && f["req_id"] == "u1" {
            break;
        }
    }

    // Message should be unlocked (available).
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/nack-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert!(
        stats["available"].as_u64().unwrap_or(0) > 0,
        "message should be back in the queue after nack"
    );
}

// ---------------------------------------------------------------------------
// WS-0013 – batch-ack
// ---------------------------------------------------------------------------

// Verifies: WS-0013
#[tokio::test]
async fn test_ws_batch_ack() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "batch-ack-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Publish two messages.
    http_publish(&http_base, "batch-ack-q", "m1").await;
    http_publish(&http_base, "batch-ack-q", "m2").await;

    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "batch-ack-q"}),
    )
    .await;
    let _ = recv(&mut ws).await; // ok

    // Collect two delivery frames.
    let mut ids = vec![];
    for _ in 0..2 {
        let f = timeout(Duration::from_secs(3), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .unwrap();
        ids.push(f["id"].as_str().unwrap().to_string());
    }

    send(
        &mut ws,
        json!({"type": "batch-ack", "queue": "batch-ack-q", "ids": ids, "req_id": "ba1"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert_eq!(resp["req_id"], "ba1");
    assert_eq!(resp["acked"], 2, "both messages should be acked");
}

// ---------------------------------------------------------------------------
// WS-0014 – batch-nack
// ---------------------------------------------------------------------------

// Verifies: WS-0014
#[tokio::test]
async fn test_ws_batch_nack() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "batch-nack-q").await;
    let mut ws = ws_connect(&ws_base).await;

    http_publish(&http_base, "batch-nack-q", "n1").await;
    http_publish(&http_base, "batch-nack-q", "n2").await;

    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "batch-nack-q"}),
    )
    .await;
    let _ = recv(&mut ws).await; // ok

    let mut ids = vec![];
    for _ in 0..2 {
        let f = timeout(Duration::from_secs(3), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .unwrap();
        ids.push(f["id"].as_str().unwrap().to_string());
    }

    send(
        &mut ws,
        json!({"type": "batch-nack", "queue": "batch-nack-q", "ids": ids, "req_id": "bn1"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert_eq!(resp["req_id"], "bn1");
    // unlocked + dropped = total messages nacked
    let unlocked = resp["unlocked"].as_u64().unwrap_or(0);
    let dropped = resp["dropped"].as_u64().unwrap_or(0);
    assert_eq!(
        unlocked + dropped,
        2,
        "all messages should be accounted for"
    );
}

// ---------------------------------------------------------------------------
// WS-0016 – connection close releases subscriptions (no crash)
// ---------------------------------------------------------------------------

// Verifies: WS-0016
#[tokio::test]
async fn test_ws_connection_close_releases_subscriptions() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "close-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "queue": "close-q"})).await;
    let _ = recv(&mut ws).await; // ok

    // Explicitly close.
    ws.close(None).await.unwrap();

    // Brief pause so the server task can clean up.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Server should still be healthy (didn't crash).
    let status = Client::new()
        .get(format!("{}/health", http_base))
        .send()
        .await
        .unwrap()
        .status();
    assert_eq!(status, 200);
}

// ---------------------------------------------------------------------------
// WS-0005 – error frame echoes req_id
// ---------------------------------------------------------------------------

// Verifies: WS-0005
#[tokio::test]
async fn test_ws_error_frame_echoes_req_id() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "nonexistent-q", "req_id": "err-123"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "err-123");
    assert!(resp["code"].is_string());
    assert!(resp["message"].is_string());
}

// ---------------------------------------------------------------------------
// WS-0006 – publish missing required field → error frame
// ---------------------------------------------------------------------------

// Verifies: WS-0006
#[tokio::test]
async fn test_ws_publish_missing_queue_field() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    // No "queue" field in the publish frame.
    send(&mut ws, json!({"type": "publish", "payload": "oops"})).await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
}

// ---------------------------------------------------------------------------
// WS-0020 – renew extends the message lock
// ---------------------------------------------------------------------------

// Verifies: WS-0020
#[tokio::test]
async fn test_ws_renew_extends_lock() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "renew-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "queue": "renew-q"})).await;
    let _ = recv(&mut ws).await; // ok

    http_publish(&http_base, "renew-q", "renew-me").await;

    // Wait for delivery to get the locked message id.
    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .expect("did not receive delivery frame in time");

    let msg_id = delivery["id"].as_str().unwrap().to_string();

    // Renew the lock – should succeed with "ok".
    send(
        &mut ws,
        json!({"type": "renew", "queue": "renew-q", "id": msg_id, "req_id": "rn1"}),
    )
    .await;

    let renew_resp = recv(&mut ws).await;
    assert_eq!(renew_resp["type"], "ok");
    assert_eq!(renew_resp["req_id"], "rn1");

    // Connection must still be open.
    send(
        &mut ws,
        json!({"type": "ack", "queue": "renew-q", "id": msg_id}),
    )
    .await;
    let ack_resp = recv(&mut ws).await;
    assert_eq!(ack_resp["type"], "ok");
}

// Verifies: WS-0020
#[tokio::test]
async fn test_ws_renew_not_locked_returns_error() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "renew-err-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Publish a message but do not subscribe – message is unlocked.
    let msg_id = http_publish(&http_base, "renew-err-q", "not-locked").await;

    // Attempt renew on a message this connection doesn't hold a lock on.
    send(
        &mut ws,
        json!({"type": "renew", "queue": "renew-err-q", "id": msg_id, "req_id": "rn2"}),
    )
    .await;

    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["code"], "not_locked");
    assert_eq!(resp["req_id"], "rn2");

    // Connection must remain open after the error.
    send(&mut ws, json!({"type": "totally-unknown"})).await;
    let still_open = recv(&mut ws).await;
    assert_eq!(still_open["type"], "error");
}

// WS publish and ack operations must update the shared rate counters so that
// /queue-stats reports non-zero publish_rate_per_sec and ack_rate_per_sec.
// Verifies: WS-0021
#[tokio::test]
async fn test_ws_operations_update_rate_counters() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "rate-ws").await;

    // First /queue-stats call establishes the baseline snapshot (all zeros).
    let _: Value = Client::new()
        .get(format!("{}/queue-stats/rate-ws", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();

    let mut ws = ws_connect(&ws_base).await;

    // Publish 3 messages via the WebSocket API.
    for i in 0..3u64 {
        send(
            &mut ws,
            json!({"type": "publish", "queue": "rate-ws", "priority": i, "payload": format!("msg-{}", i)}),
        )
        .await;
        let resp = recv(&mut ws).await;
        assert_eq!(resp["type"], "ok", "publish frame {i} failed: {resp}");
    }

    // Subscribe so the server starts delivering.
    send(&mut ws, json!({"type": "subscribe", "queue": "rate-ws"})).await;
    let sub_resp = recv(&mut ws).await;
    assert_eq!(sub_resp["type"], "ok");

    // Receive all 3 deliveries and ack each one.
    // After sending an ack the next frame may be another delivery (subscriber
    // is racing), so we loop until we see the "ok" ack response.
    let mut pending_acks = 0u32;
    let mut acked = 0u32;
    let total = 3u32;

    while acked < total {
        let f = timeout(Duration::from_secs(5), recv(&mut ws))
            .await
            .expect("timed out waiting for deliver/ok frame");

        match f["type"].as_str().unwrap_or("") {
            "deliver" => {
                let msg_id = f["id"].as_str().unwrap().to_owned();
                send(
                    &mut ws,
                    json!({"type": "ack", "queue": "rate-ws", "id": msg_id}),
                )
                .await;
                pending_acks += 1;
            }
            "ok" => {
                assert!(pending_acks > 0, "unexpected ok with no pending acks");
                pending_acks -= 1;
                acked += 1;
            }
            other => panic!("unexpected frame type: {other} — {f}"),
        }
    }

    // Poll until the background sampler (1 Hz) picks up the activity.
    // Rates are stored in the metrics ring-buffer; the sampler fires within
    // ~1 second of seeding the baseline on first activity.
    let deadline = std::time::Instant::now() + Duration::from_secs(3);
    let mut pub_rate = 0.0f64;
    let mut ack_rate = 0.0f64;
    while std::time::Instant::now() < deadline {
        let stats: Value = Client::new()
            .get(format!("{}/queue-stats/rate-ws", http_base))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        pub_rate = stats["publish_rate_per_sec"].as_f64().unwrap_or(0.0);
        ack_rate = stats["ack_rate_per_sec"].as_f64().unwrap_or(0.0);
        if pub_rate > 0.0 && ack_rate > 0.0 {
            break;
        }
        tokio::time::sleep(Duration::from_millis(100)).await;
    }

    assert!(
        pub_rate > 0.0,
        "publish_rate_per_sec must be non-zero after WS publishes; got {pub_rate}"
    );
    assert!(
        ack_rate > 0.0,
        "ack_rate_per_sec must be non-zero after WS acks; got {ack_rate}"
    );
}

// ---------------------------------------------------------------------------
// WS-0024 – no_reply suppresses server response
// ---------------------------------------------------------------------------

// Verifies: WS-0024
#[tokio::test]
async fn test_ws_no_reply_suppresses_response() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "noreply-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Publish a message so we have something to ack.
    http_publish(&http_base, "noreply-q", "msg").await;

    // Subscribe, consume the ok and then wait for the delivery.
    send(&mut ws, json!({"type": "subscribe", "queue": "noreply-q"})).await;
    let _ = recv(&mut ws).await; // subscribe ok

    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .unwrap();
    let msg_id = delivery["id"].as_str().unwrap().to_string();

    // Unsubscribe first so no redelivery frames arrive.
    send(
        &mut ws,
        json!({"type": "unsubscribe", "queue": "noreply-q"}),
    )
    .await;
    let _ = recv(&mut ws).await; // unsubscribe ok

    // Ack with no_reply: true — server should NOT send a response.
    send(
        &mut ws,
        json!({
            "type": "ack",
            "queue": "noreply-q",
            "id": msg_id,
            "no_reply": true
        }),
    )
    .await;

    // Send a probe frame; its response should be the NEXT frame we receive
    // (i.e. no ack response appeared between the ack send and the probe).
    send(
        &mut ws,
        json!({"type": "totally-unknown", "req_id": "probe"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(
        resp["req_id"], "probe",
        "no_reply ack should not produce a response; got: {resp}"
    );

    // Verify the ack was actually processed.
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/noreply-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(
        stats["total"], 0,
        "message should be acked despite no_reply"
    );
}

// ---------------------------------------------------------------------------
// WS-0023 – credit-based flow control
// ---------------------------------------------------------------------------

// Verifies: WS-0023
#[tokio::test]
async fn test_ws_credit_flow_control() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "credit-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Publish 5 messages.
    for i in 0..5 {
        http_publish(&http_base, "credit-q", &format!("msg-{}", i)).await;
    }

    // Subscribe with credits=2.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "credit-q", "credits": 2, "req_id": "sub1"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert_eq!(resp["req_id"], "sub1");

    // Should receive exactly 2 deliveries.
    for _ in 0..2 {
        let f = timeout(Duration::from_secs(3), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .expect("expected delivery");
        assert_eq!(f["queue"], "credit-q");
    }

    // Should NOT receive a 3rd delivery within a reasonable time.
    let no_more = timeout(Duration::from_millis(500), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await;
    assert!(
        no_more.is_err(),
        "should not receive more than 2 deliveries"
    );

    // Replenish 2 credits.
    send(
        &mut ws,
        json!({"type": "credit", "queue": "credit-q", "credits": 2, "req_id": "cr1"}),
    )
    .await;
    let cr_resp = recv(&mut ws).await;
    assert_eq!(cr_resp["type"], "ok");
    assert_eq!(cr_resp["req_id"], "cr1");

    // Should now receive 2 more deliveries.
    for _ in 0..2 {
        let f = timeout(Duration::from_secs(3), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .expect("expected delivery after credit replenishment");
        assert_eq!(f["queue"], "credit-q");
    }
}

// Verifies: WS-0023
#[tokio::test]
async fn test_ws_subscribe_without_credits_is_unlimited() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "unlim-q").await;
    let mut ws = ws_connect(&ws_base).await;

    for i in 0..5 {
        http_publish(&http_base, "unlim-q", &format!("msg-{}", i)).await;
    }

    // Subscribe without credits field — unlimited delivery.
    send(&mut ws, json!({"type": "subscribe", "queue": "unlim-q"})).await;
    let _ = recv(&mut ws).await; // ok

    // Should receive all 5.
    for _ in 0..5 {
        let f = timeout(Duration::from_secs(5), async {
            loop {
                let f = recv(&mut ws).await;
                if f["type"] == "deliver" {
                    return f;
                }
            }
        })
        .await
        .expect("expected delivery");
        assert_eq!(f["queue"], "unlim-q");
    }
}

// Verifies: WS-0023
#[tokio::test]
async fn test_ws_credit_on_unsubscribed_queue_returns_error() {
    let (ws_base, _http_base, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "credit", "queue": "no-such-sub", "credits": 5, "req_id": "c1"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["code"], "not_subscribed");
}

// ---------------------------------------------------------------------------
// WS-0025 – public send_frame via WsSession
// ---------------------------------------------------------------------------

// Verifies: WS-0025
#[tokio::test]
async fn test_ws_session_send_frame_public() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "sf-q").await;
    let session = qrusty_client::ws::WsSession::connect(&format!(
        "ws://127.0.0.1:{}",
        extract_port(&ws_base)
    ))
    .await
    .unwrap();

    // send_frame is public and should succeed (fire-and-forget publish).
    session
        .send_frame(json!({
            "type": "publish",
            "queue": "sf-q",
            "payload": "raw-frame",
            "priority": 0,
        }))
        .await
        .unwrap();

    // Give the server a moment to process.
    tokio::time::sleep(Duration::from_millis(200)).await;

    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/sf-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(
        stats["total"], 1,
        "send_frame should have published a message"
    );

    session.close().await.unwrap();
}

// ---------------------------------------------------------------------------
// WS-0026 – request timeout via WsSession
// ---------------------------------------------------------------------------

// Verifies: WS-0026
#[tokio::test]
async fn test_ws_session_request_timeout() {
    // Spin up a raw TCP listener that accepts the WS handshake but never
    // sends any application-level response.  This guarantees a timeout.
    use tokio_tungstenite::accept_async;

    let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move {
        if let Ok((stream, _)) = listener.accept().await {
            // Complete the WebSocket handshake, then hold the connection.
            let _ws = accept_async(stream).await;
            tokio::time::sleep(Duration::from_secs(600)).await;
        }
    });

    let session = qrusty_client::ws::WsSession::connect_with_timeout(
        &format!("ws://127.0.0.1:{}", port),
        Duration::from_millis(100),
    )
    .await
    .unwrap();

    let result = session.publish("any-q", "hello", None).await;
    assert!(
        matches!(
            result,
            Err(qrusty_client::error::QrustyClientError::RequestTimeout(_))
        ),
        "expected RequestTimeout, got: {:?}",
        result,
    );
}

// ---------------------------------------------------------------------------
// WS-0024 – ack_noreply / nack_noreply via WsSession
// ---------------------------------------------------------------------------

// Verifies: WS-0024
#[tokio::test]
async fn test_ws_session_ack_noreply() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "noreply-client-q").await;

    let session = qrusty_client::ws::WsSession::connect(&format!(
        "ws://127.0.0.1:{}",
        extract_port(&ws_base)
    ))
    .await
    .unwrap();

    session
        .publish("noreply-client-q", "msg", None)
        .await
        .unwrap();
    let mut rx = session.subscribe("noreply-client-q").await.unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(3), rx.recv())
        .await
        .expect("delivery timeout")
        .expect("channel closed")
        .expect("delivery error");

    // ack_noreply should return immediately.
    session.ack_noreply(&msg.queue, &msg.id).await.unwrap();

    tokio::time::sleep(Duration::from_millis(200)).await;

    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/noreply-client-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(stats["total"], 0, "message should be acked via ack_noreply");
}

// ---------------------------------------------------------------------------
// WS-0017 – qrusty_client WsSession full public-API surface
// ---------------------------------------------------------------------------

// Exercises every WsSession method the requirement enumerates against a live
// server: connect, publish (returns id), subscribe (yields DeliveredMessage),
// ack, nack, batch_ack, batch_nack, unsubscribe, close — plus the
// DeliveredMessage struct fields.  Each call returning Ok witnesses that the
// method exists with the specified signature and round-trips correctly.
// Verifies: WS-0017
#[tokio::test]
async fn test_ws_session_full_api_round_trip() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "wsapi-q").await;

    let session = qrusty_client::ws::WsSession::connect(&format!(
        "ws://127.0.0.1:{}",
        extract_port(&ws_base)
    ))
    .await
    .unwrap();

    // publish returns the assigned message id.
    for i in 0..4 {
        let id = session
            .publish("wsapi-q", &format!("payload-{i}"), None)
            .await
            .unwrap();
        assert!(!id.is_empty(), "publish must return a non-empty message id");
    }

    // subscribe yields DeliveredMessage values; receiving locks each message.
    let mut rx = session.subscribe("wsapi-q").await.unwrap();
    let mut delivered = Vec::new();
    for _ in 0..4 {
        let msg = tokio::time::timeout(Duration::from_secs(5), rx.recv())
            .await
            .expect("delivery timeout")
            .expect("subscription channel closed")
            .expect("delivery error");
        delivered.push(msg);
    }

    // DeliveredMessage exposes queue / id / payload / priority / created_at.
    let first = &delivered[0];
    assert_eq!(first.queue, "wsapi-q");
    assert!(!first.id.is_empty());
    assert!(first.payload.starts_with("payload-"));
    let _ = &first.priority;
    let _ = &first.created_at;

    // Single ack and single nack.
    session
        .ack(&delivered[0].queue, &delivered[0].id)
        .await
        .unwrap();
    session
        .nack(&delivered[1].queue, &delivered[1].id)
        .await
        .unwrap();

    // batch_ack returns the count acked.
    let acked = session
        .batch_ack("wsapi-q", &[delivered[2].id.as_str()])
        .await
        .unwrap();
    assert_eq!(acked, 1, "batch_ack should ack the one locked message");

    // batch_nack returns (unlocked, dropped).
    let (unlocked, _dropped) = session
        .batch_nack("wsapi-q", &[delivered[3].id.as_str()])
        .await
        .unwrap();
    assert_eq!(
        unlocked, 1,
        "batch_nack should unlock the one locked message"
    );

    // unsubscribe and close complete the API surface.
    session.unsubscribe("wsapi-q").await.unwrap();
    session.close().await.unwrap();
}

// ---------------------------------------------------------------------------
// WS-0023 – WsSession credit API
// ---------------------------------------------------------------------------

// Verifies: WS-0023
#[tokio::test]
async fn test_ws_session_credit_flow_control() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "credit-session-q").await;

    let session = qrusty_client::ws::WsSession::connect(&format!(
        "ws://127.0.0.1:{}",
        extract_port(&ws_base)
    ))
    .await
    .unwrap();

    for i in 0..4 {
        session
            .publish("credit-session-q", &format!("msg-{}", i), None)
            .await
            .unwrap();
    }

    let mut rx = session
        .subscribe_with_credits("credit-session-q", Some(2))
        .await
        .unwrap();

    // Should receive 2.
    for _ in 0..2 {
        let msg = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .expect("delivery timeout")
            .expect("channel closed")
            .expect("delivery error");
        session.ack_noreply(&msg.queue, &msg.id).await.unwrap();
    }

    // Third should not arrive.
    let no_more = tokio::time::timeout(Duration::from_millis(500), rx.recv()).await;
    assert!(
        no_more.is_err(),
        "should not receive more than 2 deliveries"
    );

    // Grant 2 more credits.
    session.grant_credits("credit-session-q", 2).await.unwrap();

    // Should receive 2 more.
    for _ in 0..2 {
        let msg = tokio::time::timeout(Duration::from_secs(3), rx.recv())
            .await
            .expect("delivery timeout after credit grant")
            .expect("channel closed")
            .expect("delivery error");
        session.ack_noreply(&msg.queue, &msg.id).await.unwrap();
    }
}

// ---------------------------------------------------------------------------
// WS-0015 – server-side ping timeout closes with 1001
// ---------------------------------------------------------------------------

/// Perform a bare WebSocket handshake over a raw TCP stream and return it.
/// The caller controls all subsequent reads/writes — no auto-pong.
async fn raw_ws_upgrade(addr: &str) -> tokio::net::TcpStream {
    use tokio::io::{AsyncReadExt, AsyncWriteExt};
    let mut stream = tokio::net::TcpStream::connect(addr).await.unwrap();
    let req = format!(
        "GET /ws HTTP/1.1\r\nHost: {addr}\r\nUpgrade: websocket\r\n\
         Connection: Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n\
         Sec-WebSocket-Version: 13\r\n\r\n"
    );
    stream.write_all(req.as_bytes()).await.unwrap();

    // Drain HTTP response headers.
    let mut buf = Vec::new();
    let mut b = [0u8; 1];
    loop {
        stream.read_exact(&mut b).await.unwrap();
        buf.push(b[0]);
        if buf.ends_with(b"\r\n\r\n") {
            break;
        }
    }
    let headers = String::from_utf8_lossy(&buf);
    assert!(
        headers.contains("101"),
        "expected 101 Switching Protocols, got: {headers}"
    );
    stream
}

/// Read one unmasked WebSocket frame and return `(opcode, payload)`.
async fn read_raw_ws_frame(stream: &mut tokio::net::TcpStream) -> (u8, Vec<u8>) {
    use tokio::io::AsyncReadExt;
    let mut hdr = [0u8; 2];
    stream.read_exact(&mut hdr).await.unwrap();
    let opcode = hdr[0] & 0x0F;
    let len_byte = (hdr[1] & 0x7F) as usize;
    let payload_len = match len_byte {
        126 => {
            let mut e = [0u8; 2];
            stream.read_exact(&mut e).await.unwrap();
            u16::from_be_bytes(e) as usize
        }
        127 => {
            let mut e = [0u8; 8];
            stream.read_exact(&mut e).await.unwrap();
            u64::from_be_bytes(e) as usize
        }
        n => n,
    };
    let mut payload = vec![0u8; payload_len];
    if payload_len > 0 {
        stream.read_exact(&mut payload).await.unwrap();
    }
    (opcode, payload)
}

// Verifies: WS-0015
#[tokio::test]
async fn test_ws_ping_timeout_closes_with_1001() {
    // Use very short intervals so the test completes quickly.
    // Safety: these env vars are only read inside handle_connection() at
    // connection setup time; no other concurrent test uses them.
    #[allow(deprecated)]
    unsafe {
        std::env::set_var("WS_PING_INTERVAL_SECS", "1");
        std::env::set_var("WS_PING_TIMEOUT_SECS", "1");
    }

    let (ws_base, _http_base, _dir) = start_test_server().await;
    // Strip "ws://" prefix to get "host:port".
    let addr = ws_base.trim_start_matches("ws://");
    let mut stream = raw_ws_upgrade(addr).await;

    // Read frames without sending any pongs.  The server tracks last_pong
    // from connection time.  After ping_iv elapses with no pong received,
    // on the next tick (ping_iv + ping_to later) it closes with 1001 (AWAY).
    // With intervals of 1 s each that should happen by ~3 s; allow 8 s total.
    let close_code = tokio::time::timeout(Duration::from_secs(8), async {
        loop {
            let (opcode, payload) = read_raw_ws_frame(&mut stream).await;
            match opcode {
                0x8 => {
                    // Close frame — extract 2-byte status code.
                    return if payload.len() >= 2 {
                        u16::from_be_bytes([payload[0], payload[1]])
                    } else {
                        0u16
                    };
                }
                0x9 => {} // Ping — intentionally not responded to
                _ => {}   // other frames
            }
        }
    })
    .await
    .expect("timed out: server did not send Close within 8 s");

    assert_eq!(close_code, 1001, "expected close code 1001 (Going Away)");

    #[allow(deprecated)]
    unsafe {
        std::env::remove_var("WS_PING_INTERVAL_SECS");
        std::env::remove_var("WS_PING_TIMEOUT_SECS");
    }
}

// ---------------------------------------------------------------------------
// Helpers for WsSession integration tests
// ---------------------------------------------------------------------------

/// Extract the port from a base URL like "ws://127.0.0.1:12345".
fn extract_port(base: &str) -> u16 {
    base.rsplit(':').next().unwrap().parse().unwrap()
}

// ---------------------------------------------------------------------------
// Additional coverage tests
// ---------------------------------------------------------------------------

/// Start a server that has a LogBuffer attached (for subscribe-logs tests).
async fn start_test_server_with_logs() -> (String, String, qrusty::log_buffer::LogBuffer, TempDir) {
    let temp_dir = TempDir::new().unwrap();
    let storage = Arc::new(Storage::new(temp_dir.path().to_str().unwrap()).unwrap());
    let log_buffer = qrusty::log_buffer::LogBuffer::new();
    let api = ApiServer::new(storage).with_log_buffer(log_buffer.clone());
    let router = api
        .router()
        .layer(tower_http::cors::CorsLayer::permissive());
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();
    tokio::spawn(async move {
        axum::serve(listener, router).await.unwrap();
    });
    (
        format!("ws://127.0.0.1:{}", port),
        format!("http://127.0.0.1:{}", port),
        log_buffer,
        temp_dir,
    )
}

// Verifies: WS-0022
#[tokio::test]
async fn test_ws_subscribe_logs_delivers_live_entries() {
    let (ws_base, _http, log_buffer, _dir) = start_test_server_with_logs().await;
    let mut ws = ws_connect(&ws_base).await;

    // Push a history entry before subscribing so the history path is exercised.
    log_buffer.push_sync("INFO", "history-entry");

    send(&mut ws, json!({"type": "subscribe-logs", "req_id": "sl1"})).await;
    let resp = recv(&mut ws).await;
    assert_eq!(
        resp["type"], "ok",
        "subscribe-logs should return ok: {resp}"
    );

    // The history entry should arrive as a log frame immediately after subscribe.
    let log_frame = timeout(Duration::from_secs(2), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "log" {
                return f;
            }
        }
    })
    .await
    .expect("expected log frame from history");

    assert_eq!(log_frame["type"], "log");
    assert!(log_frame["message"].is_string());
    assert!(log_frame["timestamp"].is_string());
    assert!(log_frame["level"].is_string());

    // Push a live entry and verify it is forwarded too.
    log_buffer.push_sync("WARN", "live-entry");
    let live_frame = timeout(Duration::from_secs(2), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "log" && f["message"] == "live-entry" {
                return f;
            }
        }
    })
    .await
    .expect("expected live log frame");
    assert_eq!(live_frame["level"], "WARN");
}

// Regression: large history must not deadlock the WS main loop.
//
// Before the fix, handle_subscribe_logs performed the history replay inline
// via log_tx.send(entry).await. log_tx has capacity 256; the main select!
// loop drains log_rx from a different branch that cannot fire while
// handle_subscribe_logs is awaiting. With history > 256 entries, entry #256
// blocked forever and no log frames reached the client. This is what caused
// the UI Logs tab to appear "connected" but show nothing after navigating
// away and back.
//
// Verifies: WS-0022
#[tokio::test]
async fn test_ws_subscribe_logs_with_large_history_does_not_deadlock() {
    let (ws_base, _http, log_buffer, _dir) = start_test_server_with_logs().await;
    let mut ws = ws_connect(&ws_base).await;

    // Push well past log_tx capacity (256) — up to HISTORY_CAPACITY=1000.
    for i in 0..500 {
        log_buffer.push_sync("INFO", format!("hist-{i}"));
    }

    send(
        &mut ws,
        json!({"type": "subscribe-logs", "req_id": "sl-big"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");

    // We should be able to receive all 500 history entries without timing out.
    // Use a generous timeout because draining + serializing 500 frames takes
    // a moment, but far less than the 5 seconds we allow.
    let first = b"hist-0".to_vec();
    let last = b"hist-499".to_vec();
    let mut saw_first = false;
    let mut saw_last = false;
    timeout(Duration::from_secs(5), async {
        while !(saw_first && saw_last) {
            let f = recv(&mut ws).await;
            if f["type"] != "log" {
                continue;
            }
            let msg = f["message"].as_str().unwrap_or("");
            if msg.as_bytes() == first.as_slice() {
                saw_first = true;
            }
            if msg.as_bytes() == last.as_slice() {
                saw_last = true;
            }
        }
    })
    .await
    .expect("history replay stalled — handle_subscribe_logs deadlocked");

    // A fresh live event should still be forwarded after the bulk replay.
    log_buffer.push_sync("WARN", "after-large-history");
    timeout(Duration::from_secs(2), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "log" && f["message"] == "after-large-history" {
                return;
            }
        }
    })
    .await
    .expect("live entry after large-history replay did not arrive");
}

// Verifies: WS-0022
#[tokio::test]
async fn test_ws_unsubscribe_logs_stops_delivery() {
    let (ws_base, _http, log_buffer, _dir) = start_test_server_with_logs().await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe-logs"})).await;
    let _ = recv(&mut ws).await; // ok

    send(
        &mut ws,
        json!({"type": "unsubscribe-logs", "req_id": "ul1"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "ok");
    assert_eq!(resp["req_id"], "ul1");

    // Push a log entry after unsubscribing — should NOT arrive.
    log_buffer.push_sync("INFO", "after-unsub");
    let no_log = timeout(Duration::from_millis(300), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "log" {
                return f;
            }
        }
    })
    .await;
    assert!(
        no_log.is_err(),
        "should not receive log frames after unsubscribe"
    );
}

// Verifies: WS-0007
#[tokio::test]
async fn test_ws_subscribe_same_queue_twice_is_ok() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "dup-sub-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // First subscribe
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "dup-sub-q", "req_id": "s1"}),
    )
    .await;
    let r1 = recv(&mut ws).await;
    assert_eq!(r1["type"], "ok");

    // Second subscribe to the same queue — silently ignored, still returns ok.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "dup-sub-q", "req_id": "s2"}),
    )
    .await;
    let r2 = recv(&mut ws).await;
    assert_eq!(r2["type"], "ok", "duplicate subscribe should be idempotent");
    assert_eq!(r2["req_id"], "s2");
}

// Verifies: WS-0006
#[tokio::test]
async fn test_ws_publish_missing_payload_field() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "miss-payload-q").await;
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "publish", "queue": "miss-payload-q", "req_id": "r1"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "r1");
}

// Verifies: WS-0007
#[tokio::test]
async fn test_ws_subscribe_missing_queue_field() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"type": "subscribe", "req_id": "r2"})).await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "r2");
}

// Covers the dispatch "missing 'type' field" path.
#[tokio::test]
async fn test_ws_frame_missing_type_field_returns_error() {
    let (ws_base, _http, _dir) = start_test_server().await;
    let mut ws = ws_connect(&ws_base).await;

    send(&mut ws, json!({"req_id": "r3"})).await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "r3");
    assert_eq!(resp["code"], "invalid_request");
}

// Verifies that subscribe-logs returns an error when no log buffer is configured.
#[tokio::test]
async fn test_ws_subscribe_logs_without_buffer_returns_error() {
    let (ws_base, _http, _dir) = start_test_server().await; // no log buffer
    let mut ws = ws_connect(&ws_base).await;

    send(
        &mut ws,
        json!({"type": "subscribe-logs", "req_id": "sl-err"}),
    )
    .await;
    let resp = recv(&mut ws).await;
    assert_eq!(resp["type"], "error");
    assert_eq!(resp["req_id"], "sl-err");
}

// ---------------------------------------------------------------------------
// WS-0007 / WS-0020 – lock_timeout_secs on subscribe and renew
// ---------------------------------------------------------------------------

/// Subscribe with a custom lock_timeout_secs (120s).  Verify the delivered
/// message has a lock expiry far beyond the default 30 seconds, proving
/// the lock_timeout_secs was honoured.
// Verifies: WS-0007
#[tokio::test]
async fn test_ws_subscribe_with_lock_timeout_secs() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "long-lock-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Subscribe with a 120-second lock.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "long-lock-q", "lock_timeout_secs": 120}),
    )
    .await;
    let sub_resp = recv(&mut ws).await;
    assert_eq!(sub_resp["type"], "ok");

    http_publish(&http_base, "long-lock-q", "long-lock-payload").await;

    // Receive delivery.
    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .expect("did not receive delivery frame in time");

    let msg_id = delivery["id"].as_str().unwrap().to_string();

    // The message should be locked with a timeout well beyond 30s.
    // Consume the same queue via HTTP to inspect the locked_until field.
    // Since the message is already locked, HTTP consume should return no
    // message (the only message is locked).
    let consume_resp: Value = Client::new()
        .post(format!("{}/consume/long-lock-q", http_base))
        .json(&json!({"consumer_id": "probe", "timeout_seconds": 5}))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    // No message available — confirming the only message is locked.
    assert!(
        consume_resp.get("id").is_none(),
        "message should be locked and not consumable by another consumer"
    );

    // Stats should show locked=1.
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/long-lock-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(
        stats["locked"], 1,
        "message should be locked after delivery"
    );

    // Ack to clean up.
    send(
        &mut ws,
        json!({"type": "ack", "queue": "long-lock-q", "id": msg_id}),
    )
    .await;
    let ack_resp = recv(&mut ws).await;
    assert_eq!(ack_resp["type"], "ok");
}

/// Subscribe with default lock, then renew with a custom lock_timeout_secs.
/// The renewed lock should use the specified timeout, not the default 30s.
// Verifies: WS-0020
#[tokio::test]
async fn test_ws_renew_with_lock_timeout_secs() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "renew-timeout-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Subscribe with default lock.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "renew-timeout-q"}),
    )
    .await;
    let _ = recv(&mut ws).await; // ok

    http_publish(&http_base, "renew-timeout-q", "renew-timeout-payload").await;

    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .expect("did not receive delivery frame in time");

    let msg_id = delivery["id"].as_str().unwrap().to_string();

    // Renew with a custom timeout (120 seconds).
    send(
        &mut ws,
        json!({
            "type": "renew",
            "queue": "renew-timeout-q",
            "id": msg_id,
            "lock_timeout_secs": 120,
            "req_id": "rn-timeout"
        }),
    )
    .await;
    let renew_resp = recv(&mut ws).await;
    assert_eq!(renew_resp["type"], "ok");
    assert_eq!(renew_resp["req_id"], "rn-timeout");

    // Message should still be locked (renewed for 120s).
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/renew-timeout-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(
        stats["locked"], 1,
        "message should still be locked after renew with 120s timeout"
    );

    // Ack the message to clean up.
    send(
        &mut ws,
        json!({"type": "ack", "queue": "renew-timeout-q", "id": msg_id}),
    )
    .await;
    let ack_resp = recv(&mut ws).await;
    assert_eq!(ack_resp["type"], "ok");
}

/// Subscribe without lock_timeout_secs should use the default (30s).
/// Verify backward compatibility.
// Verifies: WS-0007
#[tokio::test]
async fn test_ws_subscribe_default_lock_timeout() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "default-lock-q").await;
    let mut ws = ws_connect(&ws_base).await;

    // Subscribe without lock_timeout_secs.
    send(
        &mut ws,
        json!({"type": "subscribe", "queue": "default-lock-q"}),
    )
    .await;
    let sub_resp = recv(&mut ws).await;
    assert_eq!(sub_resp["type"], "ok");

    http_publish(&http_base, "default-lock-q", "default-lock-payload").await;

    let delivery = timeout(Duration::from_secs(3), async {
        loop {
            let f = recv(&mut ws).await;
            if f["type"] == "deliver" {
                return f;
            }
        }
    })
    .await
    .expect("did not receive delivery frame in time");

    let msg_id = delivery["id"].as_str().unwrap();

    // Message should be locked.
    let stats: Value = Client::new()
        .get(format!("{}/queue-stats/default-lock-q", http_base))
        .send()
        .await
        .unwrap()
        .json()
        .await
        .unwrap();
    assert_eq!(stats["locked"], 1);

    // Ack to clean up.
    send(
        &mut ws,
        json!({"type": "ack", "queue": "default-lock-q", "id": msg_id}),
    )
    .await;
    let _ = recv(&mut ws).await;
}

// ---------------------------------------------------------------------------
// WS-0027 – outbound write path decoupled from the inbound receive loop
// ---------------------------------------------------------------------------

/// The inbound receive loop must keep processing client frames while the
/// outbound write path is busy delivering a large backlog. We subscribe to a
/// "sink" queue and give it several hundred deliveries — enough to keep the
/// server's sender task continuously busy — then, *concurrently*, send inbound
/// `publish` frames to a *second* queue and confirm they take effect.
///
/// A background reader keeps our end of the socket serviced so the connection
/// stays alive and server pings are answered; the delivery frames themselves
/// are discarded, which WS-0027 explicitly permits under back-pressure. The
/// probe effect is observed over HTTP — never via an outbound WS frame — so the
/// assertion is deterministic regardless of delivery timing or drops.
///
/// (The narrower "outbound channel full ⇒ frame dropped, inbound still served"
/// path is a unit-level concern of the 512-slot channel; filling it here would
/// require ~512 durable publishes, so this integration test targets the
/// inbound/outbound *decoupling* instead.)
///
/// Verifies: WS-0027
#[tokio::test]
async fn test_ws_outbound_backpressure_does_not_stall_inbound() {
    let (ws_base, http_base, _dir) = start_test_server().await;
    http_create_queue(&http_base, "bp-sink").await;
    http_create_queue(&http_base, "bp-probe").await;

    let ws = ws_connect(&ws_base).await;
    let (mut ws_tx, mut ws_rx) = ws.split();

    let subscribe = json!({"type": "subscribe", "queue": "bp-sink"}).to_string();
    ws_tx
        .send(TungsteniteMessage::Text(subscribe.into()))
        .await
        .unwrap();

    // Keep the socket serviced (so the connection stays alive and pings are
    // answered) while the server delivers the backlog; drop the delivery frames.
    let drainer = tokio::spawn(async move { while let Some(Ok(_)) = ws_rx.next().await {} });

    // Hand the server a large outbound backlog to deliver.
    for i in 0..500 {
        http_publish(&http_base, "bp-sink", &format!("flood-{i}")).await;
    }

    // Inbound must still be processed while all that outbound work is in flight.
    const PROBE_COUNT: u64 = 10;
    for i in 0..PROBE_COUNT {
        let frame =
            json!({"type": "publish", "queue": "bp-probe", "payload": format!("probe-{i}")})
                .to_string();
        ws_tx
            .send(TungsteniteMessage::Text(frame.into()))
            .await
            .unwrap();
    }

    // Observe the side effect over HTTP, independent of the WS outbound path.
    let client = Client::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
    let mut total = 0u64;
    while tokio::time::Instant::now() < deadline {
        let stats: Value = client
            .get(format!("{}/queue-stats/bp-probe", http_base))
            .send()
            .await
            .unwrap()
            .json()
            .await
            .unwrap();
        total = stats["total"].as_u64().unwrap_or(0);
        if total >= PROBE_COUNT {
            break;
        }
        tokio::time::sleep(Duration::from_millis(50)).await;
    }
    assert_eq!(
        total, PROBE_COUNT,
        "inbound WS publishes must be processed while the outbound path is busy"
    );

    drainer.abort();
    let _ = ws_tx.close().await;
}