mod common;

use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::time::Duration;

use bytes::Bytes;
use tempfile::TempDir;
use tokio::task::JoinSet;

use fafb_q::codec::Frame;
use fafb_q::codec::frame::{
    CAP_GOING_AWAY, CAP_RECONNECT, ErrorCode, ReconnectStatus, SubscriptionEntry,
    SubscriptionStatus,
};
use fafb_q::config::{DeployConfig, ReloadState, ServeArgs};
use fafb_q::queue::{EngineConfig, QueueEngine};
use fafb_q::server::state::ServerState;
use fafb_q::types::{ApiKey, GroupName, MessageId, NamespaceId, TopicName, TopicOffset};

use common::{TEST_KEY, TEST_NS, TestClient, start_admin_test_server, start_deploy_test_server};

// Re-export for partial failure test that builds server manually
use fafb_q::auth::{MockAuthenticator, PermissionSet};
use fafb_q::namespace::{DefaultNamespaceManager, ManagerConfig};
use fafb_q::storage::StorageBackend;
use fafb_q::storage::fs::FsStorage;

// -- Local helpers --

fn default_engine_config() -> EngineConfig {
    EngineConfig {
        max_depth: 100_000,
        ..Default::default()
    }
}

/// Publish `count` messages to `topic` and collect all PublishAcks.
async fn publish_messages(client: &mut TestClient, topic: &str, count: usize) {
    let topic_name = TopicName(topic.into());
    let payload = vec![0xABu8; 16];
    for _ in 0..count {
        client
            .send(Frame::Publish {
                topic: topic_name.clone(),
                payload: Bytes::from(payload.clone()),
                headers: vec![],
                ttl_ms: 0,
            })
            .await;
        let ack = client.recv().await;
        assert!(
            matches!(ack, Frame::PublishAck { .. }),
            "expected PublishAck, got: {ack:?}"
        );
    }
}

/// Subscribe at a specific offset and consume up to `count` Deliver frames, ack each.
async fn subscribe_at_and_consume(
    client: &mut TestClient,
    topic: &str,
    offset: TopicOffset,
    group: Option<&str>,
    count: usize,
    timeout: Duration,
) -> Vec<(TopicOffset, MessageId)> {
    client
        .send(Frame::SubscribeAt {
            topic: TopicName(topic.into()),
            offset,
            group: group.map(|g| GroupName(g.into())),
        })
        .await;

    // May get OffsetNotAvailable warning before SubscribeAck
    loop {
        let frame = client.recv_timeout(Duration::from_secs(5)).await;
        match frame {
            Some(Frame::SubscribeAck { .. }) => break,
            Some(Frame::Error {
                code: ErrorCode::OffsetNotAvailable,
                ..
            }) => continue,
            other => panic!("expected SubscribeAck, got: {other:?}"),
        }
    }

    let mut results = Vec::new();
    for _ in 0..count {
        match client.recv_timeout(timeout).await {
            Some(Frame::Deliver {
                offset, message_id, ..
            }) => {
                client.send(Frame::Ack { message_id }).await;
                results.push((offset, message_id));
            }
            _ => break,
        }
    }
    results
}

/// Collect Deliver frames without acking (for post-restart verification).
/// Returns (offset, message_id) pairs.
async fn collect_delivers(
    client: &mut TestClient,
    count: usize,
    timeout: Duration,
) -> Vec<(TopicOffset, MessageId)> {
    let mut results = Vec::new();
    for _ in 0..count {
        match client.recv_timeout(timeout).await {
            Some(Frame::Deliver {
                offset, message_id, ..
            }) => {
                results.push((offset, message_id));
            }
            _ => break,
        }
    }
    results
}

// =============================================================================
// Test 14.4.1: Single-instance deploy — basic
// =============================================================================

#[tokio::test]
async fn deploy_basic_single_instance() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let topics = ["topic-a", "topic-b", "topic-c"];

    // Connect 5 publishers
    let mut publishers = Vec::new();
    for _ in 0..5 {
        publishers.push(
            TestClient::connect_and_auth_with_caps(
                &server.tcp_addr,
                TEST_NS,
                TEST_KEY,
                CAP_GOING_AWAY,
            )
            .await,
        );
    }

    // Connect 10 subscribers with CAP_GOING_AWAY, spread across 3 topics
    let mut subscribers = Vec::new();
    for i in 0..10 {
        let mut client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        let topic = TopicName(topics[i % 3].into());
        client.send(Frame::Subscribe { topic, group: None }).await;
        let ack = client.recv().await;
        assert!(matches!(ack, Frame::SubscribeAck { .. }));
        subscribers.push(client);
    }

    // Publish 200 messages per publisher (1000 total)
    for (i, pub_client) in publishers.iter_mut().enumerate() {
        let topic = topics[i % 3];
        publish_messages(pub_client, topic, 200).await;
    }

    // Let subscribers receive some messages
    tokio::time::sleep(Duration::from_millis(500)).await;

    // Check /health/ready is 200 before drain
    let resp = reqwest::get(format!("{}/health/ready", server.http_base_url))
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Trigger drain
    let _ = server.engine.force_flush().await;
    server.lifecycle.transition_to(ServerState::Draining);

    // Check /health/ready becomes 503
    tokio::time::sleep(Duration::from_millis(100)).await;
    let resp = reqwest::get(format!("{}/health/ready", server.http_base_url))
        .await
        .unwrap();
    assert_eq!(resp.status(), 503);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "draining");

    // All subscribers should receive GoingAway (may get Deliver frames first)
    for sub in &mut subscribers {
        loop {
            let frame = sub.recv_timeout(Duration::from_secs(3)).await;
            match frame {
                Some(Frame::GoingAway {
                    deadline_ms,
                    message,
                    ..
                }) => {
                    assert!(deadline_ms > 0);
                    assert!(!message.is_empty());
                    break;
                }
                Some(Frame::Deliver { .. }) => continue, // skip pending delivers
                other => panic!("expected GoingAway (or Deliver), got: {other:?}"),
            }
        }
    }

    // Transition to ShuttingDown
    server.lifecycle.transition_to(ServerState::ShuttingDown);
    tokio::time::sleep(Duration::from_millis(300)).await;
}

// =============================================================================
// Test 14.4.6: Legacy client (no caps) during deploy
// =============================================================================

#[tokio::test]
async fn deploy_legacy_vs_capable_client() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Legacy client — no capabilities
    let mut legacy = TestClient::connect_and_auth(&server.tcp_addr, TEST_NS, TEST_KEY).await;
    legacy
        .send(Frame::Subscribe {
            topic: TopicName("test-topic".into()),
            group: None,
        })
        .await;
    let _ = legacy.recv().await; // SubscribeAck

    // Capable client — CAP_GOING_AWAY
    let mut capable =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    capable
        .send(Frame::Subscribe {
            topic: TopicName("test-topic".into()),
            group: None,
        })
        .await;
    let _ = capable.recv().await; // SubscribeAck

    // Trigger drain
    server.lifecycle.transition_to(ServerState::Draining);
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Legacy client should receive Error with ServerDraining
    let frame = legacy.recv_timeout(Duration::from_secs(2)).await;
    match frame {
        Some(Frame::Error { code, .. }) => {
            assert_eq!(code, ErrorCode::ServerDraining);
        }
        other => panic!("legacy client expected Error(ServerDraining), got: {other:?}"),
    }

    // Capable client should receive GoingAway (connection stays open)
    let frame = capable.recv_timeout(Duration::from_secs(2)).await;
    assert!(
        matches!(frame, Some(Frame::GoingAway { .. })),
        "capable client expected GoingAway, got: {frame:?}"
    );

    // Capable client can still publish during drain
    capable
        .send(Frame::Publish {
            topic: TopicName("test-topic".into()),
            payload: Bytes::from_static(b"drain-msg"),
            headers: vec![],
            ttl_ms: 0,
        })
        .await;
    let ack = capable.recv_timeout(Duration::from_secs(2)).await;
    assert!(
        matches!(ack, Some(Frame::PublishAck { .. })),
        "expected PublishAck during drain, got: {ack:?}"
    );

    // ShuttingDown force-closes capable client
    server.lifecycle.transition_to(ServerState::ShuttingDown);
    tokio::time::sleep(Duration::from_millis(200)).await;
}

// =============================================================================
// Test 14.4.5: Publish during drain
// =============================================================================

#[tokio::test]
async fn publish_during_drain() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Publisher + subscriber, both capable
    let mut publisher =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;

    let mut subscriber =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    subscriber
        .send(Frame::Subscribe {
            topic: TopicName("drain-pub".into()),
            group: None,
        })
        .await;
    let _ = subscriber.recv().await; // SubscribeAck

    // Trigger drain
    server.lifecycle.transition_to(ServerState::Draining);
    tokio::time::sleep(Duration::from_millis(200)).await;

    // Both should get GoingAway
    for client in [&mut publisher, &mut subscriber] {
        let frame = client.recv_timeout(Duration::from_secs(2)).await;
        assert!(
            matches!(frame, Some(Frame::GoingAway { .. })),
            "expected GoingAway, got: {frame:?}"
        );
    }

    // Publish 100 messages during drain
    publish_messages(&mut publisher, "drain-pub", 100).await;

    // Subscriber should receive all 100 messages
    let delivers = collect_delivers(&mut subscriber, 100, Duration::from_secs(5)).await;
    assert_eq!(
        delivers.len(),
        100,
        "subscriber should receive all 100 messages during drain"
    );

    // Subscribe attempt during drain should be rejected (using an already-connected client)
    subscriber
        .send(Frame::Subscribe {
            topic: TopicName("drain-pub-2".into()),
            group: None,
        })
        .await;
    let frame = subscriber.recv_timeout(Duration::from_secs(2)).await;
    match frame {
        Some(Frame::Error { code, .. }) => {
            assert_eq!(code, ErrorCode::ServerDraining);
        }
        other => panic!("subscribe during drain should return ServerDraining, got: {other:?}"),
    }

    server.lifecycle.transition_to(ServerState::ShuttingDown);
    tokio::time::sleep(Duration::from_millis(100)).await;
}

// =============================================================================
// Test 14.4.4: Drain timeout force-closes slow client
// =============================================================================

#[tokio::test]
async fn drain_timeout_force_closes_slow_client() {
    let tmp = TempDir::new().unwrap();
    let server = start_deploy_test_server(
        tmp.path(),
        DeployConfig {
            drain_timeout_secs: 3,
            ..Default::default()
        },
        default_engine_config(),
    )
    .await;

    let mut client =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    client
        .send(Frame::Subscribe {
            topic: TopicName("slow".into()),
            group: None,
        })
        .await;
    let _ = client.recv().await; // SubscribeAck

    let start = tokio::time::Instant::now();

    // Trigger graceful drain (blocks until drain_timeout or all connections close).
    // Client will receive GoingAway but won't disconnect, so drain_timeout fires.
    server.graceful_drain().await;

    let elapsed = start.elapsed();

    // Should have waited at least drain_timeout (3s)
    assert!(
        elapsed.as_secs() >= 3,
        "should wait at least drain_timeout (3s), got {:.1}s",
        elapsed.as_secs_f64()
    );
    assert!(
        elapsed.as_secs() <= 6,
        "should not wait much longer than drain_timeout, got {:.1}s",
        elapsed.as_secs_f64()
    );

    // After ShuttingDown, drain any remaining frames (GoingAway, Error, etc.)
    // and verify the connection eventually closes
    tokio::time::sleep(Duration::from_millis(500)).await;
    loop {
        match client.recv_timeout(Duration::from_secs(2)).await {
            Some(Frame::GoingAway { .. }) => continue,
            Some(Frame::Error { .. }) => break, // server force-closed
            None => break,                      // TCP connection closed
            Some(other) => panic!("unexpected frame after shutdown: {other:?}"),
        }
    }
}

// =============================================================================
// Test 14.4.10: Storage durability through drain
// =============================================================================

#[tokio::test]
async fn storage_durability_through_drain() {
    let tmp = TempDir::new().unwrap();

    // Phase 1: Publish with auto-flush disabled
    {
        let server = start_deploy_test_server(
            tmp.path(),
            DeployConfig::default(),
            EngineConfig {
                max_depth: 100_000,
                flush_interval_ms: 60_000,
                batch_flush_size_bytes: u64::MAX,
                ..Default::default()
            },
        )
        .await;

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

        publish_messages(&mut client, "durable", 10_000).await;

        // shutdown calls force_flush internally
        server.shutdown().await;
    }

    // Phase 2: Restart and verify all data survived
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

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

        client
            .send(Frame::SubscribeAt {
                topic: TopicName("durable".into()),
                offset: TopicOffset(0),
                group: None,
            })
            .await;
        let ack = client.recv().await;
        assert!(matches!(ack, Frame::SubscribeAck { .. }));

        let delivers = collect_delivers(&mut client, 10_000, Duration::from_secs(10)).await;
        assert_eq!(
            delivers.len(),
            10_000,
            "all 10,000 messages should survive drain (force_flush)"
        );

        server.shutdown().await;
    }
}

// =============================================================================
// Test 14.4.2: Offset continuity across restart
// =============================================================================

#[tokio::test]
async fn offset_continuity_across_restart() {
    let tmp = TempDir::new().unwrap();

    // Phase 1: Publish 1000, consume 500, ack all
    let last_acked_offset;
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        publish_messages(&mut pub_client, "events", 1000).await;

        // Wait for engine to have messages available
        tokio::time::sleep(Duration::from_millis(200)).await;

        let mut sub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        // Subscribe from offset 0 (messages were published before subscribe)
        let consumed = subscribe_at_and_consume(
            &mut sub_client,
            "events",
            TopicOffset(0),
            None,
            500,
            Duration::from_secs(5),
        )
        .await;
        assert!(
            consumed.len() >= 500,
            "should consume at least 500 messages, got {}",
            consumed.len()
        );
        last_acked_offset = consumed.last().unwrap().0;

        server.shutdown().await;
    }

    // Phase 2: Restart, subscribe at last_acked + 1
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

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

        let resume_offset = TopicOffset(last_acked_offset.0 + 1);
        let remaining_count = 1000 - resume_offset.0 as usize;
        let consumed = subscribe_at_and_consume(
            &mut client,
            "events",
            resume_offset,
            None,
            remaining_count,
            Duration::from_secs(5),
        )
        .await;

        assert_eq!(
            consumed.len(),
            remaining_count,
            "should receive remaining {} messages",
            remaining_count
        );

        // First delivered offset should match resume_offset
        assert_eq!(
            consumed[0].0, resume_offset,
            "first delivered offset should be {}",
            resume_offset.0
        );

        // Verify contiguous offsets (no gaps)
        for i in 1..consumed.len() {
            assert_eq!(
                consumed[i].0.0,
                consumed[i - 1].0.0 + 1,
                "offsets should be contiguous at index {}",
                i
            );
        }

        server.shutdown().await;
    }
}

// =============================================================================
// Test 14.4.3: Consumer group continuity across restart
//
// Consumer group state is in-memory only — not persisted. After restart, the
// group is fresh. This test verifies that 3 independent subscribers (simulating
// group members that track their own offsets) can collectively consume all
// messages across a restart with no gaps.
// =============================================================================

#[tokio::test]
async fn consumer_group_continuity_across_restart() {
    let tmp = TempDir::new().unwrap();
    let topic = "orders";
    let total_messages: usize = 1000;

    // Phase 1: Publish 1000 messages, have 3 subscribers each consume a portion
    let mut last_offsets: Vec<u64> = Vec::new(); // last acked offset per subscriber
    let mut total_pre = 0usize;
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        publish_messages(&mut pub_client, topic, total_messages).await;

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

        // 3 individual subscribers, each consuming a contiguous range.
        // Sub 1: offsets 0..199 (200 msgs), Sub 2: 200..399 (200 msgs), Sub 3: 400..599 (200 msgs)
        let ranges = [(0, 200), (200, 200), (400, 200)];
        for (start, count) in &ranges {
            let mut sub = TestClient::connect_and_auth_with_caps(
                &server.tcp_addr,
                TEST_NS,
                TEST_KEY,
                CAP_GOING_AWAY,
            )
            .await;
            let consumed = subscribe_at_and_consume(
                &mut sub,
                topic,
                TopicOffset(*start),
                None,
                *count,
                Duration::from_secs(5),
            )
            .await;
            assert_eq!(
                consumed.len(),
                *count,
                "subscriber at offset {start} should consume {count}"
            );
            last_offsets.push(consumed.last().unwrap().0.0);
            total_pre += consumed.len();
        }

        server.shutdown().await;
    }

    // Phase 2: Restart, each subscriber resumes from its last acked offset + 1
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        // Subscriber resumes from the highest acked offset + 1
        let max_acked = *last_offsets.iter().max().unwrap();
        let resume_offset = TopicOffset(max_acked + 1);
        let remaining = total_messages - resume_offset.0 as usize;

        let mut sub = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        let consumed = subscribe_at_and_consume(
            &mut sub,
            topic,
            resume_offset,
            None,
            remaining,
            Duration::from_secs(10),
        )
        .await;

        assert_eq!(
            consumed.len(),
            remaining,
            "should receive remaining {remaining} messages after restart"
        );

        // Verify first offset matches resume point
        assert_eq!(consumed[0].0, resume_offset);

        // Verify contiguous offsets
        for i in 1..consumed.len() {
            assert_eq!(consumed[i].0.0, consumed[i - 1].0.0 + 1);
        }

        // Total coverage: pre + post = all messages
        assert_eq!(
            total_pre + consumed.len(),
            total_messages,
            "pre ({total_pre}) + post ({}) should equal {total_messages}",
            consumed.len()
        );

        server.shutdown().await;
    }
}

// =============================================================================
// Test 14.4.7: Reconnect frame — full restore (multi-topic)
// =============================================================================

#[tokio::test]
async fn reconnect_full_restore_multi_topic() {
    let tmp = TempDir::new().unwrap();
    let topics = ["alpha", "beta", "gamma"];
    let msgs_per_topic = 100;
    let consume_count = 50;

    // Phase 1: Subscribe to 3 topics, consume 50 from each, ack
    let mut last_acked: Vec<(String, u64)> = Vec::new();
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        // Publish to each topic
        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY | CAP_RECONNECT,
        )
        .await;
        for topic in &topics {
            publish_messages(&mut pub_client, topic, msgs_per_topic).await;
        }

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

        // Subscribe from offset 0 and consume from each topic
        for topic in &topics {
            let mut sub = TestClient::connect_and_auth_with_caps(
                &server.tcp_addr,
                TEST_NS,
                TEST_KEY,
                CAP_GOING_AWAY | CAP_RECONNECT,
            )
            .await;
            let consumed = subscribe_at_and_consume(
                &mut sub,
                topic,
                TopicOffset(0),
                None,
                consume_count,
                Duration::from_secs(5),
            )
            .await;
            assert!(!consumed.is_empty(), "should consume messages from {topic}");
            last_acked.push((topic.to_string(), consumed.last().unwrap().0.0));
        }

        server.shutdown().await;
    }

    // Phase 2: Restart, send Reconnect frame
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut client = TestClient::connect(&server.tcp_addr).await;

        let subscriptions: Vec<SubscriptionEntry> = last_acked
            .iter()
            .map(|(topic, offset)| SubscriptionEntry {
                topic: TopicName(topic.clone()),
                group: None,
                last_acked_offset: TopicOffset(*offset),
            })
            .collect();

        client
            .send(Frame::Reconnect {
                version: 1,
                namespace: NamespaceId(TEST_NS.into()),
                api_key: ApiKey(TEST_KEY.into()),
                capabilities: CAP_GOING_AWAY | CAP_RECONNECT,
                subscriptions,
            })
            .await;

        let ack = client.recv().await;
        match &ack {
            Frame::ReconnectAck {
                status,
                results,
                server_capabilities,
                ..
            } => {
                assert_eq!(*status, ReconnectStatus::Success);
                assert_eq!(results.len(), 3);
                assert!(*server_capabilities & CAP_GOING_AWAY != 0);
                assert!(*server_capabilities & CAP_RECONNECT != 0);

                for (i, result) in results.iter().enumerate() {
                    assert_eq!(
                        result.status,
                        SubscriptionStatus::Restored,
                        "topic {} should be Restored",
                        topics[i]
                    );
                    assert_eq!(
                        result.actual_start_offset.0,
                        last_acked[i].1 + 1,
                        "topic {} should resume at offset {}",
                        topics[i],
                        last_acked[i].1 + 1
                    );
                }
            }
            other => panic!("expected ReconnectAck, got: {other:?}"),
        }

        // Collect remaining delivers from all 3 topics
        let total_remaining: usize = last_acked
            .iter()
            .map(|(_, offset)| msgs_per_topic - (*offset as usize + 1))
            .sum();
        let delivers =
            collect_delivers(&mut client, total_remaining, Duration::from_secs(10)).await;
        assert_eq!(
            delivers.len(),
            total_remaining,
            "should receive {} remaining messages across all topics",
            total_remaining
        );

        server.shutdown().await;
    }
}

// =============================================================================
// Test 14.4.8: Reconnect frame — partial failure (permission denied)
// =============================================================================

#[tokio::test]
async fn reconnect_partial_failure() {
    let tmp = TempDir::new().unwrap();
    let topics = ["alpha", "beta"];
    let msgs_per_topic = 50;
    let consume_count = 25;

    // Phase 1: Publish and consume from 2 topics (with full permissions)
    let mut last_acked: Vec<(String, u64)> = Vec::new();
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY | CAP_RECONNECT,
        )
        .await;
        for topic in &topics {
            publish_messages(&mut pub_client, topic, msgs_per_topic).await;
        }

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

        for topic in &topics {
            let mut sub = TestClient::connect_and_auth_with_caps(
                &server.tcp_addr,
                TEST_NS,
                TEST_KEY,
                CAP_GOING_AWAY | CAP_RECONNECT,
            )
            .await;
            let consumed = subscribe_at_and_consume(
                &mut sub,
                topic,
                TopicOffset(0),
                None,
                consume_count,
                Duration::from_secs(5),
            )
            .await;
            assert!(!consumed.is_empty(), "should consume messages from {topic}");
            last_acked.push((topic.to_string(), consumed.last().unwrap().0.0));
        }

        server.shutdown().await;
    }

    // Phase 2: Restart with publish-only permissions (no Subscribe), send Reconnect
    // This simulates a scenario where permissions changed between deploys.
    {
        let storage = std::sync::Arc::new(FsStorage::new(tmp.path()));
        storage.init().await.expect("storage init failed");
        let engine = std::sync::Arc::new(fafb_q::queue::InMemoryQueueEngine::<FsStorage>::new(
            std::sync::Arc::clone(&storage),
            default_engine_config(),
        ));
        engine.init().await.expect("engine init failed");

        let manager = std::sync::Arc::new(DefaultNamespaceManager::new(
            std::sync::Arc::clone(&engine),
            vec![NamespaceId(TEST_NS.into())],
            common::test_shared_config(),
            ManagerConfig::default(),
        ));

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

        let (lifecycle, _lifecycle_rx) = fafb_q::server::state::ServerLifecycle::new();
        let lifecycle = std::sync::Arc::new(lifecycle);
        lifecycle.set_initialized();

        let msg_counter = std::sync::Arc::new(std::sync::atomic::AtomicU64::new(1));

        let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
        let tcp_addr = tcp_listener.local_addr().unwrap().to_string();
        let tcp_config = fafb_q::config::TcpConfig {
            bind_address: "127.0.0.1".into(),
            port: tcp_listener.local_addr().unwrap().port(),
            ..Default::default()
        };
        let tcp_server = fafb_q::server::tcp::TcpServer::new(
            tcp_config,
            DeployConfig::default(),
            common::test_shared_config(),
            std::sync::Arc::clone(&manager),
            std::sync::Arc::clone(&auth),
            std::sync::Arc::clone(&msg_counter),
            std::sync::Arc::clone(&lifecycle),
            None, // no TLS
        );
        tokio::spawn(async move {
            let _ = tcp_server.run_with_listener(tcp_listener).await;
        });

        let mut client = TestClient::connect(&tcp_addr).await;

        let subscriptions: Vec<SubscriptionEntry> = last_acked
            .iter()
            .map(|(topic, offset)| SubscriptionEntry {
                topic: TopicName(topic.clone()),
                group: None,
                last_acked_offset: TopicOffset(*offset),
            })
            .collect();

        client
            .send(Frame::Reconnect {
                version: 1,
                namespace: NamespaceId(TEST_NS.into()),
                api_key: ApiKey(TEST_KEY.into()),
                capabilities: CAP_GOING_AWAY | CAP_RECONNECT,
                subscriptions,
            })
            .await;

        let ack = client.recv().await;
        match &ack {
            Frame::ReconnectAck {
                status, results, ..
            } => {
                assert_eq!(
                    *status,
                    ReconnectStatus::Partial,
                    "should be Partial due to missing Subscribe permission"
                );

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

        // No delivers should arrive (atomic rollback — no active subscriptions)
        let delivers = collect_delivers(&mut client, 10, Duration::from_secs(2)).await;
        assert!(
            delivers.is_empty(),
            "no delivers should arrive after Partial (rollback), got {}",
            delivers.len()
        );

        lifecycle.transition_to(ServerState::ShuttingDown);
        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = engine.shutdown().await;
    }
}

// =============================================================================
// Test: Data-loss recovery across restart
//
// Complements reconnect_partial_failure by testing the plan's intended scenario:
// topic data deleted from disk between deploys. Since the engine auto-creates
// topics (so Reconnect can't produce TopicNotFound), this test verifies the
// manual recovery path: SubscribeAt to surviving topics after data loss.
// =============================================================================

#[tokio::test]
async fn data_loss_recovery_across_restart() {
    let tmp = TempDir::new().unwrap();
    let topics = ["alpha", "beta", "gamma"];
    let msgs_per_topic = 50;

    // Phase 1: Publish to 3 topics
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        for topic in &topics {
            publish_messages(&mut pub_client, topic, msgs_per_topic).await;
        }

        server.shutdown().await;
    }

    // Simulate data loss: delete "beta" from disk
    let beta_dir = tmp.path().join(TEST_NS).join("beta");
    if beta_dir.exists() {
        std::fs::remove_dir_all(&beta_dir).unwrap();
    }

    // Phase 2: Restart and verify
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

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

        // alpha and gamma should still have all their messages
        for topic in ["alpha", "gamma"] {
            let consumed = subscribe_at_and_consume(
                &mut client,
                topic,
                TopicOffset(0),
                None,
                msgs_per_topic,
                Duration::from_secs(5),
            )
            .await;
            assert_eq!(
                consumed.len(),
                msgs_per_topic,
                "{topic} should have all {msgs_per_topic} messages after restart"
            );
        }

        // beta was deleted — subscribing from offset 0 should yield 0 messages
        // (the engine auto-creates an empty topic)
        let beta_delivers = subscribe_at_and_consume(
            &mut client,
            "beta",
            TopicOffset(0),
            None,
            10,
            Duration::from_secs(1),
        )
        .await;
        assert!(
            beta_delivers.is_empty(),
            "deleted topic beta should have no messages, got {}",
            beta_delivers.len()
        );

        server.shutdown().await;
    }
}

// =============================================================================
// Test 14.4.9: Thundering herd reconnect (stress, #[ignore])
// =============================================================================

#[ignore]
#[tokio::test]
async fn thundering_herd_reconnect() {
    let tmp = TempDir::new().unwrap();
    let num_clients: usize = 1000;

    // Phase 1: Start server, publish some messages
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let mut pub_client = TestClient::connect_and_auth_with_caps(
            &server.tcp_addr,
            TEST_NS,
            TEST_KEY,
            CAP_GOING_AWAY,
        )
        .await;
        publish_messages(&mut pub_client, "stress", 100).await;

        server.shutdown().await;
    }

    // Phase 2: Restart, reconnect all with jitter
    {
        let server =
            start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
                .await;

        let addr = server.tcp_addr.clone();
        let mut join_set = JoinSet::new();

        for i in 0..num_clients {
            let addr = addr.clone();
            join_set.spawn(async move {
                // Deterministic pseudo-random jitter: 0-500ms
                let jitter_ms = (i * 37 + i * i * 13) % 500;
                tokio::time::sleep(Duration::from_millis(jitter_ms as u64)).await;

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

                client
                    .send(Frame::SubscribeAt {
                        topic: TopicName("stress".into()),
                        offset: TopicOffset(0),
                        group: None,
                    })
                    .await;
                let ack = client.recv().await;
                assert!(matches!(ack, Frame::SubscribeAck { .. }));
                true
            });
        }

        // Wait for all to complete (30s timeout)
        let mut success_count = 0;
        let deadline = tokio::time::Instant::now() + Duration::from_secs(30);
        while let Ok(Some(result)) =
            tokio::time::timeout(deadline - tokio::time::Instant::now(), join_set.join_next()).await
        {
            if let Ok(true) = result {
                success_count += 1;
            }
        }

        assert_eq!(
            success_count, num_clients,
            "all {} clients should reconnect successfully",
            num_clients
        );

        server.shutdown().await;
    }
}

// -- Phase 14.5: Admin endpoint integration tests --

const ADMIN_KEY: &str = "test-admin-key-at-least-24-chars-long";

fn admin_deploy_config() -> DeployConfig {
    DeployConfig {
        drain_timeout_secs: 5,
        admin_api_key: Some(ADMIN_KEY.into()),
        ..Default::default()
    }
}

/// Spawn a background drain coordinator that handles a single DrainRequest,
/// replicating the main.rs drain logic. Takes individual components to avoid
/// constructing a full DeployTestServer.
async fn handle_drain_request(
    lifecycle: Arc<fafb_q::server::state::ServerLifecycle>,
    engine: Arc<fafb_q::queue::InMemoryQueueEngine<fafb_q::storage::fs::FsStorage>>,
    manager: Arc<
        fafb_q::namespace::DefaultNamespaceManager<
            fafb_q::queue::InMemoryQueueEngine<fafb_q::storage::fs::FsStorage>,
        >,
    >,
    default_timeout_secs: u64,
    mut drain_rx: tokio::sync::mpsc::Receiver<fafb_q::server::http::DrainRequest>,
) {
    if let Some(req) = drain_rx.recv().await {
        let timeout_secs = req.timeout_secs.unwrap_or(default_timeout_secs);

        let _ = engine.force_flush().await;
        lifecycle.begin_drain(timeout_secs);

        let accelerated = timeout_secs.saturating_div(2).clamp(1, 30);
        manager
            .set_inflight_timeout(Duration::from_secs(accelerated))
            .await;

        let _ = req.response.send(Ok(fafb_q::server::http::DrainResponse {
            active_connections: lifecycle.active_connections(),
            drain_started_wall: lifecycle.drain_started_wall().unwrap_or(0),
            drain_deadline_wall: lifecycle.drain_deadline_wall().unwrap_or(0),
        }));

        lifecycle
            .run_drain_wait(Duration::from_secs(timeout_secs))
            .await;
    }
}

#[tokio::test]
async fn admin_drain_403_no_key() {
    let tmp = TempDir::new().unwrap();
    let config = DeployConfig {
        admin_api_key: None,
        ..Default::default()
    };
    let (server, _drain_rx) =
        common::start_admin_test_server(tmp.path(), config, default_engine_config()).await;

    let client = reqwest::Client::new();

    // POST /admin/drain without admin key configured → 403
    let resp = client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth("anything")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert!(body["error"].as_str().unwrap().contains("not configured"));

    // GET /admin/drain/status → 403
    let resp = client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth("anything")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
}

#[tokio::test]
async fn admin_drain_403_wrong_key() {
    let tmp = TempDir::new().unwrap();
    let (server, _drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    let client = reqwest::Client::new();

    let resp = client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth("wrong-key-that-is-also-long-enough")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert!(body["error"].as_str().unwrap().contains("invalid"));
}

#[tokio::test]
async fn admin_drain_status_shows_running() {
    let tmp = TempDir::new().unwrap();
    let (server, _drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    let client = reqwest::Client::new();

    let resp = client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["state"], "running");
    assert_eq!(body["active_connections"], 0);
    assert!(body["drain_started_at"].is_null());
    assert!(body["drain_deadline"].is_null());
}

#[tokio::test]
async fn admin_drain_triggers_shutdown() {
    let tmp = TempDir::new().unwrap();
    let (server, drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    // Connect a TCP client with CAP_GOING_AWAY
    let mut tcp_client =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Spawn drain coordinator in background
    let drain_handle = tokio::spawn(handle_drain_request(
        Arc::clone(&server.lifecycle),
        Arc::clone(&server.engine),
        Arc::clone(&server.manager),
        server.deploy_config.drain_timeout_secs,
        drain_rx,
    ));

    let http_client = reqwest::Client::new();

    // POST /admin/drain → should succeed
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "draining");
    assert!(body["deadline"].is_string());

    // TCP client should receive GoingAway
    let frame = tcp_client.recv().await;
    assert!(
        matches!(frame, Frame::GoingAway { .. }),
        "expected GoingAway, got: {frame:?}"
    );

    // /health/ready should return 503
    let resp = http_client
        .get(format!("{}/health/ready", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 503);

    // /admin/drain/status should show draining
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["state"], "draining");

    // Disconnect TCP client to allow shutdown to proceed
    drop(tcp_client);

    // Wait for drain coordinator to complete
    tokio::time::timeout(Duration::from_secs(10), drain_handle)
        .await
        .expect("drain coordinator timed out")
        .unwrap();
}

#[tokio::test]
async fn admin_drain_409_already_draining() {
    let tmp = TempDir::new().unwrap();
    let (server, drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    // Connect a client so drain coordinator waits for connections to close
    let _client =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Spawn drain coordinator
    let drain_handle = tokio::spawn(handle_drain_request(
        Arc::clone(&server.lifecycle),
        Arc::clone(&server.engine),
        Arc::clone(&server.manager),
        server.deploy_config.drain_timeout_secs,
        drain_rx,
    ));

    let http_client = reqwest::Client::new();

    // First drain → 200
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Wait for lifecycle to transition (but client keeps connection open)
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Second drain → 409 (server is still up because client is connected)
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 409);

    // Drop client so drain can complete
    drop(_client);
    let _ = drain_handle.await;
}

#[tokio::test]
async fn admin_drain_status_reflects_progress() {
    let tmp = TempDir::new().unwrap();
    let (server, drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    // Connect 2 TCP clients
    let mut client1 =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    let mut client2 =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Spawn drain coordinator
    let drain_handle = tokio::spawn(handle_drain_request(
        Arc::clone(&server.lifecycle),
        Arc::clone(&server.engine),
        Arc::clone(&server.manager),
        server.deploy_config.drain_timeout_secs,
        drain_rx,
    ));

    let http_client = reqwest::Client::new();

    // Trigger drain
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Consume GoingAway frames
    let _ = client1.recv().await;
    let _ = client2.recv().await;

    // Check status — both still connected
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["state"], "draining");
    assert_eq!(body["active_connections"], 2);
    assert_eq!(body["connections_drained"], 0);

    // Disconnect one client
    drop(client1);
    tokio::time::sleep(Duration::from_millis(100)).await;

    // Check status — one drained
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["active_connections"], 1);
    assert_eq!(body["connections_drained"], 1);

    // Disconnect remaining client
    drop(client2);

    let _ = drain_handle.await;
}

// -- Phase 17.6: Config reload integration tests --

/// Write a valid config TOML to a temp file and return the path.
fn write_config_file(dir: &Path, toml: &str) -> PathBuf {
    let path = dir.join("config.toml");
    std::fs::write(&path, toml).expect("failed to write config file");
    path
}

/// A config key that is long enough to pass validation (>= 24 chars).
const RELOAD_TEST_KEY: &str = "reload-test-key-at-least-24-characters";

fn base_config_toml() -> String {
    format!(
        r#"
[tcp]
bind_address = "127.0.0.1"
port = 9876

[http]
bind_address = "127.0.0.1"
port = 8080

[storage]
data_dir = "./data"

[queue]
max_depth = 100000

[deploy]
admin_api_key = "{ADMIN_KEY}"

[namespaces.{TEST_NS}]
api_key = "{RELOAD_TEST_KEY}"
permissions = ["Subscribe", "Publish", "Query"]
"#,
    )
}

/// Build a minimal HTTP server with reload_state pointing to a real config file.
async fn start_reload_test_server(
    data_dir: &Path,
    config_path: PathBuf,
) -> (
    common::DeployTestServer,
    tokio::sync::mpsc::Receiver<fafb_q::server::http::DrainRequest>,
) {
    use fafb_q::config::{ServerConfig, shared_config};
    use fafb_q::server::http::{AppState, DrainRequest, HttpServer};
    use fafb_q::storage::fs::FsStorage;

    let config: ServerConfig =
        toml::from_str(&std::fs::read_to_string(&config_path).unwrap()).unwrap();

    let storage = Arc::new(FsStorage::new(data_dir));
    storage.init().await.expect("storage init failed");

    let engine = Arc::new(fafb_q::queue::InMemoryQueueEngine::new(
        Arc::clone(&storage),
        default_engine_config(),
    ));
    engine.init().await.expect("engine init failed");

    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
        common::test_shared_config(),
        ManagerConfig::default(),
    ));

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

    let (lifecycle, _lifecycle_rx) = fafb_q::server::state::ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();

    let message_id_counter = Arc::new(std::sync::atomic::AtomicU64::new(1));

    // TCP server
    let tcp_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let tcp_addr = tcp_listener.local_addr().unwrap().to_string();
    let tcp_config = fafb_q::config::TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: tcp_listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let deploy_config = config.deploy.clone();
    let tcp_server = fafb_q::server::tcp::TcpServer::new(
        tcp_config,
        deploy_config.clone(),
        common::test_shared_config(),
        Arc::clone(&manager),
        Arc::clone(&auth),
        Arc::clone(&message_id_counter),
        Arc::clone(&lifecycle),
        None,
    );
    tokio::spawn(async move {
        let _ = tcp_server.run_with_listener(tcp_listener).await;
    });

    // HTTP server with reload state
    let (drain_tx, drain_rx) = tokio::sync::mpsc::channel::<DrainRequest>(1);

    let http_listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let http_addr = http_listener.local_addr().unwrap();
    let http_base_url = format!("http://{http_addr}");
    let http_config = fafb_q::config::HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: http_addr.port(),
        ..Default::default()
    };

    let shared_cfg = shared_config(&config);
    let config_path_arc = Arc::new(config_path);

    let (reload_notify_tx, _reload_notify_rx) =
        tokio::sync::watch::channel::<Option<fafb_q::config::ReloadNotification>>(None);

    let reload_state = Arc::new(ReloadState::new(
        Arc::clone(&config_path_arc),
        Arc::clone(&shared_cfg),
        config.clone(),
        reload_notify_tx,
        ServeArgs {
            config: None,
            dump_config: false,
            tcp_bind: None,
            http_bind: None,
            data_dir: None,
            log_level: None,
        },
    ));

    let dummy_reload_handle = {
        use tracing_subscriber::layer::SubscriberExt;
        let (_, handle) =
            tracing_subscriber::reload::Layer::new(tracing_subscriber::EnvFilter::new("info"));
        Arc::new(handle)
    };
    let http_server = HttpServer::new(
        http_config,
        AppState {
            manager: Arc::clone(&manager),
            auth: Arc::clone(&auth),
            message_id_counter: Arc::clone(&message_id_counter),
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: deploy_config.admin_api_key.clone(),
            drain_trigger: Some(drain_tx),
            deploy_config: deploy_config.clone(),
            shared_config: Arc::clone(&shared_cfg),
            config_path: Some(config_path_arc),
            reload_state: Some(reload_state),
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None,
    );
    tokio::spawn(async move {
        let _ = http_server.run_with_listener(http_listener).await;
    });

    let server = common::DeployTestServer {
        tcp_addr,
        http_base_url,
        lifecycle,
        engine,
        manager,
        data_dir: data_dir.to_path_buf(),
        deploy_config,
    };
    (server, drain_rx)
}

#[tokio::test]
async fn admin_reload_unchanged() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path).await;

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "unchanged");
}

#[tokio::test]
async fn admin_reload_applies_changes() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path.clone()).await;

    // Modify config — change max_depth (Tier 1)
    let modified = base_config_toml().replace("max_depth = 100000", "max_depth = 50000");
    std::fs::write(&config_path, modified).unwrap();

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "applied");
    assert!(body["changes_applied"].as_u64().unwrap() >= 1);

    // Verify the applied changes mention queue.max_depth
    let applied = body["applied"].as_array().unwrap();
    let fields: Vec<&str> = applied.iter().filter_map(|c| c["field"].as_str()).collect();
    assert!(fields.contains(&"queue.max_depth"));
}

#[tokio::test]
async fn admin_reload_reports_tier3_as_ignored() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path.clone()).await;

    // Modify config — change bind_address (Tier 3) and max_depth (Tier 1)
    let modified = base_config_toml()
        .replace("max_depth = 100000", "max_depth = 50000")
        .replace(
            "[tcp]\nbind_address = \"127.0.0.1\"",
            "[tcp]\nbind_address = \"0.0.0.0\"",
        );
    std::fs::write(&config_path, modified).unwrap();

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "applied");

    // Tier 3 changes should appear in "ignored"
    let ignored = body["ignored"].as_array().unwrap();
    let ignored_fields: Vec<&str> = ignored.iter().filter_map(|c| c["field"].as_str()).collect();
    assert!(ignored_fields.contains(&"tcp.bind_address"));
    assert!(body["changes_ignored"].as_u64().unwrap() >= 1);
}

#[tokio::test]
async fn admin_reload_rejects_invalid_config() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path.clone()).await;

    // Write invalid TOML (missing namespaces entirely)
    std::fs::write(
        &config_path,
        r#"
[tcp]
bind_address = "127.0.0.1"
port = 9876

[storage]
data_dir = "./data"
"#,
    )
    .unwrap();

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "error");
    assert!(body["message"].as_str().unwrap().contains("validation"));
}

#[tokio::test]
async fn admin_reload_rejects_malformed_toml() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path.clone()).await;

    // Write malformed TOML
    std::fs::write(&config_path, "this is not valid [[[toml").unwrap();

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "error");
    assert!(body["message"].as_str().unwrap().contains("parse"));
}

#[tokio::test]
async fn admin_reload_no_config_file() {
    let tmp = TempDir::new().unwrap();
    // Use the regular admin server which has reload_state: None
    let (server, _drain_rx) =
        common::start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config())
            .await;

    let client = reqwest::Client::new();
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "error");
    assert!(body["message"].as_str().unwrap().contains("defaults"));
}

#[tokio::test]
async fn admin_reload_requires_auth() {
    let tmp = TempDir::new().unwrap();
    let config_path = write_config_file(tmp.path(), &base_config_toml());
    let data_dir = tmp.path().join("data");
    std::fs::create_dir_all(&data_dir).unwrap();
    let (server, _drain_rx) = start_reload_test_server(&data_dir, config_path).await;

    let client = reqwest::Client::new();

    // No auth header
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);

    // Wrong key
    let resp = client
        .post(format!("{}/admin/reload", server.http_base_url))
        .bearer_auth("wrong-key-that-is-definitely-not-right")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
}

// ---------------------------------------------------------------------------
// CLI info / health / drain integration tests (Phase 23.4)
// ---------------------------------------------------------------------------

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Publish a few messages so the topic has data.
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let topic = TopicName("info-test".into());
    for i in 0..5u32 {
        client
            .publish(topic.clone(), Bytes::from(format!("msg-{i}")), vec![], 0)
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Query topic info via HTTP.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/info-test/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["namespace"], TEST_NS);
    assert_eq!(body["topic"], "info-test");
    let depth = body["depth"].as_u64().unwrap();
    assert_eq!(depth, 5, "expected 5 messages, got {depth}");
    assert!(body["head_offset"].as_u64().is_some());
}

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Publish to two topics to create them.
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    client
        .publish(TopicName("topic-a".into()), Bytes::from("a"), vec![], 0)
        .await
        .unwrap();
    client
        .publish(TopicName("topic-b".into()), Bytes::from("b"), vec![], 0)
        .await
        .unwrap();
    client.close().await.unwrap();

    // List topics in namespace.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!("{}/v1/{}/topics", server.http_base_url, TEST_NS))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["namespace"], TEST_NS);
    let topics = body["topics"].as_array().unwrap();
    assert!(
        topics.len() >= 2,
        "expected at least 2 topics, got {}",
        topics.len()
    );
    let names: Vec<&str> = topics.iter().filter_map(|v| v.as_str()).collect();
    assert!(names.contains(&"topic-a"), "missing topic-a in {names:?}");
    assert!(names.contains(&"topic-b"), "missing topic-b in {names:?}");
}

#[tokio::test]
async fn cli_health_ready() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let http_client = reqwest::Client::new();

    // Readiness check — server is initialized.
    let resp = http_client
        .get(format!("{}/health/ready", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "ready");

    // Liveness check — always 200.
    let resp = http_client
        .get(format!("{}/health/live", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "alive");
}

#[tokio::test]
async fn cli_health_returns_503_during_drain() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Transition to draining.
    server.lifecycle.begin_drain(5);

    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!("{}/health/ready", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 503);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "draining");

    // Liveness is still 200 during drain.
    let resp = http_client
        .get(format!("{}/health/live", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
}

#[tokio::test]
async fn cli_drain_trigger_and_status() {
    let tmp = TempDir::new().unwrap();
    let (server, drain_rx) =
        start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config()).await;

    // Connect a TCP client to keep the server alive during drain.
    let tcp_client =
        TestClient::connect_and_auth_with_caps(&server.tcp_addr, TEST_NS, TEST_KEY, CAP_GOING_AWAY)
            .await;
    tokio::time::sleep(Duration::from_millis(50)).await;

    // Spawn drain coordinator.
    let drain_handle = tokio::spawn(handle_drain_request(
        Arc::clone(&server.lifecycle),
        Arc::clone(&server.engine),
        Arc::clone(&server.manager),
        server.deploy_config.drain_timeout_secs,
        drain_rx,
    ));

    let http_client = reqwest::Client::new();

    // Check status before drain — should be running.
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["state"], "running");

    // Trigger drain.
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .json(&serde_json::json!({"reason": "cli-test"}))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "draining");

    // Check drain status — should be draining (TCP client still connected).
    tokio::time::sleep(Duration::from_millis(100)).await;
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .bearer_auth(ADMIN_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["state"], "draining");

    // Drain without auth → 403.
    let resp = http_client
        .get(format!("{}/admin/drain/status", server.http_base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);

    // Disconnect TCP client to allow drain to complete.
    drop(tcp_client);

    // Allow drain coordinator to finish.
    tokio::time::timeout(Duration::from_secs(10), drain_handle)
        .await
        .expect("drain coordinator timed out")
        .unwrap();
}

#[tokio::test]
async fn cli_info_nonexistent_topic() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/no-such-topic/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 404);
}

#[tokio::test]
async fn cli_info_wrong_auth() {
    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/anything/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth("wrong-key")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
}

#[tokio::test]
async fn cli_drain_wrong_admin_key() {
    let tmp = TempDir::new().unwrap();
    let (server, _drain_rx) =
        start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config()).await;

    let http_client = reqwest::Client::new();
    let resp = http_client
        .post(format!("{}/admin/drain", server.http_base_url))
        .bearer_auth("wrong-admin-key-that-is-long-enough")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
}

// ---------------------------------------------------------------------------
// CLI config & bench integration tests (Phase 23.5)
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_config_validate_valid_file() {
    use fafb_q::cli::{CliConfig, ConfigAction};

    let result = fafb_q::cli::run_config_cmd(
        CliConfig::default(),
        ConfigAction::Validate {
            file: "config/config.example.toml".into(),
        },
    )
    .await;
    assert!(result.is_ok(), "example config should be valid");
}

#[tokio::test]
async fn cli_config_validate_invalid_file() {
    use fafb_q::cli::{CliConfig, ConfigAction};
    use std::io::Write;

    let mut tmpfile = tempfile::NamedTempFile::new().unwrap();
    write!(
        tmpfile,
        r#"
[tcp]
port = 999999

[namespaces.test]
api_key = "short"
permissions = ["Subscribe"]
"#
    )
    .unwrap();

    let result = fafb_q::cli::run_config_cmd(
        CliConfig::default(),
        ConfigAction::Validate {
            file: tmpfile.path().to_string_lossy().into_owned(),
        },
    )
    .await;
    assert!(result.is_err(), "invalid config should produce error");
}

#[tokio::test]
async fn cli_config_validate_nonexistent_file() {
    use fafb_q::cli::{CliConfig, ConfigAction};

    let result = fafb_q::cli::run_config_cmd(
        CliConfig::default(),
        ConfigAction::Validate {
            file: "/nonexistent/path/config.toml".into(),
        },
    )
    .await;
    assert!(result.is_err(), "nonexistent file should produce error");
}

#[tokio::test]
async fn cli_config_dump_valid_file() {
    use fafb_q::cli::{CliConfig, ConfigAction};

    let result = fafb_q::cli::run_config_cmd(
        CliConfig::default(),
        ConfigAction::Dump {
            file: Some("config/config.example.toml".into()),
            admin_key: None,
        },
    )
    .await;
    assert!(result.is_ok(), "dump of valid config should succeed");
}

#[tokio::test]
async fn cli_config_example() {
    use fafb_q::cli::{CliConfig, ConfigAction};

    let result = fafb_q::cli::run_config_cmd(CliConfig::default(), ConfigAction::Example).await;
    assert!(result.is_ok(), "example config output should succeed");
}

#[tokio::test]
async fn cli_config_dump_remote() {
    use fafb_q::cli::{CliConfig, ConfigAction};

    let tmp = TempDir::new().unwrap();
    let admin_key = "test-admin-key-long-enough-for-validation";
    let deploy_config = DeployConfig {
        admin_api_key: Some(admin_key.into()),
        ..DeployConfig::default()
    };
    let (server, _drain_rx) =
        start_admin_test_server(tmp.path(), deploy_config, default_engine_config()).await;

    let http_addr = server
        .http_base_url
        .strip_prefix("http://")
        .unwrap()
        .to_string();

    let cli_config = CliConfig {
        http_server: http_addr,
        json: true,
        ..Default::default()
    };

    let result = fafb_q::cli::run_config_cmd(
        cli_config,
        ConfigAction::Dump {
            file: None,
            admin_key: Some(admin_key.into()),
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "remote config dump should succeed: {result:?}"
    );
}

#[tokio::test]
async fn cli_bench_pub_publishes_messages() {
    use fafb_q::cli::{BenchAction, CliConfig};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_bench(
        cli_config,
        BenchAction::Pub {
            topic: format!("{TEST_NS}/bench-topic"),
            messages: 50,
            payload_size: 64,
            concurrency: 1,
            no_ack: false,
        },
    )
    .await;
    assert!(result.is_ok(), "bench pub should succeed: {result:?}");

    // Verify messages were published.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/bench-topic/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let depth = body["depth"].as_u64().unwrap();
    assert_eq!(depth, 50, "expected 50 messages, got {depth}");
}

#[tokio::test]
async fn cli_bench_pub_concurrent() {
    use fafb_q::cli::{BenchAction, CliConfig};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_bench(
        cli_config,
        BenchAction::Pub {
            topic: format!("{TEST_NS}/bench-concurrent"),
            messages: 100,
            payload_size: 32,
            concurrency: 4,
            no_ack: false,
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "concurrent bench pub should succeed: {result:?}"
    );

    // Verify all messages arrived.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/bench-concurrent/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let depth = body["depth"].as_u64().unwrap();
    assert_eq!(depth, 100, "expected 100 messages, got {depth}");
}

#[tokio::test]
async fn cli_bench_pub_no_ack() {
    use fafb_q::cli::{BenchAction, CliConfig};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_bench(
        cli_config,
        BenchAction::Pub {
            topic: format!("{TEST_NS}/bench-noack"),
            messages: 50,
            payload_size: 64,
            concurrency: 1,
            no_ack: true,
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "bench pub no-ack should succeed: {result:?}"
    );

    // Wait for server to process pipelined publishes (no-ack mode sends all
    // frames before closing, so the server needs time to drain the TCP buffer).
    tokio::time::sleep(Duration::from_secs(1)).await;

    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/bench-noack/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let depth = body["depth"].as_u64().unwrap();
    assert!(
        depth > 0,
        "expected some messages published (no-ack), got 0"
    );
}

#[tokio::test]
async fn cli_bench_sub_with_count() {
    use fafb_q::cli::{BenchAction, CliConfig};
    use fafb_q::client::{ClientConfig, FafbClient};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish messages.
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    for i in 0..20u32 {
        client
            .publish(
                TopicName("bench-sub-count".into()),
                Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Consume with --from 0 --count 10.
    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_bench(
        cli_config,
        BenchAction::Sub {
            topic: format!("{TEST_NS}/bench-sub-count"),
            group: None,
            from: Some(0),
            count: Some(10),
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "bench sub with count should succeed: {result:?}"
    );
}

// ---------------------------------------------------------------------------
// CLI pub command tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_pub_string_payload() {
    use fafb_q::cli::{CliConfig, PubArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_publish(
        cli_config,
        PubArgs {
            topic: format!("{TEST_NS}/pub-test"),
            payload: Some("hello from cli".into()),
            file: None,
            lines: false,
            ttl_secs: None,
            headers: vec![],
            no_ack: false,
        },
    )
    .await;
    assert!(result.is_ok(), "pub should succeed: {result:?}");

    // Verify via HTTP info.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/pub-test/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["depth"].as_u64().unwrap(), 1);
}

#[tokio::test]
async fn cli_pub_with_headers_and_ttl() {
    use fafb_q::cli::{CliConfig, FetchArgs, PubArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    // Publish with headers and TTL.
    let result = fafb_q::cli::run_publish(
        cli_config.clone(),
        PubArgs {
            topic: format!("{TEST_NS}/hdr-ttl"),
            payload: Some("payload-with-meta".into()),
            file: None,
            lines: false,
            ttl_secs: Some(300),
            headers: vec!["X-Trace: abc123".into(), "X-Source: cli-test".into()],
            no_ack: false,
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "pub with headers/ttl should succeed: {result:?}"
    );

    // Fetch back and verify headers + TTL survived.
    use fafb_q::client::{ClientConfig, FafbClient};
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let fetch = client
        .fetch(TopicName("hdr-ttl".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    let msg = &fetch.messages[0];
    assert_eq!(msg.payload.as_ref(), b"payload-with-meta");
    assert_eq!(msg.ttl_ms, 300_000);
    assert!(msg.headers.len() >= 2, "expected at least 2 headers");
    client.close().await.unwrap();
}

#[tokio::test]
async fn cli_pub_no_ack_mode() {
    use fafb_q::cli::{CliConfig, PubArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_publish(
        cli_config,
        PubArgs {
            topic: format!("{TEST_NS}/noack-test"),
            payload: Some("fire-and-forget".into()),
            file: None,
            lines: false,
            ttl_secs: None,
            headers: vec![],
            no_ack: true,
        },
    )
    .await;
    assert!(result.is_ok(), "pub no-ack should succeed: {result:?}");

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

    // Verify it was published.
    let http_client = reqwest::Client::new();
    let resp = http_client
        .get(format!(
            "{}/v1/{}/noack-test/info",
            server.http_base_url, TEST_NS
        ))
        .bearer_auth(TEST_KEY)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["depth"].as_u64().unwrap(), 1);
}

#[tokio::test]
async fn cli_pub_from_file() {
    use fafb_q::cli::{CliConfig, PubArgs};
    use std::io::Write;

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Write a payload file.
    let payload_file = tmp.path().join("payload.txt");
    {
        let mut f = std::fs::File::create(&payload_file).unwrap();
        f.write_all(b"file-payload-content").unwrap();
    }

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_publish(
        cli_config,
        PubArgs {
            topic: format!("{TEST_NS}/file-pub"),
            payload: None,
            file: Some(payload_file.to_str().unwrap().into()),
            lines: false,
            ttl_secs: None,
            headers: vec![],
            no_ack: false,
        },
    )
    .await;
    assert!(result.is_ok(), "pub from file should succeed: {result:?}");

    // Fetch back to verify content.
    use fafb_q::client::{ClientConfig, FafbClient};
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let fetch = client
        .fetch(TopicName("file-pub".into()), TopicOffset(0), 10)
        .await
        .unwrap();
    assert_eq!(fetch.messages.len(), 1);
    assert_eq!(fetch.messages[0].payload.as_ref(), b"file-payload-content");
    client.close().await.unwrap();
}

// ---------------------------------------------------------------------------
// CLI fetch command tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_fetch_from_offset() {
    use fafb_q::cli::{CliConfig, FetchArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish messages.
    let tcp_config = fafb_q::client::ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = fafb_q::client::FafbClient::connect(&tcp_config)
        .await
        .unwrap();
    for i in 0..5u32 {
        client
            .publish(
                TopicName("fetch-test".into()),
                Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Fetch via CLI runner.
    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_fetch(
        cli_config,
        FetchArgs {
            topic: format!("{TEST_NS}/fetch-test"),
            from: Some(0),
            since: None,
            tail: None,
            count: 100,
            all: false,
            payload_only: false,
        },
    )
    .await;
    assert!(result.is_ok(), "fetch should succeed: {result:?}");
}

#[tokio::test]
async fn cli_fetch_all_pagination() {
    use fafb_q::cli::{CliConfig, FetchArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish enough messages to require multiple pages.
    let tcp_config = fafb_q::client::ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = fafb_q::client::FafbClient::connect(&tcp_config)
        .await
        .unwrap();
    for i in 0..25u32 {
        client
            .publish(
                TopicName("fetch-page".into()),
                Bytes::from(format!("page-msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Fetch all with small page size (count=10) to force pagination.
    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_fetch(
        cli_config,
        FetchArgs {
            topic: format!("{TEST_NS}/fetch-page"),
            from: None,
            since: None,
            tail: None,
            count: 10,
            all: true,
            payload_only: false,
        },
    )
    .await;
    assert!(result.is_ok(), "fetch --all should succeed: {result:?}");
}

#[tokio::test]
async fn cli_fetch_since_timestamp() {
    use fafb_q::cli::{CliConfig, FetchArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish messages.
    let tcp_config = fafb_q::client::ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = fafb_q::client::FafbClient::connect(&tcp_config)
        .await
        .unwrap();
    for i in 0..3u32 {
        client
            .publish(
                TopicName("fetch-ts".into()),
                Bytes::from(format!("ts-msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Fetch from timestamp "0" (epoch) — should get all messages.
    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_fetch(
        cli_config,
        FetchArgs {
            topic: format!("{TEST_NS}/fetch-ts"),
            from: None,
            since: Some("0".into()),
            tail: None,
            count: 100,
            all: false,
            payload_only: false,
        },
    )
    .await;
    assert!(result.is_ok(), "fetch --since should succeed: {result:?}");
}

// ---------------------------------------------------------------------------
// CLI health command tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_health_live() {
    use fafb_q::cli::{CliConfig, HealthArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        http_server: server.http_base_url.trim_start_matches("http://").into(),
        ..Default::default()
    };

    // Liveness probe should succeed.
    let result = fafb_q::cli::run_health(cli_config.clone(), HealthArgs { live: true }).await;
    assert!(result.is_ok(), "health --live should succeed: {result:?}");

    // Liveness should still succeed during drain.
    server.lifecycle.begin_drain(5);

    let result = fafb_q::cli::run_health(cli_config, HealthArgs { live: true }).await;
    assert!(
        result.is_ok(),
        "health --live should succeed during drain: {result:?}"
    );
}

// ---------------------------------------------------------------------------
// CLI bench latency test
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_bench_latency() {
    use fafb_q::cli::{BenchAction, CliConfig};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let cli_config = CliConfig {
        server: server.tcp_addr.clone(),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        ..Default::default()
    };

    let result = fafb_q::cli::run_bench(
        cli_config,
        BenchAction::Latency {
            topic: format!("{TEST_NS}/bench-latency"),
            messages: 20,
            warmup: 5,
        },
    )
    .await;
    assert!(result.is_ok(), "bench latency should succeed: {result:?}");
}

// ---------------------------------------------------------------------------
// CLI drain --wait test
// ---------------------------------------------------------------------------

#[tokio::test]
async fn cli_drain_wait_completes() {
    use fafb_q::cli::{CliConfig, DrainArgs};

    let tmp = TempDir::new().unwrap();
    let (server, drain_rx) =
        start_admin_test_server(tmp.path(), admin_deploy_config(), default_engine_config()).await;

    let cli_config = CliConfig {
        http_server: server.http_base_url.trim_start_matches("http://").into(),
        timeout_secs: 15,
        ..Default::default()
    };

    // Spawn the standard drain coordinator.
    tokio::spawn(handle_drain_request(
        Arc::clone(&server.lifecycle),
        Arc::clone(&server.engine),
        Arc::clone(&server.manager),
        server.deploy_config.drain_timeout_secs,
        drain_rx,
    ));

    // CLI drain --wait should complete when server finishes draining.
    let result = fafb_q::cli::run_drain(
        cli_config,
        DrainArgs {
            admin_key: Some(ADMIN_KEY.into()),
            timeout: Some(5),
            status: false,
            reason: Some("test-drain".into()),
            wait: true,
        },
    )
    .await;
    // The drain should either succeed or fail with a connection error
    // (server shut down during polling, which is treated as success).
    // Either way it should not timeout.
    assert!(result.is_ok(), "drain --wait should succeed: {result:?}");
}

// -- CLI --http publish and fetch tests --

#[tokio::test]
async fn cli_http_publish_and_fetch_roundtrip() {
    use fafb_q::cli::{CliConfig, FetchArgs, PubArgs};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Publish via --http mode.
    let cli_config = CliConfig {
        http_server: server.http_base_url.replace("http://", ""),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        use_http: true,
        ..Default::default()
    };

    let result = fafb_q::cli::run_publish(
        cli_config.clone(),
        PubArgs {
            topic: format!("{TEST_NS}/http-rt"),
            payload: Some("http-hello".into()),
            file: None,
            lines: false,
            ttl_secs: None,
            headers: vec![],
            no_ack: false,
        },
    )
    .await;
    assert!(result.is_ok(), "HTTP publish should succeed: {result:?}");

    // Fetch back via --http mode.
    let result = fafb_q::cli::run_fetch(
        cli_config,
        FetchArgs {
            topic: format!("{TEST_NS}/http-rt"),
            from: Some(0),
            since: None,
            tail: None,
            count: 100,
            all: false,
            payload_only: false,
        },
    )
    .await;
    assert!(result.is_ok(), "HTTP fetch should succeed: {result:?}");
}

#[tokio::test]
async fn cli_http_fetch_negative_offset() {
    use fafb_q::cli::{CliConfig, FetchArgs};
    use fafb_q::client::{ClientConfig, FafbClient};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish via TCP.
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    for i in 0..5u32 {
        client
            .publish(
                TopicName("http-neg".into()),
                Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Fetch with negative offset via --http (server resolves natively).
    let cli_config = CliConfig {
        http_server: server.http_base_url.replace("http://", ""),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        use_http: true,
        ..Default::default()
    };

    let result = fafb_q::cli::run_fetch(
        cli_config.clone(),
        FetchArgs {
            topic: format!("{TEST_NS}/http-neg"),
            from: Some(-1),
            since: None,
            tail: None,
            count: 1,
            all: false,
            payload_only: false,
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "HTTP fetch with negative offset should succeed: {result:?}"
    );
}

#[tokio::test]
async fn cli_http_fetch_tail() {
    use fafb_q::cli::{CliConfig, FetchArgs};
    use fafb_q::client::{ClientConfig, FafbClient};

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    // Pre-publish via TCP.
    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    for i in 0..10u32 {
        client
            .publish(
                TopicName("http-tail".into()),
                Bytes::from(format!("msg-{i}")),
                vec![],
                0,
            )
            .await
            .unwrap();
    }
    client.close().await.unwrap();

    // Fetch with --tail via --http (server resolves negative offset natively).
    let cli_config = CliConfig {
        http_server: server.http_base_url.replace("http://", ""),
        api_key: Some(TEST_KEY.into()),
        namespace: Some(TEST_NS.into()),
        use_http: true,
        ..Default::default()
    };

    let result = fafb_q::cli::run_fetch(
        cli_config,
        FetchArgs {
            topic: format!("{TEST_NS}/http-tail"),
            from: None,
            since: None,
            tail: Some(3),
            count: 100,
            all: false,
            payload_only: false,
        },
    )
    .await;
    assert!(
        result.is_ok(),
        "HTTP fetch with --tail should succeed: {result:?}"
    );
}

// -- FafbClient::fetch_tail tests --

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let topic = TopicName("tail-last".into());

    // Publish 5 messages.
    for i in 0..5u32 {
        client
            .publish(topic.clone(), Bytes::from(format!("msg-{i}")), vec![], 0)
            .await
            .unwrap();
    }

    // fetch_tail(1) should return only the last message (offset 4).
    let result = client.fetch_tail(topic.clone(), 1).await.unwrap();
    assert_eq!(result.messages.len(), 1);
    assert_eq!(result.messages[0].offset, TopicOffset(4));
    assert_eq!(result.messages[0].payload.as_ref(), b"msg-4");

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

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let topic = TopicName("tail-last-n".into());

    // Publish 20 messages.
    for i in 0..20u32 {
        client
            .publish(topic.clone(), Bytes::from(format!("msg-{i}")), vec![], 0)
            .await
            .unwrap();
    }

    // fetch_tail(5) should return messages at offsets 15-19.
    let result = client.fetch_tail(topic.clone(), 5).await.unwrap();
    assert_eq!(result.messages.len(), 5);
    assert_eq!(result.messages[0].offset, TopicOffset(15));
    assert_eq!(result.messages[4].offset, TopicOffset(19));

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

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let topic = TopicName("tail-exceeds".into());

    // Publish 3 messages.
    for i in 0..3u32 {
        client
            .publish(topic.clone(), Bytes::from(format!("msg-{i}")), vec![], 0)
            .await
            .unwrap();
    }

    // fetch_tail(100) on a 3-message topic should return all 3.
    let result = client.fetch_tail(topic.clone(), 100).await.unwrap();
    assert_eq!(result.messages.len(), 3);
    assert_eq!(result.messages[0].offset, TopicOffset(0));
    assert_eq!(result.messages[2].offset, TopicOffset(2));

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

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

    let tmp = TempDir::new().unwrap();
    let server =
        start_deploy_test_server(tmp.path(), DeployConfig::default(), default_engine_config())
            .await;

    let tcp_config = ClientConfig::new(&server.tcp_addr, TEST_NS, TEST_KEY);
    let mut client = FafbClient::connect(&tcp_config).await.unwrap();
    let topic = TopicName("tail-empty".into());

    // fetch_tail on a nonexistent topic should return empty.
    let result = client.fetch_tail(topic.clone(), 10).await.unwrap();
    assert!(result.messages.is_empty());

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