mod common;

use std::sync::Arc;

use base64::Engine as _;
use fafb_q::queue::QueueEngine;
use fafb_q::types::{Header, MessageEnvelope, MessageId, NamespaceId, Permission, TopicName};

use common::{
    TEST_KEY, TEST_NS, start_test_http_server, start_test_http_server_all_perms,
    start_test_http_server_with_perms, start_test_http_server_with_shared_config,
};

fn client() -> reqwest::Client {
    reqwest::Client::new()
}

// -- Health endpoint --

#[tokio::test]
async fn health_returns_200() {
    let server = start_test_http_server().await;
    let resp = client()
        .get(format!("{}/health", server.base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    server.shutdown();
}

// -- Publish endpoint --

#[tokio::test]
async fn publish_returns_offset_and_message_id() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("hello world")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["offset"], 0);
    assert!(body["message_id"].as_u64().is_some());

    server.shutdown();
}

#[tokio::test]
async fn publish_message_lands_in_queue_engine() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!("{}/v1/{}/events/publish", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("payload-data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Verify the message is in the queue engine.
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, 0u64.into())
        .await
        .unwrap()
        .expect("message should exist in engine");
    assert_eq!(msg.envelope.payload, b"payload-data");

    server.shutdown();
}

#[tokio::test]
async fn publish_sequential_offsets() {
    let server = start_test_http_server().await;
    for i in 0..5u64 {
        let resp = client()
            .post(format!(
                "{}/v1/{}/seq-topic/publish",
                server.base_url, TEST_NS
            ))
            .header("Authorization", format!("Bearer {TEST_KEY}"))
            .body(format!("msg-{i}"))
            .send()
            .await
            .unwrap();
        let body: serde_json::Value = resp.json().await.unwrap();
        assert_eq!(body["offset"], i);
    }
    server.shutdown();
}

// -- Auth: missing / invalid key --

#[tokio::test]
async fn publish_without_auth_returns_401() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .body("data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert!(body["error"]["name"].as_str().unwrap().contains("AUTH"));
    server.shutdown();
}

#[tokio::test]
async fn publish_with_bad_key_returns_401() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", "Bearer wrong-key")
        .body("data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
    server.shutdown();
}

// -- Auth: permission denied --

#[tokio::test]
async fn publish_without_publish_permission_returns_403() {
    let server =
        start_test_http_server_with_perms(&[Permission::Subscribe, Permission::Query]).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    server.shutdown();
}

#[tokio::test]
async fn query_without_query_permission_returns_403() {
    let server =
        start_test_http_server_with_perms(&[Permission::Subscribe, Permission::Publish]).await;
    let resp = client()
        .get(format!("{}/v1/{}/my-topic/info", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    server.shutdown();
}

// -- Topic info endpoint --

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

    // Publish 3 messages.
    for i in 0..3 {
        client()
            .post(format!(
                "{}/v1/{}/info-topic/publish",
                server.base_url, TEST_NS
            ))
            .header("Authorization", format!("Bearer {TEST_KEY}"))
            .body(format!("msg-{i}"))
            .send()
            .await
            .unwrap();
    }

    let resp = client()
        .get(format!(
            "{}/v1/{}/info-topic/info",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["depth"], 3);
    assert_eq!(body["head_offset"], 3);
    assert_eq!(body["base_offset"], 0);
    assert!(body["bytes_used"].as_u64().unwrap() > 0);

    server.shutdown();
}

#[tokio::test]
async fn topic_info_nonexistent_returns_404() {
    let server = start_test_http_server().await;
    let resp = client()
        .get(format!(
            "{}/v1/{}/nonexistent/info",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 404);
    server.shutdown();
}

// -- List topics endpoint --

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

    // Publish to two different topics.
    for topic in &["alpha", "beta"] {
        client()
            .post(format!(
                "{}/v1/{}/{topic}/publish",
                server.base_url, TEST_NS
            ))
            .header("Authorization", format!("Bearer {TEST_KEY}"))
            .body("data")
            .send()
            .await
            .unwrap();
    }

    let resp = client()
        .get(format!("{}/v1/{}/topics", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let topics = body["topics"].as_array().unwrap();
    assert_eq!(topics.len(), 2);
    assert!(topics.contains(&serde_json::json!("alpha")));
    assert!(topics.contains(&serde_json::json!("beta")));

    server.shutdown();
}

#[tokio::test]
async fn list_topics_empty_namespace() {
    let server = start_test_http_server().await;
    let resp = client()
        .get(format!("{}/v1/{}/topics", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["topics"].as_array().unwrap().len(), 0);
    server.shutdown();
}

// -- Namespace mismatch --

#[tokio::test]
async fn publish_to_wrong_namespace_returns_401() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!(
            "{}/v1/other-namespace/my-topic/publish",
            server.base_url
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("data")
        .send()
        .await
        .unwrap();
    // The auth helper authenticates the key (which belongs to TEST_NS),
    // but the handler rejects because path namespace doesn't match session namespace.
    assert_eq!(resp.status(), 401);
    server.shutdown();
}

// -- Edge cases --

#[tokio::test]
async fn publish_empty_body() {
    let server = start_test_http_server().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/empty-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["offset"], 0);

    // Verify empty payload is stored.
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("empty-topic".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, 0u64.into())
        .await
        .unwrap()
        .unwrap();
    assert!(msg.envelope.payload.is_empty());
    server.shutdown();
}

#[tokio::test]
async fn publish_large_payload() {
    let server = start_test_http_server().await;
    let payload = vec![0xABu8; 100_000]; // 100KB
    let resp = client()
        .post(format!(
            "{}/v1/{}/large-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body(payload.clone())
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("large-topic".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, 0u64.into())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.payload.len(), 100_000);
    server.shutdown();
}

#[tokio::test]
async fn concurrent_publishes_get_unique_offsets() {
    let server = start_test_http_server().await;
    let n = 50;
    let mut handles = Vec::new();

    for i in 0..n {
        let base_url = server.base_url.clone();
        handles.push(tokio::spawn(async move {
            let resp = reqwest::Client::new()
                .post(format!("{base_url}/v1/{TEST_NS}/concurrent-topic/publish"))
                .header("Authorization", format!("Bearer {TEST_KEY}"))
                .body(format!("msg-{i}"))
                .send()
                .await
                .unwrap();
            assert_eq!(resp.status(), 200);
            let body: serde_json::Value = resp.json().await.unwrap();
            body["offset"].as_u64().unwrap()
        }));
    }

    let mut offsets = Vec::new();
    for h in handles {
        offsets.push(h.await.unwrap());
    }
    offsets.sort();

    // All offsets should be unique and sequential 0..n.
    let expected: Vec<u64> = (0..n).collect();
    assert_eq!(offsets, expected);
    server.shutdown();
}

// -- Error response structure --

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

    // Missing auth → 401 with NOT_AUTHENTICATED
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .body("data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 401);
    let body: serde_json::Value = resp.json().await.unwrap();
    let err = &body["error"];
    assert!(err["code"].is_u64(), "error.code should be a number");
    assert!(err["name"].is_string(), "error.name should be a string");
    assert!(
        err["message"].is_string(),
        "error.message should be a string"
    );
    assert_eq!(err["name"], "NOT_AUTHENTICATED");

    // Bad key → 401 with AUTH_FAILED
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", "Bearer bad-key")
        .body("data")
        .send()
        .await
        .unwrap();
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "AUTH_FAILED");

    // Permission denied → 403 with PERMISSION_DENIED
    let server_query_only = start_test_http_server_with_perms(&[Permission::Query]).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/my-topic/publish",
            server_query_only.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("data")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "PERMISSION_DENIED");
    assert_eq!(body["error"]["code"], 0x0002);

    // Topic not found → 404 with TOPIC_NOT_FOUND
    let resp = client()
        .get(format!(
            "{}/v1/{}/nonexistent/info",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 404);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "TOPIC_NOT_FOUND");

    server.shutdown();
    server_query_only.shutdown();
}

// -- Fetch endpoint --

/// Helper: publish N messages to a topic via HTTP.
async fn publish_messages(base_url: &str, topic: &str, count: usize) {
    for i in 0..count {
        let resp = client()
            .post(format!("{base_url}/v1/{TEST_NS}/{topic}/publish"))
            .header("Authorization", format!("Bearer {TEST_KEY}"))
            .body(format!("msg-{i}"))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 200);
    }
}

#[tokio::test]
async fn http_fetch_basic() {
    let server = start_test_http_server_all_perms().await;
    publish_messages(&server.base_url, "fetch-basic", 5).await;

    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-basic/fetch?offset=0&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["topic"], "fetch-basic");
    assert_eq!(body["next_offset"], 5);
    assert_eq!(body["head_offset"], 5);
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 5);
    assert_eq!(messages[0]["offset"], 0);
    assert_eq!(messages[4]["offset"], 4);
    // Verify payload is base64 encoded.
    let payload_b64 = messages[0]["payload_base64"].as_str().unwrap();
    let decoded = base64::engine::general_purpose::STANDARD
        .decode(payload_b64)
        .unwrap();
    assert_eq!(decoded, b"msg-0");

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_by_timestamp() {
    use fafb_q::types::{MessageEnvelope, MessageId};

    let server = start_test_http_server_all_perms().await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("fetch-ts".into());

    // Publish 3 messages with an explicit early timestamp.
    let early_ts = 1_000_000u64;
    for i in 0..3 {
        let envelope = MessageEnvelope::new(
            MessageId(i + 1),
            ns.clone(),
            topic.clone(),
            format!("early-{i}").into_bytes(),
            early_ts + i,
        );
        server.engine.publish(&ns, &topic, envelope).await.unwrap();
    }

    // Publish 2 messages with a later timestamp.
    let late_ts = 2_000_000u64;
    for i in 0..2 {
        let envelope = MessageEnvelope::new(
            MessageId(10 + i),
            ns.clone(),
            topic.clone(),
            format!("late-{i}").into_bytes(),
            late_ts + i,
        );
        server.engine.publish(&ns, &topic, envelope).await.unwrap();
    }

    // Fetch from a timestamp between the two groups — should return only the late messages.
    let mid_ts = 1_500_000u64;
    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-ts/fetch?timestamp={mid_ts}&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 2);
    // First returned message should be at offset 3 (the first "late" message).
    assert_eq!(messages[0]["offset"], 3);
    assert_eq!(body["next_offset"], 5);
    assert_eq!(body["head_offset"], 5);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_pagination() {
    let server = start_test_http_server_all_perms().await;
    publish_messages(&server.base_url, "fetch-page", 50).await;

    let mut offset = 0u64;
    let mut total_fetched = 0;
    loop {
        let resp = client()
            .get(format!(
                "{}/v1/{}/fetch-page/fetch?offset={offset}&max_count=10",
                server.base_url, TEST_NS
            ))
            .header("Authorization", format!("Bearer {TEST_KEY}"))
            .send()
            .await
            .unwrap();
        assert_eq!(resp.status(), 200);
        let body: serde_json::Value = resp.json().await.unwrap();
        let messages = body["messages"].as_array().unwrap();
        total_fetched += messages.len();
        let next = body["next_offset"].as_u64().unwrap();
        let head = body["head_offset"].as_u64().unwrap();
        if next >= head {
            break;
        }
        offset = next;
    }
    assert_eq!(total_fetched, 50);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_permission_denied() {
    // Server with only Subscribe + Publish + Query (no Fetch).
    let server = start_test_http_server().await;

    let resp = client()
        .get(format!(
            "{}/v1/{}/some-topic/fetch?offset=0",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "PERMISSION_DENIED");

    server.shutdown();
}

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

    // No offset or timestamp — should get 400.
    let resp = client()
        .get(format!(
            "{}/v1/{}/some-topic/fetch",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "INVALID_REQUEST");

    server.shutdown();
}

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

    let resp = client()
        .get(format!(
            "{}/v1/{}/some-topic/fetch?offset=0&timestamp=1000",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "INVALID_REQUEST");

    server.shutdown();
}

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

    // Publish binary payload (not valid UTF-8).
    let binary_payload: Vec<u8> = (0..=255).collect();
    let resp = client()
        .post(format!(
            "{}/v1/{}/fetch-b64/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body(binary_payload.clone())
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Fetch and verify base64 roundtrip.
    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-b64/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 1);
    let payload_b64 = messages[0]["payload_base64"].as_str().unwrap();
    let decoded = base64::engine::general_purpose::STANDARD
        .decode(payload_b64)
        .unwrap();
    assert_eq!(decoded, binary_payload);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_message_too_large() {
    use fafb_q::types::{MessageEnvelope, MessageId};

    let server = start_test_http_server_all_perms().await;

    // Publish a message larger than the 4MB fetch response budget directly
    // through the engine (HTTP publish may reject bodies this large).
    let large_payload = vec![0xABu8; 4_200_000];
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("fetch-large".into());
    let envelope = MessageEnvelope::new(
        MessageId(1),
        ns.clone(),
        topic.clone(),
        large_payload,
        fafb_q::types::now_millis(),
    );
    server.engine.publish(&ns, &topic, envelope).await.unwrap();

    // Fetch should return 413 (MessageTooLarge).
    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-large/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 413);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "MESSAGE_TOO_LARGE");

    server.shutdown();
}

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

    // max_count > 10000 should return 400.
    let resp = client()
        .get(format!(
            "{}/v1/{}/some-topic/fetch?offset=0&max_count=10001",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["error"]["name"], "INVALID_REQUEST");

    server.shutdown();
}

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

    // Fetch from a topic that has never been published to — should return 200
    // with empty messages, not 404. Topics auto-create on publish, so
    // "not found" and "empty" are indistinguishable by design.
    let resp = client()
        .get(format!(
            "{}/v1/{}/nonexistent-topic/fetch?offset=0&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["topic"], "nonexistent-topic");
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 0);
    assert_eq!(body["next_offset"], body["head_offset"]);

    server.shutdown();
}

// -- Admin config introspection endpoint (Phase 17.8) --

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

#[tokio::test]
async fn admin_config_returns_runtime_config() {
    let server = common::start_test_http_admin_server(ADMIN_KEY).await;
    let resp = client()
        .get(format!("{}/admin/config", server.base_url))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    // Should contain runtime config fields
    assert!(
        body.get("max_depth").is_some(),
        "max_depth should be present"
    );
    assert!(body.get("global_max_memory_bytes").is_some());
    assert!(body.get("idle_timeout_secs").is_some());
    assert!(body.get("namespace_auth").is_some());
    server.shutdown();
}

#[tokio::test]
async fn admin_config_redacts_api_keys() {
    let server = common::start_test_http_admin_server(ADMIN_KEY).await;
    let resp = client()
        .get(format!("{}/admin/config", server.base_url))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let auth = body.get("namespace_auth").unwrap();
    // The test namespace's API key should be redacted
    let full_text = serde_json::to_string(&auth).unwrap();
    assert!(
        !full_text.contains(TEST_KEY),
        "raw API key should not appear in response"
    );
    assert!(full_text.contains("****"), "should contain redacted marker");
    server.shutdown();
}

#[tokio::test]
async fn admin_config_requires_auth() {
    let server = common::start_test_http_admin_server(ADMIN_KEY).await;
    // No auth header
    let resp = client()
        .get(format!("{}/admin/config", server.base_url))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);

    // Wrong key
    let resp = client()
        .get(format!("{}/admin/config", server.base_url))
        .header("Authorization", "Bearer wrong-key")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 403);
    server.shutdown();
}

// -- Reload-during-drain rejection --

// Config files need API keys >= 24 chars to pass validation.
const CONFIG_KEY: &str = "test-key-secret-at-least-24-chars-long";

#[tokio::test]
async fn admin_reload_rejected_during_drain() {
    use std::io::Write;

    let dir = tempfile::tempdir().unwrap();
    let config_file = dir.path().join("config.toml");
    {
        let mut f = std::fs::File::create(&config_file).unwrap();
        write!(
            f,
            "[namespaces.{TEST_NS}]\napi_key = \"{CONFIG_KEY}\"\npermissions = [\"Subscribe\", \"Publish\", \"Query\"]\n"
        )
        .unwrap();
    }

    let server = common::start_test_http_reload_server(ADMIN_KEY, &config_file).await;

    // Reload works before drain
    let resp = client()
        .post(format!("{}/admin/reload", server.base_url))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .send()
        .await
        .unwrap();
    let status = resp.status();
    let body_text = resp.text().await.unwrap();
    assert_eq!(
        status, 200,
        "reload before drain should succeed: {body_text}"
    );

    // Trigger drain
    server.lifecycle.begin_drain(30);

    // Reload rejected during drain
    let resp = client()
        .post(format!("{}/admin/reload", server.base_url))
        .header("Authorization", format!("Bearer {ADMIN_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 503);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["status"], "error");
    assert!(body["message"].as_str().unwrap().contains("draining"));

    server.shutdown();
}

// -- File watcher end-to-end tests --

#[tokio::test]
async fn file_watcher_detects_config_change() {
    use std::io::Write;

    let dir = tempfile::tempdir().unwrap();
    let config_file = dir.path().join("config.toml");
    {
        let mut f = std::fs::File::create(&config_file).unwrap();
        write!(
            f,
            "[queue]\nmax_depth = 50000\n\n[namespaces.{TEST_NS}]\napi_key = \"{CONFIG_KEY}\"\npermissions = [\"Subscribe\", \"Publish\", \"Query\"]\n"
        )
        .unwrap();
    }

    let server = common::start_test_http_reload_server(ADMIN_KEY, &config_file).await;

    // Verify initial max_depth
    let initial = server.shared_config.load();
    assert_eq!(initial.max_depth, 50000);

    // Spawn file watcher with 1s debounce
    let _watcher_handle = fafb_q::config::spawn_config_watcher(
        Arc::clone(&server.reload_state),
        1,
        Arc::clone(&server.lifecycle),
    );

    // Give the watcher time to register with the OS
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    // Modify the config file — change max_depth
    {
        let mut f = std::fs::File::create(&config_file).unwrap();
        write!(
            f,
            "[queue]\nmax_depth = 99999\n\n[namespaces.{TEST_NS}]\napi_key = \"{CONFIG_KEY}\"\npermissions = [\"Subscribe\", \"Publish\", \"Query\"]\n"
        )
        .unwrap();
    }

    // Wait for debounce (1s) + processing time
    tokio::time::sleep(std::time::Duration::from_secs(4)).await;

    // Verify config was updated
    let updated = server.shared_config.load();
    assert_eq!(
        updated.max_depth, 99999,
        "file watcher should have detected the change and reloaded"
    );

    server.shutdown();
}

#[tokio::test]
async fn file_watcher_survives_invalid_config() {
    use std::io::Write;

    let dir = tempfile::tempdir().unwrap();
    let config_file = dir.path().join("config.toml");
    {
        let mut f = std::fs::File::create(&config_file).unwrap();
        write!(
            f,
            "[queue]\nmax_depth = 50000\n\n[namespaces.{TEST_NS}]\napi_key = \"{CONFIG_KEY}\"\npermissions = [\"Subscribe\", \"Publish\", \"Query\"]\n"
        )
        .unwrap();
    }

    let server = common::start_test_http_reload_server(ADMIN_KEY, &config_file).await;

    let _watcher_handle = fafb_q::config::spawn_config_watcher(
        Arc::clone(&server.reload_state),
        1,
        Arc::clone(&server.lifecycle),
    );

    // Give the watcher time to register with the OS
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;

    // Write invalid TOML
    std::fs::write(&config_file, "this is not valid toml {{{{").unwrap();

    // Wait for debounce + processing
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;

    // Config should be unchanged (old value preserved)
    let current = server.shared_config.load();
    assert_eq!(
        current.max_depth, 50000,
        "invalid config should not overwrite previous config"
    );

    // Now write valid config again
    {
        let mut f = std::fs::File::create(&config_file).unwrap();
        write!(
            f,
            "[queue]\nmax_depth = 77777\n\n[namespaces.{TEST_NS}]\napi_key = \"{CONFIG_KEY}\"\npermissions = [\"Subscribe\", \"Publish\", \"Query\"]\n"
        )
        .unwrap();
    }

    // Wait for debounce + processing
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;

    // Config should now reflect the valid change
    let current = server.shared_config.load();
    assert_eq!(
        current.max_depth, 77777,
        "watcher should recover and apply valid config after invalid one"
    );

    server.shutdown();
}

// -- Message headers via HTTP --

#[tokio::test]
async fn http_publish_with_headers_lands_in_engine() {
    let server = start_test_http_server_all_perms().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/hdr-topic/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .header("X-Fafb-Content-Type", "application/json")
        .header("X-Fafb-Correlation-Id", "req-42")
        .body(r#"{"order":1}"#)
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Verify headers are stored in the engine.
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("hdr-topic".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, 0u64.into())
        .await
        .unwrap()
        .expect("message should exist");
    assert_eq!(msg.envelope.payload, br#"{"order":1}"#);
    assert_eq!(msg.envelope.headers.len(), 2);

    // HTTP header names are lowercased by hyper, so keys are lowercase after
    // stripping the x-fafb- prefix.
    let keys: Vec<&[u8]> = msg
        .envelope
        .headers
        .iter()
        .map(|h| h.key.as_slice())
        .collect();
    assert!(keys.contains(&b"content-type".as_slice()));
    assert!(keys.contains(&b"correlation-id".as_slice()));

    let ct = msg
        .envelope
        .headers
        .iter()
        .find(|h| h.key == b"content-type")
        .unwrap();
    assert_eq!(ct.value, b"application/json");

    server.shutdown();
}

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

    // Publish with headers.
    let resp = client()
        .post(format!(
            "{}/v1/{}/hdr-fetch/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .header("X-Fafb-Trace-Parent", "00-abc123")
        .header("X-Fafb-Priority", "high")
        .body("payload")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Fetch and verify headers in JSON response.
    let resp = client()
        .get(format!(
            "{}/v1/{}/hdr-fetch/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 1);

    let hdrs = messages[0]["headers"].as_array().unwrap();
    assert_eq!(hdrs.len(), 2);

    // Find headers by key (order may vary due to HeaderMap iteration).
    let find_hdr = |key: &str| -> &serde_json::Value {
        hdrs.iter()
            .find(|h| h["key"] == key)
            .unwrap_or_else(|| panic!("header '{key}' not found"))
    };
    assert_eq!(find_hdr("trace-parent")["value"], "00-abc123");
    assert_eq!(find_hdr("priority")["value"], "high");

    // Payload is still correct.
    let payload = base64::engine::general_purpose::STANDARD
        .decode(messages[0]["payload_base64"].as_str().unwrap())
        .unwrap();
    assert_eq!(payload, b"payload");

    server.shutdown();
}

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

    // Publish without any X-Fafb-* headers.
    let resp = client()
        .post(format!("{}/v1/{}/no-hdr/publish", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("plain")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);

    // Fetch — no headers field expected (empty headers omitted).
    let resp = client()
        .get(format!(
            "{}/v1/{}/no-hdr/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let msg = &body["messages"].as_array().unwrap()[0];
    assert!(
        msg.get("headers").is_none(),
        "empty headers should be omitted"
    );

    server.shutdown();
}

#[tokio::test]
async fn http_publish_reserved_prefix_rejected() {
    let server = start_test_http_server_all_perms().await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/reserved/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .header("X-Fafb-_fafb.dlq.reason", "hacked")
        .body("bad")
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    let msg = body["error"]["message"].as_str().unwrap();
    assert!(
        msg.contains("reserved prefix"),
        "error should mention reserved prefix, got: {msg}"
    );

    server.shutdown();
}

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

    // Default limit is 64 headers. Send 65.
    let mut req = client()
        .post(format!(
            "{}/v1/{}/too-many/publish",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"));

    for i in 0..65 {
        req = req.header(format!("X-Fafb-Key-{i}"), format!("val-{i}"));
    }
    let resp = req.body("data").send().await.unwrap();
    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    let msg = body["error"]["message"].as_str().unwrap();
    assert!(
        msg.contains("too many headers"),
        "error should mention too many headers, got: {msg}"
    );

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_binary_header_values_base64_encoded() {
    let server = start_test_http_server_all_perms().await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("bin-hdr".into());

    // Publish directly via engine with a binary (non-UTF-8) header value.
    let envelope = MessageEnvelope::new(
        MessageId::new(1),
        ns.clone(),
        topic.clone(),
        b"binary-test".to_vec(),
        fafb_q::types::now_millis(),
    )
    .with_headers(vec![
        Header {
            key: b"text-key".to_vec(),
            value: b"text-value".to_vec(),
        },
        Header {
            key: b"binary-key".to_vec(),
            value: vec![0xFF, 0xFE, 0x00, 0x01], // not valid UTF-8
        },
    ]);
    server.engine.publish(&ns, &topic, envelope).await.unwrap();

    // Fetch via HTTP.
    let resp = client()
        .get(format!(
            "{}/v1/{}/bin-hdr/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let msg = &body["messages"].as_array().unwrap()[0];
    let hdrs = msg["headers"].as_array().unwrap();
    assert_eq!(hdrs.len(), 2);

    // Text value should have "value" field.
    let text_hdr = hdrs.iter().find(|h| h["key"] == "text-key").unwrap();
    assert_eq!(text_hdr["value"], "text-value");
    assert!(text_hdr.get("value_base64").is_none());

    // Binary value should have "value_base64" field, no "value".
    let bin_hdr = hdrs.iter().find(|h| h["key"] == "binary-key").unwrap();
    assert!(
        bin_hdr.get("value").is_none(),
        "binary header should not have 'value' field"
    );
    let b64_val = bin_hdr["value_base64"].as_str().unwrap();
    let decoded = base64::engine::general_purpose::STANDARD
        .decode(b64_val)
        .unwrap();
    assert_eq!(decoded, vec![0xFF, 0xFE, 0x00, 0x01]);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_duplicate_header_keys_preserved() {
    let server = start_test_http_server_all_perms().await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("dup-hdr".into());

    // Publish directly via engine with duplicate header keys.
    let envelope = MessageEnvelope::new(
        MessageId::new(1),
        ns.clone(),
        topic.clone(),
        b"dup-test".to_vec(),
        fafb_q::types::now_millis(),
    )
    .with_headers(vec![
        Header {
            key: b"Tag".to_vec(),
            value: b"env:prod".to_vec(),
        },
        Header {
            key: b"Tag".to_vec(),
            value: b"region:us-east".to_vec(),
        },
        Header {
            key: b"Tag".to_vec(),
            value: b"team:payments".to_vec(),
        },
    ]);
    server.engine.publish(&ns, &topic, envelope).await.unwrap();

    // Fetch via HTTP — all three duplicate keys should be present.
    let resp = client()
        .get(format!(
            "{}/v1/{}/dup-hdr/fetch?offset=0&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let hdrs = body["messages"][0]["headers"].as_array().unwrap();
    assert_eq!(
        hdrs.len(),
        3,
        "all three duplicate-key headers should be present"
    );

    let tag_values: Vec<&str> = hdrs
        .iter()
        .filter(|h| h["key"] == "Tag")
        .map(|h| h["value"].as_str().unwrap())
        .collect();
    assert_eq!(
        tag_values,
        vec!["env:prod", "region:us-east", "team:payments"]
    );

    server.shutdown();
}

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

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

#[tokio::test]
async fn http_publish_with_ttl_query_param() {
    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/events/publish?ttl=30",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("hello with ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    assert_eq!(body["offset"], 0);
    assert!(body["message_id"].as_u64().is_some());

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

    server.shutdown();
}

#[tokio::test]
async fn http_publish_without_ttl_gets_zero() {
    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let resp = client()
        .post(format!("{}/v1/{}/events/publish", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("no ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);

    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, fafb_q::types::TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 0);

    server.shutdown();
}

#[tokio::test]
async fn http_publish_namespace_default_ttl_applied() {
    // Namespace default TTL = 120 seconds.
    let server = start_test_http_server_with_shared_config(ttl_shared_config(120, 0, 0)).await;
    let resp = client()
        .post(format!("{}/v1/{}/events/publish", server.base_url, TEST_NS))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("default ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);

    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, fafb_q::types::TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 120_000);

    server.shutdown();
}

#[tokio::test]
async fn http_publish_explicit_ttl_overrides_default() {
    // Namespace default TTL = 120s, but client sends ?ttl=30.
    let server = start_test_http_server_with_shared_config(ttl_shared_config(120, 0, 0)).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/events/publish?ttl=30",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("override ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);

    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, fafb_q::types::TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 30_000);

    server.shutdown();
}

#[tokio::test]
async fn http_publish_ttl_clamped_to_max() {
    // max_ttl = 10s. Client sends ?ttl=60.
    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 10, 0)).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/events/publish?ttl=60",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("clamped ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 200);

    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("events".into());
    let msg = server
        .engine
        .consume_from(&ns, &topic, fafb_q::types::TopicOffset(0))
        .await
        .unwrap()
        .unwrap();
    assert_eq!(msg.envelope.ttl_ms, 10_000); // clamped to max 10s

    server.shutdown();
}

#[tokio::test]
async fn http_publish_ttl_below_min_rejected() {
    // min_ttl = 10s. Client sends ?ttl=5.
    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 10)).await;
    let resp = client()
        .post(format!(
            "{}/v1/{}/events/publish?ttl=5",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .body("rejected ttl")
        .send()
        .await
        .unwrap();

    assert_eq!(resp.status(), 400);
    let body: serde_json::Value = resp.json().await.unwrap();
    let msg = body["error"]["message"].as_str().unwrap_or("");
    assert!(msg.contains("below minimum"), "error body: {body}");

    server.shutdown();
}

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

/// HTTP fetch filters expired messages — only live messages returned.
#[tokio::test]
async fn http_fetch_filters_expired_messages() {
    use fafb_q::types::{MessageId, TopicOffset, now_millis};

    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("fetch-ttl-filter".into());

    // Publish 3 messages directly via engine: 2 with short TTL, 1 without.
    let now = now_millis();
    for i in 0..3u64 {
        let mut envelope = MessageEnvelope::new(
            MessageId(i + 1),
            ns.clone(),
            topic.clone(),
            format!("msg-{i}").into_bytes(),
            now,
        );
        if i < 2 {
            envelope = envelope.with_ttl(100); // 100ms TTL — will expire
        }
        server.engine.publish(&ns, &topic, envelope).await.unwrap();
    }

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

    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-ttl-filter/fetch?offset=0&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();

    let messages = body["messages"].as_array().unwrap();
    // Only 1 message should be returned (offset 2, the one without TTL).
    assert_eq!(
        messages.len(),
        1,
        "expected 1 live message, got: {messages:?}"
    );
    assert_eq!(messages[0]["offset"], 2);
    // next_offset should be past all messages.
    assert_eq!(body["next_offset"], 3);
    assert_eq!(body["head_offset"], 3);

    server.shutdown();
}

/// HTTP fetch returns ttl_ms=0 messages regardless of age.
#[tokio::test]
async fn http_fetch_no_ttl_always_returned() {
    use fafb_q::types::{MessageId, TopicOffset, now_millis};

    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("fetch-no-ttl".into());

    // Publish messages without TTL (ttl_ms = 0).
    let now = now_millis();
    for i in 0..3u64 {
        let envelope = MessageEnvelope::new(
            MessageId(i + 1),
            ns.clone(),
            topic.clone(),
            format!("msg-{i}").into_bytes(),
            now,
        );
        server.engine.publish(&ns, &topic, envelope).await.unwrap();
    }

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

    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-no-ttl/fetch?offset=0&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();

    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 3, "all messages should be returned");

    server.shutdown();
}

/// Verify HTTP fetch JSON response includes ttl_ms field.
#[tokio::test]
async fn http_fetch_includes_ttl_ms_in_response() {
    use fafb_q::types::{MessageId, TopicOffset, now_millis};

    let server = start_test_http_server_with_shared_config(ttl_shared_config(0, 0, 0)).await;
    let ns = NamespaceId(TEST_NS.into());
    let topic = TopicName("fetch-ttl-json".into());

    let now = now_millis();
    // Publish message with TTL.
    let envelope = MessageEnvelope::new(
        MessageId(1),
        ns.clone(),
        topic.clone(),
        b"with-ttl".to_vec(),
        now,
    )
    .with_ttl(30_000);
    server.engine.publish(&ns, &topic, envelope).await.unwrap();

    // Publish message without TTL.
    let envelope = MessageEnvelope::new(
        MessageId(2),
        ns.clone(),
        topic.clone(),
        b"no-ttl".to_vec(),
        now,
    );
    server.engine.publish(&ns, &topic, envelope).await.unwrap();

    let resp = client()
        .get(format!(
            "{}/v1/{}/fetch-ttl-json/fetch?offset=0&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();

    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 2);
    assert_eq!(messages[0]["ttl_ms"], 30_000);
    assert_eq!(messages[1]["ttl_ms"], 0);

    server.shutdown();
}

// -- Negative offset (relative from head) tests --

#[tokio::test]
async fn http_fetch_negative_offset_last() {
    let server = start_test_http_server_all_perms().await;
    publish_messages(&server.base_url, "neg-last", 5).await;

    // offset=-1 should return the last message (offset 4).
    let resp = client()
        .get(format!(
            "{}/v1/{}/neg-last/fetch?offset=-1&max_count=1",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 1);
    assert_eq!(messages[0]["offset"], 4);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_negative_offset_last_n() {
    let server = start_test_http_server_all_perms().await;
    publish_messages(&server.base_url, "neg-last-n", 20).await;

    // offset=-5 with max_count=10 should start at offset 15.
    let resp = client()
        .get(format!(
            "{}/v1/{}/neg-last-n/fetch?offset=-5&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 5);
    assert_eq!(messages[0]["offset"], 15);
    assert_eq!(messages[4]["offset"], 19);

    server.shutdown();
}

#[tokio::test]
async fn http_fetch_negative_offset_exceeds_depth() {
    let server = start_test_http_server_all_perms().await;
    publish_messages(&server.base_url, "neg-exceeds", 3).await;

    // offset=-100 on a 3-message topic should clamp to offset 0.
    let resp = client()
        .get(format!(
            "{}/v1/{}/neg-exceeds/fetch?offset=-100&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 3);
    assert_eq!(messages[0]["offset"], 0);

    server.shutdown();
}

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

    // offset=-1 on a nonexistent (empty) topic should return empty 200.
    let resp = client()
        .get(format!(
            "{}/v1/{}/neg-empty/fetch?offset=-1&max_count=10",
            server.base_url, TEST_NS
        ))
        .header("Authorization", format!("Bearer {TEST_KEY}"))
        .send()
        .await
        .unwrap();
    assert_eq!(resp.status(), 200);
    let body: serde_json::Value = resp.json().await.unwrap();
    let messages = body["messages"].as_array().unwrap();
    assert_eq!(messages.len(), 0);

    server.shutdown();
}
