use std::sync::Arc;

use criterion::{BenchmarkId, Criterion, criterion_group, criterion_main};
use tokio::runtime::Runtime;

use fafb_q::queue::QueueEngine;
use fafb_q::queue::{EngineConfig, InMemoryQueueEngine};
use fafb_q::storage::mock::MockStorage;
use fafb_q::types::{
    Header, MessageEnvelope, MessageId, NamespaceId, TopicName, TopicOffset, now_millis,
};

fn make_envelope(id: u64, ns: &str, topic: &str, payload_size: usize) -> MessageEnvelope {
    MessageEnvelope {
        id: MessageId(id),
        namespace: NamespaceId(ns.into()),
        topic: TopicName(topic.into()),
        payload: vec![0xAB; payload_size],
        timestamp: now_millis(),
        headers: vec![],
        ttl_ms: 0,
    }
}

fn make_engine() -> Arc<InMemoryQueueEngine<MockStorage>> {
    Arc::new(InMemoryQueueEngine::new(
        Arc::new(MockStorage::new()),
        EngineConfig::default(),
    ))
}

fn single_topic_publish(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_publish_single_topic");

    for payload_size in [64, 1024, 65536] {
        group.bench_with_input(
            BenchmarkId::new("payload", payload_size),
            &payload_size,
            |b, &size| {
                let engine = make_engine();
                let ns = NamespaceId("bench-ns".into());
                let topic = TopicName("bench-topic".into());
                let mut id = 0u64;

                b.iter(|| {
                    let env = make_envelope(id, "bench-ns", "bench-topic", size);
                    id += 1;
                    rt.block_on(engine.publish(&ns, &topic, env)).unwrap();
                });
            },
        );
    }

    group.finish();
}

fn single_topic_consume(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_consume_single_topic");

    let engine = make_engine();
    let ns = NamespaceId("bench-ns".into());
    let topic = TopicName("bench-topic".into());

    // Pre-fill with 10,000 messages
    rt.block_on(async {
        for i in 0..10_000u64 {
            let env = make_envelope(i, "bench-ns", "bench-topic", 256);
            engine.publish(&ns, &topic, env).await.unwrap();
        }
    });

    let mut offset = 0u64;
    group.bench_function("consume_256b", |b| {
        b.iter(|| {
            let o = TopicOffset(offset % 10_000);
            offset += 1;
            rt.block_on(engine.consume_from(&ns, &topic, o)).unwrap();
        });
    });

    group.finish();
}

fn concurrent_publish_same_topic(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_concurrent_same_topic");

    for num_producers in [10, 100] {
        group.bench_with_input(
            BenchmarkId::new("producers", num_producers),
            &num_producers,
            |b, &n| {
                let engine = make_engine();
                let ns = NamespaceId("bench-ns".into());
                let topic = TopicName("bench-topic".into());

                b.iter(|| {
                    rt.block_on(async {
                        let mut handles = Vec::with_capacity(n);
                        for producer_id in 0..n {
                            let engine = Arc::clone(&engine);
                            let ns = ns.clone();
                            let topic = topic.clone();
                            handles.push(tokio::spawn(async move {
                                for i in 0..100u64 {
                                    let id = producer_id as u64 * 1000 + i;
                                    let env = make_envelope(id, &ns.0, &topic.0, 128);
                                    engine.publish(&ns, &topic, env).await.unwrap();
                                }
                            }));
                        }
                        for h in handles {
                            h.await.unwrap();
                        }
                    });
                });
            },
        );
    }

    group.finish();
}

/// Measures cross-topic contention with topics pre-created (steady-state).
/// Isolates lock contention from topic auto-creation overhead.
fn concurrent_publish_cross_topic_steady(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_cross_topic_steady");

    for num_topics in [10, 100] {
        group.bench_with_input(
            BenchmarkId::new("topics", num_topics),
            &num_topics,
            |b, &n| {
                let engine = make_engine();
                let ns = NamespaceId("bench-ns".into());

                // Pre-create all topics with a seed message
                rt.block_on(async {
                    for topic_id in 0..n {
                        let topic = TopicName(format!("topic-{topic_id}"));
                        let env = make_envelope(0, "bench-ns", &topic.0, 128);
                        engine.publish(&ns, &topic, env).await.unwrap();
                    }
                });

                let mut round = 1u64;
                b.iter(|| {
                    rt.block_on(async {
                        let mut handles = Vec::with_capacity(n);
                        for topic_id in 0..n {
                            let engine = Arc::clone(&engine);
                            let ns = ns.clone();
                            let topic = TopicName(format!("topic-{topic_id}"));
                            let base = round;
                            handles.push(tokio::spawn(async move {
                                for i in 0..100u64 {
                                    let env = make_envelope(base + i, &ns.0, &topic.0, 128);
                                    engine.publish(&ns, &topic, env).await.unwrap();
                                }
                            }));
                        }
                        for h in handles {
                            h.await.unwrap();
                        }
                    });
                    round += 100;
                });
            },
        );
    }

    group.finish();
}

/// Measures namespace contention with topics pre-created (steady-state).
fn namespace_contention_steady(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_ns_contention_steady");
    group.sample_size(10);

    for num_ns in [100, 1000] {
        group.bench_with_input(BenchmarkId::new("namespaces", num_ns), &num_ns, |b, &n| {
            let engine = make_engine();

            // Pre-create all namespace/topic pairs
            rt.block_on(async {
                for ns_id in 0..n {
                    let ns = NamespaceId(format!("ns-{ns_id}"));
                    let topic = TopicName("events".into());
                    let env = make_envelope(0, &ns.0, &topic.0, 128);
                    engine.publish(&ns, &topic, env).await.unwrap();
                }
            });

            let mut round = 1u64;
            b.iter(|| {
                rt.block_on(async {
                    let mut handles = Vec::with_capacity(n);
                    for ns_id in 0..n {
                        let engine = Arc::clone(&engine);
                        let ns = NamespaceId(format!("ns-{ns_id}"));
                        let topic = TopicName("events".into());
                        let base = round;
                        handles.push(tokio::spawn(async move {
                            for i in 0..50u64 {
                                let env = make_envelope(base + i, &ns.0, &topic.0, 128);
                                engine.publish(&ns, &topic, env).await.unwrap();
                            }
                        }));
                    }
                    for h in handles {
                        h.await.unwrap();
                    }
                });
                round += 50;
            });
        });
    }

    group.finish();
}

fn compact_throughput(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_compact");

    let engine = make_engine();
    let ns = NamespaceId("bench-ns".into());
    let topic = TopicName("bench-topic".into());

    // Pre-fill
    rt.block_on(async {
        for i in 0..50_000u64 {
            let env = make_envelope(i, "bench-ns", "bench-topic", 256);
            engine.publish(&ns, &topic, env).await.unwrap();
        }
    });

    let mut compact_offset = 0u64;
    group.bench_function("compact_1000", |b| {
        b.iter(|| {
            compact_offset += 1000;
            if compact_offset > 49_000 {
                compact_offset = 1000;
            }
            rt.block_on(engine.compact(&ns, &topic, TopicOffset(compact_offset)))
                .unwrap();
        });
    });

    group.finish();
}

fn single_topic_publish_with_headers(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();
    let mut group = c.benchmark_group("queue_publish_headers");

    let make_headers = |n: usize| -> Vec<Header> {
        (0..n)
            .map(|i| Header {
                key: format!("x-custom-header-{:03}", i).into_bytes(),
                value: format!("value-{:0>96}", i).into_bytes(),
            })
            .collect()
    };

    for header_count in [0, 1, 5, 10] {
        group.bench_with_input(
            BenchmarkId::new("headers", header_count),
            &header_count,
            |b, &count| {
                let engine = make_engine();
                let ns = NamespaceId("bench-ns".into());
                let topic = TopicName("bench-topic".into());
                let headers = make_headers(count);
                let mut id = 0u64;

                b.iter(|| {
                    let env = MessageEnvelope {
                        id: MessageId(id),
                        namespace: NamespaceId("bench-ns".into()),
                        topic: TopicName("bench-topic".into()),
                        payload: vec![0xAB; 1024],
                        timestamp: now_millis(),
                        headers: headers.clone(),
                        ttl_ms: 0,
                    };
                    id += 1;
                    rt.block_on(engine.publish(&ns, &topic, env)).unwrap();
                });
            },
        );
    }

    group.finish();
}

criterion_group!(
    benches,
    single_topic_publish,
    single_topic_consume,
    concurrent_publish_same_topic,
    concurrent_publish_cross_topic_steady,
    namespace_contention_steady,
    compact_throughput,
    single_topic_publish_with_headers,
);
criterion_main!(benches);
