//! # Streaming Performance Benchmarks
//!
//! Comprehensive benchmarks to validate sub-millisecond latency and
//! high-throughput streaming requirements.

use criterion::{BenchmarkId, Criterion, black_box, criterion_group, criterion_main};
use rtx_streaming::{StreamingConfig, StreamingServer};
use std::time::{Duration, Instant};
use tokio::runtime::Runtime;

/// Benchmark sub-millisecond latency requirement
fn bench_sub_millisecond_latency(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("latency");
    group.measurement_time(Duration::from_secs(10));
    group.sample_size(1000);

    // Setup streaming server
    let server = rt.block_on(async {
        let config = StreamingConfig {
            target_latency: Duration::from_micros(500), // 0.5ms target
            ..Default::default()
        };
        StreamingServer::new(config).await.unwrap()
    });

    group.bench_function("single_token_generation", |b| {
        b.to_async(&rt).iter(|| async {
            let start = Instant::now();
            let _result = server.stream_inference(black_box("Hello world")).await;
            let latency = start.elapsed();

            // Assert sub-millisecond latency
            assert!(
                latency < Duration::from_millis(1),
                "Latency {} exceeds 1ms requirement",
                latency.as_micros()
            );

            latency
        });
    });

    // Test different prompt lengths
    for prompt_length in [10, 50, 100, 500].iter() {
        let prompt = "test ".repeat(*prompt_length);
        group.bench_with_input(
            BenchmarkId::new("prompt_length", prompt_length),
            prompt_length,
            |b, _| {
                b.to_async(&rt).iter(|| async {
                    let start = Instant::now();
                    let _result = server.stream_inference(black_box(&prompt)).await;
                    let latency = start.elapsed();

                    // Latency should remain sub-millisecond regardless of prompt length
                    assert!(latency < Duration::from_millis(1));

                    latency
                });
            },
        );
    }

    group.finish();
}

/// Benchmark high throughput (>1000 QPS) requirement
fn bench_high_throughput(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("throughput");
    group.measurement_time(Duration::from_secs(30));
    group.sample_size(100);

    let server = rt.block_on(async {
        let config = StreamingConfig {
            max_connections: 2000,
            ..Default::default()
        };
        StreamingServer::new(config).await.unwrap()
    });

    // Test concurrent streaming connections
    for concurrent_connections in [100, 500, 1000, 1500].iter() {
        group.bench_with_input(
            BenchmarkId::new("concurrent_connections", concurrent_connections),
            concurrent_connections,
            |b, &conn_count| {
                b.to_async(&rt).iter(|| async {
                    let start = Instant::now();

                    let mut handles = Vec::new();
                    for i in 0..conn_count {
                        let server_clone = server.clone();
                        let handle = tokio::spawn(async move {
                            server_clone
                                .stream_inference(&format!("prompt_{}", i))
                                .await
                        });
                        handles.push(handle);
                    }

                    // Wait for all to complete
                    for handle in handles {
                        handle.await.unwrap().unwrap();
                    }

                    let elapsed = start.elapsed();
                    let qps = conn_count as f64 / elapsed.as_secs_f64();

                    // Assert >1000 QPS requirement
                    if conn_count >= 1000 {
                        assert!(
                            qps > 1000.0,
                            "Throughput {} QPS is below 1000 QPS requirement",
                            qps
                        );
                    }

                    qps
                });
            },
        );
    }

    group.finish();
}

/// Benchmark memory efficiency (<10% overhead per connection)
fn bench_memory_efficiency(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("memory_efficiency");
    group.measurement_time(Duration::from_secs(15));
    group.sample_size(50);

    let server = rt.block_on(async {
        let config = StreamingConfig::default();
        StreamingServer::new(config).await.unwrap()
    });

    // Measure memory overhead per connection
    for connection_count in [10, 50, 100, 500].iter() {
        group.bench_with_input(
            BenchmarkId::new("connection_count", connection_count),
            connection_count,
            |b, &conn_count| {
                b.to_async(&rt).iter(|| async {
                    // Measure baseline memory
                    let baseline_memory = get_memory_usage();

                    // Create connections
                    let mut connections = Vec::new();
                    for i in 0..conn_count {
                        let connection = server
                            .create_connection(&format!("client_{}", i))
                            .await
                            .unwrap();
                        connections.push(connection);
                    }

                    let streaming_memory = get_memory_usage();
                    let overhead_per_connection = (streaming_memory - baseline_memory) / conn_count;
                    let overhead_percentage =
                        (overhead_per_connection as f64 / baseline_memory as f64) * 100.0;

                    // Assert <10% overhead per connection
                    assert!(
                        overhead_percentage < 10.0,
                        "Memory overhead {}% exceeds 10% requirement",
                        overhead_percentage
                    );

                    overhead_percentage
                });
            },
        );
    }

    group.finish();
}

/// Benchmark connection stability (zero drops)
fn bench_connection_stability(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("connection_stability");
    group.measurement_time(Duration::from_secs(20));
    group.sample_size(100);

    let server = rt.block_on(async {
        let config = StreamingConfig::default();
        StreamingServer::new(config).await.unwrap()
    });

    group.bench_function("connection_stability_test", |b| {
        b.to_async(&rt).iter(|| async {
            let mut connection_drops = 0;
            let total_connections = black_box(1000);

            for i in 0..total_connections {
                match server
                    .create_connection(&format!("stable_client_{}", i))
                    .await
                {
                    Ok(_) => {
                        // Simulate some streaming activity
                        tokio::time::sleep(Duration::from_micros(100)).await;
                    }
                    Err(_) => {
                        connection_drops += 1;
                    }
                }
            }

            // Assert zero connection drops
            assert_eq!(
                connection_drops, 0,
                "Had {} connection drops, expected 0",
                connection_drops
            );

            connection_drops
        });
    });

    group.finish();
}

/// Benchmark backpressure handling
fn bench_backpressure_handling(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("backpressure");
    group.measurement_time(Duration::from_secs(10));
    group.sample_size(50);

    let server = rt.block_on(async {
        let config = StreamingConfig {
            backpressure_threshold: 0.7,
            ..Default::default()
        };
        StreamingServer::new(config).await.unwrap()
    });

    group.bench_function("overload_scenario", |b| {
        b.to_async(&rt).iter(|| async {
            // Create overload scenario
            let result = server.handle_overload_scenario().await;

            // Assert graceful handling
            assert!(
                result.is_ok(),
                "Backpressure handling failed: {:?}",
                result.err()
            );

            let handled_gracefully = result.unwrap();
            assert!(
                handled_gracefully,
                "System did not handle overload gracefully"
            );

            handled_gracefully
        });
    });

    group.finish();
}

/// Benchmark end-to-end streaming performance
fn bench_end_to_end_streaming(c: &mut Criterion) {
    let rt = Runtime::new().unwrap();

    let mut group = c.benchmark_group("end_to_end");
    group.measurement_time(Duration::from_secs(30));
    group.sample_size(200);

    let server = rt.block_on(async {
        let config = StreamingConfig {
            target_latency: Duration::from_micros(800),
            max_connections: 1500,
            ..Default::default()
        };
        StreamingServer::new(config).await.unwrap()
    });

    group.bench_function("realistic_workload", |b| {
        b.to_async(&rt).iter(|| async {
            let start = Instant::now();

            // Simulate realistic mixed workload
            let mut handles = Vec::new();

            // 70% normal requests
            for i in 0..700 {
                let server_clone = server.clone();
                let handle = tokio::spawn(async move {
                    server_clone
                        .stream_inference(&format!("normal_prompt_{}", i))
                        .await
                });
                handles.push(handle);
            }

            // 20% concurrent batch requests
            for i in 0..200 {
                let server_clone = server.clone();
                let handle = tokio::spawn(async move {
                    server_clone
                        .stream_inference(&format!("batch_prompt_{}", i))
                        .await
                });
                handles.push(handle);
            }

            // 10% high-priority requests
            for i in 0..100 {
                let server_clone = server.clone();
                let handle = tokio::spawn(async move {
                    server_clone
                        .stream_inference(&format!("priority_prompt_{}", i))
                        .await
                });
                handles.push(handle);
            }

            // Wait for all to complete
            for handle in handles {
                handle.await.unwrap().unwrap();
            }

            let elapsed = start.elapsed();
            let total_requests = 1000;
            let qps = total_requests as f64 / elapsed.as_secs_f64();

            // Validate performance requirements
            assert!(qps > 1000.0, "End-to-end QPS {} below requirement", qps);

            let avg_latency = elapsed / total_requests;
            assert!(
                avg_latency < Duration::from_millis(1),
                "Average latency {} exceeds requirement",
                avg_latency.as_micros()
            );

            (qps, avg_latency)
        });
    });

    group.finish();
}

/// Mock helper function for memory measurement
fn get_memory_usage() -> usize {
    // In real implementation, this would use system APIs
    // For benchmarking, return a realistic mock value
    1024 * 1024 * 50 // 50MB baseline
}

criterion_group!(
    streaming_benches,
    bench_sub_millisecond_latency,
    bench_high_throughput,
    bench_memory_efficiency,
    bench_connection_stability,
    bench_backpressure_handling,
    bench_end_to_end_streaming
);

criterion_main!(streaming_benches);