// benches/http_benchmarks.rs
//
// Criterion benchmarks that send requests through the full Axum HTTP layer
// using tower's in-process `oneshot` transport (no real TCP socket).
//
// Run with:
//   cargo bench --bench http_benchmarks
//
// Comparing these results to storage_benchmarks isolates the overhead of:
//   - JSON serialization / deserialization
//   - Axum routing and middleware
//   - tokio task scheduling
//   - Rate-tracker and ApiServer bookkeeping
//
// Web overhead ≈ http_benchmark_time − storage_benchmark_time

use axum::{
    body::Body,
    http::{Request, StatusCode},
};
use chrono::Utc;
use criterion::{criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion};
use qrusty::{
    api::ApiServer,
    message::{Message, Priority, PriorityOrdering, QueueConfig},
    storage::Storage,
};
use serde_json::{json, Value};
use std::sync::atomic::{AtomicU64, Ordering};
use std::sync::Arc;
use tempfile::TempDir;
use tower::ServiceExt; // for `oneshot`

const DEPTHS: &[usize] = &[0, 100, 1_000, 10_000];

// Batch sizes for batch ACK/NACK HTTP benchmarks.
const BATCH_SIZES: &[usize] = &[1, 10, 50, 100];

// Fixed background queue depth for batch benchmarks.
const BATCH_DEPTH: usize = 1_000;

const QUEUE_ALLOW_DUP: &str = "http_allow_dup";
const QUEUE_NO_DUP: &str = "http_no_dup";
const CONSUMER: &str = "bench-consumer";

static COUNTER: AtomicU64 = AtomicU64::new(0);

fn next_id() -> u64 {
    COUNTER.fetch_add(1, Ordering::Relaxed)
}

// ---------------------------------------------------------------------------
// HTTP helpers
// ---------------------------------------------------------------------------

/// Issue one HTTP request through the Axum router using tower's in-process
/// oneshot transport.  Returns (status, parsed JSON body).
async fn http(
    api: &ApiServer,
    method: &str,
    path: &str,
    body: Option<Value>,
) -> (StatusCode, Value) {
    let app = api.clone().router();

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

    let request = match body {
        Some(b) => builder.body(Body::from(b.to_string())).unwrap(),
        None => builder.body(Body::empty()).unwrap(),
    };

    let response = app.oneshot(request).await.unwrap();
    let status = response.status();
    let bytes = axum::body::to_bytes(response.into_body(), usize::MAX)
        .await
        .unwrap();
    let text = String::from_utf8(bytes.to_vec()).unwrap();
    let json: Value = if text.is_empty() {
        Value::Null
    } else {
        serde_json::from_str(&text).unwrap_or(Value::String(text))
    };
    (status, json)
}

// ---------------------------------------------------------------------------
// Setup helpers
// ---------------------------------------------------------------------------

fn make_storage_message(queue: &str, payload: &str, priority: u64) -> Message {
    Message {
        id: format!("http-bench-{}", next_id()),
        queue: queue.to_string(),
        priority: Priority::Numeric(priority),
        payload: payload.to_string(),
        created_at: Utc::now(),
        locked_until: None,
        locked_by: None,
        retry_count: 0,
        max_retries: 10,
        payload_ref: None,
        payload_hash: None,
    }
}

/// Creates an ApiServer backed by real Storage, pre-populates `depth`
/// background messages, and returns (ApiServer, TempDir).
/// TempDir must stay alive for the benchmark.
fn setup_api(queue: &str, allow_duplicates: bool, depth: usize) -> (ApiServer, TempDir) {
    let temp_dir = TempDir::new().expect("tempdir");
    let storage = Storage::new(temp_dir.path().to_str().unwrap()).expect("storage");

    storage.configure_queue(
        queue,
        QueueConfig {
            ordering: PriorityOrdering::MaxFirst,
            allow_duplicates,
            ..Default::default()
        },
    );

    for i in 0..depth {
        let msg = make_storage_message(queue, &format!("background-{}", i), 100);
        storage.push(msg).expect("pre-populate push");
    }

    let api = ApiServer::new(Arc::new(storage));
    (api, temp_dir)
}

// ---------------------------------------------------------------------------
// Scenario 1: POST /publish — allow_duplicates = true
// ---------------------------------------------------------------------------
fn bench_http_publish_allow_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/publish/allow_duplicates");
    group.sample_size(200);

    for &depth in DEPTHS {
        let (api, _dir) = setup_api(QUEUE_ALLOW_DUP, true, depth);

        group.bench_with_input(
            BenchmarkId::new("queue_depth", depth),
            &depth,
            |b, _depth| {
                b.iter(|| {
                    let n = next_id();
                    rt.block_on(http(
                        &api,
                        "POST",
                        "/publish",
                        Some(json!({
                            "queue": QUEUE_ALLOW_DUP,
                            "priority": 50,
                            "payload": format!("payload-{}", n),
                        })),
                    ))
                });
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 2: POST /publish — allow_duplicates = false
// ---------------------------------------------------------------------------
fn bench_http_publish_no_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/publish/no_duplicates");
    group.sample_size(200);

    for &depth in DEPTHS {
        let (api, _dir) = setup_api(QUEUE_NO_DUP, false, depth);

        group.bench_with_input(
            BenchmarkId::new("queue_depth", depth),
            &depth,
            |b, _depth| {
                b.iter(|| {
                    let n = next_id();
                    rt.block_on(http(
                        &api,
                        "POST",
                        "/publish",
                        Some(json!({
                            "queue": QUEUE_NO_DUP,
                            "priority": 50,
                            "payload": format!("unique-{}", n),
                        })),
                    ))
                });
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 3: POST /ack/{queue}/{id}
//
// Setup (not measured): publish a low-priority message + consume it to lock it.
// Measured: POST /ack with the locked message's ID.
// ---------------------------------------------------------------------------
fn bench_http_ack(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/ack");
    group.sample_size(200);

    for &depth in DEPTHS {
        let (api, _dir) = setup_api(QUEUE_ALLOW_DUP, true, depth);

        group.bench_with_input(
            BenchmarkId::new("queue_depth", depth),
            &depth,
            |b, _depth| {
                b.iter_batched(
                    || {
                        // Setup (not measured): publish + consume a low-priority message
                        rt.block_on(async {
                            let n = next_id();

                            // Publish at priority 1 — scans past all background messages
                            let (status, body) = http(
                                &api,
                                "POST",
                                "/publish",
                                Some(json!({
                                    "queue": QUEUE_ALLOW_DUP,
                                    "priority": 1,
                                    "payload": format!("ack-target-{}", n),
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "publish failed: {:?}", body);

                            // Consume to lock it
                            let (status, body) = http(
                                &api,
                                "POST",
                                &format!("/consume/{}", QUEUE_ALLOW_DUP),
                                Some(json!({
                                    "consumer_id": CONSUMER,
                                    "timeout_seconds": 3600,
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "consume failed: {:?}", body);

                            // Return the message ID
                            body["id"].as_str().unwrap().to_string()
                        })
                    },
                    |msg_id| {
                        // Measured: POST /ack
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/ack/{}/{}", QUEUE_ALLOW_DUP, msg_id),
                            Some(json!({ "consumer_id": CONSUMER })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 4a: POST /nack/{queue}/{id} — allow_duplicates = true
// ---------------------------------------------------------------------------
fn bench_http_nack_allow_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/nack/allow_duplicates");
    group.sample_size(200);

    for &depth in DEPTHS {
        let (api, _dir) = setup_api(QUEUE_ALLOW_DUP, true, depth);

        group.bench_with_input(
            BenchmarkId::new("queue_depth", depth),
            &depth,
            |b, _depth| {
                b.iter_batched(
                    || {
                        rt.block_on(async {
                            let n = next_id();

                            let (status, body) = http(
                                &api,
                                "POST",
                                "/publish",
                                Some(json!({
                                    "queue": QUEUE_ALLOW_DUP,
                                    "priority": 1,
                                    "payload": format!("nack-target-{}", n),
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "publish failed: {:?}", body);

                            let (status, body) = http(
                                &api,
                                "POST",
                                &format!("/consume/{}", QUEUE_ALLOW_DUP),
                                Some(json!({
                                    "consumer_id": CONSUMER,
                                    "timeout_seconds": 3600,
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "consume failed: {:?}", body);

                            body["id"].as_str().unwrap().to_string()
                        })
                    },
                    |msg_id| {
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/nack/{}/{}", QUEUE_ALLOW_DUP, msg_id),
                            Some(json!({ "consumer_id": CONSUMER })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 4b: POST /nack/{queue}/{id} — allow_duplicates = false
//
// After the ID scan, NACK also calls has_duplicate_payload() — O(n) extra scan.
// ---------------------------------------------------------------------------
fn bench_http_nack_no_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/nack/no_duplicates");
    group.sample_size(200);

    for &depth in DEPTHS {
        let (api, _dir) = setup_api(QUEUE_NO_DUP, false, depth);

        group.bench_with_input(
            BenchmarkId::new("queue_depth", depth),
            &depth,
            |b, _depth| {
                b.iter_batched(
                    || {
                        rt.block_on(async {
                            let n = next_id();

                            let (status, body) = http(
                                &api,
                                "POST",
                                "/publish",
                                Some(json!({
                                    "queue": QUEUE_NO_DUP,
                                    "priority": 1,
                                    "payload": format!("nack-unique-{}", n),
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "publish failed: {:?}", body);

                            let (status, body) = http(
                                &api,
                                "POST",
                                &format!("/consume/{}", QUEUE_NO_DUP),
                                Some(json!({
                                    "consumer_id": CONSUMER,
                                    "timeout_seconds": 3600,
                                })),
                            )
                            .await;
                            assert_eq!(status, StatusCode::OK, "consume failed: {:?}", body);

                            body["id"].as_str().unwrap().to_string()
                        })
                    },
                    |msg_id| {
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/nack/{}/{}", QUEUE_NO_DUP, msg_id),
                            Some(json!({ "consumer_id": CONSUMER })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 5: POST /ack-batch/{queue}
//
// Setup (not measured): publish + consume `batch_size` low-priority messages.
// Measured: one POST /ack-batch with all message IDs in the body.
//
// Quantifies the HTTP + JSON serialization overhead for batch operations and
// shows how per-request fixed costs amortize across larger batches.
// ---------------------------------------------------------------------------
fn bench_http_batch_ack(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/batch_ack");
    group.sample_size(50);

    for &batch_size in BATCH_SIZES {
        let (api, _dir) = setup_api(QUEUE_ALLOW_DUP, true, BATCH_DEPTH);

        group.bench_with_input(
            BenchmarkId::new("batch_size", batch_size),
            &batch_size,
            |b, &bs| {
                b.iter_batched(
                    || {
                        // Setup (not measured): publish bs messages and consume them all.
                        rt.block_on(async {
                            let mut ids = Vec::with_capacity(bs);
                            for _ in 0..bs {
                                let n = next_id();

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    "/publish",
                                    Some(json!({
                                        "queue": QUEUE_ALLOW_DUP,
                                        "priority": 1,
                                        "payload": format!("batch-ack-{}", n),
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "publish: {:?}", body);

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    &format!("/consume/{}", QUEUE_ALLOW_DUP),
                                    Some(json!({
                                        "consumer_id": CONSUMER,
                                        "timeout_seconds": 3600,
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "consume: {:?}", body);
                                ids.push(body["id"].as_str().unwrap().to_string());
                            }
                            ids
                        })
                    },
                    |ids| {
                        // Measured: one POST /ack-batch for all `bs` message IDs.
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/ack-batch/{}", QUEUE_ALLOW_DUP),
                            Some(json!({
                                "consumer_id": CONSUMER,
                                "message_ids": ids,
                            })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 6a: POST /nack-batch/{queue} — allow_duplicates = true
// ---------------------------------------------------------------------------
fn bench_http_batch_nack_allow_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/batch_nack/allow_duplicates");
    group.sample_size(50);

    for &batch_size in BATCH_SIZES {
        let (api, _dir) = setup_api(QUEUE_ALLOW_DUP, true, BATCH_DEPTH);

        group.bench_with_input(
            BenchmarkId::new("batch_size", batch_size),
            &batch_size,
            |b, &bs| {
                b.iter_batched(
                    || {
                        rt.block_on(async {
                            let mut ids = Vec::with_capacity(bs);
                            for _ in 0..bs {
                                let n = next_id();

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    "/publish",
                                    Some(json!({
                                        "queue": QUEUE_ALLOW_DUP,
                                        "priority": 1,
                                        "payload": format!("batch-nack-dup-{}", n),
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "publish: {:?}", body);

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    &format!("/consume/{}", QUEUE_ALLOW_DUP),
                                    Some(json!({
                                        "consumer_id": CONSUMER,
                                        "timeout_seconds": 3600,
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "consume: {:?}", body);
                                ids.push(body["id"].as_str().unwrap().to_string());
                            }
                            ids
                        })
                    },
                    |ids| {
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/nack-batch/{}", QUEUE_ALLOW_DUP),
                            Some(json!({
                                "consumer_id": CONSUMER,
                                "message_ids": ids,
                            })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

// ---------------------------------------------------------------------------
// Scenario 6b: POST /nack-batch/{queue} — allow_duplicates = false
//
// The batch nack builds the duplicate-payload set during its single scan,
// so there is no per-message second scan. Compare to bench_http_nack_no_duplicates.
// ---------------------------------------------------------------------------
fn bench_http_batch_nack_no_duplicates(c: &mut Criterion) {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let mut group = c.benchmark_group("http/batch_nack/no_duplicates");
    group.sample_size(50);

    for &batch_size in BATCH_SIZES {
        let (api, _dir) = setup_api(QUEUE_NO_DUP, false, BATCH_DEPTH);

        group.bench_with_input(
            BenchmarkId::new("batch_size", batch_size),
            &batch_size,
            |b, &bs| {
                b.iter_batched(
                    || {
                        rt.block_on(async {
                            let mut ids = Vec::with_capacity(bs);
                            for _ in 0..bs {
                                let n = next_id();

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    "/publish",
                                    Some(json!({
                                        "queue": QUEUE_NO_DUP,
                                        "priority": 1,
                                        "payload": format!("batch-nack-nodup-{}", n),
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "publish: {:?}", body);

                                let (status, body) = http(
                                    &api,
                                    "POST",
                                    &format!("/consume/{}", QUEUE_NO_DUP),
                                    Some(json!({
                                        "consumer_id": CONSUMER,
                                        "timeout_seconds": 3600,
                                    })),
                                )
                                .await;
                                assert_eq!(status, StatusCode::OK, "consume: {:?}", body);
                                ids.push(body["id"].as_str().unwrap().to_string());
                            }
                            ids
                        })
                    },
                    |ids| {
                        rt.block_on(http(
                            &api,
                            "POST",
                            &format!("/nack-batch/{}", QUEUE_NO_DUP),
                            Some(json!({
                                "consumer_id": CONSUMER,
                                "message_ids": ids,
                            })),
                        ))
                    },
                    BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

criterion_group!(
    benches,
    bench_http_publish_allow_duplicates,
    bench_http_publish_no_duplicates,
    bench_http_ack,
    bench_http_nack_allow_duplicates,
    bench_http_nack_no_duplicates,
    bench_http_batch_ack,
    bench_http_batch_nack_allow_duplicates,
    bench_http_batch_nack_no_duplicates,
);
criterion_main!(benches);
