//! gRPC Streaming Throughput Benchmarks
//!
//! Validates streaming performance targets:
//! - Message throughput: >10,000 msg/sec
//! - Stream latency: p99 <1ms
//! - Backpressure handling: graceful degradation
//! - Concurrent stream capacity: >100 streams
//!
//! Critical for real-time market data and order flow.

use criterion::{black_box, criterion_group, criterion_main, BenchmarkId, Criterion, Throughput};
use std::sync::{
    atomic::{AtomicU64, Ordering},
    Arc,
};
use std::time::Duration;

/// Mock streaming message
#[derive(Clone, Debug)]
struct StreamMessage {
    payload: Vec<u8>,
}

impl StreamMessage {
    fn new(_sequence: u64, payload_size: usize) -> Self {
        Self {
            payload: vec![0u8; payload_size],
        }
    }
}

/// Mock streaming channel
struct StreamChannel {
    buffer: Vec<StreamMessage>,
    capacity: usize,
    sent_count: Arc<AtomicU64>,
}

impl StreamChannel {
    fn new(capacity: usize) -> Self {
        Self {
            buffer: Vec::with_capacity(capacity),
            capacity,
            sent_count: Arc::new(AtomicU64::new(0)),
        }
    }

    fn send(&mut self, message: StreamMessage) -> Result<(), &'static str> {
        if self.buffer.len() < self.capacity {
            self.buffer.push(message);
            self.sent_count.fetch_add(1, Ordering::Relaxed);
            Ok(())
        } else {
            Err("Channel full")
        }
    }

    fn receive(&mut self) -> Option<StreamMessage> {
        if !self.buffer.is_empty() {
            Some(self.buffer.remove(0))
        } else {
            None
        }
    }
}

/// Benchmark message throughput
fn bench_message_throughput(c: &mut Criterion) {
    let mut group = c.benchmark_group("message_throughput");

    for payload_size in &[64, 256, 1024, 4096] {
        group.throughput(Throughput::Bytes(*payload_size as u64));
        group.bench_with_input(
            BenchmarkId::new("payload_bytes", payload_size),
            payload_size,
            |b, &size| {
                b.iter_batched(
                    || StreamChannel::new(10000),
                    |mut channel| {
                        for i in 0..1000 {
                            let msg = StreamMessage::new(i, size);
                            let _ = channel.send(msg);
                        }
                        black_box(channel)
                    },
                    criterion::BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

/// Benchmark stream latency (send to receive)
fn bench_stream_latency(c: &mut Criterion) {
    let mut group = c.benchmark_group("stream_latency");
    group.throughput(Throughput::Elements(1));

    group.bench_function("send_receive_latency", |b| {
        b.iter_batched(
            || StreamChannel::new(1000),
            |mut channel| {
                let msg = StreamMessage::new(1, 256);
                let _ = channel.send(msg);
                let received = channel.receive();
                black_box(received)
            },
            criterion::BatchSize::SmallInput,
        );
    });

    group.finish();
}

/// Benchmark backpressure handling
fn bench_backpressure(c: &mut Criterion) {
    let mut group = c.benchmark_group("backpressure_handling");

    for buffer_size in &[100, 1000, 10000] {
        group.bench_with_input(
            BenchmarkId::new("buffer_size", buffer_size),
            buffer_size,
            |b, &size| {
                b.iter_batched(
                    || StreamChannel::new(size),
                    |mut channel| {
                        let mut successful = 0;
                        let mut failed = 0;

                        // Try to send more than capacity
                        for i in 0..(size * 2) {
                            let msg = StreamMessage::new(i as u64, 256);
                            match channel.send(msg) {
                                Ok(_) => successful += 1,
                                Err(_) => failed += 1,
                            }
                        }

                        black_box((successful, failed, channel))
                    },
                    criterion::BatchSize::SmallInput,
                );
            },
        );
    }

    group.finish();
}

/// Benchmark concurrent streams
fn bench_concurrent_streams(c: &mut Criterion) {
    let mut group = c.benchmark_group("concurrent_streams");

    for num_streams in &[10, 50, 100, 200] {
        group.bench_with_input(
            BenchmarkId::new("streams", num_streams),
            num_streams,
            |b, &streams| {
                b.iter(|| {
                    let mut channels: Vec<StreamChannel> = Vec::new();

                    // Create multiple streams
                    for _ in 0..streams {
                        channels.push(StreamChannel::new(1000));
                    }

                    // Send messages to all streams
                    for channel in &mut channels {
                        for i in 0..10 {
                            let msg = StreamMessage::new(i, 256);
                            let _ = channel.send(msg);
                        }
                    }

                    black_box(channels)
                });
            },
        );
    }

    group.finish();
}

/// Benchmark message serialization overhead
fn bench_serialization(c: &mut Criterion) {
    let mut group = c.benchmark_group("message_serialization");

    for payload_size in &[64, 256, 1024] {
        group.throughput(Throughput::Bytes(*payload_size as u64));
        group.bench_with_input(
            BenchmarkId::new("bytes", payload_size),
            payload_size,
            |b, &size| {
                b.iter(|| {
                    let msg = StreamMessage::new(1, size);
                    // Simulate serialization (copy payload)
                    let serialized = msg.payload.clone();
                    black_box(serialized)
                });
            },
        );
    }

    group.finish();
}

/// Benchmark flow control
fn bench_flow_control(c: &mut Criterion) {
    let mut group = c.benchmark_group("flow_control");

    group.bench_function("windowed_send", |b| {
        b.iter(|| {
            let mut channel = StreamChannel::new(1000);
            let window_size = 100;

            // Send in windows with flow control
            for window in 0..10 {
                // Send window
                for i in 0..window_size {
                    let msg = StreamMessage::new(window * window_size + i, 256);
                    let _ = channel.send(msg);
                }

                // Receive window (simulate acknowledgment)
                for _ in 0..window_size {
                    let _ = channel.receive();
                }
            }

            black_box(channel)
        });
    });

    group.bench_function("continuous_send", |b| {
        b.iter(|| {
            let mut channel = StreamChannel::new(1000);

            // Send continuously
            for i in 0..1000 {
                let msg = StreamMessage::new(i, 256);
                let _ = channel.send(msg);
            }

            black_box(channel)
        });
    });

    group.finish();
}

criterion_group! {
    name = streaming_benchmarks;
    config = Criterion::default()
        .measurement_time(Duration::from_secs(10))
        .sample_size(500)
        .warm_up_time(Duration::from_secs(2))
        .with_plots();
    targets =
        bench_message_throughput,
        bench_stream_latency,
        bench_backpressure,
        bench_concurrent_streams,
        bench_serialization,
        bench_flow_control
}

criterion_main!(streaming_benchmarks);

#[cfg(test)]
mod throughput_validation {
    #[allow(unused_imports)]
    use super::*;
    #[allow(unused_imports)]
    use std::time::{Duration, Instant};

    #[test]
    fn validate_message_throughput() {
        let mut channel = StreamChannel::new(100000);
        let message_count = 100000;

        let start = Instant::now();
        for i in 0..message_count {
            let msg = StreamMessage::new(i, 256);
            channel.send(msg).unwrap();
        }
        let elapsed = start.elapsed();

        let messages_per_sec = (message_count as f64 / elapsed.as_secs_f64()) as u64;
        println!("Throughput: {} msg/sec", messages_per_sec);

        // Target: >10,000 msg/sec
        assert!(
            messages_per_sec > 10000,
            "Throughput below target: {} msg/sec (target: >10,000)",
            messages_per_sec
        );
    }

    #[test]
    fn validate_stream_latency() {
        let mut channel = StreamChannel::new(1000);
        let iterations = 10000;

        let start = Instant::now();
        for i in 0..iterations {
            let msg = StreamMessage::new(i, 256);
            channel.send(msg).unwrap();
            let _ = channel.receive();
        }
        let elapsed = start.elapsed();

        let avg_latency_us = elapsed.as_micros() / iterations;
        println!("Average stream latency: {}us", avg_latency_us);

        // Target: p99 <1ms = 1000us
        assert!(
            avg_latency_us < 1000,
            "Stream latency exceeds 1ms target: {}us",
            avg_latency_us
        );
    }

    #[test]
    fn validate_backpressure_handling() {
        let capacity = 1000;
        let mut channel = StreamChannel::new(capacity);

        let mut successful = 0;
        let mut failed = 0;

        // Attempt to send 2x capacity
        for i in 0..(capacity * 2) {
            let msg = StreamMessage::new(i as u64, 256);
            match channel.send(msg) {
                Ok(_) => successful += 1,
                Err(_) => failed += 1,
            }
        }

        assert_eq!(
            successful, capacity,
            "Should accept exactly capacity messages"
        );
        assert_eq!(failed, capacity, "Should reject messages beyond capacity");

        println!(
            "Backpressure: accepted {}, rejected {} (capacity: {})",
            successful, failed, capacity
        );
    }

    #[test]
    fn validate_concurrent_streams() {
        let num_streams = 100;
        let msgs_per_stream = 100;

        let start = Instant::now();
        let mut channels: Vec<StreamChannel> = Vec::new();

        for _ in 0..num_streams {
            let mut channel = StreamChannel::new(1000);
            for i in 0..msgs_per_stream {
                let msg = StreamMessage::new(i, 256);
                channel.send(msg).unwrap();
            }
            channels.push(channel);
        }

        let elapsed = start.elapsed();
        let total_messages = num_streams * msgs_per_stream;

        println!(
            "{} streams, {} total messages in {:?}",
            num_streams, total_messages, elapsed
        );

        assert!(
            elapsed < Duration::from_secs(1),
            "Concurrent stream creation too slow: {:?}",
            elapsed
        );
    }
}