qrusty 0.20.9

// tests/api_tests.rs

//! Integration tests for the API layer
//!
//! These tests verify the HTTP API endpoints work correctly
//! with real storage backend and proper request/response handling.
//!
//! All tests in this file implicitly verify SYS-0016 because the
//! `StorageApi` async trait wraps every RocksDB call in
//! `spawn_blocking`, and the tests exercise that async wrapper end-
//! to-end through the Axum HTTP handler layer.
// Verifies: SYS-0016

use axum::{
    body::Body,
    http::{Request, StatusCode},
};
use qrusty::api::ApiServer;
use qrusty::storage::Storage;
use serde_json::{json, Value};
use std::sync::Arc;
use tempfile::TempDir;
use tower::ServiceExt; // for `oneshot` and `ready`

/// Helper function to create a test API server with storage
async fn create_test_api() -> (ApiServer, TempDir) {
    let temp_dir = TempDir::new().expect("Failed to create temp directory");
    let storage =
        Storage::new(temp_dir.path().to_str().unwrap()).expect("Failed to create test storage");
    let api = ApiServer::new(Arc::new(storage));
    (api, temp_dir)
}

/// Helper function to make HTTP requests to the API
async fn make_request(
    api: ApiServer,
    method: &str,
    path: &str,
    body: Option<Value>,
) -> (StatusCode, Value) {
    let app = api.router();

    let request_builder = Request::builder()
        .method(method)
        .uri(path)
        .header("content-type", "application/json");

    let request = if let Some(body_json) = body {
        request_builder
            .body(Body::from(body_json.to_string()))
            .unwrap()
    } else {
        request_builder.body(Body::empty()).unwrap()
    };

    let response = app.oneshot(request).await.unwrap();
    let status = response.status();

    let body = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    let body_str = String::from_utf8(body.to_vec()).unwrap();

    let json_body: Value = if body_str.is_empty() {
        Value::Null
    } else {
        serde_json::from_str(&body_str).unwrap_or(Value::String(body_str))
    };

    (status, json_body)
}

#[tokio::test]
async fn test_health_endpoint() {
    // Verifies: SYS-0002
    let (api, _temp_dir) = create_test_api().await;
    let (status, body) = make_request(api, "GET", "/health", None).await;

    assert_eq!(status, StatusCode::OK);
    // Health returns JSON with a status field (SYS-0002, SYS-0019).
    // In tests the server is created fully initialised, so the status is "ok".
    assert_eq!(body["status"], Value::String("ok".to_string()));
}

#[tokio::test]
async fn test_stats_endpoint() {
    // Verifies: API-0006
    let (api, _temp_dir) = create_test_api().await;

    // Initially, stats should show no queues
    let (status, body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());
    assert_eq!(body["queues"].as_array().unwrap().len(), 0);
    assert_eq!(body["summary"]["total_queues"], 0);
    assert_eq!(body["summary"]["total_messages"], 0);

    // Publish some messages to different queues
    let publish_request1 = json!({
        "queue": "orders",
        "priority": 100,
        "payload": "{\"order_id\": 123}"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request1)).await;

    let publish_request2 = json!({
        "queue": "orders",
        "priority": 50,
        "payload": "{\"order_id\": 124}"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request2)).await;

    let publish_request3 = json!({
        "queue": "notifications",
        "priority": 200,
        "payload": "{\"user_id\": 456}"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request3)).await;

    // Check stats after publishing
    let (status, body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());

    let queues = body["queues"].as_array().unwrap();
    assert_eq!(queues.len(), 2); // orders and notifications

    // Check summary stats
    assert_eq!(body["summary"]["total_queues"], 2);
    assert_eq!(body["summary"]["total_messages"], 3);
    assert_eq!(body["summary"]["total_available"], 3);
    assert_eq!(body["summary"]["total_locked"], 0);

    // Consume a message and check stats again
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });
    make_request(
        api.clone(),
        "POST",
        "/consume/orders",
        Some(consume_request),
    )
    .await;

    let (status, body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(status, StatusCode::OK);

    // Now should have 1 locked message
    assert_eq!(body["summary"]["total_messages"], 3);
    assert_eq!(body["summary"]["total_available"], 2);
    assert_eq!(body["summary"]["total_locked"], 1);
}

#[tokio::test]
async fn test_publish_message() {
    // Verifies: API-0002
    let (api, _temp_dir) = create_test_api().await;

    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": \"data\"}",
        "max_retries": 3
    });

    let (status, body) = make_request(api, "POST", "/publish", Some(publish_request)).await;

    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());
    assert!(body["id"].is_string());
    assert!(!body["id"].as_str().unwrap().is_empty());
}

#[tokio::test]
async fn test_publish_message_minimal() {
    // Verifies: API-0002
    let (api, _temp_dir) = create_test_api().await;

    // Test with minimal required fields (max_retries is optional)
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 50,
        "payload": "{\"minimal\": true}"
    });

    let (status, body) = make_request(api, "POST", "/publish", Some(publish_request)).await;

    assert_eq!(status, StatusCode::OK);
    assert!(body["id"].is_string());
}

#[tokio::test]
async fn test_webui_served_from_configurable_directory() {
    // Verifies: SYS-0003
    let (api, _temp_dir) = create_test_api().await;

    // Create a temporary web UI directory structure
    let webui_dir = TempDir::new().expect("Failed to create WEBUI_DIR temp dir");
    let assets_dir = webui_dir.path().join("assets");
    std::fs::create_dir_all(&assets_dir).expect("Failed to create assets dir");

    std::fs::write(webui_dir.path().join("index.html"), "<html>ok</html>")
        .expect("Failed to write index.html");
    std::fs::write(assets_dir.join("test.txt"), "asset-ok").expect("Failed to write test asset");

    std::env::set_var("WEBUI_DIR", webui_dir.path());

    let (status, body) = make_request(api.clone(), "GET", "/ui", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.as_str().unwrap_or("").contains("<html>"));

    let (status, body) = make_request(api, "GET", "/assets/test.txt", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body, Value::String("asset-ok".to_string()));
}

#[tokio::test]
async fn test_publish_and_consume() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish a message
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"order_id\": 12345}",
        "max_retries": 3
    });

    let (status, publish_body) =
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    assert_eq!(status, StatusCode::OK);
    let message_id = publish_body["id"].as_str().unwrap();

    // Consume the message
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    let (status, consume_body) =
        make_request(api, "POST", "/consume/test_queue", Some(consume_request)).await;

    assert_eq!(status, StatusCode::OK);
    assert!(consume_body.is_object());
    assert_eq!(consume_body["id"].as_str().unwrap(), message_id);
    assert_eq!(
        consume_body["payload"].as_str().unwrap(),
        "{\"order_id\": 12345}"
    );
    assert_eq!(consume_body["retry_count"].as_u64().unwrap(), 1);
}

#[tokio::test]
async fn test_consume_empty_queue() {
    let (api, _temp_dir) = create_test_api().await;

    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    let (status, body) =
        make_request(api, "POST", "/consume/empty_queue", Some(consume_request)).await;

    assert_eq!(status, StatusCode::OK);
    assert_eq!(body, Value::Null);
}

#[tokio::test]
async fn test_consume_with_default_timeout() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish a message first
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": true}"
    });

    make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;

    // Consume without specifying timeout (should use default 30 seconds)
    let consume_request = json!({
        "consumer_id": "worker-1"
    });

    let (status, body) =
        make_request(api, "POST", "/consume/test_queue", Some(consume_request)).await;

    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());
    assert!(!body["id"].as_str().unwrap().is_empty());
}

#[tokio::test]
async fn test_ack_message() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish and consume a message
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": true}"
    });

    let (_, publish_body) =
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    let message_id = publish_body["id"].as_str().unwrap();

    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    make_request(
        api.clone(),
        "POST",
        "/consume/test_queue",
        Some(consume_request),
    )
    .await;

    // Ack the message
    let ack_request = json!({
        "consumer_id": "worker-1"
    });

    let ack_path = format!("/ack/test_queue/{}", message_id);
    let (status, _) = make_request(api.clone(), "POST", &ack_path, Some(ack_request)).await;

    assert_eq!(status, StatusCode::OK);

    // Try to consume again - should be empty
    let consume_request2 = json!({
        "consumer_id": "worker-2",
        "timeout_seconds": 30
    });

    let (status, body) =
        make_request(api, "POST", "/consume/test_queue", Some(consume_request2)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body, Value::Null);
}

#[tokio::test]
async fn test_ack_wrong_consumer() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish and consume a message
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": true}"
    });

    let (_, publish_body) =
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    let message_id = publish_body["id"].as_str().unwrap();

    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    make_request(
        api.clone(),
        "POST",
        "/consume/test_queue",
        Some(consume_request),
    )
    .await;

    // Try to ack with wrong consumer
    let ack_request = json!({
        "consumer_id": "worker-2"
    });

    let ack_path = format!("/ack/test_queue/{}", message_id);
    let (status, _) = make_request(api, "POST", &ack_path, Some(ack_request)).await;

    assert_eq!(status, StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn test_nack_message() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish and consume a message
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": true}"
    });

    let (_, publish_body) =
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    let message_id = publish_body["id"].as_str().unwrap();

    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    make_request(
        api.clone(),
        "POST",
        "/consume/test_queue",
        Some(consume_request),
    )
    .await;

    // Nack the message
    let nack_request = json!({
        "consumer_id": "worker-1"
    });

    let nack_path = format!("/nack/test_queue/{}", message_id);
    let (status, _) = make_request(api.clone(), "POST", &nack_path, Some(nack_request)).await;

    assert_eq!(status, StatusCode::OK);

    // Message should be available for consumption again
    let consume_request2 = json!({
        "consumer_id": "worker-2",
        "timeout_seconds": 30
    });

    let (status, body) =
        make_request(api, "POST", "/consume/test_queue", Some(consume_request2)).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());
    assert_eq!(body["id"].as_str().unwrap(), message_id);
    assert_eq!(body["retry_count"].as_u64().unwrap(), 2); // Should be incremented
}

#[tokio::test]
async fn test_nack_wrong_consumer() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish and consume a message
    let publish_request = json!({
        "queue": "test_queue",
        "priority": 100,
        "payload": "{\"test\": true}"
    });

    let (_, publish_body) =
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    let message_id = publish_body["id"].as_str().unwrap();

    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    make_request(
        api.clone(),
        "POST",
        "/consume/test_queue",
        Some(consume_request),
    )
    .await;

    // Try to nack with wrong consumer
    let nack_request = json!({
        "consumer_id": "worker-2"
    });

    let nack_path = format!("/nack/test_queue/{}", message_id);
    let (status, _) = make_request(api, "POST", &nack_path, Some(nack_request)).await;

    assert_eq!(status, StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn test_priority_ordering_through_api() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish messages with different priorities
    let messages = vec![
        (10, "low priority"),
        (100, "high priority"),
        (50, "medium priority"),
    ];

    for (priority, payload) in &messages {
        let publish_request = json!({
            "queue": "priority_test",
            "priority": priority,
            "payload": format!("{{\"message\": \"{}\"}}", payload)
        });

        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    }

    // Consume messages and verify they come out in priority order
    let expected_order = vec!["high priority", "medium priority", "low priority"];

    for expected_message in expected_order {
        let consume_request = json!({
            "consumer_id": "worker-1",
            "timeout_seconds": 30
        });

        let (status, body) = make_request(
            api.clone(),
            "POST",
            "/consume/priority_test",
            Some(consume_request),
        )
        .await;
        assert_eq!(status, StatusCode::OK);

        let payload = body["payload"].as_str().unwrap();
        assert!(payload.contains(expected_message));
    }
}

#[tokio::test]
async fn test_queue_isolation_through_api() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish to different queues
    let queue1_request = json!({
        "queue": "queue1",
        "priority": 100,
        "payload": "{\"queue\": \"queue1\"}"
    });

    let queue2_request = json!({
        "queue": "queue2",
        "priority": 100,
        "payload": "{\"queue\": \"queue2\"}"
    });

    make_request(api.clone(), "POST", "/publish", Some(queue1_request)).await;
    make_request(api.clone(), "POST", "/publish", Some(queue2_request)).await;

    // Consume from queue1
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/queue1",
        Some(consume_request.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert!(body["payload"].as_str().unwrap().contains("queue1"));

    // Consume from queue2
    let consume_request2 = json!({
        "consumer_id": "worker-2",
        "timeout_seconds": 30
    });

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/queue2",
        Some(consume_request2.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert!(body["payload"].as_str().unwrap().contains("queue2"));

    // Both queues should now be empty
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/queue1",
        Some(consume_request),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body, Value::Null);

    let (status, body) = make_request(api, "POST", "/consume/queue2", Some(consume_request2)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body, Value::Null);
}

#[tokio::test]
async fn test_invalid_json_handling() {
    let (_api, _temp_dir) = create_test_api().await;

    // This test would require more sophisticated error handling in the API
    // For now, we just verify that invalid requests fail gracefully
    // The actual implementation might need improvements for better error handling
}

#[tokio::test]
async fn test_concurrent_api_operations() {
    let (api, _temp_dir) = create_test_api().await;
    let api = Arc::new(api);

    // Publish multiple messages concurrently
    let mut publish_handles = vec![];

    for i in 0..10 {
        let api_clone = api.clone();
        let handle = tokio::spawn(async move {
            let publish_request = json!({
                "queue": "concurrent_test",
                "priority": i * 10,
                "payload": format!("{{\"message_number\": {}}}", i)
            });

            make_request(
                (*api_clone).clone(),
                "POST",
                "/publish",
                Some(publish_request),
            )
            .await
        });
        publish_handles.push(handle);
    }

    // Wait for all publishes to complete
    for handle in publish_handles {
        let (status, _) = handle.await.unwrap();
        assert_eq!(status, StatusCode::OK);
    }

    // Consume all messages concurrently
    let mut consume_handles = vec![];

    for i in 0..10 {
        let api_clone = api.clone();
        let handle = tokio::spawn(async move {
            let consume_request = json!({
                "consumer_id": format!("worker-{}", i),
                "timeout_seconds": 30
            });

            make_request(
                (*api_clone).clone(),
                "POST",
                "/consume/concurrent_test",
                Some(consume_request),
            )
            .await
        });
        consume_handles.push(handle);
    }

    // Collect results
    let mut consumed_messages = vec![];
    for handle in consume_handles {
        let (status, body) = handle.await.unwrap();
        if status == StatusCode::OK && body != Value::Null {
            consumed_messages.push(body);
        }
    }

    // Should have consumed all 10 messages
    assert_eq!(consumed_messages.len(), 10);
}

#[tokio::test]
async fn test_create_queue_max_first() {
    // Verifies: API-0001
    let (api, _temp_dir) = create_test_api().await;

    // Create a max-first queue
    let create_request = json!({
        "name": "urgent_queue",
        "config": {
            "ordering": "MaxFirst"
        }
    });

    let (status, _body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK);

    // Test that the queue behaves with max-first ordering
    // Publish messages with different priorities
    let publish_low = json!({
        "queue": "urgent_queue",
        "priority": 10,
        "payload": "low priority"
    });
    let publish_high = json!({
        "queue": "urgent_queue",
        "priority": 100,
        "payload": "high priority"
    });

    make_request(api.clone(), "POST", "/publish", Some(publish_low)).await;
    make_request(api.clone(), "POST", "/publish", Some(publish_high)).await;

    // Consume should return high priority first
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 30
    });

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/urgent_queue",
        Some(consume_request.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "high priority");

    let (status, body) =
        make_request(api, "POST", "/consume/urgent_queue", Some(consume_request)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "low priority");
}

#[tokio::test]
async fn test_create_queue_min_first() {
    // Verifies: API-0001
    let (api, _temp_dir) = create_test_api().await;

    // Create a min-first queue
    let create_request = json!({
        "name": "batch_queue",
        "config": {
            "ordering": "MinFirst"
        }
    });

    let (status, _body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK);

    // Test that the queue behaves with min-first ordering
    // Publish messages with different priorities
    let publish_low = json!({
        "queue": "batch_queue",
        "priority": 10,
        "payload": "low priority"
    });
    let publish_high = json!({
        "queue": "batch_queue",
        "priority": 100,
        "payload": "high priority"
    });

    make_request(api.clone(), "POST", "/publish", Some(publish_low)).await;
    make_request(api.clone(), "POST", "/publish", Some(publish_high)).await;

    // Consume should return low priority first
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 30
    });

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/batch_queue",
        Some(consume_request.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "low priority");

    let (status, body) =
        make_request(api, "POST", "/consume/batch_queue", Some(consume_request)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "high priority");
}

#[tokio::test]
async fn test_create_queue_conflict() {
    // Verifies: API-0001 (creating an already-existing queue returns 409)
    let (api, _temp_dir) = create_test_api().await;

    // Create a queue
    let create_request = json!({
        "name": "duplicate_test",
        "config": {
            "ordering": "MaxFirst"
        }
    });

    let (status, _body) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(create_request.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Try to create the same queue again - should get 409 Conflict
    let (status, _body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::CONFLICT);

    // Verify we can still update via update-queue
    let update_request = json!({
        "name": "duplicate_test",
        "config": {
            "allow_duplicates": false
        }
    });

    let (status, _body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::OK);

    // Verify the update worked
    let (status, stats) = make_request(api, "GET", "/queue-stats/duplicate_test", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(stats["config"]["allow_duplicates"], false);
}

#[tokio::test]
async fn test_queue_stats_endpoint() {
    // Verifies: API-0007
    let (api, _temp_dir) = create_test_api().await;

    // Publish two messages
    let publish_request1 = json!({
        "queue": "stats_queue",
        "priority": 100,
        "payload": "a"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request1)).await;

    let publish_request2 = json!({
        "queue": "stats_queue",
        "priority": 50,
        "payload": "b"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request2)).await;

    // Lock one message
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });
    make_request(
        api.clone(),
        "POST",
        "/consume/stats_queue",
        Some(consume_request),
    )
    .await;

    let (status, body) = make_request(api, "GET", "/queue-stats/stats_queue", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object());
    assert_eq!(body["available"], 1);
    assert_eq!(body["locked"], 1);
    assert_eq!(body["total"], 2);
}

#[tokio::test]
async fn test_queue_stats_includes_live_rate_fields() {
    // Verifies: API-0007 (rate fields for live UI graphs)
    let (api, _temp_dir) = create_test_api().await;

    // Establish a baseline snapshot (rates may be 0 on first call).
    let (status, body) = make_request(api.clone(), "GET", "/queue-stats/rate_queue", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.get("publish_rate_per_sec").is_some());
    assert!(body.get("ack_rate_per_sec").is_some());

    // Drive publish activity, then sample again.
    let publish_request = json!({
        "queue": "rate_queue",
        "priority": 1,
        "payload": "x"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    let mut publish_rate = 0.0;
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
    while std::time::Instant::now() < deadline {
        let (_status, body) =
            make_request(api.clone(), "GET", "/queue-stats/rate_queue", None).await;
        publish_rate = body["publish_rate_per_sec"].as_f64().unwrap_or(0.0);
        if publish_rate > 0.0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }
    assert!(
        publish_rate > 0.0,
        "expected publish_rate_per_sec > 0 after activity, got {publish_rate}"
    );

    // Drive ack activity, then sample again.
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });
    let (_status, consumed) = make_request(
        api.clone(),
        "POST",
        "/consume/rate_queue",
        Some(consume_request),
    )
    .await;
    let msg_id = consumed["id"]
        .as_str()
        .expect("expected consumed message id");

    let ack_request = json!({
        "consumer_id": "worker-1"
    });
    make_request(
        api.clone(),
        "POST",
        &format!("/ack/rate_queue/{msg_id}"),
        Some(ack_request),
    )
    .await;
    let mut ack_rate = 0.0;
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(2);
    while std::time::Instant::now() < deadline {
        let (_status, body) =
            make_request(api.clone(), "GET", "/queue-stats/rate_queue", None).await;
        ack_rate = body["ack_rate_per_sec"].as_f64().unwrap_or(0.0);
        if ack_rate > 0.0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    assert!(
        ack_rate > 0.0,
        "expected ack_rate_per_sec > 0 after activity, got {ack_rate}"
    );
}

#[tokio::test]
async fn test_queue_metrics_endpoint_returns_time_series() {
    // Sanity check for API-0011 (verification is marked as analysis in Doorstop,
    // but this test prevents accidental regressions in the endpoint shape).
    let (api, _temp_dir) = create_test_api().await;

    // Create activity so the queue exists and has non-zero totals.
    let publish_request = json!({
        "queue": "metrics_queue",
        "priority": 1,
        "payload": "hello"
    });
    let (status, body) = make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    assert_eq!(status, StatusCode::OK, "publish failed: {body}");

    // Lock the message so we can observe both total + locked counts.
    let consume_request = json!({
        "consumer_id": "worker-1",
        "timeout_seconds": 30
    });
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/metrics_queue",
        Some(consume_request),
    )
    .await;
    assert_eq!(status, StatusCode::OK, "consume failed: {body}");

    // Wait for the 1Hz sampler to run after the queue has activity.
    let mut body = Value::Null;
    let deadline = std::time::Instant::now() + std::time::Duration::from_secs(3);
    while std::time::Instant::now() < deadline {
        let (status, next_body) =
            make_request(api.clone(), "GET", "/queues/metrics_queue/metrics", None).await;
        assert_eq!(status, StatusCode::OK);

        let points_len = next_body["points"]
            .as_array()
            .expect("expected points array")
            .len();
        body = next_body;
        if points_len > 0 {
            break;
        }
        tokio::time::sleep(std::time::Duration::from_millis(100)).await;
    }

    assert_eq!(body["queue"], "metrics_queue");
    assert_eq!(body["resolution_seconds"], 1);
    assert_eq!(body["window_seconds"], 10);

    let points = body["points"].as_array().expect("expected points array");
    assert!(!points.is_empty(), "expected at least one sampled point");

    let last = points.last().expect("non-empty points");
    assert!(last["t_ms"].as_i64().is_some());
    let total = last["total"].as_u64().unwrap_or(0);
    let locked = last["locked"].as_u64().unwrap_or(0);
    assert!(total >= 1);
    assert!(locked <= total);
    assert!(last["publish_rate_per_sec"].as_f64().is_some());
    assert!(last["ack_rate_per_sec"].as_f64().is_some());
    assert!(last["nack_rate_per_sec"].as_f64().is_some());
}

#[tokio::test]
async fn test_list_queues_endpoint() {
    // Verifies: API-0008
    let (api, _temp_dir) = create_test_api().await;

    // Create activity in multiple queues
    let publish_orders = json!({
        "queue": "orders",
        "priority": 1,
        "payload": "x"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_orders)).await;

    let publish_notifications = json!({
        "queue": "notifications",
        "priority": 1,
        "payload": "y"
    });
    make_request(api.clone(), "POST", "/publish", Some(publish_notifications)).await;

    let (status, body) = make_request(api, "GET", "/queues", None).await;
    assert_eq!(status, StatusCode::OK);

    let queues = body.as_array().expect("Expected JSON array");
    let queue_names: Vec<&str> = queues.iter().filter_map(|v| v.as_str()).collect();

    assert!(queue_names.contains(&"orders"));
    assert!(queue_names.contains(&"notifications"));
}

#[tokio::test]
async fn test_default_queue_behavior() {
    let (api, _temp_dir) = create_test_api().await;

    // Don't create queue explicitly - should use default max-first behavior
    let publish_low = json!({
        "queue": "default_queue",
        "priority": 10,
        "payload": "low priority"
    });
    let publish_high = json!({
        "queue": "default_queue",
        "priority": 100,
        "payload": "high priority"
    });

    make_request(api.clone(), "POST", "/publish", Some(publish_low)).await;
    make_request(api.clone(), "POST", "/publish", Some(publish_high)).await;

    // Should default to max-first behavior (high priority first)
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 30
    });

    let (status, body) =
        make_request(api, "POST", "/consume/default_queue", Some(consume_request)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "high priority");
}

#[tokio::test]
async fn test_multiple_queue_configurations() {
    let (api, _temp_dir) = create_test_api().await;

    // Create queues with different configurations
    let create_urgent = json!({
        "name": "urgent",
        "config": {"ordering": "MaxFirst"}
    });
    let create_batch = json!({
        "name": "batch",
        "config": {"ordering": "MinFirst"}
    });

    make_request(api.clone(), "POST", "/create-queue", Some(create_urgent)).await;
    make_request(api.clone(), "POST", "/create-queue", Some(create_batch)).await;

    // Publish same priorities to both queues
    let urgent_msg = json!({
        "queue": "urgent",
        "priority": 50,
        "payload": "urgent task"
    });
    let batch_msg = json!({
        "queue": "batch",
        "priority": 50,
        "payload": "batch task"
    });

    make_request(api.clone(), "POST", "/publish", Some(urgent_msg)).await;
    make_request(api.clone(), "POST", "/publish", Some(batch_msg)).await;

    // Both should return their messages since they're the only ones
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 30
    });

    let (status, urgent_body) = make_request(
        api.clone(),
        "POST",
        "/consume/urgent",
        Some(consume_request.clone()),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(urgent_body["payload"], "urgent task");

    let (status, batch_body) =
        make_request(api, "POST", "/consume/batch", Some(consume_request)).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(batch_body["payload"], "batch task");
}

#[tokio::test]
async fn test_priority_ordering_comprehensive() {
    let (api, _temp_dir) = create_test_api().await;

    // Create min-first queue
    let create_request = json!({
        "name": "priority_test",
        "config": {"ordering": "MinFirst"}
    });
    make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;

    // Publish messages in non-sequential order
    let priorities_and_payloads = vec![
        (50, "medium"),
        (10, "low"),
        (100, "high"),
        (1, "very_low"),
        (90, "very_high"),
    ];

    for (priority, payload) in &priorities_and_payloads {
        let publish_request = json!({
            "queue": "priority_test",
            "priority": priority,
            "payload": payload
        });
        make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    }

    // Consume all messages and verify they come in min-first order
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 30
    });

    let expected_order = vec!["very_low", "low", "medium", "very_high", "high"];
    let mut consumed_payloads = vec![];

    for _ in 0..5 {
        let (status, body) = make_request(
            api.clone(),
            "POST",
            "/consume/priority_test",
            Some(consume_request.clone()),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        consumed_payloads.push(body["payload"].as_str().unwrap().to_string());
    }

    assert_eq!(consumed_payloads, expected_order);
}

#[tokio::test]
async fn test_delete_queue_endpoint() {
    // Verifies: API-0010
    let (api, _temp_dir) = create_test_api().await;

    // Create a queue first
    let create_request = json!({
        "name": "delete_test",
        "config": {
            "ordering": "MaxFirst"
        }
    });

    let (status, body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;

    if status != StatusCode::OK {
        panic!(
            "Create queue failed with status {} and body: {}",
            status, body
        );
    }

    // Add some messages to the queue
    for i in 1..=3 {
        let publish_request = json!({
            "queue": "delete_test",
            "priority": i * 100,
            "payload": format!("message {}", i)
        });

        let (status, _body) =
            make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
        assert_eq!(status, StatusCode::OK);
    }

    // Verify messages exist
    let (stats_status, stats_body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(stats_status, StatusCode::OK);

    // Find the delete_test queue in the array
    let queues = stats_body["queues"].as_array().unwrap();
    let delete_test_queue = queues.iter().find(|q| q["name"] == "delete_test").unwrap();
    assert_eq!(delete_test_queue["total"], 3);

    // Delete the queue
    let (delete_status, delete_body) =
        make_request(api.clone(), "DELETE", "/delete-queue/delete_test", None).await;

    if delete_status != StatusCode::OK {
        panic!(
            "Delete failed with status {} and body: {}",
            delete_status, delete_body
        );
    }

    assert_eq!(delete_body["queue"], "delete_test");
    assert_eq!(delete_body["deleted_messages"], 3);

    // Verify queue no longer exists
    let (final_status, final_body) = make_request(api, "GET", "/stats", None).await;
    assert_eq!(final_status, StatusCode::OK);

    // The queue should not exist in the stats anymore
    let queues = final_body["queues"].as_array().unwrap();
    assert!(
        !queues.iter().any(|q| q["name"] == "delete_test"),
        "Queue should not exist after deletion"
    );
}

#[tokio::test]
async fn test_purge_queue_endpoint() {
    // Verifies: API-0009
    let (api, _temp_dir) = create_test_api().await;

    // Create a queue with custom configuration
    let create_request = json!({
        "name": "purge_test",
        "config": {
            "ordering": "MinFirst"
        }
    });

    let (status, _body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK);

    // Add messages to the queue
    for i in 1..=4 {
        let publish_request = json!({
            "queue": "purge_test",
            "priority": i * 50,
            "payload": format!("message {}", i)
        });

        let (status, _body) =
            make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
        assert_eq!(status, StatusCode::OK);
    }

    // Verify messages exist
    let (stats_status, stats_body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(stats_status, StatusCode::OK);

    // Find the purge_test queue in the array
    let queues = stats_body["queues"].as_array().unwrap();
    let purge_test_queue = queues.iter().find(|q| q["name"] == "purge_test").unwrap();
    assert_eq!(purge_test_queue["total"], 4);

    // Purge the queue
    let (purge_status, purge_body) =
        make_request(api.clone(), "POST", "/purge-queue/purge_test", None).await;
    assert_eq!(purge_status, StatusCode::OK);
    assert_eq!(purge_body["queue"], "purge_test");
    assert_eq!(purge_body["purged_messages"], 4);

    // Verify queue is empty but still exists (can add new messages)
    let (empty_status, empty_body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(empty_status, StatusCode::OK);

    // Find the purge_test queue in the array - it should still exist but be empty
    let queues = empty_body["queues"].as_array().unwrap();
    let purge_test_queue = queues.iter().find(|q| q["name"] == "purge_test").unwrap();
    assert_eq!(purge_test_queue["total"], 0);

    // Add a new message to verify queue configuration is preserved
    let new_message_request = json!({
        "queue": "purge_test",
        "priority": 25,
        "payload": "new message after purge"
    });

    let (status, _body) =
        make_request(api.clone(), "POST", "/publish", Some(new_message_request)).await;
    assert_eq!(status, StatusCode::OK);

    // Verify the queue configuration is preserved (MinFirst ordering)
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 5
    });

    let (consume_status, consume_body) =
        make_request(api, "POST", "/consume/purge_test", Some(consume_request)).await;
    assert_eq!(consume_status, StatusCode::OK);
    assert_eq!(consume_body["payload"], "new message after purge");
}

#[tokio::test]
async fn test_rename_queue_via_update() {
    // Verifies: API-0012 (renaming via update-queue)
    let (api, _temp_dir) = create_test_api().await;

    // Create queue and publish a message
    let create_request = json!({
        "name": "rename_from",
        "config": { "ordering": "Fifo", "allow_duplicates": true }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK, "create failed: {}", body);

    let publish_request = json!({
        "queue": "rename_from",
        "priority": 1,
        "payload": "hello"
    });
    let (status, body) = make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    assert_eq!(status, StatusCode::OK, "publish failed: {}", body);

    // Rename queue using update-queue
    let update_request = json!({
        "name": "rename_from",
        "config": { "name": "rename_to" }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::OK, "update failed: {}", body);

    // Verify stats show new queue and not old
    let (stats_status, stats_body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(stats_status, StatusCode::OK);
    let queues = stats_body["queues"].as_array().unwrap();
    assert!(queues.iter().any(|q| q["name"] == "rename_to"));
    assert!(!queues.iter().any(|q| q["name"] == "rename_from"));

    // Verify message is consumable from new queue
    let consume_request = json!({
        "consumer_id": "test_consumer",
        "timeout_seconds": 5
    });
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/rename_to",
        Some(consume_request),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "hello");
}

#[tokio::test]
async fn test_update_queue_allow_duplicates() {
    // Verifies: API-0012 (updating allow_duplicates only)
    let (api, _temp_dir) = create_test_api().await;

    // Create queue with allow_duplicates=true
    let create_request = json!({
        "name": "test_queue",
        "config": { "ordering": "Fifo", "allow_duplicates": true }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK, "create failed: {}", body);

    // Update to allow_duplicates=false
    let update_request = json!({
        "name": "test_queue",
        "config": { "allow_duplicates": false }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::OK, "update failed: {}", body);

    // Verify setting was updated
    let (status, stats) = make_request(api.clone(), "GET", "/queue-stats/test_queue", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(stats["config"]["allow_duplicates"], false);
    assert_eq!(stats["config"]["ordering"], "Fifo"); // Should preserve ordering
}

#[tokio::test]
async fn test_update_queue_rename() {
    // Verifies: API-0012 (renaming via update-queue)
    let (api, _temp_dir) = create_test_api().await;

    // Create queue
    let create_request = json!({
        "name": "old_name",
        "config": { "ordering": "MaxFirst", "allow_duplicates": true }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK, "create failed: {}", body);

    // Publish a message to verify it moves with rename
    let publish_request = json!({
        "queue": "old_name",
        "priority": 1,
        "payload": "test"
    });
    let (status, body) = make_request(api.clone(), "POST", "/publish", Some(publish_request)).await;
    assert_eq!(status, StatusCode::OK, "publish failed: {}", body);

    // Rename via update-queue
    let update_request = json!({
        "name": "old_name",
        "config": { "name": "new_name" }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::OK, "update failed: {}", body);

    // Verify old queue is gone and new queue exists
    let (status, stats_body) = make_request(api.clone(), "GET", "/stats", None).await;
    assert_eq!(status, StatusCode::OK);
    let queues = stats_body["queues"].as_array().unwrap();
    assert!(queues.iter().any(|q| q["name"] == "new_name"));
    assert!(!queues.iter().any(|q| q["name"] == "old_name"));

    // Verify new queue has correct config
    let (status, stats) = make_request(api.clone(), "GET", "/queue-stats/new_name", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(stats["config"]["ordering"], "MaxFirst"); // Preserved
    assert_eq!(stats["config"]["allow_duplicates"], true); // Preserved

    // Verify message moved to new queue
    let consume_request = json!({
        "consumer_id": "test",
        "timeout_seconds": 1
    });
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/new_name",
        Some(consume_request),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "test");
}

#[tokio::test]
async fn test_update_queue_rename_and_allow_duplicates() {
    // Verifies: API-0012 (updating both name and allow_duplicates)
    let (api, _temp_dir) = create_test_api().await;

    // Create queue
    let create_request = json!({
        "name": "queue_a",
        "config": { "ordering": "MinFirst", "allow_duplicates": true }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/create-queue", Some(create_request)).await;
    assert_eq!(status, StatusCode::OK, "create failed: {}", body);

    // Update both name and allow_duplicates
    let update_request = json!({
        "name": "queue_a",
        "config": {
            "name": "queue_b",
            "allow_duplicates": false
        }
    });
    let (status, body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::OK, "update failed: {}", body);

    // Verify new queue has updated config
    let (status, stats) = make_request(api.clone(), "GET", "/queue-stats/queue_b", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(stats["config"]["ordering"], "MinFirst"); // Preserved
    assert_eq!(stats["config"]["allow_duplicates"], false); // Updated
}

#[tokio::test]
async fn test_update_queue_nonexistent() {
    // Verifies: API-0012 (error handling for non-existent queue)
    let (api, _temp_dir) = create_test_api().await;

    let update_request = json!({
        "name": "nonexistent",
        "config": { "allow_duplicates": false }
    });
    let (status, _body) =
        make_request(api.clone(), "POST", "/update-queue", Some(update_request)).await;
    assert_eq!(status, StatusCode::NOT_FOUND);
}

#[tokio::test]
async fn test_delete_nonexistent_queue() {
    let (api, _temp_dir) = create_test_api().await;

    // Try to delete a queue that doesn't exist
    let (delete_status, delete_body) =
        make_request(api, "DELETE", "/delete-queue/nonexistent", None).await;
    assert_eq!(delete_status, StatusCode::OK);
    assert_eq!(delete_body["queue"], "nonexistent");
    assert_eq!(delete_body["deleted_messages"], 0);
}

#[tokio::test]
async fn test_purge_nonexistent_queue() {
    let (api, _temp_dir) = create_test_api().await;

    // Try to purge a queue that doesn't exist
    let (purge_status, purge_body) =
        make_request(api, "POST", "/purge-queue/nonexistent", None).await;
    assert_eq!(purge_status, StatusCode::OK);
    assert_eq!(purge_body["queue"], "nonexistent");
    assert_eq!(purge_body["purged_messages"], 0);
}

// ---------------------------------------------------------------------------
// Batch ACK endpoint tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_batch_ack_all_found() {
    let (api, _temp_dir) = create_test_api().await;

    // Publish three messages
    let mut ids = Vec::new();
    for i in 0..3 {
        let (_, body) = make_request(
            api.clone(),
            "POST",
            "/publish",
            Some(json!({ "queue": "batch_q", "priority": 100, "payload": format!("p-{}", i) })),
        )
        .await;
        ids.push(body["id"].as_str().unwrap().to_string());
    }

    // Consume all three to lock them
    for _ in 0..3 {
        make_request(
            api.clone(),
            "POST",
            "/consume/batch_q",
            Some(json!({ "consumer_id": "worker-1", "timeout_seconds": 3600 })),
        )
        .await;
    }

    // Batch ACK
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/ack-batch/batch_q",
        Some(json!({ "consumer_id": "worker-1", "message_ids": ids })),
    )
    .await;

    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["acked"].as_array().unwrap().len(), 3);
    assert_eq!(body["not_found"].as_array().unwrap().len(), 0);
}

#[tokio::test]
async fn test_batch_ack_partial_not_found() {
    let (api, _temp_dir) = create_test_api().await;

    let (_, body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({ "queue": "batch_q", "priority": 100, "payload": "real" })),
    )
    .await;
    let real_id = body["id"].as_str().unwrap().to_string();

    make_request(
        api.clone(),
        "POST",
        "/consume/batch_q",
        Some(json!({ "consumer_id": "worker-1", "timeout_seconds": 3600 })),
    )
    .await;

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/ack-batch/batch_q",
        Some(json!({ "consumer_id": "worker-1", "message_ids": [real_id, "fake-id"] })),
    )
    .await;

    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["acked"].as_array().unwrap().len(), 1);
    assert_eq!(body["not_found"].as_array().unwrap().len(), 1);
    assert_eq!(body["not_found"][0], "fake-id");
}

#[tokio::test]
async fn test_batch_ack_empty_list() {
    let (api, _temp_dir) = create_test_api().await;

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/ack-batch/batch_q",
        Some(json!({ "consumer_id": "worker-1", "message_ids": [] })),
    )
    .await;

    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["acked"].as_array().unwrap().len(), 0);
    assert_eq!(body["not_found"].as_array().unwrap().len(), 0);
}

// ---------------------------------------------------------------------------
// Batch NACK endpoint tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_batch_nack_all_unlocked() {
    let (api, _temp_dir) = create_test_api().await;

    let mut ids = Vec::new();
    for i in 0..3 {
        let (_, body) = make_request(
            api.clone(),
            "POST",
            "/publish",
            Some(json!({ "queue": "batch_q", "priority": 100, "payload": format!("p-{}", i) })),
        )
        .await;
        ids.push(body["id"].as_str().unwrap().to_string());
    }

    for _ in 0..3 {
        make_request(
            api.clone(),
            "POST",
            "/consume/batch_q",
            Some(json!({ "consumer_id": "worker-1", "timeout_seconds": 3600 })),
        )
        .await;
    }

    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/nack-batch/batch_q",
        Some(json!({ "consumer_id": "worker-1", "message_ids": ids })),
    )
    .await;

    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["unlocked"].as_array().unwrap().len(), 3);
    assert_eq!(body["dead_lettered"].as_array().unwrap().len(), 0);
    assert_eq!(body["dropped"].as_array().unwrap().len(), 0);
    assert_eq!(body["not_found"].as_array().unwrap().len(), 0);
}

#[tokio::test]
async fn test_batch_nack_missing_fields_rejected() {
    let (api, _temp_dir) = create_test_api().await;

    // Missing consumer_id
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/nack-batch/batch_q",
        Some(json!({ "message_ids": ["id1"] })),
    )
    .await;
    assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
}

// ---------------------------------------------------------------------------
// Purge-all / Delete-all endpoint tests
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_purge_all_empties_all_queues() {
    let (api, _temp_dir) = create_test_api().await;

    // Create 3 queues with messages
    for q in &["pa_q1", "pa_q2", "pa_q3"] {
        make_request(
            api.clone(),
            "POST",
            "/create-queue",
            Some(json!({ "name": q, "config": { "ordering": "MaxFirst" } })),
        )
        .await;
        for i in 0..3 {
            make_request(
                api.clone(),
                "POST",
                "/publish",
                Some(json!({ "queue": q, "priority": i, "payload": format!("msg-{}", i) })),
            )
            .await;
        }
    }

    // Purge all
    let (status, body) = make_request(api.clone(), "POST", "/purge-all", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["purged_queues"], 3);
    assert_eq!(body["purged_messages"], 9);

    // Verify all queues are empty but still exist
    let (_, stats) = make_request(api.clone(), "GET", "/stats", None).await;
    for q in stats["queues"].as_array().unwrap() {
        assert_eq!(q["available"], 0);
        assert_eq!(q["total"], 0);
    }
}

#[tokio::test]
async fn test_purge_all_empty_server() {
    let (api, _temp_dir) = create_test_api().await;

    let (status, body) = make_request(api, "POST", "/purge-all", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["purged_queues"], 0);
    assert_eq!(body["purged_messages"], 0);
}

#[tokio::test]
async fn test_delete_all_removes_all_queues() {
    let (api, _temp_dir) = create_test_api().await;

    // Create 3 queues with messages
    for q in &["da_q1", "da_q2", "da_q3"] {
        make_request(
            api.clone(),
            "POST",
            "/create-queue",
            Some(json!({ "name": q, "config": { "ordering": "MaxFirst" } })),
        )
        .await;
        make_request(
            api.clone(),
            "POST",
            "/publish",
            Some(json!({ "queue": q, "priority": 1, "payload": "msg" })),
        )
        .await;
    }

    // Delete all
    let (status, body) = make_request(api.clone(), "POST", "/delete-all", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["deleted_queues"], 3);
    assert_eq!(body["deleted_messages"], 3);

    // Verify no queues remain
    let (_, queues) = make_request(api.clone(), "GET", "/queues", None).await;
    assert_eq!(queues.as_array().unwrap().len(), 0);
}

#[tokio::test]
async fn test_delete_all_empty_server() {
    let (api, _temp_dir) = create_test_api().await;

    let (status, body) = make_request(api, "POST", "/delete-all", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["deleted_queues"], 0);
    assert_eq!(body["deleted_messages"], 0);
}

// Verifies: PER-0013, API-0010
#[tokio::test]
async fn test_delete_queue_with_dlq_allows_recreate() {
    let (api, _temp_dir) = create_test_api().await;

    // Create queue
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "dlq_q", "config": { "ordering": "MaxFirst" } })),
    )
    .await;

    // Publish a message with max_retries=0 so first nack sends to DLQ
    let (_, pub_body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({ "queue": "dlq_q", "priority": 1, "payload": "dlq_msg", "max_retries": 0 })),
    )
    .await;
    let msg_id = pub_body["id"].as_str().unwrap();

    // Consume it
    make_request(
        api.clone(),
        "POST",
        "/consume/dlq_q",
        Some(json!({ "consumer_id": "c1", "timeout_seconds": 30 })),
    )
    .await;

    // Nack → should go to DLQ (max_retries=0, retry_count incremented by pop)
    let nack_path = format!("/nack/dlq_q/{}", msg_id);
    make_request(
        api.clone(),
        "POST",
        &nack_path,
        Some(json!({ "consumer_id": "c1" })),
    )
    .await;

    // Delete the queue
    let (status, _) = make_request(api.clone(), "DELETE", "/delete-queue/dlq_q", None).await;
    assert_eq!(status, StatusCode::OK);

    // Re-create must succeed (not 409)
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "dlq_q", "config": { "ordering": "MaxFirst" } })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
}

// Verifies: PER-0013
#[tokio::test]
async fn test_delete_all_with_dlq_allows_recreate() {
    let (api, _temp_dir) = create_test_api().await;

    // Create queue, publish with max_retries=0, consume, nack → DLQ
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "da_dlq", "config": { "ordering": "MaxFirst" } })),
    )
    .await;
    let (_, pub_body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({ "queue": "da_dlq", "priority": 1, "payload": "dlq_msg", "max_retries": 0 })),
    )
    .await;
    let msg_id = pub_body["id"].as_str().unwrap();
    make_request(
        api.clone(),
        "POST",
        "/consume/da_dlq",
        Some(json!({ "consumer_id": "c1", "timeout_seconds": 30 })),
    )
    .await;
    let nack_path = format!("/nack/da_dlq/{}", msg_id);
    make_request(
        api.clone(),
        "POST",
        &nack_path,
        Some(json!({ "consumer_id": "c1" })),
    )
    .await;

    // Delete all
    let (status, _) = make_request(api.clone(), "POST", "/delete-all", None).await;
    assert_eq!(status, StatusCode::OK);

    // Re-create must succeed
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "da_dlq", "config": { "ordering": "MaxFirst" } })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
}

// ---------------------------------------------------------------------------
// update_queue validation error paths
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_update_queue_empty_name_returns_bad_request() {
    let (api, _temp_dir) = create_test_api().await;

    let (status, _) = make_request(
        api,
        "POST",
        "/update-queue",
        Some(json!({"name": "   ", "config": {}})),
    )
    .await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
}

#[tokio::test]
async fn test_update_queue_rename_to_existing_returns_conflict() {
    let (api, _temp_dir) = create_test_api().await;

    // Create two queues.
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({"name": "src-q", "config": {"ordering": "MaxFirst"}})),
    )
    .await;
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({"name": "dst-q", "config": {"ordering": "MaxFirst"}})),
    )
    .await;

    // Try to rename src-q to dst-q (already exists) → conflict.
    let (status, _) = make_request(
        api,
        "POST",
        "/update-queue",
        Some(json!({"name": "src-q", "config": {"name": "dst-q"}})),
    )
    .await;
    assert_eq!(status, StatusCode::CONFLICT);
}

#[tokio::test]
async fn test_update_queue_rename_to_empty_string_returns_bad_request() {
    let (api, _temp_dir) = create_test_api().await;

    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({"name": "renameq", "config": {"ordering": "MaxFirst"}})),
    )
    .await;

    let (status, _) = make_request(
        api,
        "POST",
        "/update-queue",
        Some(json!({"name": "renameq", "config": {"name": ""}})),
    )
    .await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
}

// ---------------------------------------------------------------------------
// Text priority through API
// ---------------------------------------------------------------------------

#[tokio::test]
async fn test_text_priority_ordering_through_api() {
    let (api, _temp_dir) = create_test_api().await;

    // Create a MinFirst queue with text priority
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({
            "name": "text_q",
            "config": {
                "ordering": "MinFirst",
                "priority_kind": "Text"
            }
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Publish with string priorities
    for (prio, payload) in &[("charlie", "c"), ("alpha", "a"), ("bravo", "b")] {
        let (status, _) = make_request(
            api.clone(),
            "POST",
            "/publish",
            Some(json!({
                "queue": "text_q",
                "priority": prio,
                "payload": payload
            })),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
    }

    // Consume — should come out in ascending lex order
    for expected in &["a", "b", "c"] {
        let (status, body) = make_request(
            api.clone(),
            "POST",
            "/consume/text_q",
            Some(json!({"consumer_id": "w", "timeout_seconds": 30})),
        )
        .await;
        assert_eq!(status, StatusCode::OK);
        assert_eq!(body["payload"].as_str().unwrap(), *expected);
    }
}

#[tokio::test]
async fn test_text_priority_kind_mismatch_through_api() {
    let (api, _temp_dir) = create_test_api().await;

    // Create text priority queue
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({
            "name": "tq",
            "config": {
                "ordering": "MinFirst",
                "priority_kind": "Text"
            }
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Publish with numeric priority — should fail
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({
            "queue": "tq",
            "priority": 42,
            "payload": "nope"
        })),
    )
    .await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
}

/// Verifies: SCH-0002
#[tokio::test]
async fn test_text_priority_with_slash_succeeds() {
    let (api, _temp_dir) = create_test_api().await;

    // Create text priority queue
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({
            "name": "slash_q",
            "config": {
                "ordering": "MinFirst",
                "priority_kind": "Text"
            }
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Publish with text priority containing '/' — should succeed
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({
            "queue": "slash_q",
            "priority": "/usr/local/bin",
            "payload": "path_payload"
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert!(
        body["id"].as_str().is_some(),
        "Expected message id in response"
    );

    // Consume and verify payload round-trips correctly
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/consume/slash_q",
        Some(json!({
            "consumer_id": "c1",
            "timeout": 30
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["payload"], "path_payload");

    // Queue stats should show correct counts (1 locked)
    let (status, body) = make_request(api.clone(), "GET", "/queue-stats/slash_q", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["total"], 1);
}

/// Verifies: PER-0010
#[tokio::test]
async fn test_text_priority_empty_returns_400() {
    let (api, _temp_dir) = create_test_api().await;

    // Create text priority queue
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({
            "name": "empty_q",
            "config": {
                "ordering": "MinFirst",
                "priority_kind": "Text"
            }
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Publish with empty text priority — should be 400
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({
            "queue": "empty_q",
            "priority": "",
            "payload": "test"
        })),
    )
    .await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
    let error_msg = body["error"].as_str().unwrap();
    assert!(
        error_msg.contains("Text priority must not be empty"),
        "Expected validation error, got: {}",
        error_msg
    );
}

/// Verifies: PER-0012
#[tokio::test]
async fn test_duplicate_payload_returns_400() {
    let (api, _temp_dir) = create_test_api().await;

    // Create queue with duplicates disabled
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({
            "name": "nodup_q",
            "config": {
                "ordering": "MaxFirst",
                "allow_duplicates": false
            }
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // First publish — should succeed
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({
            "queue": "nodup_q",
            "priority": 1,
            "payload": "same"
        })),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Second publish with same payload — should be 400 (not 500)
    let (status, body) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({
            "queue": "nodup_q",
            "priority": 2,
            "payload": "same"
        })),
    )
    .await;
    assert_eq!(status, StatusCode::BAD_REQUEST);
    let error_msg = body["error"].as_str().unwrap();
    assert!(
        error_msg.contains("Duplicate payload rejected"),
        "Expected duplicate error, got: {}",
        error_msg
    );
}

// ─── Operation Timings (API-0013, SYS-0017) ────────────────────────

/// Verifies: API-0013 — endpoint returns 200 with empty data when idle
#[tokio::test]
async fn test_operation_timings_returns_200_when_idle() {
    let (api, _dir) = create_test_api().await;
    let (status, body) = make_request(api, "GET", "/operation-timings", None).await;
    assert_eq!(status, StatusCode::OK);
    assert!(body.is_object(), "expected JSON object, got: {body}");
}

/// Verifies: API-0013 — default seconds=300
#[tokio::test]
async fn test_operation_timings_default_window() {
    let (api, _dir) = create_test_api().await;

    // Publish a message to generate timing data.
    let (status, _) = make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({"queue":"t","priority":1,"payload":"x"})),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    let (status, body) = make_request(api, "GET", "/operation-timings", None).await;
    assert_eq!(status, StatusCode::OK);

    // Should contain publish and storage_push operations.
    let obj = body.as_object().unwrap();
    assert!(
        obj.contains_key("publish"),
        "missing 'publish' key: {obj:?}"
    );
    assert!(
        obj.contains_key("storage_push"),
        "missing 'storage_push' key: {obj:?}"
    );

    // Each operation should have buckets with the right fields.
    let buckets = obj["publish"].as_array().unwrap();
    assert!(!buckets.is_empty());
    let b = &buckets[0];
    assert!(b.get("t_epoch_s").is_some());
    assert!(b.get("count").is_some());
    assert!(b.get("total_us").is_some());
    assert!(b.get("min_us").is_some());
    assert!(b.get("max_us").is_some());
}

/// Verifies: API-0013 — custom seconds parameter
#[tokio::test]
async fn test_operation_timings_custom_seconds() {
    let (api, _dir) = create_test_api().await;

    // Publish to generate data.
    make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({"queue":"t","priority":1,"payload":"x"})),
    )
    .await;

    let (status, body) = make_request(api, "GET", "/operation-timings?seconds=10", None).await;
    assert_eq!(status, StatusCode::OK);

    let obj = body.as_object().unwrap();
    // With seconds=10, each operation should have at most 10 buckets.
    if let Some(buckets) = obj.get("publish").and_then(|v| v.as_array()) {
        assert!(
            buckets.len() <= 10,
            "expected <=10 buckets, got {}",
            buckets.len()
        );
    }
}

/// Verifies: SYS-0017 — timing recorded for consume, ack, nack operations
#[tokio::test]
async fn test_operation_timings_covers_all_api_operations() {
    let (api, _dir) = create_test_api().await;

    // Create queue and publish.
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({"name":"ops","ordering":"MaxFirst"})),
    )
    .await;
    make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({"queue":"ops","priority":1,"payload":"p1"})),
    )
    .await;
    make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({"queue":"ops","priority":2,"payload":"p2"})),
    )
    .await;

    // Consume.
    let (status, consume_body) = make_request(
        api.clone(),
        "POST",
        "/consume/ops",
        Some(json!({"consumer_id":"c1","timeout_seconds":30})),
    )
    .await;
    assert_eq!(status, StatusCode::OK);
    let msg_id = consume_body["id"].as_str().unwrap().to_string();

    // Ack.
    let (status, _) = make_request(
        api.clone(),
        "POST",
        &format!("/ack/ops/{msg_id}"),
        Some(json!({"consumer_id":"c1"})),
    )
    .await;
    assert_eq!(status, StatusCode::OK);

    // Consume another and nack.
    let (_, consume_body2) = make_request(
        api.clone(),
        "POST",
        "/consume/ops",
        Some(json!({"consumer_id":"c1","timeout_seconds":30})),
    )
    .await;
    let msg_id2 = consume_body2["id"].as_str().unwrap().to_string();
    make_request(
        api.clone(),
        "POST",
        &format!("/nack/ops/{msg_id2}"),
        Some(json!({"consumer_id":"c1"})),
    )
    .await;

    // Queue stats.
    make_request(api.clone(), "GET", "/queue-stats/ops", None).await;

    // Now check timings.
    let (status, body) = make_request(api, "GET", "/operation-timings?seconds=60", None).await;
    assert_eq!(status, StatusCode::OK);

    let obj = body.as_object().unwrap();
    let expected_ops = [
        "publish",
        "consume",
        "ack",
        "nack",
        "queue_stats",
        "storage_push",
        "storage_pop",
        "storage_ack",
        "storage_nack",
        "storage_get_queue_stats",
    ];

    for op in &expected_ops {
        assert!(
            obj.contains_key(*op),
            "missing operation '{}' in timings: {:?}",
            op,
            obj.keys().collect::<Vec<_>>()
        );

        let buckets = obj[*op].as_array().unwrap();
        let total_count: u64 = buckets.iter().map(|b| b["count"].as_u64().unwrap()).sum();
        assert!(
            total_count > 0,
            "operation '{}' should have at least 1 sample, got 0",
            op
        );
    }
}

/// Verifies: API-0013 — endpoint responds without touching storage
#[tokio::test]
async fn test_operation_timings_responsive_without_storage() {
    let (api, _dir) = create_test_api().await;

    // This test just verifies the endpoint is fast and doesn't error.
    // The real guarantee (no storage access) is architectural — the handler
    // reads only from the in-memory OperationTimingStore.
    let start = std::time::Instant::now();
    let (status, _) = make_request(api, "GET", "/operation-timings?seconds=3600", None).await;
    let elapsed = start.elapsed();
    assert_eq!(status, StatusCode::OK);
    assert!(
        elapsed.as_millis() < 500,
        "endpoint took too long: {:?}",
        elapsed
    );
}

// ---------------------------------------------------------------------------
// Force-unlock endpoint tests (API-0014)
// ---------------------------------------------------------------------------

/// Verifies: API-0014 — POST /force-unlock/{queue} unlocks all locked messages.
#[tokio::test]
async fn test_force_unlock_queue_endpoint() {
    let (api, _temp_dir) = create_test_api().await;

    // Create queue and publish + consume (lock) 3 messages.
    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "fu_api_q", "config": { "ordering": "MaxFirst" } })),
    )
    .await;
    for i in 0..3 {
        make_request(
            api.clone(),
            "POST",
            "/publish",
            Some(json!({ "queue": "fu_api_q", "priority": i, "payload": format!("msg-{}", i) })),
        )
        .await;
    }
    for _ in 0..3 {
        make_request(
            api.clone(),
            "POST",
            "/consume/fu_api_q",
            Some(json!({ "consumer_id": "test", "timeout_seconds": 300 })),
        )
        .await;
    }

    // Verify locked.
    let (_, stats) = make_request(api.clone(), "GET", "/queue-stats/fu_api_q", None).await;
    assert_eq!(stats["locked"], 3);

    // Force-unlock.
    let (status, body) = make_request(api.clone(), "POST", "/force-unlock/fu_api_q", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["queue"], "fu_api_q");
    assert_eq!(body["unlocked"], 3);

    // Verify unlocked.
    let (_, stats) = make_request(api.clone(), "GET", "/queue-stats/fu_api_q", None).await;
    assert_eq!(stats["locked"], 0);
    assert_eq!(stats["available"], 3);
}

/// Verifies: API-0014 — POST /force-unlock on a queue with no locked messages returns 0.
#[tokio::test]
async fn test_force_unlock_queue_endpoint_nothing_locked() {
    let (api, _temp_dir) = create_test_api().await;

    make_request(
        api.clone(),
        "POST",
        "/create-queue",
        Some(json!({ "name": "fu_empty", "config": { "ordering": "MaxFirst" } })),
    )
    .await;
    make_request(
        api.clone(),
        "POST",
        "/publish",
        Some(json!({ "queue": "fu_empty", "priority": 1, "payload": "test" })),
    )
    .await;

    let (status, body) = make_request(api.clone(), "POST", "/force-unlock/fu_empty", None).await;
    assert_eq!(status, StatusCode::OK);
    assert_eq!(body["unlocked"], 0);
}

// ---------------------------------------------------------------------------
// Stats memory fields test (SYS-0023)
// ---------------------------------------------------------------------------

/// Verifies: SYS-0023 — /stats summary includes memory fields.
#[tokio::test]
async fn test_stats_includes_memory_fields() {
    let (api, _temp_dir) = create_test_api().await;

    let (status, body) = make_request(api, "GET", "/stats", None).await;
    assert_eq!(status, StatusCode::OK);

    let summary = &body["summary"];
    // Memory fields must be present and numeric.
    assert!(
        summary["memory_usage_bytes"].is_number(),
        "summary.memory_usage_bytes should be a number, got {:?}",
        summary["memory_usage_bytes"]
    );
    assert!(
        summary["memory_limit_bytes"].is_number(),
        "summary.memory_limit_bytes should be a number, got {:?}",
        summary["memory_limit_bytes"]
    );
    // memory_pressure is a boolean.
    assert!(
        summary["memory_pressure"].is_boolean(),
        "summary.memory_pressure should be a boolean, got {:?}",
        summary["memory_pressure"]
    );
}