// Shared test helpers and fixtures.

use std::io::Write;
use std::path::{Path, PathBuf};
use std::sync::Arc;
use std::sync::atomic::AtomicU64;
use std::time::Duration;

use futures::SinkExt;
use futures::StreamExt;
use tokio::net::TcpStream;
use tokio_util::codec::Framed;

use fafb_q::auth::{MockAuthenticator, PermissionSet};
use fafb_q::codec::frame::Capabilities;
use fafb_q::codec::{Frame, MessageCodec};
use fafb_q::config::DeployConfig;
use fafb_q::config::HttpConfig;
use fafb_q::config::ServerConfig;
use fafb_q::config::SharedConfig;
use fafb_q::config::TcpConfig;
use fafb_q::config::TlsConfig;
use fafb_q::namespace::NamespaceManager;
use fafb_q::namespace::{DefaultNamespaceManager, ManagerConfig};
use fafb_q::queue::{EngineConfig, InMemoryQueueEngine, QueueEngine};
use fafb_q::server::http::{AppState, DrainRequest, DrainResponse, HttpServer};
use fafb_q::server::state::{ServerLifecycle, ServerState};
use fafb_q::server::tcp::TcpServer;
use fafb_q::storage::StorageBackend;
use fafb_q::storage::fs::FsStorage;
use fafb_q::storage::mock::MockStorage;
use fafb_q::tls::TlsProvider;
use fafb_q::types::{
    ApiKey, MessageEnvelope, MessageId, NamespaceId, Permission, TopicName, now_millis,
};

/// Create a default SharedConfig for tests.
pub fn test_shared_config() -> SharedConfig {
    fafb_q::config::shared_config(&ServerConfig::default())
}

/// Create a default config path for tests (None = running on defaults).
pub fn test_config_path() -> Option<Arc<PathBuf>> {
    None
}

pub fn make_envelope(id: u64, ns: &str, topic: &str, payload: &[u8]) -> MessageEnvelope {
    MessageEnvelope::new(
        MessageId(id + 1),
        NamespaceId(ns.into()),
        TopicName(topic.into()),
        payload.to_vec(),
        now_millis(),
    )
}

pub fn make_engine() -> Arc<InMemoryQueueEngine<MockStorage>> {
    Arc::new(InMemoryQueueEngine::new(
        Arc::new(MockStorage::new()),
        EngineConfig {
            max_depth: 100_000,
            backpressure: fafb_q::queue::BackpressurePolicy::Reject,
            ..Default::default()
        },
    ))
}

pub fn make_namespace_manager(
    engine: Arc<InMemoryQueueEngine<MockStorage>>,
    namespace_ids: Vec<NamespaceId>,
) -> DefaultNamespaceManager<InMemoryQueueEngine<MockStorage>> {
    DefaultNamespaceManager::new(
        engine,
        namespace_ids,
        test_shared_config(),
        ManagerConfig::default(),
    )
}

// -- TCP integration test infrastructure --

pub const TEST_NS: &str = "test-ns";
pub const TEST_KEY: &str = "test-key-secret";

pub struct TestServer {
    pub addr: String,
    pub lifecycle: Arc<ServerLifecycle>,
    pub engine: Option<Arc<InMemoryQueueEngine<MockStorage>>>,
}

impl TestServer {
    pub fn shutdown(&self) {
        self.lifecycle.transition_to(ServerState::ShuttingDown);
    }
}

/// Start a TCP server on a random port with a MockAuthenticator.
///
/// Uses `run_with_listener` with a pre-bound listener to avoid the
/// bind-drop-rebind race and the need for sleep.
pub async fn start_test_tcp_server() -> TestServer {
    start_test_tcp_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
        Permission::Fetch,
    ])
    .await
}

/// Start a TCP server with specific permissions for the test API key.
pub async fn start_test_tcp_server_with_perms(perms: &[Permission]) -> TestServer {
    start_test_tcp_server_with_config(perms, TcpConfig::default()).await
}

/// Start a TCP server with specific permissions and config overrides.
pub async fn start_test_tcp_server_with_config(
    perms: &[Permission],
    mut config: TcpConfig,
) -> TestServer {
    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        engine,
        vec![NamespaceId(TEST_NS.into())],
    ));

    let auth = Arc::new(MockAuthenticator {
        namespace: NamespaceId(TEST_NS.into()),
        key: ApiKey(TEST_KEY.into()),
        permissions: PermissionSet::from_slice(perms),
    });

    // Bind to port 0 to get a random available port — keep the listener.
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();

    // Override config to match bound address (port doesn't matter since we
    // pass the listener directly, but keep it consistent).
    config.bind_address = "127.0.0.1".into();
    config.port = listener.local_addr().unwrap().port();

    // Build a SharedConfig that reflects the TcpConfig values
    // (e.g. idle_timeout_secs, max_connections) so the accept loop reads
    // correct values from SharedConfig.
    let shared_cfg = {
        let mut rt = fafb_q::config::RuntimeConfig::from_server_config(&ServerConfig::default());
        rt.idle_timeout_secs = config.idle_timeout_secs;
        rt.max_connections = config.max_connections;
        rt.max_message_size = config.max_message_size;
        rt.subscriber_channel_buffer = config.subscriber_channel_buffer;
        rt.slow_consumer_policy = config.slow_consumer_policy;
        rt.slow_consumer_threshold = config.slow_consumer_threshold;
        rt.slow_consumer_disconnect_after = config.slow_consumer_disconnect_after;
        std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        shared_cfg,
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestServer {
        addr,
        lifecycle,
        engine: None,
    }
}

/// Start a TCP server with custom engine config (e.g., small max_depth + DropOldest for eviction tests).
pub async fn start_test_tcp_server_with_engine_config(
    perms: &[Permission],
    mut config: TcpConfig,
    engine_config: EngineConfig,
) -> TestServer {
    let engine = Arc::new(InMemoryQueueEngine::new(
        Arc::new(MockStorage::new()),
        engine_config,
    ));
    let manager = Arc::new(make_namespace_manager(
        engine,
        vec![NamespaceId(TEST_NS.into())],
    ));

    let auth = Arc::new(MockAuthenticator {
        namespace: NamespaceId(TEST_NS.into()),
        key: ApiKey(TEST_KEY.into()),
        permissions: PermissionSet::from_slice(perms),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();
    config.bind_address = "127.0.0.1".into();
    config.port = listener.local_addr().unwrap().port();

    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        test_shared_config(),
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestServer {
        addr,
        lifecycle,
        engine: None,
    }
}

/// Generic test client that wraps a framed connection.
///
/// The default type parameter `S = TcpStream` preserves backward compatibility:
/// existing code using `TestClient` without a type parameter continues to work.
/// TLS clients use `TestClient<TlsStream<TcpStream>>` (aliased as `TlsTestClient`).
pub struct TestClient<S = TcpStream> {
    framed: Framed<S, MessageCodec>,
}

/// Type alias for TLS test clients.
pub type TlsTestClient = TestClient<tokio_rustls::client::TlsStream<TcpStream>>;

// Shared methods for any stream type.
impl<S: tokio::io::AsyncRead + tokio::io::AsyncWrite + Unpin> TestClient<S> {
    pub async fn send(&mut self, frame: Frame) {
        self.framed.send(frame).await.unwrap();
    }

    pub async fn recv(&mut self) -> Frame {
        self.framed
            .next()
            .await
            .expect("stream closed unexpectedly")
            .expect("codec error")
    }

    pub async fn recv_timeout(&mut self, duration: Duration) -> Option<Frame> {
        match tokio::time::timeout(duration, self.framed.next()).await {
            Ok(Some(Ok(frame))) => Some(frame),
            _ => None,
        }
    }

    /// Enable header encoding/decoding on this client's codec.
    pub fn enable_headers(&mut self) {
        self.framed.codec_mut().set_headers_enabled(true);
    }

    pub fn enable_ttl(&mut self) {
        self.framed.codec_mut().set_ttl_enabled(true);
    }
}

// Plaintext-specific constructors.
impl TestClient<TcpStream> {
    pub async fn connect(addr: &str) -> Self {
        let stream = TcpStream::connect(addr).await.unwrap();
        let framed = Framed::new(stream, MessageCodec::new(1_048_576));
        Self { framed }
    }

    /// Connect to the server and authenticate with the test credentials.
    pub async fn connect_and_auth(addr: &str, ns: &str, key: &str) -> Self {
        Self::connect_and_auth_with_caps(addr, ns, key, 0).await
    }

    /// Connect to the server and authenticate with specific capabilities.
    pub async fn connect_and_auth_with_caps(
        addr: &str,
        ns: &str,
        key: &str,
        caps: Capabilities,
    ) -> Self {
        let mut client = Self::connect(addr).await;
        client
            .send(Frame::Connect {
                version: 1,
                namespace: NamespaceId(ns.into()),
                api_key: ApiKey(key.into()),
                capabilities: caps,
            })
            .await;
        let ack = client.recv().await;
        let server_caps = match &ack {
            Frame::ConnectAck {
                success: true,
                server_capabilities,
                ..
            } => *server_capabilities,
            _ => panic!("expected ConnectAck success, got: {ack:?}"),
        };
        // Enable headers on the client codec if both sides support it.
        if caps & fafb_q::codec::frame::CAP_HEADERS != 0
            && server_caps & fafb_q::codec::frame::CAP_HEADERS != 0
        {
            client.enable_headers();
        }
        // Enable TTL on the client codec if both sides support it.
        if caps & fafb_q::codec::frame::CAP_MESSAGE_TTL != 0
            && server_caps & fafb_q::codec::frame::CAP_MESSAGE_TTL != 0
        {
            client.enable_ttl();
        }
        client
    }
}

// TLS-specific constructors.
impl TestClient<tokio_rustls::client::TlsStream<TcpStream>> {
    /// Connect over TLS to the given address using the provided connector.
    pub async fn connect_tls(addr: &str, connector: &tokio_rustls::TlsConnector) -> Self {
        let stream = TcpStream::connect(addr).await.unwrap();
        let server_name = rustls::pki_types::ServerName::try_from("localhost").unwrap();
        let tls_stream = connector.connect(server_name, stream).await.unwrap();
        let framed = Framed::new(tls_stream, MessageCodec::new(1_048_576));
        Self { framed }
    }

    /// Connect over TLS and authenticate with the test credentials.
    pub async fn connect_and_auth_tls(
        addr: &str,
        connector: &tokio_rustls::TlsConnector,
        ns: &str,
        key: &str,
    ) -> Self {
        let mut client = Self::connect_tls(addr, connector).await;
        client
            .send(Frame::Connect {
                version: 1,
                namespace: NamespaceId(ns.into()),
                api_key: ApiKey(key.into()),
                capabilities: 0,
            })
            .await;
        let ack = client.recv().await;
        assert!(
            matches!(ack, Frame::ConnectAck { success: true, .. }),
            "expected ConnectAck success, got: {ack:?}"
        );
        client
    }
}

// -- Load test infrastructure --

/// Percentile-based latency statistics computed from raw samples.
pub struct LatencyStats {
    pub p50: f64,
    pub p95: f64,
    pub p99: f64,
    pub max: f64,
    pub count: usize,
}

impl LatencyStats {
    pub fn from_samples(mut samples: Vec<f64>) -> Self {
        if samples.is_empty() {
            return Self {
                p50: 0.0,
                p95: 0.0,
                p99: 0.0,
                max: 0.0,
                count: 0,
            };
        }
        samples.sort_by(|a, b| a.partial_cmp(b).unwrap_or(std::cmp::Ordering::Equal));
        let len = samples.len();
        Self {
            p50: samples[len * 50 / 100],
            p95: samples[len * 95 / 100],
            p99: samples[len * 99 / 100],
            max: samples[len - 1],
            count: len,
        }
    }
}

/// A pool of authenticated test clients for load testing.
pub struct ClientPool {
    pub clients: Vec<TestClient>,
    ns: String,
    #[allow(dead_code)]
    key: String,
}

impl ClientPool {
    /// Connect `n` clients to the server and authenticate each.
    pub async fn connect(addr: &str, n: usize, ns: &str, key: &str) -> Self {
        let mut clients = Vec::with_capacity(n);
        for _ in 0..n {
            clients.push(TestClient::connect_and_auth(addr, ns, key).await);
        }
        Self {
            clients,
            ns: ns.to_string(),
            key: key.to_string(),
        }
    }

    /// Publish messages at a sustained rate for `duration`, distributing across clients.
    ///
    /// Returns `(total_published, latency_samples_secs)`.
    pub async fn sustained_publish(
        &mut self,
        topic: &str,
        rate_per_sec: u64,
        duration: Duration,
        payload_size: usize,
    ) -> (u64, Vec<f64>) {
        let payload = vec![0xABu8; payload_size];
        let topic_name = TopicName(topic.into());
        let interval_ns = if rate_per_sec > 0 {
            1_000_000_000 / rate_per_sec
        } else {
            1_000_000
        };

        let deadline = tokio::time::Instant::now() + duration;
        let n_clients = self.clients.len();

        // Drain clients out of self temporarily for the spawned tasks
        let mut clients: Vec<TestClient> = std::mem::take(&mut self.clients);

        let mut handles = tokio::task::JoinSet::new();
        let per_client_interval = Duration::from_nanos((interval_ns as usize * n_clients) as u64);

        for mut client in clients.drain(..) {
            let payload = payload.clone();
            let topic_name = topic_name.clone();

            // Each task collects its own latency samples — no shared lock on hot path
            handles.spawn(async move {
                let mut local_latencies = Vec::new();
                let mut local_count = 0u64;
                let mut interval = tokio::time::interval(per_client_interval);
                interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
                loop {
                    interval.tick().await;
                    if tokio::time::Instant::now() >= deadline {
                        break;
                    }
                    let start = std::time::Instant::now();
                    client
                        .send(Frame::Publish {
                            topic: topic_name.clone(),
                            payload: bytes::Bytes::from(payload.clone()),
                            headers: vec![],
                            ttl_ms: 0,
                        })
                        .await;
                    // Wait for PublishAck
                    let _ack = client.recv().await;
                    local_latencies.push(start.elapsed().as_secs_f64());
                    local_count += 1;
                }
                (client, local_count, local_latencies)
            });
        }

        // Merge per-task results
        let mut returned_clients = Vec::new();
        let mut total = 0u64;
        let mut all_latencies = Vec::new();
        while let Some(result) = handles.join_next().await {
            if let Ok((client, count, latencies)) = result {
                returned_clients.push(client);
                total += count;
                all_latencies.extend(latencies);
            }
        }
        self.clients = returned_clients;

        (total, all_latencies)
    }

    /// Subscribe all clients and collect delivered messages for `duration`.
    ///
    /// Returns `(total_received, out_of_order_count)`. Each subscriber tracks
    /// message IDs and counts any that arrive out of order relative to the
    /// previous delivery on that subscriber.
    pub async fn subscribe_and_collect(&mut self, topic: &str, duration: Duration) -> (u64, u64) {
        let topic_name = TopicName(topic.into());

        // Subscribe each client
        for client in &mut self.clients {
            client
                .send(Frame::Subscribe {
                    topic: topic_name.clone(),
                    group: None,
                })
                .await;
            let _ack = client.recv().await;
        }

        let deadline = tokio::time::Instant::now() + duration;
        let mut clients: Vec<TestClient> = std::mem::take(&mut self.clients);
        let mut handles = tokio::task::JoinSet::new();

        for mut client in clients.drain(..) {
            handles.spawn(async move {
                let mut count = 0u64;
                let mut out_of_order = 0u64;
                let mut last_id: Option<u64> = None;
                loop {
                    match client.recv_timeout(Duration::from_millis(100)).await {
                        Some(Frame::Deliver { message_id, .. }) => {
                            count += 1;
                            // Check ordering: message IDs should be monotonically increasing
                            if let Some(prev) = last_id {
                                if message_id.0 <= prev {
                                    out_of_order += 1;
                                }
                            }
                            last_id = Some(message_id.0);
                        }
                        _ => {}
                    }
                    if tokio::time::Instant::now() >= deadline {
                        break;
                    }
                }
                (client, count, out_of_order)
            });
        }

        let mut returned_clients = Vec::new();
        let mut total = 0u64;
        let mut total_out_of_order = 0u64;
        while let Some(result) = handles.join_next().await {
            if let Ok((client, count, ooo)) = result {
                returned_clients.push(client);
                total += count;
                total_out_of_order += ooo;
            }
        }
        self.clients = returned_clients;

        (total, total_out_of_order)
    }
}

/// Multi-namespace mock authenticator. Maps multiple API keys to
/// different namespaces with their own permission sets.
///
/// Lives in test code (not `src/auth.rs`) since it's only needed by
/// integration/load tests.
pub struct MultiMockAuthenticator {
    entries: Vec<(NamespaceId, ApiKey, PermissionSet)>,
}

impl MultiMockAuthenticator {
    pub fn new(entries: Vec<(NamespaceId, ApiKey, PermissionSet)>) -> Self {
        Self { entries }
    }
}

impl fafb_q::auth::Authenticator for MultiMockAuthenticator {
    fn authenticate(&self, key: &ApiKey) -> fafb_q::error::Result<fafb_q::auth::AuthSession> {
        for (ns, k, perms) in &self.entries {
            if k.0 == key.0 {
                return Ok(fafb_q::auth::AuthSession {
                    namespace: ns.clone(),
                    permissions: *perms,
                });
            }
        }
        Err(fafb_q::error::Error::AuthFailed)
    }

    fn check_permission(
        &self,
        namespace: &NamespaceId,
        key: &ApiKey,
        perm: Permission,
    ) -> fafb_q::error::Result<()> {
        let session = <Self as fafb_q::auth::Authenticator>::authenticate(self, key)?;
        if session.namespace != *namespace {
            return Err(fafb_q::error::Error::AuthFailed);
        }
        session.permissions.check(perm)
    }
}

/// Namespace config for multi-namespace test servers.
pub struct NsConfig {
    pub namespace: String,
    pub api_key: String,
    pub quota: fafb_q::namespace::manager::NamespaceQuota,
}

/// Start a TCP server with multiple namespaces, each with its own API key and quota.
pub async fn start_test_tcp_server_multi_ns(ns_configs: Vec<NsConfig>) -> TestServer {
    let engine = make_engine();

    let namespace_ids: Vec<NamespaceId> = ns_configs
        .iter()
        .map(|c| NamespaceId(c.namespace.clone()))
        .collect();
    // Build a SharedConfig with the namespace quotas.
    let shared_cfg = {
        let mut rt = fafb_q::config::RuntimeConfig::from_server_config(&ServerConfig::default());
        for c in &ns_configs {
            rt.namespace_quotas
                .insert(NamespaceId(c.namespace.clone()), c.quota.clone());
        }
        std::sync::Arc::new(arc_swap::ArcSwap::from_pointee(rt))
    };

    let manager = Arc::new(DefaultNamespaceManager::new(
        engine,
        namespace_ids,
        Arc::clone(&shared_cfg),
        ManagerConfig::default(),
    ));

    let entries: Vec<(NamespaceId, ApiKey, PermissionSet)> = ns_configs
        .iter()
        .map(|c| {
            (
                NamespaceId(c.namespace.clone()),
                ApiKey(c.api_key.clone()),
                PermissionSet::from_slice(&[
                    Permission::Subscribe,
                    Permission::Publish,
                    Permission::Query,
                ]),
            )
        })
        .collect();

    let auth = Arc::new(MultiMockAuthenticator::new(entries));
    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();

    let config = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        shared_cfg, // use the same config given to the manager (not defaults)
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestServer {
        addr,
        lifecycle,
        engine: None,
    }
}

/// Start a TCP server with a custom SharedConfig.
/// The same shared config is passed to both the namespace manager and the TCP server,
/// ensuring namespace quotas (including TTL limits) are visible to the connection handler.
pub async fn start_test_tcp_server_with_shared_config(shared_cfg: SharedConfig) -> TestServer {
    let engine = make_engine();
    let engine_ref = Arc::clone(&engine);
    let manager = Arc::new(make_namespace_manager(
        engine,
        vec![NamespaceId(TEST_NS.into())],
    ));

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

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();

    let config = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        shared_cfg,
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestServer {
        addr,
        lifecycle,
        engine: Some(engine_ref),
    }
}

/// Start an HTTP server with a custom SharedConfig.
pub async fn start_test_http_server_with_shared_config(shared_cfg: SharedConfig) -> TestHttpServer {
    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("http://{addr}");

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: None,
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: shared_cfg,
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestHttpServer {
        base_url,
        engine,
        lifecycle,
    }
}

// -- HTTP integration test infrastructure --

pub struct TestHttpServer {
    pub base_url: String,
    pub engine: Arc<InMemoryQueueEngine<MockStorage>>,
    pub lifecycle: Arc<ServerLifecycle>,
}

impl TestHttpServer {
    pub fn shutdown(&self) {
        self.lifecycle.transition_to(ServerState::ShuttingDown);
    }
}

/// Start an HTTP server on a random port with a MockAuthenticator.
pub async fn start_test_http_server() -> TestHttpServer {
    start_test_http_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
    ])
    .await
}

/// Start an HTTP server with all permissions including Fetch.
pub async fn start_test_http_server_all_perms() -> TestHttpServer {
    start_test_http_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
        Permission::Fetch,
    ])
    .await
}

/// Start an HTTP server with specific permissions for the test API key.
pub async fn start_test_http_server_with_perms(perms: &[Permission]) -> TestHttpServer {
    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(perms),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("http://{addr}");

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    // Create a dummy reload handle for tests (the filter is never actually used
    // since the test's tracing subscriber is separate).
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: None,
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestHttpServer {
        base_url,
        engine,
        lifecycle,
    }
}

/// Start an HTTP server with admin API key enabled (no TLS).
/// Used for testing admin endpoints like /admin/config and /admin/reload.
/// The shared config includes namespace auth so /admin/config has data to redact.
pub async fn start_test_http_admin_server(admin_key: &str) -> TestHttpServer {
    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("http://{addr}");

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    // Build a ServerConfig with actual namespace auth so /admin/config has keys to redact.
    let server_config: ServerConfig = toml::from_str(&format!(
        r#"
[namespaces.{TEST_NS}]
api_key = "{TEST_KEY}"
permissions = ["Subscribe", "Publish", "Query"]
"#
    ))
    .unwrap();
    let shared_config = fafb_q::config::shared_config(&server_config);

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: Some(admin_key.to_owned()),
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config,
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TestHttpServer {
        base_url,
        engine,
        lifecycle,
    }
}

/// HTTP test server with a real config file and ReloadState wired.
/// Used for testing reload behavior (drain rejection, file watcher, metrics).
pub struct ReloadTestHttpServer {
    pub base_url: String,
    pub lifecycle: Arc<ServerLifecycle>,
    pub config_file: PathBuf,
    pub reload_state: Arc<fafb_q::config::ReloadState>,
    pub shared_config: SharedConfig,
}

impl ReloadTestHttpServer {
    pub fn shutdown(&self) {
        self.lifecycle.transition_to(ServerState::ShuttingDown);
    }
}

/// Start an HTTP server with a real config file and ReloadState for reload testing.
pub async fn start_test_http_reload_server(
    admin_key: &str,
    config_file: &Path,
) -> ReloadTestHttpServer {
    use fafb_q::config::{ReloadState, ServeArgs};

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("http://{addr}");

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    // Load config from the test file
    let server_config: ServerConfig =
        toml::from_str(&std::fs::read_to_string(config_file).unwrap()).unwrap();
    let shared_config = fafb_q::config::shared_config(&server_config);

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

    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::new(config_file.to_path_buf()),
            Arc::clone(&shared_config),
            server_config,
            reload_notify_tx,
            ServeArgs {
                config: Some(config_file.to_path_buf()),
                dump_config: false,
                tcp_bind: None,
                http_bind: None,
                data_dir: None,
                log_level: None,
            },
        )
        .with_lifecycle(Arc::clone(&lifecycle)),
    );

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: Some(admin_key.to_owned()),
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: Arc::clone(&shared_config),
            config_path: Some(Arc::new(config_file.to_path_buf())),
            reload_state: Some(Arc::clone(&reload_state)),
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    ReloadTestHttpServer {
        base_url,
        lifecycle,
        config_file: config_file.to_path_buf(),
        reload_state,
        shared_config,
    }
}

// -- Deploy integration test infrastructure (FsStorage-backed) --

/// A combined TCP + HTTP server backed by FsStorage for deploy integration tests.
/// Supports graceful drain/shutdown and server restarts on the same data directory.
pub struct DeployTestServer {
    pub tcp_addr: String,
    pub http_base_url: String,
    pub lifecycle: Arc<ServerLifecycle>,
    pub engine: Arc<InMemoryQueueEngine<FsStorage>>,
    pub manager: Arc<DefaultNamespaceManager<InMemoryQueueEngine<FsStorage>>>,
    pub data_dir: PathBuf,
    pub deploy_config: DeployConfig,
}

impl DeployTestServer {
    /// Replicate the main.rs graceful shutdown sequence:
    /// force-flush → Draining → wait for connections → ShuttingDown.
    pub async fn graceful_drain(&self) {
        let drain_secs = self.deploy_config.drain_timeout_secs;

        // 1. Force-flush storage for durability.
        let _ = self.engine.force_flush().await;

        // 2. Transition to Draining (with wall-clock timestamps).
        self.lifecycle.begin_drain(drain_secs);

        // 3. Accelerate inflight timeout.
        let accelerated = drain_secs.saturating_div(2).clamp(1, 30);
        self.manager
            .set_inflight_timeout(Duration::from_secs(accelerated))
            .await;

        // 4. Wait, record metric, transition to ShuttingDown.
        self.lifecycle
            .run_drain_wait(Duration::from_secs(drain_secs))
            .await;
    }

    /// Full shutdown: graceful drain then engine/manager cleanup.
    pub async fn shutdown(self) {
        self.graceful_drain().await;
        // Brief pause for connection handlers to react to ShuttingDown.
        tokio::time::sleep(Duration::from_millis(100)).await;
        let _ = self.manager.shutdown().await;
        let _ = self.engine.shutdown().await;
    }
}

/// Start a combined TCP + HTTP server backed by FsStorage.
///
/// Each call creates a new engine and manager on the given `data_dir`.
/// Reuse the same `data_dir` across calls to simulate server restarts.
pub async fn start_deploy_test_server(
    data_dir: &Path,
    deploy_config: DeployConfig,
    engine_config: EngineConfig,
) -> DeployTestServer {
    let storage = Arc::new(FsStorage::new(data_dir));
    storage.init().await.expect("storage init failed");

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

    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
        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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

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

    let message_id_counter = Arc::new(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 = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: tcp_listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let tcp_server = TcpServer::new(
        tcp_config,
        deploy_config.clone(),
        test_shared_config(),
        Arc::clone(&manager),
        Arc::clone(&auth),
        Arc::clone(&message_id_counter),
        Arc::clone(&lifecycle),
        None, // no TLS
    );
    tokio::spawn(async move {
        let _ = tcp_server.run_with_listener(tcp_listener).await;
    });

    // -- HTTP server --
    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 = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: http_addr.port(),
        ..Default::default()
    };
    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: None,
            drain_trigger: None,
            deploy_config: deploy_config.clone(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );
    tokio::spawn(async move {
        let _ = http_server.run_with_listener(http_listener).await;
    });

    DeployTestServer {
        tcp_addr,
        http_base_url,
        lifecycle,
        engine,
        manager,
        data_dir: data_dir.to_path_buf(),
        deploy_config,
    }
}

/// Start a deploy test server with an admin API key and a drain trigger channel.
///
/// The returned `mpsc::Receiver` receives drain requests from the HTTP admin
/// endpoint. The test is responsible for handling drain requests (calling
/// `engine.force_flush()`, `lifecycle.begin_drain()`, etc.) — just like main.rs.
pub async fn start_admin_test_server(
    data_dir: &Path,
    deploy_config: DeployConfig,
    engine_config: EngineConfig,
) -> (DeployTestServer, tokio::sync::mpsc::Receiver<DrainRequest>) {
    let storage = Arc::new(FsStorage::new(data_dir));
    storage.init().await.expect("storage init failed");

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

    let manager = Arc::new(DefaultNamespaceManager::new(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
        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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

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

    let message_id_counter = Arc::new(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 = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: tcp_listener.local_addr().unwrap().port(),
        ..Default::default()
    };
    let tcp_server = TcpServer::new(
        tcp_config,
        deploy_config.clone(),
        test_shared_config(),
        Arc::clone(&manager),
        Arc::clone(&auth),
        Arc::clone(&message_id_counter),
        Arc::clone(&lifecycle),
        None, // no TLS
    );
    tokio::spawn(async move {
        let _ = tcp_server.run_with_listener(tcp_listener).await;
    });

    // -- HTTP server with drain trigger --
    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 = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: http_addr.port(),
        ..Default::default()
    };
    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: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        None, // no TLS
    );
    tokio::spawn(async move {
        let _ = http_server.run_with_listener(http_listener).await;
    });

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

    (server, drain_rx)
}

// -- TLS test infrastructure --

/// Generated test certificate artifacts. All fields are PEM-encoded strings.
pub struct TestCerts {
    pub cert_pem: String,
    pub key_pem: String,
    pub cert_file: tempfile::NamedTempFile,
    pub key_file: tempfile::NamedTempFile,
    /// DER-encoded root certificate for use in client trust stores.
    pub root_cert_der: Vec<u8>,
}

/// Generate a self-signed server certificate for `localhost` with 365-day validity.
///
/// Returns cert/key as temp files (for `TlsConfig`) and the DER root cert
/// (for building a `tokio-rustls` `TlsConnector` in test clients).
pub fn generate_test_certs() -> TestCerts {
    let mut params = rcgen::CertificateParams::new(vec!["localhost".to_owned()]).unwrap();
    params
        .subject_alt_names
        .push(rcgen::SanType::IpAddress(std::net::IpAddr::V4(
            std::net::Ipv4Addr::LOCALHOST,
        )));
    let key_pair = rcgen::KeyPair::generate().unwrap();
    let cert = params.self_signed(&key_pair).unwrap();

    let cert_pem = cert.pem();
    let key_pem = key_pair.serialize_pem();
    let root_cert_der = cert.der().to_vec();

    let mut cert_file = tempfile::NamedTempFile::new().unwrap();
    cert_file.write_all(cert_pem.as_bytes()).unwrap();

    let mut key_file = tempfile::NamedTempFile::new().unwrap();
    key_file.write_all(key_pem.as_bytes()).unwrap();

    TestCerts {
        cert_pem,
        key_pem,
        cert_file,
        key_file,
        root_cert_der,
    }
}

/// Build a `TlsConfig` from test certs.
pub fn make_test_tls_config(certs: &TestCerts) -> TlsConfig {
    TlsConfig {
        enabled: true,
        cert_file: certs.cert_file.path().to_str().unwrap().to_owned(),
        key_file: certs.key_file.path().to_str().unwrap().to_owned(),
        client_ca_file: String::new(),
        min_version: "1.2".to_owned(),
        reload_interval_secs: 0,
        ciphersuite_allowlist: Vec::new(),
        handshake_timeout_secs: 10,
    }
}

/// A TLS test server with its address and certs.
pub struct TlsTestServer {
    pub addr: String,
    pub lifecycle: Arc<ServerLifecycle>,
    pub certs: TestCerts,
}

impl TlsTestServer {
    pub fn shutdown(&self) {
        self.lifecycle.transition_to(ServerState::ShuttingDown);
    }
}

/// Start a TCP server with TLS enabled on a random port.
pub async fn start_test_tls_tcp_server() -> TlsTestServer {
    start_test_tls_tcp_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
        Permission::Fetch,
    ])
    .await
}

/// Start a TLS-enabled TCP server with specific permissions.
pub async fn start_test_tls_tcp_server_with_perms(perms: &[Permission]) -> TlsTestServer {
    let certs = generate_test_certs();
    let tls_config = make_test_tls_config(&certs);
    start_test_tls_tcp_server_inner(certs, tls_config, perms).await
}

/// Start a TLS-enabled TCP server with a custom `TlsConfig`.
///
/// Generates fresh test certs and merges them into the provided config
/// (cert_file/key_file are overwritten with the generated paths).
pub async fn start_test_tls_tcp_server_with_config(mut tls_config: TlsConfig) -> TlsTestServer {
    let certs = generate_test_certs();
    // Overwrite cert/key paths with the generated certs (caller may have set
    // other fields like handshake_timeout_secs).
    tls_config.cert_file = certs.cert_file.path().to_str().unwrap().to_owned();
    tls_config.key_file = certs.key_file.path().to_str().unwrap().to_owned();
    let perms = &[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
        Permission::Fetch,
    ];
    start_test_tls_tcp_server_inner(certs, tls_config, perms).await
}

async fn start_test_tls_tcp_server_inner(
    certs: TestCerts,
    tls_config: TlsConfig,
    perms: &[Permission],
) -> TlsTestServer {
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid test TLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        engine,
        vec![NamespaceId(TEST_NS.into())],
    ));

    let auth = Arc::new(MockAuthenticator {
        namespace: NamespaceId(TEST_NS.into()),
        key: ApiKey(TEST_KEY.into()),
        permissions: PermissionSet::from_slice(perms),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();

    let config = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: listener.local_addr().unwrap().port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        test_shared_config(),
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TlsTestServer {
        addr,
        lifecycle,
        certs,
    }
}

/// Build a `tokio_rustls::TlsConnector` that trusts the given test root cert.
pub fn make_test_tls_connector(root_cert_der: &[u8]) -> tokio_rustls::TlsConnector {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(rustls::pki_types::CertificateDer::from(
            root_cert_der.to_vec(),
        ))
        .expect("add test root cert");
    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let client_config = rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .expect("valid TLS versions")
        .with_root_certificates(root_store)
        .with_no_client_auth();
    tokio_rustls::TlsConnector::from(Arc::new(client_config))
}

// -- HTTPS test infrastructure --

/// An HTTPS test server with its base URL and certs.
pub struct TlsHttpTestServer {
    pub base_url: String,
    pub engine: Arc<InMemoryQueueEngine<MockStorage>>,
    pub lifecycle: Arc<ServerLifecycle>,
    pub certs: TestCerts,
}

impl TlsHttpTestServer {
    pub fn shutdown(&self) {
        self.lifecycle.transition_to(ServerState::ShuttingDown);
    }
}

/// Start an HTTPS server on a random port with all permissions.
pub async fn start_test_https_server() -> TlsHttpTestServer {
    start_test_https_server_with_perms(&[
        Permission::Subscribe,
        Permission::Publish,
        Permission::Query,
        Permission::Fetch,
    ])
    .await
}

/// Start an HTTPS server with specific permissions for the test API key.
pub async fn start_test_https_server_with_perms(perms: &[Permission]) -> TlsHttpTestServer {
    let certs = generate_test_certs();
    let tls_config = make_test_tls_config(&certs);
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid test TLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(perms),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("https://127.0.0.1:{}", addr.port());

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: None,
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TlsHttpTestServer {
        base_url,
        engine,
        lifecycle,
        certs,
    }
}

/// Start an HTTPS server with a custom `TlsConfig`.
///
/// Generates fresh test certs and merges them into the provided config
/// (cert_file/key_file are overwritten with the generated paths).
pub async fn start_test_https_server_with_tls_config(
    mut tls_config: TlsConfig,
) -> TlsHttpTestServer {
    let certs = generate_test_certs();
    tls_config.cert_file = certs.cert_file.path().to_str().unwrap().to_owned();
    tls_config.key_file = certs.key_file.path().to_str().unwrap().to_owned();

    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid test TLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("https://127.0.0.1:{}", addr.port());

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: None,
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TlsHttpTestServer {
        base_url,
        engine,
        lifecycle,
        certs,
    }
}

/// Start an HTTPS server with admin API key configured.
pub async fn start_test_https_admin_server(admin_key: &str) -> TlsHttpTestServer {
    let certs = generate_test_certs();
    let tls_config = make_test_tls_config(&certs);
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid test TLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("https://127.0.0.1:{}", addr.port());

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: Some(admin_key.to_owned()),
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TlsHttpTestServer {
        base_url,
        engine,
        lifecycle,
        certs,
    }
}

/// An HTTPS test server with TLS provider exposed for reload testing.
pub struct TlsReloadTestServer {
    pub base_url: String,
    pub lifecycle: Arc<ServerLifecycle>,
    pub certs: TestCerts,
    pub tls_provider: Arc<TlsProvider>,
}

/// Start an HTTPS admin server that wires TlsProviders via Extension for reload testing.
pub async fn start_test_https_reload_server(admin_key: &str) -> TlsReloadTestServer {
    use fafb_q::tls::{LabeledTlsProvider, TlsProviders};

    let certs = generate_test_certs();
    let tls_config = make_test_tls_config(&certs);
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid test TLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("https://127.0.0.1:{}", addr.port());

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let providers = TlsProviders(vec![LabeledTlsProvider {
        label: "shared",
        provider: Arc::clone(&tls_provider),
    }]);
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: Some(admin_key.to_owned()),
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        Some(Arc::clone(&tls_provider)),
    )
    .with_tls_providers(providers);

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    TlsReloadTestServer {
        base_url,
        lifecycle,
        certs,
        tls_provider,
    }
}

/// Build a `reqwest::Client` that trusts the given test root cert for HTTPS.
pub fn make_test_https_client(root_cert_der: &[u8]) -> reqwest::Client {
    ensure_ring_provider();
    let cert = reqwest::tls::Certificate::from_der(root_cert_der).expect("valid DER cert");
    reqwest::Client::builder()
        .add_root_certificate(cert)
        .build()
        .expect("build reqwest client")
}

// -- mTLS test infrastructure --

/// Generated test artifacts for mTLS testing.
///
/// Contains a self-signed CA, a server cert signed by the CA, and client
/// certs (valid + untrusted) for mutual TLS testing.
pub struct MtlsTestCerts {
    /// PEM CA cert written to a temp file — used as `client_ca_file` in `TlsConfig`.
    pub ca_cert_file: tempfile::NamedTempFile,
    /// Server cert file (signed by CA).
    pub server_cert_file: tempfile::NamedTempFile,
    /// Server key file.
    pub server_key_file: tempfile::NamedTempFile,
    /// DER-encoded CA cert — clients trust this to verify the server (same CA signs server cert).
    pub server_ca_der: Vec<u8>,
    /// Client cert PEM (signed by CA).
    pub client_cert_pem: String,
    /// Client key PEM.
    pub client_key_pem: String,
    /// Untrusted client cert PEM (signed by a different CA).
    pub untrusted_client_cert_pem: String,
    /// Untrusted client key PEM.
    pub untrusted_client_key_pem: String,
}

/// Ensure the ring crypto provider is installed as the process-level default.
///
/// With `reqwest`'s `rustls-tls` feature, both `ring` and `aws-lc-rs` may be
/// available. Call this before any code that needs an implicit provider.
fn ensure_ring_provider() {
    let _ = rustls::crypto::ring::default_provider().install_default();
}

/// Generate a full set of mTLS test certificates:
/// - CA (self-signed root)
/// - Server cert (signed by CA, SAN=localhost+127.0.0.1)
/// - Valid client cert (signed by CA, CN=test-client)
/// - Untrusted client cert (signed by a *different* CA)
pub fn generate_mtls_test_certs() -> MtlsTestCerts {
    ensure_ring_provider();
    // 1. Create CA
    let ca_key_pair = rcgen::KeyPair::generate().unwrap();
    let mut ca_params = rcgen::CertificateParams::new(vec![]).unwrap();
    ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
    ca_params
        .distinguished_name
        .push(rcgen::DnType::CommonName, "Test CA");
    let ca_cert = ca_params.self_signed(&ca_key_pair).unwrap();
    let ca_cert_der = ca_cert.der().to_vec();

    // Write CA cert to temp file for client_ca_file config
    let mut ca_cert_file = tempfile::NamedTempFile::new().unwrap();
    ca_cert_file.write_all(ca_cert.pem().as_bytes()).unwrap();

    // 2. Create server cert signed by CA
    let server_key_pair = rcgen::KeyPair::generate().unwrap();
    let mut server_params = rcgen::CertificateParams::new(vec!["localhost".to_owned()]).unwrap();
    server_params
        .subject_alt_names
        .push(rcgen::SanType::IpAddress(std::net::IpAddr::V4(
            std::net::Ipv4Addr::LOCALHOST,
        )));
    let server_cert = server_params
        .signed_by(&server_key_pair, &ca_cert, &ca_key_pair)
        .unwrap();

    let mut server_cert_file = tempfile::NamedTempFile::new().unwrap();
    server_cert_file
        .write_all(server_cert.pem().as_bytes())
        .unwrap();

    let mut server_key_file = tempfile::NamedTempFile::new().unwrap();
    server_key_file
        .write_all(server_key_pair.serialize_pem().as_bytes())
        .unwrap();

    // 3. Create valid client cert signed by CA
    let client_key_pair = rcgen::KeyPair::generate().unwrap();
    let mut client_params = rcgen::CertificateParams::new(vec![]).unwrap();
    client_params
        .distinguished_name
        .push(rcgen::DnType::CommonName, "test-client");
    let client_cert = client_params
        .signed_by(&client_key_pair, &ca_cert, &ca_key_pair)
        .unwrap();
    let client_cert_pem = client_cert.pem();
    let client_key_pem = client_key_pair.serialize_pem();

    // 4. Create untrusted client cert (signed by a different CA)
    let rogue_ca_key = rcgen::KeyPair::generate().unwrap();
    let mut rogue_ca_params = rcgen::CertificateParams::new(vec![]).unwrap();
    rogue_ca_params.is_ca = rcgen::IsCa::Ca(rcgen::BasicConstraints::Unconstrained);
    rogue_ca_params
        .distinguished_name
        .push(rcgen::DnType::CommonName, "Rogue CA");
    let rogue_ca_cert = rogue_ca_params.self_signed(&rogue_ca_key).unwrap();

    let untrusted_key = rcgen::KeyPair::generate().unwrap();
    let mut untrusted_params = rcgen::CertificateParams::new(vec![]).unwrap();
    untrusted_params
        .distinguished_name
        .push(rcgen::DnType::CommonName, "untrusted-client");
    let untrusted_cert = untrusted_params
        .signed_by(&untrusted_key, &rogue_ca_cert, &rogue_ca_key)
        .unwrap();

    MtlsTestCerts {
        ca_cert_file,
        server_cert_file,
        server_key_file,
        server_ca_der: ca_cert_der,
        client_cert_pem,
        client_key_pem,
        untrusted_client_cert_pem: untrusted_cert.pem(),
        untrusted_client_key_pem: untrusted_key.serialize_pem(),
    }
}

/// Build a TLS config for mTLS server testing.
pub fn make_test_mtls_tls_config(certs: &MtlsTestCerts) -> TlsConfig {
    TlsConfig {
        enabled: true,
        cert_file: certs.server_cert_file.path().to_str().unwrap().to_owned(),
        key_file: certs.server_key_file.path().to_str().unwrap().to_owned(),
        client_ca_file: certs.ca_cert_file.path().to_str().unwrap().to_owned(),
        min_version: "1.2".to_owned(),
        reload_interval_secs: 0,
        ciphersuite_allowlist: Vec::new(),
        handshake_timeout_secs: 10,
    }
}

/// Build a `tokio_rustls::TlsConnector` with a client certificate for mTLS.
///
/// The connector trusts the server's CA and presents the given client cert.
pub fn make_test_mtls_connector(
    server_ca_der: &[u8],
    client_cert_pem: &str,
    client_key_pem: &str,
) -> tokio_rustls::TlsConnector {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(rustls::pki_types::CertificateDer::from(
            server_ca_der.to_vec(),
        ))
        .expect("add server CA cert");

    // Parse client cert chain
    let client_certs: Vec<rustls::pki_types::CertificateDer<'static>> =
        rustls_pemfile::certs(&mut client_cert_pem.as_bytes())
            .collect::<std::result::Result<Vec<_>, _>>()
            .expect("parse client cert PEM");

    // Parse client private key
    let client_key = rustls_pemfile::private_key(&mut client_key_pem.as_bytes())
        .expect("parse client key PEM")
        .expect("no key found in PEM");

    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let client_config = rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .expect("valid TLS versions")
        .with_root_certificates(root_store)
        .with_client_auth_cert(client_certs, client_key)
        .expect("valid client cert/key");

    tokio_rustls::TlsConnector::from(Arc::new(client_config))
}

/// Build a `tokio_rustls::TlsConnector` that trusts the server CA but has NO client cert.
///
/// Used to test mTLS rejection when client doesn't present a certificate.
pub fn make_test_no_client_cert_connector(server_ca_der: &[u8]) -> tokio_rustls::TlsConnector {
    let mut root_store = rustls::RootCertStore::empty();
    root_store
        .add(rustls::pki_types::CertificateDer::from(
            server_ca_der.to_vec(),
        ))
        .expect("add server CA cert");

    let provider = Arc::new(rustls::crypto::ring::default_provider());
    let client_config = rustls::ClientConfig::builder_with_provider(provider)
        .with_safe_default_protocol_versions()
        .expect("valid TLS versions")
        .with_root_certificates(root_store)
        .with_no_client_auth();

    tokio_rustls::TlsConnector::from(Arc::new(client_config))
}

/// Start an mTLS-enabled TCP server on a random port.
pub async fn start_test_mtls_tcp_server() -> (TlsTestServer, MtlsTestCerts) {
    let certs = generate_mtls_test_certs();
    let tls_config = make_test_mtls_tls_config(&certs);
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid mTLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        engine,
        vec![NamespaceId(TEST_NS.into())],
    ));

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

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap().to_string();

    let config = TcpConfig {
        bind_address: "127.0.0.1".into(),
        port: listener.local_addr().unwrap().port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    let (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = TcpServer::new(
        config,
        DeployConfig::default(),
        test_shared_config(),
        manager,
        auth,
        message_id_counter,
        Arc::clone(&lifecycle),
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    let test_server = TlsTestServer {
        addr,
        lifecycle,
        // Store a dummy TestCerts — the real certs are in MtlsTestCerts.
        // We need to keep MtlsTestCerts alive so temp files aren't deleted.
        certs: generate_test_certs(),
    };
    (test_server, certs)
}

/// Start an mTLS-enabled HTTPS server on a random port.
pub async fn start_test_mtls_https_server() -> (TlsHttpTestServer, MtlsTestCerts) {
    let certs = generate_mtls_test_certs();
    let tls_config = make_test_mtls_tls_config(&certs);
    let tls_provider = TlsProvider::from_config(&tls_config)
        .expect("valid mTLS config")
        .expect("TLS enabled");
    let tls_provider = Arc::new(tls_provider);

    let engine = make_engine();
    let manager = Arc::new(make_namespace_manager(
        Arc::clone(&engine),
        vec![NamespaceId(TEST_NS.into())],
    ));

    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(&[
            Permission::Subscribe,
            Permission::Publish,
            Permission::Query,
            Permission::Fetch,
        ]),
    });

    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
    let addr = listener.local_addr().unwrap();
    let base_url = format!("https://127.0.0.1:{}", addr.port());

    let config = HttpConfig {
        bind_address: "127.0.0.1".into(),
        port: addr.port(),
        ..Default::default()
    };

    let message_id_counter = Arc::new(AtomicU64::new(1));
    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 (lifecycle, _lifecycle_rx) = ServerLifecycle::new();
    let lifecycle = Arc::new(lifecycle);
    lifecycle.set_initialized();
    let server = HttpServer::new(
        config,
        AppState {
            manager,
            auth,
            message_id_counter,
            prom_handle: None,
            log_reload_handle: dummy_reload_handle,
            lifecycle: Arc::clone(&lifecycle),
            admin_api_key: None,
            drain_trigger: None,
            deploy_config: DeployConfig::default(),
            shared_config: test_shared_config(),
            config_path: test_config_path(),
            reload_state: None,
            grace_pending: std::sync::Arc::new(tokio::sync::RwLock::new(
                std::collections::HashMap::new(),
            )),
        },
        Some(tls_provider),
    );

    tokio::spawn(async move {
        let _ = server.run_with_listener(listener).await;
    });

    let test_server = TlsHttpTestServer {
        base_url,
        engine,
        lifecycle,
        certs: generate_test_certs(),
    };
    (test_server, certs)
}

/// Build a `reqwest::Client` with mTLS client certificate for HTTPS testing.
///
/// Trusts the server CA and presents the client cert for mTLS.
pub fn make_test_mtls_https_client(mtls_certs: &MtlsTestCerts) -> reqwest::Client {
    ensure_ring_provider();
    let ca_cert =
        reqwest::tls::Certificate::from_der(&mtls_certs.server_ca_der).expect("valid CA DER cert");

    // Build PEM identity (cert + key concatenated)
    let mut identity_pem = mtls_certs.client_cert_pem.clone();
    identity_pem.push_str(&mtls_certs.client_key_pem);
    let identity =
        reqwest::tls::Identity::from_pem(identity_pem.as_bytes()).expect("valid PEM identity");

    reqwest::Client::builder()
        .add_root_certificate(ca_cert)
        .identity(identity)
        .build()
        .expect("build mTLS reqwest client")
}

/// Build a `reqwest::Client` that trusts the server CA but has NO client cert (for mTLS rejection tests).
pub fn make_test_no_client_cert_https_client(mtls_certs: &MtlsTestCerts) -> reqwest::Client {
    ensure_ring_provider();
    let ca_cert =
        reqwest::tls::Certificate::from_der(&mtls_certs.server_ca_der).expect("valid CA DER cert");

    reqwest::Client::builder()
        .add_root_certificate(ca_cert)
        .build()
        .expect("build no-client-cert reqwest client")
}
