use criterion::{
    criterion_group, criterion_main, BatchSize, BenchmarkId, Criterion, Throughput,
};
use std::sync::{Arc, Barrier};
use std::thread;
use swactor::channel::{HybridChannel, Receiver, Sender};

// ---------------------------------------------------------------------------
// Payload types
// ---------------------------------------------------------------------------

#[derive(Clone, Debug)]
struct Tiny(u64);

#[derive(Clone, Debug)]
struct Medium([u8; 256]);

impl Default for Medium {
    fn default() -> Self {
        Medium([0xAB; 256])
    }
}

#[derive(Clone, Debug)]
struct Large(Box<[u8; 4096]>);

impl Default for Large {
    fn default() -> Self {
        Large(Box::new([0xCD; 4096]))
    }
}

// ===========================================================================
// 1. SINGLE-THREADED — raw HybridChannel push/pop
// ===========================================================================

fn st_push_pop(c: &mut Criterion) {
    let mut group = c.benchmark_group("st_push_pop");

    // 1a — Single push + pop (hot path: ring only)
    for cap in [64, 256, 1024] {
        group.bench_with_input(
            BenchmarkId::new("single_roundtrip", cap),
            &cap,
            |b, &cap| {
                let ch = HybridChannel::new(cap);
                b.iter(|| {
                    ch.push(Tiny(42)).unwrap();
                    ch.pop().unwrap();
                });
            },
        );
    }

    // 1b — Burst: push N then pop N (all in ring)
    for n in [64, 256, 1024] {
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(BenchmarkId::new("burst_ring_only", n), &n, |b, &n| {
            let ch = HybridChannel::new(n);
            b.iter(|| {
                for i in 0..n as u64 {
                    ch.push(Tiny(i)).unwrap();
                }
                for _ in 0..n {
                    ch.pop().unwrap();
                }
            });
        });
    }

    // 1c — Burst: push N into capacity N/2 (forces overflow)
    for n in [128, 512, 2048] {
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(
            BenchmarkId::new("burst_with_overflow", n),
            &n,
            |b, &n| {
                let ch = HybridChannel::new(n / 2);
                b.iter(|| {
                    for i in 0..n as u64 {
                        ch.push(Tiny(i)).unwrap();
                    }
                    for _ in 0..n {
                        ch.pop().unwrap();
                    }
                });
            },
        );
    }

    // 1d — Burst: 100% overflow (capacity=1)
    for n in [256, 1024] {
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(
            BenchmarkId::new("burst_all_overflow", n),
            &n,
            |b, &n| {
                let ch = HybridChannel::new(1);
                b.iter(|| {
                    for i in 0..n as u64 {
                        ch.push(Tiny(i)).unwrap();
                    }
                    for _ in 0..n {
                        ch.pop().unwrap();
                    }
                });
            },
        );
    }

    group.finish();
}

// ===========================================================================
// 2. EMPTY-POP COST
// ===========================================================================

fn empty_pop(c: &mut Criterion) {
    let mut group = c.benchmark_group("empty_pop");

    group.bench_function("empty_channel", |b| {
        let ch: HybridChannel<Tiny> = HybridChannel::new(64);
        b.iter(|| {
            assert!(ch.pop().is_none());
        });
    });

    // Pop from a channel that previously had items (ring drained, overflow drained)
    group.bench_function("after_drain", |b| {
        let ch = HybridChannel::new(64);
        for i in 0..100u64 {
            ch.push(Tiny(i)).unwrap();
        }
        while ch.pop().is_some() {}
        b.iter(|| {
            assert!(ch.pop().is_none());
        });
    });

    group.finish();
}

// ===========================================================================
// 3. PAYLOAD SIZE COMPARISON
// ===========================================================================

fn payload_sizes(c: &mut Criterion) {
    let mut group = c.benchmark_group("payload_size");
    let n = 512usize;
    group.throughput(Throughput::Elements(n as u64));

    // Tiny (8 bytes)
    group.bench_function("tiny_8B", |b| {
        let ch = HybridChannel::new(n);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // Medium (256 bytes)
    group.bench_function("medium_256B", |b| {
        let ch = HybridChannel::new(n);
        b.iter(|| {
            for _ in 0..n {
                ch.push(Medium::default()).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // Large (4096 bytes, boxed)
    group.bench_function("large_4KB_boxed", |b| {
        let ch = HybridChannel::new(n);
        b.iter(|| {
            for _ in 0..n {
                ch.push(Large::default()).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // Box<dyn Any + Send> — what the runtime actually uses
    group.bench_function("boxed_any", |b| {
        let ch: HybridChannel<Box<dyn std::any::Any + Send>> = HybridChannel::new(n);
        b.iter(|| {
            for i in 0..n as u64 {
                let _ = ch.push(Box::new(Tiny(i)));
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    group.finish();
}

// ===========================================================================
// 4. SENDER / RECEIVER API
// ===========================================================================

fn sender_receiver_api(c: &mut Criterion) {
    let mut group = c.benchmark_group("sender_receiver");

    // 4a — Single send + recv
    group.bench_function("single_roundtrip", |b| {
        let rx: Receiver<Tiny> = Receiver::new(64);
        let tx = rx.new_sender();
        b.iter(|| {
            tx.try_send(Tiny(42)).unwrap();
            rx.try_recv().unwrap();
        });
    });

    // 4b — Burst through sender/receiver
    for n in [256, 1024, 4096] {
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(BenchmarkId::new("burst", n), &n, |b, &n| {
            let rx: Receiver<Tiny> = Receiver::new(n);
            let tx = rx.new_sender();
            b.iter(|| {
                for i in 0..n as u64 {
                    tx.try_send(Tiny(i)).unwrap();
                }
                for _ in 0..n {
                    rx.try_recv().unwrap();
                }
            });
        });
    }

    // 4c — Multiple senders (clone cost + contention-free since ST)
    group.bench_function("clone_sender", |b| {
        let rx: Receiver<Tiny> = Receiver::new(64);
        let tx = rx.new_sender();
        b.iter(|| {
            let _tx2 = rx.new_sender();
            tx.try_send(Tiny(1)).unwrap();
            rx.try_recv().unwrap();
        });
    });

    group.finish();
}

// ===========================================================================
// 5. SPSC — one producer thread, one consumer thread
// ===========================================================================

fn spsc(c: &mut Criterion) {
    let mut group = c.benchmark_group("spsc");

    for n in [1_000usize, 10_000, 100_000] {
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(BenchmarkId::new("throughput", n), &n, |b, &n| {
            b.iter_batched(
                || {
                    let rx: Receiver<Tiny> = Receiver::new(1024);
                    let tx = rx.new_sender();
                    (tx, rx)
                },
                |(tx, rx)| {
                    let barrier = Arc::new(Barrier::new(2));

                    let b1 = barrier.clone();
                    let producer = thread::spawn(move || {
                        b1.wait();
                        for i in 0..n as u64 {
                            tx.try_send(Tiny(i)).unwrap();
                        }
                    });

                    let b2 = barrier.clone();
                    let consumer = thread::spawn(move || {
                        b2.wait();
                        let mut received = 0u64;
                        while received < n as u64 {
                            if rx.try_recv().is_some() {
                                received += 1;
                            } else {
                                std::hint::spin_loop();
                            }
                        }
                        received
                    });

                    producer.join().unwrap();
                    let count = consumer.join().unwrap();
                    assert_eq!(count, n as u64);
                },
                BatchSize::PerIteration,
            );
        });
    }

    // SPSC with varying ring capacities (fixed message count)
    let n = 10_000usize;
    group.throughput(Throughput::Elements(n as u64));
    for cap in [16, 64, 256, 1024, 8192] {
        group.bench_with_input(
            BenchmarkId::new("cap_sweep", cap),
            &cap,
            |b, &cap| {
                b.iter_batched(
                    || {
                        let rx: Receiver<Tiny> = Receiver::new(cap);
                        let tx = rx.new_sender();
                        (tx, rx)
                    },
                    |(tx, rx)| {
                        let barrier = Arc::new(Barrier::new(2));

                        let b1 = barrier.clone();
                        let producer = thread::spawn(move || {
                            b1.wait();
                            for i in 0..n as u64 {
                                tx.try_send(Tiny(i)).unwrap();
                            }
                        });

                        let b2 = barrier.clone();
                        let consumer = thread::spawn(move || {
                            b2.wait();
                            let mut received = 0u64;
                            while received < n as u64 {
                                if rx.try_recv().is_some() {
                                    received += 1;
                                } else {
                                    std::hint::spin_loop();
                                }
                            }
                        });

                        producer.join().unwrap();
                        consumer.join().unwrap();
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }

    group.finish();
}

// ===========================================================================
// 6. MPSC — multiple producer threads, one consumer thread
// ===========================================================================

fn mpsc(c: &mut Criterion) {
    let mut group = c.benchmark_group("mpsc");

    // Vary number of producers
    let msgs_per_producer = 5_000usize;
    for num_producers in [2, 4, 8] {
        let total = num_producers * msgs_per_producer;
        group.throughput(Throughput::Elements(total as u64));
        group.bench_with_input(
            BenchmarkId::new("producers", num_producers),
            &num_producers,
            |b, &num_producers| {
                b.iter_batched(
                    || {
                        let rx: Receiver<Tiny> = Receiver::new(1024);
                        let senders: Vec<Sender<Tiny>> =
                            (0..num_producers).map(|_| rx.new_sender()).collect();
                        (senders, rx)
                    },
                    |(senders, rx)| {
                        let barrier = Arc::new(Barrier::new(num_producers + 1));

                        let handles: Vec<_> = senders
                            .into_iter()
                            .map(|tx| {
                                let b = barrier.clone();
                                thread::spawn(move || {
                                    b.wait();
                                    for i in 0..msgs_per_producer as u64 {
                                        tx.try_send(Tiny(i)).unwrap();
                                    }
                                })
                            })
                            .collect();

                        let total = num_producers * msgs_per_producer;
                        let b = barrier.clone();
                        let consumer = thread::spawn(move || {
                            b.wait();
                            let mut received = 0usize;
                            while received < total {
                                if rx.try_recv().is_some() {
                                    received += 1;
                                } else {
                                    std::hint::spin_loop();
                                }
                            }
                            received
                        });

                        for h in handles {
                            h.join().unwrap();
                        }
                        assert_eq!(consumer.join().unwrap(), total);
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }

    // Heavy contention: many producers, small ring
    for cap in [16, 256] {
        let num_producers = 8;
        let total = num_producers * msgs_per_producer;
        group.throughput(Throughput::Elements(total as u64));
        group.bench_with_input(
            BenchmarkId::new("contention_cap", cap),
            &cap,
            |b, &cap| {
                b.iter_batched(
                    || {
                        let rx: Receiver<Tiny> = Receiver::new(cap);
                        let senders: Vec<Sender<Tiny>> =
                            (0..num_producers).map(|_| rx.new_sender()).collect();
                        (senders, rx)
                    },
                    |(senders, rx)| {
                        let barrier = Arc::new(Barrier::new(num_producers + 1));

                        let handles: Vec<_> = senders
                            .into_iter()
                            .map(|tx| {
                                let b = barrier.clone();
                                thread::spawn(move || {
                                    b.wait();
                                    for i in 0..msgs_per_producer as u64 {
                                        tx.try_send(Tiny(i)).unwrap();
                                    }
                                })
                            })
                            .collect();

                        let total = num_producers * msgs_per_producer;
                        let b = barrier.clone();
                        let consumer = thread::spawn(move || {
                            b.wait();
                            let mut received = 0usize;
                            while received < total {
                                if rx.try_recv().is_some() {
                                    received += 1;
                                } else {
                                    std::hint::spin_loop();
                                }
                            }
                            received
                        });

                        for h in handles {
                            h.join().unwrap();
                        }
                        assert_eq!(consumer.join().unwrap(), total);
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }

    group.finish();
}

// ===========================================================================
// 7. OVERFLOW RATIO — measure how overflow % affects throughput
// ===========================================================================

fn overflow_ratio(c: &mut Criterion) {
    let mut group = c.benchmark_group("overflow_ratio");
    let n = 4096usize;
    group.throughput(Throughput::Elements(n as u64));

    // 0% overflow
    group.bench_function("0pct", |b| {
        let ch = HybridChannel::new(n);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // 25% overflow
    group.bench_function("25pct", |b| {
        let ch = HybridChannel::new(n * 3 / 4);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // 50% overflow
    group.bench_function("50pct", |b| {
        let ch = HybridChannel::new(n / 2);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // 75% overflow
    group.bench_function("75pct", |b| {
        let ch = HybridChannel::new(n / 4);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    // 100% overflow
    group.bench_function("100pct", |b| {
        let ch = HybridChannel::new(1);
        b.iter(|| {
            for i in 0..n as u64 {
                ch.push(Tiny(i)).unwrap();
            }
            for _ in 0..n {
                ch.pop().unwrap();
            }
        });
    });

    group.finish();
}

// ===========================================================================
// 8. INTERLEAVED — alternating push/pop (simulates steady-state runtime)
// ===========================================================================

fn interleaved(c: &mut Criterion) {
    let mut group = c.benchmark_group("interleaved");

    // Push 1, pop 1 — the most common runtime pattern
    for cap in [64, 256, 1024] {
        let n = 4096usize;
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(
            BenchmarkId::new("push1_pop1", cap),
            &cap,
            |b, &cap| {
                let ch = HybridChannel::new(cap);
                b.iter(|| {
                    for i in 0..n as u64 {
                        ch.push(Tiny(i)).unwrap();
                        ch.pop().unwrap();
                    }
                });
            },
        );
    }

    // Push K, pop K — batched steady-state
    for k in [4, 16, 64] {
        let rounds = 256usize;
        let n = rounds * k;
        group.throughput(Throughput::Elements(n as u64));
        group.bench_with_input(
            BenchmarkId::new("batch_k", k),
            &k,
            |b, &k| {
                let ch = HybridChannel::new(256);
                b.iter(|| {
                    for _ in 0..rounds {
                        for i in 0..k as u64 {
                            ch.push(Tiny(i)).unwrap();
                        }
                        for _ in 0..k {
                            ch.pop().unwrap();
                        }
                    }
                });
            },
        );
    }

    group.finish();
}

// ===========================================================================
// 9. CHANNEL CREATION COST
// ===========================================================================

fn creation(c: &mut Criterion) {
    let mut group = c.benchmark_group("creation");

    for cap in [16, 64, 256, 1024, 4096] {
        group.bench_with_input(
            BenchmarkId::new("hybrid_channel", cap),
            &cap,
            |b, &cap| {
                b.iter(|| HybridChannel::<Tiny>::new(cap));
            },
        );
    }

    for cap in [16, 64, 256, 1024, 4096] {
        group.bench_with_input(
            BenchmarkId::new("receiver_and_sender", cap),
            &cap,
            |b, &cap| {
                b.iter(|| {
                    let rx: Receiver<Tiny> = Receiver::new(cap);
                    let _tx = rx.new_sender();
                });
            },
        );
    }

    group.finish();
}

// ===========================================================================
// 10. SPSC PINGPONG — latency measurement (two channels, bounce messages)
// ===========================================================================

fn spsc_pingpong(c: &mut Criterion) {
    let mut group = c.benchmark_group("spsc_pingpong");

    for cap in [64, 1024] {
        let rounds = 10_000usize;
        group.throughput(Throughput::Elements(rounds as u64));
        group.bench_with_input(
            BenchmarkId::new("roundtrips", cap),
            &cap,
            |b, &cap| {
                b.iter_batched(
                    || {
                        // Channel A: thread1 → thread2
                        let rx_a: Receiver<Tiny> = Receiver::new(cap);
                        let tx_a = rx_a.new_sender();
                        // Channel B: thread2 → thread1
                        let rx_b: Receiver<Tiny> = Receiver::new(cap);
                        let tx_b = rx_b.new_sender();
                        (tx_a, rx_a, tx_b, rx_b)
                    },
                    |(tx_a, rx_a, tx_b, rx_b)| {
                        let barrier = Arc::new(Barrier::new(2));

                        let b1 = barrier.clone();
                        let t1 = thread::spawn(move || {
                            b1.wait();
                            for i in 0..rounds as u64 {
                                tx_a.try_send(Tiny(i)).unwrap();
                                while rx_b.try_recv().is_none() {
                                    std::hint::spin_loop();
                                }
                            }
                        });

                        let b2 = barrier.clone();
                        let t2 = thread::spawn(move || {
                            b2.wait();
                            for _ in 0..rounds {
                                while rx_a.try_recv().is_none() {
                                    std::hint::spin_loop();
                                }
                                tx_b.try_send(Tiny(0)).unwrap();
                            }
                        });

                        t1.join().unwrap();
                        t2.join().unwrap();
                    },
                    BatchSize::PerIteration,
                );
            },
        );
    }

    group.finish();
}

// ===========================================================================
// Register all groups
// ===========================================================================

criterion_group!(
    benches,
    st_push_pop,
    empty_pop,
    payload_sizes,
    sender_receiver_api,
    spsc,
    mpsc,
    overflow_ratio,
    interleaved,
    creation,
    spsc_pingpong,
);
criterion_main!(benches);