mod common;

use std::time::Duration;

use bytes::Bytes;

use fafb_q::codec::Frame;
use fafb_q::codec::frame::{
    CAP_BATCH_ACK, CAP_HEADERS, CAP_NACK, CAP_WILDCARD, ErrorCode, ReconnectStatus,
    SubscriptionEntry, SubscriptionStatus,
};
use fafb_q::config::TcpConfig;
use fafb_q::queue::{BackpressurePolicy, EngineConfig};
use fafb_q::types::{ApiKey, GroupName, Header, NamespaceId, Permission, TopicName, TopicOffset};

use common::{
    TEST_KEY, TEST_NS, TestClient, start_test_tcp_server, start_test_tcp_server_with_config,
    start_test_tcp_server_with_engine_config, start_test_tcp_server_with_perms,
    start_test_tcp_server_with_shared_config,
};

#[tokio::test]
async fn connect_success() {
    let server = start_test_tcp_server().await;
    let mut client = TestClient::connect(&server.addr).await;

    client
        .send(Frame::Connect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: 0,
        })
        .await;

    let frame = client.recv().await;
    assert!(matches!(frame, Frame::ConnectAck { success: true, .. }));

    server.shutdown();
}

#[tokio::test]
async fn connect_bad_key() {
    let server = start_test_tcp_server().await;
    let mut client = TestClient::connect(&server.addr).await;

    client
        .send(Frame::Connect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey("wrong-key".into()),
            capabilities: 0,
        })
        .await;

    let frame = client.recv().await;
    assert!(matches!(frame, Frame::ConnectAck { success: false, .. }));

    // Connection should be closed by server.
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none(), "expected connection to close");

    server.shutdown();
}

#[tokio::test]
async fn connect_wrong_version() {
    let server = start_test_tcp_server().await;
    let mut client = TestClient::connect(&server.addr).await;

    client
        .send(Frame::Connect {
            version: 2,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: 0,
        })
        .await;

    let frame = client.recv().await;
    assert!(matches!(
        frame,
        Frame::Error {
            code: ErrorCode::UnsupportedVersion,
            ..
        }
    ));

    // Connection should be closed.
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none());

    server.shutdown();
}

#[tokio::test]
async fn frame_before_auth() {
    let server = start_test_tcp_server().await;
    let mut client = TestClient::connect(&server.addr).await;

    client
        .send(Frame::Subscribe {
            topic: TopicName("t".into()),
            group: None,
        })
        .await;

    let frame = client.recv().await;
    assert!(matches!(
        frame,
        Frame::Error {
            code: ErrorCode::NotAuthenticated,
            ..
        }
    ));

    // Connection should be closed.
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none());

    server.shutdown();
}

#[tokio::test]
async fn client_sends_server_frame() {
    let server = start_test_tcp_server().await;
    let mut client = TestClient::connect(&server.addr).await;

    client
        .send(Frame::ConnectAck {
            success: true,
            message: String::new(),
            server_capabilities: 0,
        })
        .await;

    let frame = client.recv().await;
    assert!(matches!(
        frame,
        Frame::Error {
            code: ErrorCode::InvalidFrame,
            ..
        }
    ));

    // Connection should be closed.
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none());

    server.shutdown();
}

#[tokio::test]
async fn subscribe_publish_deliver() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("events".into());

    // Client A subscribes.
    let mut client_a = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client_a
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let ack = client_a.recv().await;
    assert!(matches!(ack, Frame::SubscribeAck { .. }));

    // Client B publishes.
    let mut client_b = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client_b
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"hello world"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client_b.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Client A receives the delivery.
    let deliver = client_a
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    match deliver {
        Frame::Deliver {
            topic: t,
            payload,
            message_id: _,
            ..
        } => {
            assert_eq!(t, topic);
            assert_eq!(payload.as_ref(), b"hello world");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn publish_ack_flow() {
    let server = start_test_tcp_server().await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client
        .send(Frame::Publish {
            topic: TopicName("t".into()),
            payload: Bytes::from_static(b"data"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;

    let frame = client.recv().await;
    match frame {
        Frame::PublishAck { message_id } => {
            assert!(message_id.0 > 0, "message id should be positive");
        }
        other => panic!("expected PublishAck, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn subscribe_ack_then_deliver_then_ack() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("lifecycle".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Subscribe.
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Publish (same client for simplicity).
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"lifecycle-msg"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Receive delivery.
    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver");
    let msg_id = match &deliver {
        Frame::Deliver { message_id, .. } => *message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    // Ack.
    client.send(Frame::Ack { message_id: msg_id }).await;

    // No error should come back — verify by sending a Ping.
    client.send(Frame::Ping).await;
    let pong = client
        .recv_timeout(Duration::from_secs(2))
        .await
        .expect("expected Pong");
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn unsubscribe_stops_delivery() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("unsub-test".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Subscribe.
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let _ = client.recv().await; // SubscribeAck

    // Unsubscribe.
    client
        .send(Frame::Unsubscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let unsub_ack = client.recv().await;
    assert!(matches!(unsub_ack, Frame::UnsubscribeAck { .. }));

    // Publish after unsubscribe (using a second client).
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    publisher
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"should-not-arrive"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ = publisher.recv().await; // PublishAck

    // Original client should NOT receive a Deliver.
    let maybe = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(
        maybe.is_none(),
        "should not receive delivery after unsubscribe"
    );

    server.shutdown();
}

#[tokio::test]
async fn multiple_subscriptions() {
    let server = start_test_tcp_server().await;
    let topic_a = TopicName("topic-a".into());
    let topic_b = TopicName("topic-b".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Subscribe to both topics.
    client
        .send(Frame::Subscribe {
            topic: topic_a.clone(),
            group: None,
        })
        .await;
    let _ = client.recv().await; // SubscribeAck

    client
        .send(Frame::Subscribe {
            topic: topic_b.clone(),
            group: None,
        })
        .await;
    let _ = client.recv().await; // SubscribeAck

    // Publish to both topics.
    let mut pub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    pub_client
        .send(Frame::Publish {
            topic: topic_a.clone(),
            payload: Bytes::from_static(b"msg-a"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ = pub_client.recv().await; // PublishAck

    pub_client
        .send(Frame::Publish {
            topic: topic_b.clone(),
            payload: Bytes::from_static(b"msg-b"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ = pub_client.recv().await; // PublishAck

    // Receive both deliveries.
    let mut topics_received = Vec::new();
    for _ in 0..2 {
        let frame = client
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect("expected Deliver");
        if let Frame::Deliver { topic, .. } = frame {
            topics_received.push(topic);
        } else {
            panic!("expected Deliver, got: {frame:?}");
        }
    }

    topics_received.sort_by(|a, b| a.0.cmp(&b.0));
    assert_eq!(topics_received, vec![topic_a, topic_b]);

    server.shutdown();
}

#[tokio::test]
async fn concurrent_connections() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("concurrent".into());
    let num_subscribers = 20;
    let num_messages = 100;

    // Create subscribers.
    let mut subscribers = Vec::new();
    for _ in 0..num_subscribers {
        let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
        client
            .send(Frame::Subscribe {
                topic: topic.clone(),
                group: None,
            })
            .await;
        let _ = client.recv().await; // SubscribeAck
        subscribers.push(client);
    }

    // Publish messages.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..num_messages {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ = publisher.recv().await; // PublishAck
    }

    // Each subscriber should receive all messages.
    for sub in &mut subscribers {
        let mut count = 0;
        for _ in 0..num_messages {
            let frame = sub
                .recv_timeout(Duration::from_secs(5))
                .await
                .expect("expected Deliver");
            assert!(matches!(frame, Frame::Deliver { .. }));
            count += 1;
        }
        assert_eq!(count, num_messages);
    }

    server.shutdown();
}

#[tokio::test]
async fn disconnect_cleanup() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("disconnect-test".into());

    // Client A subscribes then disconnects.
    {
        let mut client_a = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
        client_a
            .send(Frame::Subscribe {
                topic: topic.clone(),
                group: None,
            })
            .await;
        let _ = client_a.recv().await; // SubscribeAck
        // client_a dropped here — TCP connection closes.
    }

    // Give server time to process the disconnect.
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Publishing should not panic — server should handle the disconnected subscriber gracefully.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    publisher
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"after-disconnect"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let frame = publisher.recv().await;
    assert!(matches!(frame, Frame::PublishAck { .. }));

    // Another client can still subscribe and receive messages.
    let mut client_b = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client_b
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let _ = client_b.recv().await; // SubscribeAck

    publisher
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"for-client-b"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ = publisher.recv().await; // PublishAck

    let deliver = client_b
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("client_b should receive message");
    assert!(matches!(deliver, Frame::Deliver { .. }));

    server.shutdown();
}

// -- New tests --

#[tokio::test]
async fn publish_only_client_cannot_subscribe() {
    let server = start_test_tcp_server_with_perms(&[Permission::Publish, Permission::Query]).await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Subscribe {
            topic: TopicName("restricted".into()),
            group: None,
        })
        .await;

    let frame = client.recv().await;
    assert!(
        matches!(
            frame,
            Frame::Error {
                code: ErrorCode::PermissionDenied,
                ..
            }
        ),
        "publish-only client should not be able to subscribe, got: {frame:?}"
    );

    // Connection should still be alive (permission denied is non-fatal).
    client.send(Frame::Ping).await;
    let pong = client
        .recv_timeout(Duration::from_secs(1))
        .await
        .expect("expected Pong");
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn subscribe_only_client_cannot_publish() {
    let server =
        start_test_tcp_server_with_perms(&[Permission::Subscribe, Permission::Query]).await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Publish {
            topic: TopicName("restricted".into()),
            payload: Bytes::from_static(b"denied"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;

    let frame = client.recv().await;
    assert!(
        matches!(
            frame,
            Frame::Error {
                code: ErrorCode::PermissionDenied,
                ..
            }
        ),
        "subscribe-only client should not be able to publish, got: {frame:?}"
    );

    // Connection should still be alive.
    client.send(Frame::Ping).await;
    let pong = client
        .recv_timeout(Duration::from_secs(1))
        .await
        .expect("expected Pong");
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn idle_timeout_closes_connection() {
    let config = TcpConfig {
        idle_timeout_secs: 1,
        ..TcpConfig::default()
    };
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ],
        config,
    )
    .await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Wait for idle timeout to fire (1 second + buffer).
    let frame = client.recv_timeout(Duration::from_secs(3)).await;
    match frame {
        Some(Frame::Error { message, .. }) => {
            assert!(
                message.contains("idle timeout"),
                "expected idle timeout message, got: {message}"
            );
        }
        other => panic!("expected Error frame with idle timeout, got: {other:?}"),
    }

    // Connection should be closed after the error.
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(
        next.is_none(),
        "connection should be closed after idle timeout"
    );

    server.shutdown();
}

#[tokio::test]
async fn max_connections_rejects_with_error_frame() {
    let config = TcpConfig {
        max_connections: 2,
        ..TcpConfig::default()
    };
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ],
        config,
    )
    .await;

    // Fill up to the limit.
    let _client1 = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    let _client2 = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Third connection should receive an error frame explaining why.
    let mut client3 = TestClient::connect(&server.addr).await;
    let frame = client3.recv_timeout(Duration::from_secs(2)).await;
    match frame {
        Some(Frame::Error { message, .. }) => {
            assert!(
                message.contains("capacity"),
                "expected capacity message, got: {message}"
            );
        }
        None => {} // Connection closed without frame — also acceptable.
        other => panic!("expected Error or close, got: {other:?}"),
    }

    server.shutdown();
}

// -- Seek/rewind integration tests --

#[tokio::test]
async fn subscribe_at_receives_from_offset() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("seek-test".into());

    // Publish 20 messages.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..20u64 {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // Subscribe at offset 15.
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: topic.clone(),
            offset: TopicOffset(15),
            group: None,
        })
        .await;

    let ack = subscriber.recv().await;
    match &ack {
        Frame::SubscribeAck {
            start_offset,
            topic: t,
            ..
        } => {
            assert_eq!(*t, topic);
            assert_eq!(*start_offset, TopicOffset(15));
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    // Should receive messages starting from offset 15.
    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(15));
            assert_eq!(payload.as_ref(), b"msg-15");
        }
        other => panic!("expected Deliver at offset 15, got: {other:?}"),
    }

    // Verify the next message is offset 16.
    let deliver2 = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected second Deliver frame");
    match deliver2 {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(16));
            assert_eq!(payload.as_ref(), b"msg-16");
        }
        other => panic!("expected Deliver at offset 16, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn subscribe_at_returns_start_offset_in_ack() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("ack-offset-test".into());

    // Subscribe at head (no messages published yet).
    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;

    let ack = client.recv().await;
    match &ack {
        Frame::SubscribeAck { start_offset, .. } => {
            assert_eq!(*start_offset, TopicOffset(0));
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn subscribe_at_permission_denied() {
    let server = start_test_tcp_server_with_perms(&[Permission::Publish, Permission::Query]).await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::SubscribeAt {
            topic: TopicName("restricted".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;

    let frame = client.recv().await;
    assert!(
        matches!(
            frame,
            Frame::Error {
                code: ErrorCode::PermissionDenied,
                ..
            }
        ),
        "expected PermissionDenied, got: {frame:?}"
    );

    // Connection should still be alive (non-fatal).
    client.send(Frame::Ping).await;
    let pong = client
        .recv_timeout(Duration::from_secs(1))
        .await
        .expect("expected Pong");
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn subscribe_from_permission_denied() {
    let server = start_test_tcp_server_with_perms(&[Permission::Publish, Permission::Query]).await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::SubscribeFrom {
            topic: TopicName("restricted".into()),
            timestamp_millis: 1_700_000_000_000,
            group: None,
        })
        .await;

    let frame = client.recv().await;
    assert!(
        matches!(
            frame,
            Frame::Error {
                code: ErrorCode::PermissionDenied,
                ..
            }
        ),
        "expected PermissionDenied, got: {frame:?}"
    );

    server.shutdown();
}

#[tokio::test]
async fn subscribe_from_delivers_from_timestamp() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("ts-seek-test".into());

    // Publish 10 messages.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..10u64 {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // SubscribeFrom with timestamp 0 — should get everything from offset 0.
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeFrom {
            topic: topic.clone(),
            timestamp_millis: 0,
            group: None,
        })
        .await;

    let ack = subscriber.recv().await;
    match &ack {
        Frame::SubscribeAck {
            start_offset,
            topic: t,
            ..
        } => {
            assert_eq!(*t, topic);
            assert_eq!(*start_offset, TopicOffset(0));
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    // Should receive first message at offset 0.
    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(0));
            assert_eq!(payload.as_ref(), b"msg-0");
        }
        other => panic!("expected Deliver at offset 0, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn subscribe_from_future_timestamp_starts_at_head() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("ts-future-test".into());

    // Publish 5 messages.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..5u64 {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ = publisher.recv().await;
    }

    // SubscribeFrom with far-future timestamp — should start at head (offset 5).
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeFrom {
            topic: topic.clone(),
            timestamp_millis: 9_999_999_999_999,
            group: None,
        })
        .await;

    let ack = subscriber.recv().await;
    match &ack {
        Frame::SubscribeAck { start_offset, .. } => {
            // Should be at head (5), not 0
            assert_eq!(*start_offset, TopicOffset(5));
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    // Publish one more — subscriber should receive it (live tail).
    publisher
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"new-msg"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ = publisher.recv().await;

    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(5));
            assert_eq!(payload.as_ref(), b"new-msg");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn idle_timeout_reset_by_activity() {
    let config = TcpConfig {
        idle_timeout_secs: 2,
        ..TcpConfig::default()
    };
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ],
        config,
    )
    .await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Send pings at 1-second intervals — each should reset the 2s idle timer.
    for _ in 0..4 {
        tokio::time::sleep(Duration::from_secs(1)).await;
        client.send(Frame::Ping).await;
        let pong = client
            .recv_timeout(Duration::from_secs(1))
            .await
            .expect("expected Pong — connection should still be alive");
        assert!(matches!(pong, Frame::Pong));
    }

    // 4 seconds have elapsed total, but we never went 2s without activity.
    // Now stop sending and wait for the timeout.
    let frame = client.recv_timeout(Duration::from_secs(4)).await;
    match frame {
        Some(Frame::Error { message, .. }) => {
            assert!(message.contains("idle timeout"));
        }
        other => panic!("expected idle timeout Error, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn subscribe_at_clamped_offset_sends_warning() {
    // Engine with max_depth=10 + DropOldest — forces eviction after 10 messages
    let engine_config = EngineConfig {
        max_depth: 10,
        backpressure: BackpressurePolicy::DropOldest,
        ..Default::default()
    };
    let server = start_test_tcp_server_with_engine_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ],
        TcpConfig::default(),
        engine_config,
    )
    .await;

    let topic = TopicName("eviction-test".into());

    // Publish 20 messages → offsets 0..19, but only 10..19 remain in buffer
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..20u64 {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // SubscribeAt offset 0 — should be clamped to base_offset (10)
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: topic.clone(),
            offset: TopicOffset(0),
            group: None,
        })
        .await;

    // First frame: Error(OffsetNotAvailable) warning
    let warning = subscriber.recv().await;
    match &warning {
        Frame::Error { code, message } => {
            assert_eq!(*code, ErrorCode::OffsetNotAvailable);
            assert!(
                message.contains("not available"),
                "expected clamping message, got: {message}"
            );
        }
        other => panic!("expected OffsetNotAvailable warning, got: {other:?}"),
    }

    // Second frame: SubscribeAck with clamped start_offset
    let ack = subscriber.recv().await;
    match &ack {
        Frame::SubscribeAck { start_offset, .. } => {
            assert!(
                start_offset.0 >= 10,
                "expected clamped offset >= 10, got {start_offset}"
            );
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    // Should receive a delivery from the clamped offset
    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame after clamped subscribe");
    match &deliver {
        Frame::Deliver { offset, .. } => {
            assert!(
                offset.0 >= 10,
                "expected delivery from offset >= 10, got {offset}"
            );
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

// -- Reconnect frame integration tests (Phase 14.3) --

#[tokio::test]
async fn reconnect_resumes_delivery_at_correct_offset() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("reconnect-test".into());
    let caps = fafb_q::codec::frame::CAP_GOING_AWAY | fafb_q::codec::frame::CAP_RECONNECT;

    // Step 1: Subscriber subscribes at head (before any messages).
    let mut subscriber = TestClient::connect(&server.addr).await;
    subscriber
        .send(Frame::Connect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: caps,
        })
        .await;
    let connect_ack = subscriber.recv().await;
    assert!(matches!(
        connect_ack,
        Frame::ConnectAck { success: true, .. }
    ));

    subscriber
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = subscriber.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Step 2: Publisher publishes 10 messages.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..10u64 {
        publisher
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // Step 3: Consume and ack 5 messages, tracking offset.
    let mut last_acked_offset = TopicOffset(0);
    for _ in 0..5 {
        let deliver = subscriber
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect("expected Deliver frame");
        match deliver {
            Frame::Deliver {
                message_id, offset, ..
            } => {
                last_acked_offset = offset;
                subscriber.send(Frame::Ack { message_id }).await;
            }
            other => panic!("expected Deliver, got: {other:?}"),
        }
    }
    // last_acked_offset should be 4 (offsets 0-4).
    assert_eq!(last_acked_offset, TopicOffset(4));

    // Step 4: Disconnect (drop the subscriber).
    drop(subscriber);
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Step 5: Reconnect with the Reconnect frame, requesting offset 5.
    let mut reconnector = TestClient::connect(&server.addr).await;
    reconnector
        .send(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: caps,
            subscriptions: vec![SubscriptionEntry {
                topic: topic.clone(),
                group: None,
                last_acked_offset,
            }],
        })
        .await;

    let reconnect_ack = reconnector.recv().await;
    match &reconnect_ack {
        Frame::ReconnectAck {
            status,
            results,
            server_capabilities,
            ..
        } => {
            assert_eq!(*status, ReconnectStatus::Success);
            assert_eq!(results.len(), 1);
            assert_eq!(results[0].status, SubscriptionStatus::Restored);
            // Server should start from offset 5 (last_acked + 1).
            assert_eq!(results[0].actual_start_offset, TopicOffset(5));
            assert_ne!(*server_capabilities, 0);
        }
        other => panic!("expected ReconnectAck, got: {other:?}"),
    }

    // Step 6: Verify delivery resumes at offset 5.
    let deliver = reconnector
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame after reconnect");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(5));
            assert_eq!(payload.as_ref(), b"msg-5");
        }
        other => panic!("expected Deliver at offset 5, got: {other:?}"),
    }

    // Next should be offset 6.
    let deliver2 = reconnector
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected second Deliver frame");
    match deliver2 {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(6));
            assert_eq!(payload.as_ref(), b"msg-6");
        }
        other => panic!("expected Deliver at offset 6, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn reconnect_partial_no_subscribe_permission() {
    // Server with only Publish permission — no Subscribe.
    let server = start_test_tcp_server_with_perms(&[Permission::Publish]).await;
    let caps = fafb_q::codec::frame::CAP_GOING_AWAY | fafb_q::codec::frame::CAP_RECONNECT;

    // Reconnect with subscriptions — all should fail with PermissionDenied.
    let mut client = TestClient::connect(&server.addr).await;
    client
        .send(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: caps,
            subscriptions: vec![
                SubscriptionEntry {
                    topic: TopicName("events".into()),
                    group: None,
                    last_acked_offset: TopicOffset(0),
                },
                SubscriptionEntry {
                    topic: TopicName("logs".into()),
                    group: None,
                    last_acked_offset: TopicOffset(0),
                },
            ],
        })
        .await;

    let reconnect_ack = client.recv().await;
    match &reconnect_ack {
        Frame::ReconnectAck {
            status, results, ..
        } => {
            assert_eq!(*status, ReconnectStatus::Partial);
            assert_eq!(results.len(), 2);

            // All subscriptions should be PermissionDenied due to rollback.
            for r in results {
                assert_eq!(
                    r.status,
                    SubscriptionStatus::PermissionDenied,
                    "expected PermissionDenied for topic {}, got {:?}",
                    r.topic.0,
                    r.status
                );
            }
        }
        other => panic!("expected ReconnectAck, got: {other:?}"),
    }

    // No deliveries should arrive (no subscriptions activated).
    let maybe = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(
        maybe.is_none(),
        "expected no deliveries after Partial, got: {maybe:?}"
    );

    server.shutdown();
}

#[tokio::test]
async fn reconnect_auth_failed_closes_connection() {
    let server = start_test_tcp_server().await;
    let caps = fafb_q::codec::frame::CAP_RECONNECT;

    let mut client = TestClient::connect(&server.addr).await;
    client
        .send(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey("wrong-key".into()),
            capabilities: caps,
            subscriptions: vec![],
        })
        .await;

    let ack = client.recv().await;
    match &ack {
        Frame::ReconnectAck { status, .. } => {
            assert_eq!(*status, ReconnectStatus::AuthFailed);
        }
        other => panic!("expected ReconnectAck with AuthFailed, got: {other:?}"),
    }

    // Connection should be closed (ReplyAndClose).
    let next = client.recv_timeout(Duration::from_millis(500)).await;
    assert!(
        next.is_none(),
        "expected connection to close after auth failure"
    );

    server.shutdown();
}

// ──────────────────────────────────────────────────────────────────────────────
// Fetch (Plan 15 Phase B) integration tests
// ──────────────────────────────────────────────────────────────────────────────

#[tokio::test]
async fn tcp_fetch_basic() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("fetch-basic".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish 5 messages
    for i in 0..5u64 {
        client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = client.recv().await;
    }

    // Fetch all from offset 0
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 100,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse {
            topic: t,
            next_offset,
            head_offset,
            messages,
        } => {
            assert_eq!(t, topic);
            assert_eq!(messages.len(), 5);
            assert_eq!(messages[0].offset, TopicOffset(0));
            assert_eq!(messages[4].offset, TopicOffset(4));
            assert_eq!(next_offset, TopicOffset(5));
            assert_eq!(head_offset, TopicOffset(5));
            assert_eq!(messages[0].payload.as_ref(), b"msg-0");
            assert_eq!(messages[4].payload.as_ref(), b"msg-4");
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_from_timestamp() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("fetch-ts".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish 5 messages
    for i in 0..5u64 {
        client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = client.recv().await;
    }

    // FetchFrom with timestamp 0 should get all messages (all timestamps > 0)
    client
        .send(Frame::FetchFrom {
            topic: topic.clone(),
            timestamp_millis: 0,
            max_count: 100,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert_eq!(messages.len(), 5);
            for msg in &messages {
                assert!(msg.timestamp_millis > 0);
            }
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_from_no_match() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("fetch-ts-nomatch".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish a message
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from("data"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ack = client.recv().await;

    // FetchFrom with a far-future timestamp
    client
        .send(Frame::FetchFrom {
            topic: topic.clone(),
            timestamp_millis: u64::MAX,
            max_count: 10,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse {
            messages,
            next_offset,
            head_offset,
            ..
        } => {
            assert!(messages.is_empty());
            assert_eq!(next_offset, head_offset);
        }
        other => panic!("expected empty FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_permission_denied() {
    // Server without Fetch permission
    let server = start_test_tcp_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
    ])
    .await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Fetch {
            topic: TopicName("no-fetch".into()),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::Error { code, .. } => {
            assert_eq!(code, ErrorCode::PermissionDenied);
        }
        other => panic!("expected PermissionDenied error, got: {other:?}"),
    }

    // Connection should stay alive (non-fatal)
    client.send(Frame::Ping).await;
    let pong = client.recv().await;
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_zero_max_count() {
    let server = start_test_tcp_server().await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Fetch {
            topic: TopicName("zero-count".into()),
            offset: TopicOffset(0),
            max_count: 0,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::Error { code, .. } => {
            assert_eq!(code, ErrorCode::InvalidFrame);
        }
        other => panic!("expected InvalidFrame error for max_count=0, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_offset_clamping() {
    // Use DropOldest with small depth so messages get evicted
    let engine_config = EngineConfig {
        max_depth: 5,
        backpressure: BackpressurePolicy::DropOldest,
        ..Default::default()
    };
    let server = start_test_tcp_server_with_engine_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        TcpConfig::default(),
        engine_config,
    )
    .await;
    let topic = TopicName("clamp-test".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish 10 messages — first 5 will be evicted
    for i in 0..10u64 {
        client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = client.recv().await;
    }

    // Fetch at offset 0 — should be clamped
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 100,
        })
        .await;

    // First frame: OffsetNotAvailable warning
    let warning = client.recv().await;
    match &warning {
        Frame::Error { code, .. } => {
            assert_eq!(*code, ErrorCode::OffsetNotAvailable);
        }
        other => panic!("expected OffsetNotAvailable warning, got: {other:?}"),
    }

    // Second frame: FetchResponse starting from clamped offset
    let response = client.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert!(!messages.is_empty());
            assert!(messages[0].offset >= TopicOffset(5));
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_pagination() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("fetch-paginate".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish 15 messages
    for i in 0..15u64 {
        client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = client.recv().await;
    }

    // Fetch in pages of 5
    let mut all_offsets = Vec::new();
    let mut offset = TopicOffset(0);

    for _ in 0..3 {
        client
            .send(Frame::Fetch {
                topic: topic.clone(),
                offset,
                max_count: 5,
            })
            .await;

        let response = client.recv().await;
        match response {
            Frame::FetchResponse {
                messages,
                next_offset,
                head_offset,
                ..
            } => {
                assert_eq!(messages.len(), 5);
                for msg in &messages {
                    all_offsets.push(msg.offset);
                }
                assert_eq!(head_offset, TopicOffset(15));
                offset = next_offset;
            }
            other => panic!("expected FetchResponse, got: {other:?}"),
        }
    }

    // Verify we got all 15 messages with no duplicates
    assert_eq!(all_offsets.len(), 15);
    for (i, off) in all_offsets.iter().enumerate() {
        assert_eq!(*off, TopicOffset(i as u64));
    }

    // Next fetch should return empty (caught up)
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset,
            max_count: 5,
        })
        .await;
    let response = client.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert!(messages.is_empty());
        }
        other => panic!("expected empty FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_interleaved_with_subscribe() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("fetch-interleave".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish a message
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from("hello"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _ack = client.recv().await;

    // Subscribe
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Now fetch — should work independently
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert_eq!(messages.len(), 1);
            assert_eq!(messages[0].payload.as_ref(), b"hello");
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_message_too_large() {
    // Use a small frame size so a moderate message exceeds the fetch budget.
    // max_message_size = 2048 → frame overhead ~30 bytes → budget ~2018 bytes.
    // A 500-byte payload Publish frame fits (~520 bytes), but a 2000-byte payload
    // exceeds the FetchResponse budget after frame overhead.
    let mut config = TcpConfig::default();
    config.max_message_size = 2048;
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        config,
    )
    .await;
    let topic = TopicName("fetch-too-large".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish frame overhead: 4 (len) + 1 (type) + 2 + 15 (topic) = 22 bytes.
    // So payload up to 2048 - 22 = 2026 fits in a Publish.
    // FetchResponse overhead: 4 + 1 + 2 + 15 + 8 + 8 + 4 = 42 bytes.
    // Budget = 2048 - 42 = 2006. Per-message overhead = 20.
    // Need payload > 2006 - 20 = 1986 to trigger MessageTooLarge.
    let large_payload = vec![0xABu8; 2000];
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from(large_payload),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let ack = client.recv().await;
    assert!(matches!(ack, Frame::PublishAck { .. }));

    // Fetch — the single message should exceed the response budget
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::MessageTooLarge);
            assert!(
                message.contains("offset 0"),
                "error should mention the stuck offset"
            );
        }
        other => panic!("expected MessageTooLarge error, got: {other:?}"),
    }

    // Connection should stay alive (non-fatal error)
    client.send(Frame::Ping).await;
    let pong = client.recv().await;
    assert!(matches!(pong, Frame::Pong));

    server.shutdown();
}

#[tokio::test]
async fn tcp_fetch_large_batch_truncated() {
    // Use a small frame size and moderate messages to force truncation.
    // max_message_size = 2048; each message ~120 bytes payload → ~140 with overhead.
    // Budget ~2000 bytes → fits ~14 messages. Publish 20 → fetch should truncate.
    let mut config = TcpConfig::default();
    config.max_message_size = 2048;
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        config,
    )
    .await;
    let topic = TopicName("fetch-truncate".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Publish 20 messages with 120-byte payloads
    for i in 0..20u64 {
        client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(vec![i as u8; 120]),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _ack = client.recv().await;
    }

    // Fetch all 20 — should be truncated by frame size
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 100,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse {
            messages,
            next_offset,
            head_offset,
            ..
        } => {
            // Should get fewer than 20 due to frame size limit
            assert!(
                messages.len() < 20,
                "expected truncation, got all {} messages",
                messages.len()
            );
            assert!(!messages.is_empty(), "expected at least some messages");
            // Key assertion: next_offset < head_offset means more data available
            assert_eq!(head_offset, TopicOffset(20));
            assert!(
                next_offset < head_offset,
                "next_offset ({next_offset}) should be < head_offset ({head_offset}) when truncated"
            );
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn nack_routes_message_to_dlq_topic() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("nack-test".into());

    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_NACK).await;

    // Subscribe first.
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Publish a message.
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"nack-me-payload"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Receive delivery.
    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    let msg_id = match &deliver {
        Frame::Deliver { message_id, .. } => *message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    // Nack the message with a reason.
    client
        .send(Frame::Nack {
            message_id: msg_id,
            reason: "invalid format".into(),
        })
        .await;

    let nack_ack = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected NackAck frame");
    match nack_ack {
        Frame::NackAck { message_id, status } => {
            assert_eq!(message_id, msg_id);
            assert_eq!(status, 0x00, "expected dead-lettered status");
        }
        other => panic!("expected NackAck, got: {other:?}"),
    }

    // Fetch from the DLQ topic to verify the message landed there.
    let dlq_topic = TopicName("nack-test__dlq".into());
    client
        .send(Frame::Fetch {
            topic: dlq_topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let fetch_resp = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected FetchResponse");
    match fetch_resp {
        Frame::FetchResponse {
            topic: t, messages, ..
        } => {
            assert_eq!(t, dlq_topic);
            assert_eq!(messages.len(), 1, "expected exactly 1 DLQ message");
            // Payload is original (not wrapped in a binary envelope)
            assert_eq!(&messages[0].payload[..], b"nack-me-payload");
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn nack_dlq_message_has_system_headers_via_fetch() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("nack-hdr-test".into());

    // Connect with both CAP_NACK and CAP_HEADERS so fetch returns headers.
    let mut client = TestClient::connect_and_auth_with_caps(
        &server.addr,
        TEST_NS,
        TEST_KEY,
        CAP_NACK | CAP_HEADERS,
    )
    .await;

    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Publish with application headers.
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"dlq-hdr-payload"),
            headers: vec![fafb_q::types::Header {
                key: b"trace-id".to_vec(),
                value: b"abc-123".to_vec(),
            }],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Receive delivery and nack it.
    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    let msg_id = match &deliver {
        Frame::Deliver { message_id, .. } => *message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    client
        .send(Frame::Nack {
            message_id: msg_id,
            reason: "bad schema".into(),
        })
        .await;

    let nack_ack = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected NackAck");
    assert!(matches!(nack_ack, Frame::NackAck { status: 0x00, .. }));

    // Fetch from the DLQ topic with CAP_HEADERS to see system headers.
    let dlq_topic = TopicName("nack-hdr-test__dlq".into());
    client
        .send(Frame::Fetch {
            topic: dlq_topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let fetch_resp = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected FetchResponse");
    match fetch_resp {
        Frame::FetchResponse {
            topic: t, messages, ..
        } => {
            assert_eq!(t, dlq_topic);
            assert_eq!(messages.len(), 1);
            // Original payload preserved
            assert_eq!(&messages[0].payload[..], b"dlq-hdr-payload");
            // Original application header preserved
            let app_hdr = messages[0]
                .headers
                .iter()
                .find(|h| h.key == b"trace-id")
                .expect("app header should be preserved");
            assert_eq!(app_hdr.value, b"abc-123");
            // DLQ system headers present
            let meta = fafb_q::dlq::parse_dlq_headers(&messages[0].headers)
                .expect("DLQ system headers should be present");
            assert_eq!(meta.source_topic, topic.0);
            assert_eq!(meta.reason, "client_nack");
            assert_eq!(meta.nack_reason.as_deref(), Some("bad schema"));
            assert_eq!(meta.delivery_count, 1);
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn nack_unknown_message_id_returns_error() {
    let server = start_test_tcp_server().await;

    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_NACK).await;

    // Nack a message_id that was never delivered.
    client
        .send(Frame::Nack {
            message_id: fafb_q::types::MessageId(99999),
            reason: "bad".into(),
        })
        .await;

    let resp = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Error frame");
    match resp {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::InternalError);
            assert!(
                message.contains("unknown message id"),
                "expected 'unknown message id' in error, got: {message}"
            );
        }
        other => panic!("expected Error frame, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn nack_without_cap_nack_returns_invalid_frame() {
    let server = start_test_tcp_server().await;

    // Connect WITHOUT CAP_NACK (legacy client, capabilities=0).
    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Attempt to send Nack — should be rejected.
    client
        .send(Frame::Nack {
            message_id: fafb_q::types::MessageId(1),
            reason: "test".into(),
        })
        .await;

    let resp = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Error frame");
    match resp {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::InvalidFrame);
            assert!(
                message.contains("CAP_NACK"),
                "expected 'CAP_NACK' in error, got: {message}"
            );
        }
        other => panic!("expected Error frame, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn nack_with_cap_nack_succeeds() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("nack-cap-test".into());

    // Connect WITH CAP_NACK.
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_NACK).await;

    // Subscribe, publish, receive delivery.
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"cap-nack-payload"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    let msg_id = match &deliver {
        Frame::Deliver { message_id, .. } => *message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    // Nack should succeed with CAP_NACK capability.
    client
        .send(Frame::Nack {
            message_id: msg_id,
            reason: "rejected".into(),
        })
        .await;

    let nack_ack = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected NackAck frame");
    match nack_ack {
        Frame::NackAck { message_id, status } => {
            assert_eq!(message_id, msg_id);
            assert_eq!(status, 0x00, "expected dead-lettered status");
        }
        other => panic!("expected NackAck, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn reconnect_with_cap_nack_allows_nack() {
    use fafb_q::codec::frame::{CAP_GOING_AWAY, CAP_RECONNECT};

    let server = start_test_tcp_server().await;
    let topic = TopicName("reconnect-nack".into());
    let caps = CAP_GOING_AWAY | CAP_RECONNECT | CAP_NACK;

    // Step 1: Connect, subscribe, publish, consume, ack — then disconnect.
    let mut client = TestClient::connect(&server.addr).await;
    client
        .send(Frame::Connect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: caps,
        })
        .await;
    let ack = client.recv().await;
    assert!(matches!(ack, Frame::ConnectAck { success: true, .. }));

    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let sub_ack = client.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Publish a message.
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"reconnect-nack-payload"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack = client.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Consume and ack.
    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver");
    let last_offset = match deliver {
        Frame::Deliver {
            message_id, offset, ..
        } => {
            client.send(Frame::Ack { message_id }).await;
            offset
        }
        other => panic!("expected Deliver, got: {other:?}"),
    };

    drop(client);
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Step 2: Reconnect with CAP_NACK.
    let mut client2 = TestClient::connect(&server.addr).await;
    client2
        .send(Frame::Reconnect {
            version: 1,
            namespace: NamespaceId(TEST_NS.into()),
            api_key: ApiKey(TEST_KEY.into()),
            capabilities: caps,
            subscriptions: vec![SubscriptionEntry {
                topic: topic.clone(),
                group: None,
                last_acked_offset: last_offset,
            }],
        })
        .await;

    let reconnect_ack = client2.recv().await;
    match &reconnect_ack {
        Frame::ReconnectAck {
            status,
            server_capabilities,
            ..
        } => {
            assert_eq!(*status, ReconnectStatus::Success);
            // Server should negotiate CAP_NACK.
            assert_ne!(
                *server_capabilities & CAP_NACK,
                0,
                "CAP_NACK should be negotiated"
            );
        }
        other => panic!("expected ReconnectAck, got: {other:?}"),
    }

    // Step 3: Publish another message and Nack it after reconnect.
    client2
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from_static(b"nack-after-reconnect"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let pub_ack2 = client2.recv().await;
    assert!(matches!(pub_ack2, Frame::PublishAck { .. }));

    let deliver2 = client2
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver after reconnect");
    let msg_id = match deliver2 {
        Frame::Deliver { message_id, .. } => message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    // Nack should work because CAP_NACK was negotiated.
    client2
        .send(Frame::Nack {
            message_id: msg_id,
            reason: "bad after reconnect".into(),
        })
        .await;

    let nack_ack = client2
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected NackAck");
    match nack_ack {
        Frame::NackAck { message_id, status } => {
            assert_eq!(message_id, msg_id);
            assert_eq!(status, 0x00, "expected dead-lettered status");
        }
        other => panic!("expected NackAck, got: {other:?}"),
    }

    server.shutdown();
}

// ── Phase 19.3: Message Headers End-to-End ──────────────────────────────────

#[tokio::test]
async fn headers_publish_subscribe_deliver() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("header-events".into());

    let headers = vec![
        Header {
            key: b"Content-Type".to_vec(),
            value: b"application/json".to_vec(),
        },
        Header {
            key: b"Correlation-Id".to_vec(),
            value: b"req-12345".to_vec(),
        },
        Header {
            key: b"Trace-Parent".to_vec(),
            value: b"00-abcdef1234567890".to_vec(),
        },
    ];

    // Client A subscribes with CAP_HEADERS.
    let mut client_a =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    client_a
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let ack = client_a.recv().await;
    assert!(matches!(ack, Frame::SubscribeAck { .. }));

    // Client B publishes with headers.
    let mut client_b =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    client_b
        .send(Frame::Publish {
            topic: topic.clone(),
            headers: headers.clone(),
            ttl_ms: 0,
            payload: Bytes::from_static(b"{\"order\": 42}"),
        })
        .await;
    let pub_ack = client_b.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Client A receives delivery with headers intact.
    let deliver = client_a
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    match deliver {
        Frame::Deliver {
            topic: t,
            payload,
            headers: delivered_headers,
            ..
        } => {
            assert_eq!(t, topic);
            assert_eq!(payload.as_ref(), b"{\"order\": 42}");
            assert_eq!(delivered_headers.len(), 3);
            assert_eq!(delivered_headers[0].key, b"Content-Type");
            assert_eq!(delivered_headers[0].value, b"application/json");
            assert_eq!(delivered_headers[1].key, b"Correlation-Id");
            assert_eq!(delivered_headers[1].value, b"req-12345");
            assert_eq!(delivered_headers[2].key, b"Trace-Parent");
            assert_eq!(delivered_headers[2].value, b"00-abcdef1234567890");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn headers_publish_fetch() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("header-fetch".into());

    let headers = vec![
        Header {
            key: b"X-Custom".to_vec(),
            value: b"value1".to_vec(),
        },
        Header {
            key: b"X-Trace".to_vec(),
            value: vec![0xDE, 0xAD, 0xBE, 0xEF], // binary header value
        },
    ];

    // Publish with headers.
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    client
        .send(Frame::Publish {
            topic: topic.clone(),
            headers: headers.clone(),
            ttl_ms: 0,
            payload: Bytes::from_static(b"fetch-me"),
        })
        .await;
    let _ack = client.recv().await;

    // Fetch and verify headers in response.
    client
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = client.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert_eq!(messages.len(), 1);
            let msg = &messages[0];
            assert_eq!(msg.payload.as_ref(), b"fetch-me");
            assert_eq!(msg.headers.len(), 2);
            assert_eq!(msg.headers[0].key, b"X-Custom");
            assert_eq!(msg.headers[0].value, b"value1");
            assert_eq!(msg.headers[1].key, b"X-Trace");
            assert_eq!(msg.headers[1].value, vec![0xDE, 0xAD, 0xBE, 0xEF]);
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn headers_legacy_client_no_headers() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("header-legacy".into());

    // Legacy client (no CAP_HEADERS) publishes without headers.
    let mut legacy = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    legacy
        .send(Frame::Publish {
            topic: topic.clone(),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"plain-msg"),
        })
        .await;
    let _ack = legacy.recv().await;

    // Capable client fetches — message should have empty headers.
    let mut capable =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    capable
        .send(Frame::Fetch {
            topic: topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = capable.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert_eq!(messages.len(), 1);
            assert_eq!(messages[0].payload.as_ref(), b"plain-msg");
            assert!(
                messages[0].headers.is_empty(),
                "expected empty headers for legacy publish"
            );
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn headers_consumer_group_deliver() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("header-group".into());
    let group = GroupName("workers".into());

    let headers = vec![
        Header {
            key: b"Trace-Id".to_vec(),
            value: b"abc-123".to_vec(),
        },
        Header {
            key: b"Priority".to_vec(),
            value: b"high".to_vec(),
        },
    ];

    // Client A subscribes to a consumer group with CAP_HEADERS.
    let mut client_a =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    client_a
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: Some(group.clone()),
        })
        .await;
    let ack = client_a.recv().await;
    assert!(matches!(ack, Frame::SubscribeAck { .. }));

    // Client B publishes with headers.
    let mut client_b =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_HEADERS).await;
    client_b
        .send(Frame::Publish {
            topic: topic.clone(),
            headers: headers.clone(),
            ttl_ms: 0,
            payload: Bytes::from_static(b"group-msg"),
        })
        .await;
    let pub_ack = client_b.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Client A receives delivery via consumer group — headers should be intact.
    let deliver = client_a
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame via consumer group");
    match deliver {
        Frame::Deliver {
            topic: t,
            payload,
            headers: delivered_headers,
            ..
        } => {
            assert_eq!(t, topic);
            assert_eq!(payload.as_ref(), b"group-msg");
            assert_eq!(delivered_headers.len(), 2);
            assert_eq!(delivered_headers[0].key, b"Trace-Id");
            assert_eq!(delivered_headers[0].value, b"abc-123");
            assert_eq!(delivered_headers[1].key, b"Priority");
            assert_eq!(delivered_headers[1].value, b"high");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

// ── Wildcard capability tests (Phase 20.3) ──────────────────────────

/// Legacy client (no CAP_WILDCARD) sends a wildcard subscribe pattern.
/// Should receive Error(InvalidTopic) because the capability is not negotiated.
#[tokio::test]
async fn wildcard_subscribe_without_cap_wildcard_rejected() {
    let server = start_test_tcp_server().await;

    // Connect WITHOUT CAP_WILDCARD (caps = 0 → legacy client).
    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    // Try subscribing with a wildcard pattern.
    client
        .send(Frame::Subscribe {
            topic: TopicName("orders.*".into()),
            group: None,
        })
        .await;

    // Should get InvalidTopic error.
    match client.recv().await {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::InvalidTopic);
            assert!(
                message.contains("CAP_WILDCARD"),
                "error should mention CAP_WILDCARD, got: {message}"
            );
        }
        other => panic!("expected Error(InvalidTopic), got: {other:?}"),
    }

    server.shutdown();
}

/// Client with CAP_WILDCARD can subscribe to a wildcard pattern and
/// receive SubscribeAck with matched_count.
#[tokio::test]
async fn wildcard_subscribe_with_cap_wildcard_succeeds() {
    let server = start_test_tcp_server().await;

    // Publisher creates some topics first.
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for topic in &["orders.us", "orders.eu"] {
        publisher
            .send(Frame::Publish {
                topic: TopicName(topic.to_string()),
                headers: vec![],
                ttl_ms: 0,
                payload: Bytes::from_static(b"init"),
            })
            .await;
        // Consume PublishAck.
        publisher.recv().await;
    }

    // Subscriber connects with CAP_WILDCARD.
    let mut subscriber =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_WILDCARD).await;

    subscriber
        .send(Frame::Subscribe {
            topic: TopicName("orders.*".into()),
            group: None,
        })
        .await;

    match subscriber.recv().await {
        Frame::SubscribeAck {
            topic,
            matched_count,
            ..
        } => {
            assert_eq!(topic.0, "orders.*");
            assert_eq!(matched_count, 2, "should match 2 existing topics");
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    server.shutdown();
}

/// Publishing to a topic containing wildcard characters is rejected.
#[tokio::test]
async fn publish_to_wildcard_topic_rejected() {
    let server = start_test_tcp_server().await;

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Publish {
            topic: TopicName("orders.*".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"data"),
        })
        .await;

    match client.recv().await {
        Frame::Error { code, .. } => {
            assert_eq!(code, ErrorCode::InvalidTopic);
        }
        other => panic!("expected Error(InvalidTopic), got: {other:?}"),
    }

    server.shutdown();
}

/// Fetch with wildcard pattern is rejected.
#[tokio::test]
async fn fetch_with_wildcard_rejected() {
    let server = start_test_tcp_server().await;

    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_WILDCARD).await;

    client
        .send(Frame::Fetch {
            topic: TopicName("orders.*".into()),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    match client.recv().await {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::InvalidTopic);
            assert!(
                message.contains("wildcard"),
                "error should mention wildcard, got: {message}"
            );
        }
        other => panic!("expected Error(InvalidTopic), got: {other:?}"),
    }

    server.shutdown();
}

// --- BatchAck integration tests ---

#[tokio::test]
async fn batch_ack_multiple_messages() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("batch-ack-test".into());

    // Subscriber client with CAP_BATCH_ACK.
    let mut sub_client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_BATCH_ACK)
            .await;
    sub_client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let _sub_ack = sub_client.recv().await;

    // Separate publisher client.
    let mut pub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..5 {
        pub_client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _pub_ack = pub_client.recv().await;
    }

    // Receive all 5 deliveries on the subscriber.
    let mut message_ids = Vec::new();
    for _ in 0..5 {
        let deliver = sub_client
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect("expected Deliver");
        match deliver {
            Frame::Deliver { message_id, .. } => message_ids.push(message_id),
            other => panic!("expected Deliver, got: {other:?}"),
        }
    }
    assert_eq!(message_ids.len(), 5);

    // BatchAck all 5.
    sub_client
        .send(Frame::BatchAck {
            message_ids: message_ids.clone(),
        })
        .await;
    let response = sub_client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected BatchAckResponse");
    match response {
        Frame::BatchAckResponse {
            acked_count,
            unknown_count,
        } => {
            assert_eq!(acked_count, 5, "all 5 should be acked");
            assert_eq!(unknown_count, 0, "none should be unknown");
        }
        other => panic!("expected BatchAckResponse, got: {other:?}"),
    }

    server.shutdown();
}

#[tokio::test]
async fn batch_ack_with_already_acked_ids() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("batch-ack-partial".into());

    // Subscriber with CAP_BATCH_ACK.
    let mut sub_client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_BATCH_ACK)
            .await;
    sub_client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let _sub_ack = sub_client.recv().await;

    // Separate publisher.
    let mut pub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..3 {
        pub_client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _pub_ack = pub_client.recv().await;
    }

    // Receive all 3 deliveries.
    let mut message_ids = Vec::new();
    for _ in 0..3 {
        let deliver = sub_client
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect("expected Deliver");
        match deliver {
            Frame::Deliver { message_id, .. } => message_ids.push(message_id),
            other => panic!("expected Deliver, got: {other:?}"),
        }
    }

    // Single-ack the first message.
    sub_client
        .send(Frame::Ack {
            message_id: message_ids[0],
        })
        .await;

    // Now BatchAck all 3 — first is already acked (unknown), other 2 should succeed.
    sub_client
        .send(Frame::BatchAck {
            message_ids: message_ids.clone(),
        })
        .await;
    let response = sub_client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected BatchAckResponse");
    match response {
        Frame::BatchAckResponse {
            acked_count,
            unknown_count,
        } => {
            assert_eq!(acked_count, 2, "2 should be newly acked");
            assert_eq!(unknown_count, 1, "1 already acked = unknown");
        }
        other => panic!("expected BatchAckResponse, got: {other:?}"),
    }

    server.shutdown();
}

// -- Phase 21.4: Delivery Observability Metrics integration tests --

/// Publish messages to a consumer group, ack them, and verify the consumer lag
/// converges to zero. Exercises fafb_consumer_lag gauge updates on both
/// the publish path (head advances) and the ack path (inflight clears).
#[tokio::test]
async fn consumer_lag_decreases_after_ack() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("lag-test".into());

    // Subscriber client in a consumer group.
    let mut sub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    sub_client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: Some(GroupName("lag-group".into())),
        })
        .await;
    let _sub_ack = sub_client.recv().await;

    // Publisher client.
    let mut pub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..5 {
        pub_client
            .send(Frame::Publish {
                topic: topic.clone(),
                payload: Bytes::from(format!("lag-msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let _pub_ack = pub_client.recv().await;
    }

    // Receive all 5 deliveries.
    let mut message_ids = Vec::new();
    for _ in 0..5 {
        let deliver = sub_client
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect("expected Deliver");
        match deliver {
            Frame::Deliver { message_id, .. } => message_ids.push(message_id),
            other => panic!("expected Deliver, got: {other:?}"),
        }
    }

    // Ack all 5 — consumer lag should go to 0.
    for mid in &message_ids {
        sub_client.send(Frame::Ack { message_id: *mid }).await;
    }

    // Small delay to let acks process.
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Verify the system is consistent by publishing one more message and
    // receiving it — proves the group offset advanced correctly.
    pub_client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from("after-lag"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _pub_ack = pub_client.recv().await;

    let deliver = sub_client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver after ack");
    match deliver {
        Frame::Deliver { offset, .. } => {
            assert_eq!(offset, TopicOffset(5), "next message should be at offset 5");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

/// Double-ack the same message over TCP. The second ack is a no-op that
/// exercises the fafb_duplicate_ack_total counter path.
#[tokio::test]
async fn duplicate_ack_is_idempotent() {
    let server = start_test_tcp_server().await;
    let topic = TopicName("dup-ack-test".into());

    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    client
        .send(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await;
    let _sub_ack = client.recv().await;

    // Publish one message via a separate client.
    let mut pub_client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    pub_client
        .send(Frame::Publish {
            topic: topic.clone(),
            payload: Bytes::from("dup-ack"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let _pub_ack = pub_client.recv().await;

    // Receive the delivery.
    let deliver = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver");
    let message_id = match deliver {
        Frame::Deliver { message_id, .. } => message_id,
        other => panic!("expected Deliver, got: {other:?}"),
    };

    // Ack twice — second returns a non-fatal error (unknown message ID) and
    // exercises the fafb_duplicate_ack_total counter in the manager layer.
    client.send(Frame::Ack { message_id }).await;
    client.send(Frame::Ack { message_id }).await;

    // The second ack returns an InternalError (session removes from inflight_topics
    // on first ack, so the second is unknown at session level). This is non-fatal.
    let error_frame = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected error for duplicate ack");
    match error_frame {
        Frame::Error { code, .. } => {
            assert_eq!(
                code,
                ErrorCode::InternalError,
                "should be InternalError for unknown message_id"
            );
        }
        other => panic!("expected Error, got: {other:?}"),
    }

    // Verify the connection is still healthy after the non-fatal error.
    client.send(Frame::Ping).await;
    let pong = client
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Pong after duplicate ack");
    assert!(matches!(pong, Frame::Pong), "expected Pong, got: {pong:?}");

    server.shutdown();
}

// -- Message TTL tests (Phase 22.2) --

/// Helper: build a SharedConfig with TTL settings for the test namespace.
fn ttl_shared_config(
    default_ttl_secs: u64,
    max_ttl_secs: u64,
    min_ttl_secs: u64,
) -> fafb_q::config::SharedConfig {
    use fafb_q::config::{RuntimeConfig, ServerConfig};
    use fafb_q::namespace::manager::NamespaceQuota;
    let mut rt = RuntimeConfig::from_server_config(&ServerConfig::default());
    rt.namespace_quotas.insert(
        NamespaceId(TEST_NS.into()),
        NamespaceQuota {
            default_ttl_secs,
            max_ttl_secs,
            min_ttl_secs,
            ..Default::default()
        },
    );
    std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
}

#[tokio::test]
async fn publish_with_explicit_ttl() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;
    use fafb_q::queue::QueueEngine;
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    client
        .send(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 30_000,
            payload: Bytes::from_static(b"hello"),
        })
        .await;
    let ack = client.recv().await;
    assert!(
        matches!(ack, Frame::PublishAck { .. }),
        "expected PublishAck, got: {ack:?}"
    );

    // Verify the stored envelope has the correct TTL.
    let engine = server.engine.as_ref().expect("engine should be available");
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = engine
        .consume_from(&ns, &topic, TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 30_000);

    server.shutdown();
}

#[tokio::test]
async fn publish_without_ttl_cap_sends_zero_ttl() {
    use fafb_q::queue::QueueEngine;
    // Client doesn't negotiate CAP_MESSAGE_TTL — ttl_ms should be ignored/zero.
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let mut client = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;

    client
        .send(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"hello"),
        })
        .await;
    let ack = client.recv().await;
    assert!(
        matches!(ack, Frame::PublishAck { .. }),
        "expected PublishAck, got: {ack:?}"
    );

    let engine = server.engine.as_ref().expect("engine should be available");
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = engine
        .consume_from(&ns, &topic, TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 0);

    server.shutdown();
}

#[tokio::test]
async fn publish_applies_namespace_default_ttl() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;
    use fafb_q::queue::QueueEngine;
    // Namespace default TTL = 60 seconds. Client sends ttl_ms = 0 (no explicit TTL).
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(60, 0, 0)).await;
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    client
        .send(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 0, // no explicit TTL — should get default 60s = 60000ms
            payload: Bytes::from_static(b"hello"),
        })
        .await;
    let ack = client.recv().await;
    assert!(
        matches!(ack, Frame::PublishAck { .. }),
        "expected PublishAck, got: {ack:?}"
    );

    // Verify the stored envelope has the namespace default TTL applied.
    let engine = server.engine.as_ref().expect("engine should be available");
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = engine
        .consume_from(&ns, &topic, TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 60_000);

    server.shutdown();
}

#[tokio::test]
async fn publish_ttl_clamped_to_max() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;
    use fafb_q::queue::QueueEngine;
    // max_ttl_secs = 10 → max 10000ms. Client sends 30000ms.
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 10, 0)).await;
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    client
        .send(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 30_000, // exceeds max 10s — should be clamped, not rejected
            payload: Bytes::from_static(b"clamped"),
        })
        .await;
    let ack = client.recv().await;
    assert!(
        matches!(ack, Frame::PublishAck { .. }),
        "expected PublishAck (TTL clamped), got: {ack:?}"
    );

    // Verify the stored envelope has TTL clamped to the maximum.
    let engine = server.engine.as_ref().expect("engine should be available");
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = engine
        .consume_from(&ns, &topic, TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 10_000);

    server.shutdown();
}

#[tokio::test]
async fn publish_ttl_below_min_rejected() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;
    // min_ttl_secs = 10 → min 10000ms. Client sends 5000ms.
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 10)).await;
    let mut client =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    client
        .send(Frame::Publish {
            topic: TopicName("events".into()),
            headers: vec![],
            ttl_ms: 5_000, // below min 10s — should be rejected
            payload: Bytes::from_static(b"rejected"),
        })
        .await;
    let resp = client.recv().await;
    match resp {
        Frame::Error { code, message } => {
            assert_eq!(code, ErrorCode::InvalidFrame);
            assert!(message.contains("below minimum"), "error: {message}");
        }
        other => panic!("expected Error, got: {other:?}"),
    }
    server.shutdown();
}

// -- Message TTL Delivery Filtering tests (Phase 22.3) --

/// Publish a message with a short TTL, wait for it to expire, then subscribe.
/// The subscriber should NOT receive the expired message.
#[tokio::test]
async fn expired_message_not_delivered_to_subscriber() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;
    use fafb_q::queue::QueueEngine;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    // Publish with a very short TTL (100ms).
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-expire".into()),
            headers: vec![],
            ttl_ms: 100, // 100ms TTL
            payload: Bytes::from_static(b"ephemeral"),
        })
        .await;
    let ack = publisher.recv().await;
    assert!(matches!(ack, Frame::PublishAck { .. }));

    // Also publish a message with no TTL (should be delivered).
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-expire".into()),
            headers: vec![],
            ttl_ms: 0, // no expiry
            payload: Bytes::from_static(b"permanent"),
        })
        .await;
    let ack2 = publisher.recv().await;
    assert!(matches!(ack2, Frame::PublishAck { .. }));

    // Wait for the TTL to expire.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Subscribe at offset 0 — the expired message should be skipped.
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: TopicName("ttl-expire".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;
    let sub_ack = subscriber.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Should receive the permanent message (offset 1), NOT the expired one (offset 0).
    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame for permanent message");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(1));
            assert_eq!(payload.as_ref(), b"permanent");
        }
        other => panic!("expected Deliver at offset 1, got: {other:?}"),
    }

    // No more messages should be delivered.
    let next = subscriber.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none(), "expected no more messages, got: {next:?}");

    server.shutdown();
}

/// Publish a mix of messages with short and long TTLs.
/// After the short TTLs expire, only the long-TTL messages should be delivered.
#[tokio::test]
async fn mixed_ttl_messages_only_live_delivered() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    // Publish 5 messages: offsets 0,2,4 with short TTL; offsets 1,3 with no TTL.
    for i in 0..5u32 {
        let ttl = if i % 2 == 0 { 100 } else { 0 };
        publisher
            .send(Frame::Publish {
                topic: TopicName("mixed-ttl".into()),
                headers: vec![],
                ttl_ms: ttl,
                payload: Bytes::from(format!("msg-{i}")),
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // Wait for short TTLs to expire.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Subscribe at offset 0.
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: TopicName("mixed-ttl".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;
    let sub_ack = subscriber.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Should only receive offsets 1 and 3 (the ones without TTL).
    let deliver1 = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected first live message");
    match deliver1 {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(1));
            assert_eq!(payload.as_ref(), b"msg-1");
        }
        other => panic!("expected Deliver at offset 1, got: {other:?}"),
    }

    let deliver2 = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected second live message");
    match deliver2 {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(3));
            assert_eq!(payload.as_ref(), b"msg-3");
        }
        other => panic!("expected Deliver at offset 3, got: {other:?}"),
    }

    // No more messages.
    let next = subscriber.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none(), "expected no more messages, got: {next:?}");

    server.shutdown();
}

/// Fetch path: expired messages are filtered out, next_offset advances past them.
#[tokio::test]
async fn fetch_filters_expired_messages() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    // Publish 4 messages: offsets 0,1 with short TTL; offsets 2,3 with no TTL.
    for i in 0..4u32 {
        let ttl = if i < 2 { 100 } else { 0 };
        publisher
            .send(Frame::Publish {
                topic: TopicName("fetch-ttl".into()),
                headers: vec![],
                ttl_ms: ttl,
                payload: Bytes::from(format!("msg-{i}")),
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // Wait for short TTLs to expire.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Fetch from offset 0 — expired messages should be filtered.
    let mut fetcher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    fetcher
        .send(Frame::Fetch {
            topic: TopicName("fetch-ttl".into()),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let resp = fetcher.recv().await;
    match resp {
        Frame::FetchResponse {
            messages,
            next_offset,
            head_offset,
            ..
        } => {
            // Only the 2 non-expired messages should be returned.
            assert_eq!(messages.len(), 2);
            assert_eq!(messages[0].offset, TopicOffset(2));
            assert_eq!(messages[1].offset, TopicOffset(3));
            // next_offset should be past all messages (4).
            assert_eq!(next_offset, TopicOffset(4));
            assert_eq!(head_offset, TopicOffset(4));
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

/// Messages with no TTL (ttl_ms=0) are always delivered regardless of age.
#[tokio::test]
async fn messages_without_ttl_always_delivered() {
    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    for i in 0..3u32 {
        publisher
            .send(Frame::Publish {
                topic: TopicName("no-ttl".into()),
                headers: vec![],
                ttl_ms: 0,
                payload: Bytes::from(format!("msg-{i}")),
            })
            .await;
        let ack = publisher.recv().await;
        assert!(matches!(ack, Frame::PublishAck { .. }));
    }

    // Sleep a bit to ensure these aren't accidentally treated as expired.
    tokio::time::sleep(Duration::from_millis(100)).await;

    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: TopicName("no-ttl".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;
    let sub_ack = subscriber.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // All 3 messages should be delivered.
    for i in 0..3u64 {
        let deliver = subscriber
            .recv_timeout(Duration::from_secs(3))
            .await
            .expect(&format!("expected Deliver at offset {i}"));
        match deliver {
            Frame::Deliver { offset, .. } => {
                assert_eq!(offset, TopicOffset(i));
            }
            other => panic!("expected Deliver at offset {i}, got: {other:?}"),
        }
    }

    server.shutdown();
}

/// Consumer group delivery also skips expired messages.
#[tokio::test]
async fn consumer_group_skips_expired_messages() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    // Publish: offset 0 = expired, offset 1 = live.
    publisher
        .send(Frame::Publish {
            topic: TopicName("grp-ttl".into()),
            headers: vec![],
            ttl_ms: 100,
            payload: Bytes::from_static(b"expired"),
        })
        .await;
    let _ = publisher.recv().await;

    publisher
        .send(Frame::Publish {
            topic: TopicName("grp-ttl".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"live"),
        })
        .await;
    let _ = publisher.recv().await;

    // Wait for expiry.
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Subscribe with consumer group at offset 0.
    let mut consumer = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    consumer
        .send(Frame::SubscribeAt {
            topic: TopicName("grp-ttl".into()),
            offset: TopicOffset(0),
            group: Some(GroupName("g1".into())),
        })
        .await;
    let sub_ack = consumer.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    // Should receive only the live message.
    let deliver = consumer
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame for live message");
    match deliver {
        Frame::Deliver {
            offset, payload, ..
        } => {
            assert_eq!(offset, TopicOffset(1));
            assert_eq!(payload.as_ref(), b"live");
        }
        other => panic!("expected Deliver at offset 1, got: {other:?}"),
    }

    let next = consumer.recv_timeout(Duration::from_millis(500)).await;
    assert!(next.is_none(), "expected no more messages, got: {next:?}");

    server.shutdown();
}

/// A message with TTL that expires while inflight (delivered, not acked) should
/// be routed to the DLQ with `message_expired` reason instead of being redelivered.
#[tokio::test]
async fn inflight_expired_message_routed_to_dlq() {
    use fafb_q::codec::frame::{CAP_HEADERS, CAP_MESSAGE_TTL};
    use fafb_q::config::{DeployConfig, RuntimeConfig, ServerConfig};
    use fafb_q::namespace::manager::{DefaultNamespaceManager, ManagerConfig, NamespaceQuota};
    use fafb_q::server::state::ServerLifecycle;
    use fafb_q::server::tcp::TcpServer;
    use std::sync::Arc;
    use std::sync::atomic::AtomicU64;

    // Build shared config with short inflight timeout (1s) so sweep triggers fast.
    let mut rt = RuntimeConfig::from_server_config(&ServerConfig::default());
    rt.inflight_timeout_secs = 1;
    rt.max_redelivery_backoff_secs = 1;
    rt.namespace_quotas
        .insert(NamespaceId(TEST_NS.into()), NamespaceQuota::default());
    let shared_cfg: fafb_q::config::SharedConfig = Arc::new(arc_swap::ArcSwap::from_pointee(rt));

    // Build manager with the SAME shared config (so inflight sweep uses 1s timeout).
    let engine = common::make_engine();
    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
        Arc::clone(&shared_cfg),
        ManagerConfig::default(),
    ));

    let auth: Arc<dyn fafb_q::auth::Authenticator> = Arc::new(fafb_q::auth::MockAuthenticator {
        namespace: NamespaceId(TEST_NS.into()),
        key: fafb_q::types::ApiKey(TEST_KEY.into()),
        permissions: fafb_q::auth::PermissionSet::from_slice(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();
    let config = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let tcp_server = TcpServer::new(
        config,
        DeployConfig::default(),
        shared_cfg,
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        None,
    );
    tokio::spawn(async move {
        let _ = tcp_server.run_with_listener(listener).await;
    });

    let server = common::TestServer {
        addr,
        lifecycle,
        engine: None,
    };

    // Publisher with TTL + headers caps (headers needed to read DLQ system headers).
    let caps = CAP_MESSAGE_TTL | CAP_HEADERS;
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, caps).await;

    // Subscribe first, then publish a message with a short TTL (500ms).
    let mut subscriber =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, caps).await;
    subscriber
        .send(Frame::Subscribe {
            topic: TopicName("ttl-inflight".into()),
            group: None,
        })
        .await;
    let sub_ack = subscriber.recv().await;
    assert!(matches!(sub_ack, Frame::SubscribeAck { .. }));

    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-inflight".into()),
            headers: vec![],
            ttl_ms: 500, // 500ms TTL
            payload: Bytes::from_static(b"expire-while-inflight"),
        })
        .await;
    let pub_ack = publisher.recv().await;
    assert!(matches!(pub_ack, Frame::PublishAck { .. }));

    // Receive the message — do NOT ack it.
    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected Deliver frame");
    assert!(matches!(deliver, Frame::Deliver { .. }));

    // Wait for TTL to expire (500ms) + inflight timeout (1s) + sweep interval (1s) + buffer.
    tokio::time::sleep(Duration::from_millis(3500)).await;

    // The expired message should not be redelivered to us.
    let redelivery = subscriber.recv_timeout(Duration::from_millis(500)).await;
    assert!(
        redelivery.is_none(),
        "expired message should NOT be redelivered, got: {redelivery:?}"
    );

    // Fetch from the DLQ topic — the message should have been routed there.
    let dlq_topic = TopicName("ttl-inflight__dlq".into());
    let mut fetcher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, caps).await;
    fetcher
        .send(Frame::Fetch {
            topic: dlq_topic.clone(),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let fetch_resp = fetcher
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected FetchResponse from DLQ topic");
    match fetch_resp {
        Frame::FetchResponse {
            topic: t, messages, ..
        } => {
            assert_eq!(t, dlq_topic);
            assert_eq!(messages.len(), 1, "expected 1 message in DLQ");
            assert_eq!(&messages[0].payload[..], b"expire-while-inflight");

            // Verify DLQ reason header is "message_expired".
            let meta = fafb_q::dlq::parse_dlq_headers(&messages[0].headers)
                .expect("DLQ system headers should be present");
            assert_eq!(meta.reason, "message_expired");
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

/// Verify that Deliver frames include the correct `ttl_ms` when CAP_MESSAGE_TTL is negotiated.
#[tokio::test]
async fn deliver_frame_includes_ttl_ms() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    // Publisher with TTL capability.
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;

    // Publish a message with TTL=30s.
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-deliver".into()),
            headers: vec![],
            ttl_ms: 30_000,
            payload: Bytes::from_static(b"with-ttl"),
        })
        .await;
    assert!(matches!(publisher.recv().await, Frame::PublishAck { .. }));

    // Publish a message without TTL.
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-deliver".into()),
            headers: vec![],
            ttl_ms: 0,
            payload: Bytes::from_static(b"no-ttl"),
        })
        .await;
    assert!(matches!(publisher.recv().await, Frame::PublishAck { .. }));

    // Subscriber with TTL capability — should see ttl_ms in Deliver frames.
    let mut subscriber =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: TopicName("ttl-deliver".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;
    assert!(matches!(
        subscriber.recv().await,
        Frame::SubscribeAck { .. }
    ));

    // First message should have ttl_ms=30000.
    let deliver1 = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected first delivery");
    match deliver1 {
        Frame::Deliver {
            ttl_ms, payload, ..
        } => {
            assert_eq!(ttl_ms, 30_000, "first message should have ttl_ms=30000");
            assert_eq!(payload.as_ref(), b"with-ttl");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    // Second message should have ttl_ms=0.
    let deliver2 = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected second delivery");
    match deliver2 {
        Frame::Deliver {
            ttl_ms, payload, ..
        } => {
            assert_eq!(ttl_ms, 0, "second message should have ttl_ms=0");
            assert_eq!(payload.as_ref(), b"no-ttl");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

/// Verify that Deliver frames have ttl_ms=0 when subscriber doesn't negotiate CAP_MESSAGE_TTL.
#[tokio::test]
async fn deliver_frame_ttl_zero_without_cap() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    // Publisher with TTL capability — publishes with ttl_ms=30000.
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-no-cap".into()),
            headers: vec![],
            ttl_ms: 30_000,
            payload: Bytes::from_static(b"data"),
        })
        .await;
    assert!(matches!(publisher.recv().await, Frame::PublishAck { .. }));

    // Subscriber WITHOUT TTL capability — ttl_ms should be 0 (not on wire).
    let mut subscriber = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    subscriber
        .send(Frame::SubscribeAt {
            topic: TopicName("ttl-no-cap".into()),
            offset: TopicOffset(0),
            group: None,
        })
        .await;
    assert!(matches!(
        subscriber.recv().await,
        Frame::SubscribeAck { .. }
    ));

    let deliver = subscriber
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("expected delivery");
    match deliver {
        Frame::Deliver { ttl_ms, .. } => {
            assert_eq!(ttl_ms, 0, "legacy client should see ttl_ms=0");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    server.shutdown();
}

/// Verify FetchResponse includes ttl_ms when CAP_MESSAGE_TTL is negotiated.
#[tokio::test]
async fn fetch_response_includes_ttl_ms() {
    use fafb_q::codec::frame::CAP_MESSAGE_TTL;

    let server = start_test_tcp_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;

    // Publish with TTL.
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;
    publisher
        .send(Frame::Publish {
            topic: TopicName("ttl-fetch".into()),
            headers: vec![],
            ttl_ms: 45_000,
            payload: Bytes::from_static(b"fetch-me"),
        })
        .await;
    assert!(matches!(publisher.recv().await, Frame::PublishAck { .. }));

    // Fetch with TTL capability.
    let mut fetcher =
        TestClient::connect_and_auth_with_caps(&server.addr, TEST_NS, TEST_KEY, CAP_MESSAGE_TTL)
            .await;
    fetcher
        .send(Frame::Fetch {
            topic: TopicName("ttl-fetch".into()),
            offset: TopicOffset(0),
            max_count: 10,
        })
        .await;

    let response = fetcher.recv().await;
    match response {
        Frame::FetchResponse { messages, .. } => {
            assert_eq!(messages.len(), 1);
            assert_eq!(messages[0].ttl_ms, 45_000);
            assert_eq!(messages[0].payload.as_ref(), b"fetch-me");
        }
        other => panic!("expected FetchResponse, got: {other:?}"),
    }

    server.shutdown();
}

// ---------------------------------------------------------------------------
// FafbClient integration tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn fafb_client_connect_and_capabilities() {
    use fafb_q::client::{ClientConfig, FafbClient};
    use fafb_q::codec::frame::{CAP_HEADERS, CAP_MESSAGE_TTL};

    let server = start_test_tcp_server().await;

    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let client = FafbClient::connect(&config).await.unwrap();

    // Server advertises all capabilities; client advertises all; intersection = all.
    let caps = client.capabilities();
    assert_ne!(caps, 0, "should have negotiated some capabilities");
    assert_ne!(caps & CAP_HEADERS, 0, "should have CAP_HEADERS");
    assert_ne!(caps & CAP_MESSAGE_TTL, 0, "should have CAP_MESSAGE_TTL");

    assert_eq!(client.namespace(), &NamespaceId(TEST_NS.into()));

    client.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_auth_failure() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;

    let config = ClientConfig::new(&server.addr, TEST_NS, "wrong-key");
    let result = FafbClient::connect(&config).await;

    match result {
        Err(e) => {
            let err_msg = format!("{e}");
            assert!(
                err_msg.contains("authentication failed") || err_msg.contains("error"),
                "unexpected error: {err_msg}"
            );
        }
        Ok(_) => panic!("expected auth failure, but connect succeeded"),
    }

    server.shutdown();
}

// ---------------------------------------------------------------------------
// FafbClient publish & fetch integration tests (Phase 23.2)
// ---------------------------------------------------------------------------

#[tokio::test]
async fn fafb_client_publish_and_fetch_roundtrip() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&config).await.unwrap();

    // Publish a message.
    let result = client
        .publish(
            TopicName("cli-test".into()),
            bytes::Bytes::from("hello from client"),
            vec![],
            0,
        )
        .await
        .unwrap();
    assert!(result.message_id.0 > 0);

    // Fetch it back.
    let fetch = client
        .fetch(TopicName("cli-test".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    assert_eq!(fetch.messages[0].payload.as_ref(), b"hello from client");
    assert_eq!(fetch.messages[0].offset, TopicOffset(0));

    client.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_publish_with_headers() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&config).await.unwrap();

    let headers = vec![
        Header {
            key: b"Content-Type".to_vec(),
            value: b"application/json".to_vec(),
        },
        Header {
            key: b"Trace-Id".to_vec(),
            value: b"abc123".to_vec(),
        },
    ];

    client
        .publish(
            TopicName("hdr-test".into()),
            bytes::Bytes::from(r#"{"id":1}"#),
            headers.clone(),
            0,
        )
        .await
        .unwrap();

    let fetch = client
        .fetch(TopicName("hdr-test".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    assert_eq!(fetch.messages[0].headers.len(), 2);
    assert_eq!(fetch.messages[0].headers[0].key, b"Content-Type");
    assert_eq!(fetch.messages[0].headers[0].value, b"application/json");
    assert_eq!(fetch.messages[0].headers[1].key, b"Trace-Id");
    assert_eq!(fetch.messages[0].headers[1].value, b"abc123");

    client.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_publish_no_ack() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&config).await.unwrap();

    // Fire-and-forget publish — drains the ack internally so the stream
    // stays clean for subsequent operations.
    client
        .publish_no_ack(
            TopicName("noack-test".into()),
            bytes::Bytes::from("fire"),
            vec![],
            0,
        )
        .await
        .unwrap();

    // Same client can immediately fetch — no leftover ack in the stream.
    let fetch = client
        .fetch(TopicName("noack-test".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    assert_eq!(fetch.messages[0].payload.as_ref(), b"fire");

    client.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_fetch_pagination() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&config).await.unwrap();

    // Publish 15 messages.
    for i in 0..15u32 {
        client
            .publish(
                TopicName("page-test".into()),
                bytes::Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }

    // Fetch first page (5 messages).
    let page1 = client
        .fetch(TopicName("page-test".into()), TopicOffset(0), 5)
        .await
        .unwrap();
    assert_eq!(page1.messages.len(), 5);
    assert_eq!(page1.messages[0].offset, TopicOffset(0));
    assert_eq!(page1.next_offset, TopicOffset(5));

    // Fetch second page from next_offset.
    let page2 = client
        .fetch(TopicName("page-test".into()), page1.next_offset, 5)
        .await
        .unwrap();
    assert_eq!(page2.messages.len(), 5);
    assert_eq!(page2.messages[0].offset, TopicOffset(5));

    // Fetch third page.
    let page3 = client
        .fetch(TopicName("page-test".into()), page2.next_offset, 5)
        .await
        .unwrap();
    assert_eq!(page3.messages.len(), 5);
    assert_eq!(page3.messages[0].offset, TopicOffset(10));

    // No more messages.
    assert!(page3.next_offset >= page3.head_offset);

    client.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_publish_with_ttl() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&config).await.unwrap();

    client
        .publish(
            TopicName("ttl-test".into()),
            bytes::Bytes::from("expires-soon"),
            vec![],
            30_000, // 30s TTL
        )
        .await
        .unwrap();

    let fetch = client
        .fetch(TopicName("ttl-test".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    assert_eq!(fetch.messages[0].ttl_ms, 30_000);

    client.close().await.unwrap();
    server.shutdown();
}

// ---------------------------------------------------------------------------
// FafbClient subscribe integration tests (Phase 23.3)
// ---------------------------------------------------------------------------

#[tokio::test]
async fn fafb_client_subscribe_and_receive() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;

    // Subscriber client.
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut subscriber = FafbClient::connect(&config).await.unwrap();

    // Subscribe from head.
    subscriber
        .send_frame(Frame::Subscribe {
            topic: TopicName("sub-test".into()),
            group: None,
        })
        .await
        .unwrap();

    let ack = subscriber.recv_frame().await.unwrap();
    assert!(matches!(ack, Frame::SubscribeAck { .. }));

    // Publisher client.
    let pub_config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    let mut publisher = FafbClient::connect(&pub_config).await.unwrap();

    publisher
        .publish(
            TopicName("sub-test".into()),
            Bytes::from("streamed message"),
            vec![],
            0,
        )
        .await
        .unwrap();

    // Subscriber should receive the message.
    let deliver = tokio::time::timeout(Duration::from_secs(5), subscriber.recv_frame())
        .await
        .expect("timed out waiting for Deliver")
        .unwrap();

    match deliver {
        Frame::Deliver {
            topic,
            payload,
            message_id,
            ..
        } => {
            assert_eq!(topic, TopicName("sub-test".into()));
            assert_eq!(payload.as_ref(), b"streamed message");

            // Ack the message.
            subscriber
                .send_frame(Frame::Ack { message_id })
                .await
                .unwrap();
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    // Unsubscribe.
    subscriber
        .send_frame(Frame::Unsubscribe {
            topic: TopicName("sub-test".into()),
            group: None,
        })
        .await
        .unwrap();

    let unsub_ack = tokio::time::timeout(Duration::from_secs(3), subscriber.recv_frame())
        .await
        .expect("timed out waiting for UnsubscribeAck")
        .unwrap();
    assert!(matches!(unsub_ack, Frame::UnsubscribeAck { .. }));

    subscriber.close().await.unwrap();
    publisher.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_subscribe_at_offset() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);

    // First, publish some messages.
    let mut publisher = FafbClient::connect(&config).await.unwrap();
    for i in 0..5u32 {
        publisher
            .publish(
                TopicName("sub-at-test".into()),
                Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }

    // Subscribe from offset 3.
    let mut subscriber = FafbClient::connect(&config).await.unwrap();
    subscriber
        .send_frame(Frame::SubscribeAt {
            topic: TopicName("sub-at-test".into()),
            offset: TopicOffset(3),
            group: None,
        })
        .await
        .unwrap();

    let ack = subscriber.recv_frame().await.unwrap();
    match &ack {
        Frame::SubscribeAck { start_offset, .. } => {
            assert_eq!(*start_offset, TopicOffset(3));
        }
        other => panic!("expected SubscribeAck, got: {other:?}"),
    }

    // Should receive msg-3 and msg-4.
    for expected in ["msg-3", "msg-4"] {
        let deliver = tokio::time::timeout(Duration::from_secs(5), subscriber.recv_frame())
            .await
            .expect("timed out waiting for Deliver")
            .unwrap();

        match deliver {
            Frame::Deliver {
                payload,
                message_id,
                ..
            } => {
                assert_eq!(payload.as_ref(), expected.as_bytes());
                subscriber
                    .send_frame(Frame::Ack { message_id })
                    .await
                    .unwrap();
            }
            other => panic!("expected Deliver, got: {other:?}"),
        }
    }

    subscriber.close().await.unwrap();
    publisher.close().await.unwrap();
    server.shutdown();
}

#[tokio::test]
async fn fafb_client_subscribe_with_group() {
    use fafb_q::client::{ClientConfig, FafbClient};

    let server = start_test_tcp_server().await;
    let config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);

    // Two subscribers in the same consumer group.
    let mut sub_a = FafbClient::connect(&config).await.unwrap();
    sub_a
        .send_frame(Frame::Subscribe {
            topic: TopicName("group-sub-test".into()),
            group: Some(GroupName("cli-group".into())),
        })
        .await
        .unwrap();
    let ack_a = sub_a.recv_frame().await.unwrap();
    assert!(matches!(ack_a, Frame::SubscribeAck { .. }));

    let mut sub_b = FafbClient::connect(&config).await.unwrap();
    sub_b
        .send_frame(Frame::Subscribe {
            topic: TopicName("group-sub-test".into()),
            group: Some(GroupName("cli-group".into())),
        })
        .await
        .unwrap();
    let ack_b = sub_b.recv_frame().await.unwrap();
    assert!(matches!(ack_b, Frame::SubscribeAck { .. }));

    // Publish messages.
    let mut publisher = FafbClient::connect(&config).await.unwrap();
    for i in 0..4u32 {
        publisher
            .publish(
                TopicName("group-sub-test".into()),
                Bytes::from(format!("group-msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }

    // Both subscribers together should receive all messages (each message to one subscriber).
    let mut received_a = Vec::new();
    let mut received_b = Vec::new();

    for _ in 0..4 {
        tokio::select! {
            frame = sub_a.recv_frame() => {
                if let Frame::Deliver { payload, message_id, .. } = frame.unwrap() {
                    received_a.push(String::from_utf8(payload.to_vec()).unwrap());
                    let _ = sub_a.send_frame(Frame::Ack { message_id }).await;
                }
            }
            frame = sub_b.recv_frame() => {
                if let Frame::Deliver { payload, message_id, .. } = frame.unwrap() {
                    received_b.push(String::from_utf8(payload.to_vec()).unwrap());
                    let _ = sub_b.send_frame(Frame::Ack { message_id }).await;
                }
            }
        }
    }

    // Together they should have all 4 messages.
    let total = received_a.len() + received_b.len();
    assert_eq!(total, 4, "expected 4 total messages, got {total}");

    // Neither should have all 4 (consumer group distributes).
    // Note: with only 4 messages it's possible but unlikely one gets all 4.
    // Just verify the total count is correct.

    sub_a.close().await.unwrap();
    sub_b.close().await.unwrap();
    publisher.close().await.unwrap();
    server.shutdown();
}

// ---------------------------------------------------------------------------
// FafbClient auto-Pong tests (Task 3)
// ---------------------------------------------------------------------------

#[tokio::test]
async fn client_auto_pong_on_server_ping() {
    use fafb_q::client::{ClientConfig, FafbClient};
    use fafb_q::codec::MessageCodec;
    use fafb_q::codec::frame::ErrorCode;
    use futures::{SinkExt, StreamExt};
    use tokio_util::codec::Framed;

    // Bind a raw TCP listener to act as a fake server
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();

    let server_task = tokio::spawn(async move {
        let (stream, _) = listener.accept().await.unwrap();
        let codec = MessageCodec::new(1_048_576);
        let mut framed = Framed::new(stream, codec);

        // Receive Connect, send ConnectAck
        let _connect = framed.next().await.unwrap().unwrap();
        framed
            .send(Frame::ConnectAck {
                success: true,
                message: String::new(),
                server_capabilities: 0,
            })
            .await
            .unwrap();

        // Send Ping — client should auto-reply Pong
        framed.send(Frame::Ping).await.unwrap();

        // Send an Error frame so the client has a real frame to return
        framed
            .send(Frame::Error {
                code: ErrorCode::InternalError,
                message: "test-marker".into(),
            })
            .await
            .unwrap();

        // Verify we received Pong
        let pong = framed.next().await.unwrap().unwrap();
        assert!(matches!(pong, Frame::Pong), "expected Pong, got {pong:?}");
    });

    // Connect FafbClient to the fake server
    let mut config = ClientConfig::new(format!("{addr}"), "test-ns", "test-key");
    config.ping_interval = std::time::Duration::ZERO; // disable keepalive for this test

    let mut client = FafbClient::connect(&config).await.unwrap();

    // recv_frame() should skip the Ping and return the Error frame
    let frame = client.recv_frame().await.unwrap();
    assert!(
        matches!(&frame, Frame::Error { message, .. } if message == "test-marker"),
        "expected Error frame with test-marker, got {frame:?}"
    );

    server_task.await.unwrap();
}

// ---------------------------------------------------------------------------
// Keepalive integration tests
// ---------------------------------------------------------------------------

/// Verify that the client keepalive prevents the server from closing an idle
/// connection before the idle timeout expires.
#[tokio::test]
async fn keepalive_prevents_idle_timeout() {
    use fafb_q::client::{ClientConfig, FafbClient};

    // Server: idle timeout 3 seconds.
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        TcpConfig {
            idle_timeout_secs: 3,
            ..TcpConfig::default()
        },
    )
    .await;

    // Client: keepalive ping every 1 second (well below the 3s idle timeout).
    let mut config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    config.ping_interval = Duration::from_secs(1);
    let mut client = FafbClient::connect(&config).await.unwrap();

    // Publish a message and subscribe to it.
    let topic = TopicName("keepalive-test".into());
    client
        .publish(
            topic.clone(),
            bytes::Bytes::from_static(b"initial"),
            vec![],
            0,
        )
        .await
        .unwrap();

    client
        .send_frame(Frame::Subscribe {
            topic: topic.clone(),
            group: None,
        })
        .await
        .unwrap();
    let sub_ack = client.recv_frame().await.unwrap();
    assert!(
        matches!(sub_ack, Frame::SubscribeAck { .. }),
        "expected SubscribeAck, got: {sub_ack:?}"
    );

    // Sleep well past the idle timeout — the keepalive should keep the
    // connection alive throughout.
    tokio::time::sleep(Duration::from_secs(8)).await;

    // Connection should still be alive: publish another message successfully.
    let result = client
        .publish(
            topic.clone(),
            bytes::Bytes::from_static(b"after idle"),
            vec![],
            0,
        )
        .await;
    assert!(
        result.is_ok(),
        "expected publish to succeed after 8s idle (keepalive active), got: {result:?}"
    );

    client.close().await.unwrap();
    server.shutdown();
}

/// Verify that a connection with keepalive disabled is closed by the server
/// after the idle timeout.
#[tokio::test]
async fn no_keepalive_when_disabled() {
    use fafb_q::client::{ClientConfig, FafbClient};

    // Server: idle timeout 2 seconds.
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        TcpConfig {
            idle_timeout_secs: 2,
            ..TcpConfig::default()
        },
    )
    .await;

    // Client: keepalive disabled.
    let mut config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    config.ping_interval = Duration::ZERO;
    let mut client = FafbClient::connect(&config).await.unwrap();

    // Sleep past the idle timeout — server should close the connection.
    tokio::time::sleep(Duration::from_secs(4)).await;

    // The connection should now be dead: either send or recv should fail.
    let result = client
        .publish(
            TopicName("dead-test".into()),
            bytes::Bytes::from_static(b"should fail"),
            vec![],
            0,
        )
        .await;
    assert!(
        result.is_err(),
        "expected connection to be dead after idle timeout with keepalive disabled, but publish succeeded"
    );

    server.shutdown();
}

/// Verify that active traffic (publishing faster than the ping interval)
/// keeps the connection alive without needing explicit Ping frames.
#[tokio::test]
async fn keepalive_suppressed_by_traffic() {
    use fafb_q::client::{ClientConfig, FafbClient};

    // Server: idle timeout 3 seconds.
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        TcpConfig {
            idle_timeout_secs: 3,
            ..TcpConfig::default()
        },
    )
    .await;

    // Client: ping interval 1 second, but we will publish every 500ms —
    // traffic should reset the idle timer without relying on Ping.
    let mut config = ClientConfig::new(&server.addr, TEST_NS, TEST_KEY);
    config.ping_interval = Duration::from_secs(1);
    let mut client = FafbClient::connect(&config).await.unwrap();

    let topic = TopicName("traffic-test".into());

    // Send Publish frames every 500ms for 5 seconds.
    let iterations = 10u32;
    for i in 0..iterations {
        client
            .send_frame(Frame::Publish {
                topic: topic.clone(),
                payload: bytes::Bytes::from(format!("msg-{i}")),
                headers: vec![],
                ttl_ms: 0,
            })
            .await
            .unwrap();
        // Drain the PublishAck so the stream stays clean.
        let ack = client.recv_frame().await.unwrap();
        assert!(
            matches!(ack, Frame::PublishAck { .. }),
            "expected PublishAck for msg-{i}, got: {ack:?}"
        );
        tokio::time::sleep(Duration::from_millis(500)).await;
    }

    // After the loop the connection should still be alive.
    let result = client
        .publish(
            topic.clone(),
            bytes::Bytes::from_static(b"final"),
            vec![],
            0,
        )
        .await;
    assert!(
        result.is_ok(),
        "expected connection to be alive after continuous traffic, got: {result:?}"
    );

    client.close().await.unwrap();
    server.shutdown();
}

/// Verify that inflight messages belonging to a dead consumer are redelivered to
/// another member of the same consumer group once the idle timeout fires.
///
/// Flow:
///   1. Start server with a short idle_timeout_secs (2s).
///   2. Publish one message to a topic.
///   3. Consumer A subscribes with a consumer group at offset 0, receives the
///      message, but does NOT ack — simulating a crash.
///   4. Drop Consumer A (simulates dead/crashed client; no clean unsubscribe).
///   5. Wait long enough for the idle timeout to fire and the inflight sweep to
///      detect the orphaned message (~4s).
///   6. Consumer B joins the same group at offset 0 and must receive the
///      redelivered message within a reasonable timeout.
#[tokio::test]
async fn dead_client_inflight_redelivered_to_consumer_group() {
    let server = start_test_tcp_server_with_config(
        &[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ],
        TcpConfig {
            idle_timeout_secs: 2,
            ..TcpConfig::default()
        },
    )
    .await;

    let topic = TopicName("redeliver-group".into());
    let group = GroupName("redeliver-workers".into());

    // -- Step 1: publish a message --
    let mut publisher = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    publisher
        .send(Frame::Publish {
            topic: topic.clone(),
            headers: vec![],
            ttl_ms: 0,
            payload: bytes::Bytes::from_static(b"must-redeliver"),
        })
        .await;
    let pub_ack = publisher.recv().await;
    assert!(
        matches!(pub_ack, Frame::PublishAck { .. }),
        "expected PublishAck, got: {pub_ack:?}"
    );

    // -- Step 2: Consumer A subscribes at offset 0 with a consumer group --
    let mut consumer_a = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    consumer_a
        .send(Frame::SubscribeAt {
            topic: topic.clone(),
            offset: TopicOffset(0),
            group: Some(group.clone()),
        })
        .await;
    let sub_ack = consumer_a.recv().await;
    assert!(
        matches!(sub_ack, Frame::SubscribeAck { .. }),
        "expected SubscribeAck, got: {sub_ack:?}"
    );

    // Consumer A receives the message — deliberately does NOT send Ack.
    let deliver = consumer_a
        .recv_timeout(Duration::from_secs(3))
        .await
        .expect("Consumer A expected Deliver frame");
    match &deliver {
        Frame::Deliver { payload, .. } => {
            assert_eq!(payload.as_ref(), b"must-redeliver");
        }
        other => panic!("expected Deliver, got: {other:?}"),
    }

    // -- Step 3: drop Consumer A — simulates a crashed/dead client --
    drop(consumer_a);

    // -- Step 4: wait for idle timeout (2s) + inflight sweep (runs every ~1s) + buffer --
    tokio::time::sleep(Duration::from_secs(4)).await;

    // -- Step 5: Consumer B joins the same group at offset 0 --
    let mut consumer_b = TestClient::connect_and_auth(&server.addr, TEST_NS, TEST_KEY).await;
    consumer_b
        .send(Frame::SubscribeAt {
            topic: topic.clone(),
            offset: TopicOffset(0),
            group: Some(group.clone()),
        })
        .await;
    // The server may send a non-fatal OffsetNotAvailable warning before the SubscribeAck
    // when the requested offset has already been consumed/advanced.  Skip it.
    let sub_ack_b = loop {
        let frame = consumer_b.recv().await;
        match frame {
            Frame::SubscribeAck { .. } => break frame,
            Frame::Error {
                code: fafb_q::codec::frame::ErrorCode::OffsetNotAvailable,
                ..
            } => {
                // Non-fatal warning — the server will still send a SubscribeAck.
                continue;
            }
            other => panic!("expected SubscribeAck for Consumer B, got: {other:?}"),
        }
    };
    assert!(
        matches!(sub_ack_b, Frame::SubscribeAck { .. }),
        "expected SubscribeAck for Consumer B, got: {sub_ack_b:?}"
    );

    // Consumer B should receive the redelivered message within a reasonable timeout.
    let redelivered = tokio::time::timeout(Duration::from_secs(5), async {
        loop {
            if let Some(frame) = consumer_b.recv_timeout(Duration::from_secs(5)).await {
                if matches!(frame, Frame::Deliver { .. }) {
                    return frame;
                }
                // Skip any non-Deliver frames (e.g. Warning).
            } else {
                panic!("stream closed before redelivery");
            }
        }
    })
    .await
    .expect("timed out waiting for redelivered message on Consumer B");

    match redelivered {
        Frame::Deliver { payload, .. } => {
            assert_eq!(
                payload.as_ref(),
                b"must-redeliver",
                "redelivered payload mismatch"
            );
        }
        other => panic!("expected Deliver for redelivery, got: {other:?}"),
    }

    server.shutdown();
}
