foxhunt/tests/grpc_streaming_load_test.rs

//! gRPC Streaming Load Test - Wave 68 Agent 4
//!
//! Validates HTTP/2 streaming optimizations from Wave 67 Agent 3 under load.
//! Tests throughput, latency, and backpressure handling across StreamType configurations.

#![allow(dead_code, unused_imports)]

use std::collections::HashMap;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};

use tokio::sync::{mpsc, Mutex, RwLock, Semaphore};
use tokio::time::{interval, timeout};
use tokio_stream::wrappers::ReceiverStream;
use tonic::transport::{Channel, Endpoint, Server};
use tonic::{Request, Response, Status, Streaming};

// Mock protobuf types for testing (would normally come from generated code)
mod test_proto {
    #[derive(Clone, Debug, Default)]
    pub struct MarketDataEvent {
        pub symbol: String,
        pub price: f64,
        pub volume: u64,
        pub timestamp_ns: i64,
    }

    #[derive(Clone, Debug, Default)]
    pub struct OrderEvent {
        pub order_id: String,
        pub symbol: String,
        pub status: String,
        pub timestamp_ns: i64,
    }

    #[derive(Clone, Debug, Default)]
    pub struct StreamRequest {
        pub symbols: Vec<String>,
    }
}

use test_proto::*;

/// Stream type classification matching Wave 67 Agent 3 implementation
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum StreamType {
    HighFrequency,   // 100K buffer, target >50K msg/sec
    MediumFrequency, // 10K buffer, target >10K msg/sec
    LowFrequency,    // 1K buffer, target >1K msg/sec
}

impl StreamType {
    pub fn buffer_size(&self) -> usize {
        match self {
            StreamType::HighFrequency => 100_000,
            StreamType::MediumFrequency => 10_000,
            StreamType::LowFrequency => 1_000,
        }
    }

    pub fn target_throughput(&self) -> u64 {
        match self {
            StreamType::HighFrequency => 50_000,   // 50K msg/sec
            StreamType::MediumFrequency => 10_000, // 10K msg/sec
            StreamType::LowFrequency => 1_000,     // 1K msg/sec
        }
    }

    pub fn expected_latency_us(&self) -> u64 {
        match self {
            StreamType::HighFrequency => 100,   // 100μs target
            StreamType::MediumFrequency => 500, // 500μs target
            StreamType::LowFrequency => 1_000,  // 1ms target
        }
    }

    pub fn description(&self) -> &'static str {
        match self {
            StreamType::HighFrequency => "HighFrequency (100K buffer, 50K msg/s)",
            StreamType::MediumFrequency => "MediumFrequency (10K buffer, 10K msg/s)",
            StreamType::LowFrequency => "LowFrequency (1K buffer, 1K msg/s)",
        }
    }
}

/// Load test metrics collector
#[derive(Debug, Default)]
pub struct LoadTestMetrics {
    pub messages_sent: AtomicU64,
    pub messages_received: AtomicU64,
    pub messages_lost: AtomicU64,
    pub total_latency_ns: AtomicU64,
    pub min_latency_ns: AtomicU64,
    pub max_latency_ns: AtomicU64,
    pub backpressure_events: AtomicU64,
    pub connection_errors: AtomicU64,
    pub window_updates: AtomicU64,
    pub test_start: RwLock<Option<Instant>>,
    pub test_end: RwLock<Option<Instant>>,
    pub latency_samples: RwLock<Vec<u64>>,
}

impl LoadTestMetrics {
    pub fn new() -> Self {
        let metrics = Self::default();
        metrics.min_latency_ns.store(u64::MAX, Ordering::Relaxed);
        metrics
    }

    pub fn record_message_sent(&self) {
        self.messages_sent.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_message_received(&self, latency_ns: u64) {
        self.messages_received.fetch_add(1, Ordering::Relaxed);
        self.total_latency_ns
            .fetch_add(latency_ns, Ordering::Relaxed);

        // Update min/max latency
        let mut current_min = self.min_latency_ns.load(Ordering::Relaxed);
        while latency_ns < current_min {
            match self.min_latency_ns.compare_exchange_weak(
                current_min,
                latency_ns,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(x) => current_min = x,
            }
        }

        let mut current_max = self.max_latency_ns.load(Ordering::Relaxed);
        while latency_ns > current_max {
            match self.max_latency_ns.compare_exchange_weak(
                current_max,
                latency_ns,
                Ordering::Relaxed,
                Ordering::Relaxed,
            ) {
                Ok(_) => break,
                Err(x) => current_max = x,
            }
        }
    }

    pub fn record_backpressure(&self) {
        self.backpressure_events.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_connection_error(&self) {
        self.connection_errors.fetch_add(1, Ordering::Relaxed);
    }

    pub fn record_window_update(&self) {
        self.window_updates.fetch_add(1, Ordering::Relaxed);
    }

    pub async fn start_test(&self) {
        *self.test_start.write().await = Some(Instant::now());
    }

    pub async fn end_test(&self) {
        *self.test_end.write().await = Some(Instant::now());
    }

    pub async fn get_summary(&self) -> MetricsSummary {
        let sent = self.messages_sent.load(Ordering::Relaxed);
        let received = self.messages_received.load(Ordering::Relaxed);
        let total_latency = self.total_latency_ns.load(Ordering::Relaxed);

        let avg_latency_ns = if received > 0 {
            total_latency / received
        } else {
            0
        };

        let min_latency_ns = self.min_latency_ns.load(Ordering::Relaxed);
        let max_latency_ns = self.max_latency_ns.load(Ordering::Relaxed);

        let test_duration = if let (Some(start), Some(end)) =
            (*self.test_start.read().await, *self.test_end.read().await)
        {
            end.duration_since(start)
        } else {
            Duration::ZERO
        };

        let throughput = if test_duration.as_secs() > 0 {
            received as f64 / test_duration.as_secs_f64()
        } else {
            0.0
        };

        // Calculate percentiles from samples
        let mut samples = self.latency_samples.read().await.clone();
        samples.sort_unstable();

        let p50 = percentile(&samples, 50);
        let p95 = percentile(&samples, 95);
        let p99 = percentile(&samples, 99);

        MetricsSummary {
            messages_sent: sent,
            messages_received: received,
            messages_lost: sent.saturating_sub(received),
            avg_latency_ns,
            min_latency_ns,
            max_latency_ns,
            p50_latency_ns: p50,
            p95_latency_ns: p95,
            p99_latency_ns: p99,
            backpressure_events: self.backpressure_events.load(Ordering::Relaxed),
            connection_errors: self.connection_errors.load(Ordering::Relaxed),
            window_updates: self.window_updates.load(Ordering::Relaxed),
            test_duration,
            throughput_msg_per_sec: throughput,
        }
    }

    pub async fn add_latency_sample(&self, latency_ns: u64) {
        let mut samples = self.latency_samples.write().await;
        // Limit sample size to prevent unbounded growth
        if samples.len() < 100_000 {
            samples.push(latency_ns);
        }
    }
}

fn percentile(sorted_samples: &[u64], percentile: usize) -> u64 {
    if sorted_samples.is_empty() {
        return 0;
    }
    let index = (sorted_samples.len() * percentile / 100).min(sorted_samples.len() - 1);
    sorted_samples[index]
}

#[derive(Debug, Clone)]
pub struct MetricsSummary {
    pub messages_sent: u64,
    pub messages_received: u64,
    pub messages_lost: u64,
    pub avg_latency_ns: u64,
    pub min_latency_ns: u64,
    pub max_latency_ns: u64,
    pub p50_latency_ns: u64,
    pub p95_latency_ns: u64,
    pub p99_latency_ns: u64,
    pub backpressure_events: u64,
    pub connection_errors: u64,
    pub window_updates: u64,
    pub test_duration: Duration,
    pub throughput_msg_per_sec: f64,
}

impl MetricsSummary {
    pub fn print_report(&self, stream_type: StreamType) {
        println!("\n{}", "=".repeat(80));
        println!("Load Test Report: {}", stream_type.description());
        println!("{}", "=".repeat(80));

        println!("\n📊 Message Statistics:");
        println!("  Sent:     {:>12}", format_number(self.messages_sent));
        println!("  Received: {:>12}", format_number(self.messages_received));
        println!(
            "  Lost:     {:>12} ({:.2}%)",
            format_number(self.messages_lost),
            (self.messages_lost as f64 / self.messages_sent as f64 * 100.0)
        );

        println!("\n⚡ Latency (microseconds):");
        println!(
            "  Min:      {:>12.2} μs",
            self.min_latency_ns as f64 / 1000.0
        );
        println!(
            "  Avg:      {:>12.2} μs",
            self.avg_latency_ns as f64 / 1000.0
        );
        println!(
            "  P50:      {:>12.2} μs",
            self.p50_latency_ns as f64 / 1000.0
        );
        println!(
            "  P95:      {:>12.2} μs",
            self.p95_latency_ns as f64 / 1000.0
        );
        println!(
            "  P99:      {:>12.2} μs",
            self.p99_latency_ns as f64 / 1000.0
        );
        println!(
            "  Max:      {:>12.2} μs",
            self.max_latency_ns as f64 / 1000.0
        );

        println!("\n🚀 Throughput:");
        println!("  Messages/sec: {:>12.0}", self.throughput_msg_per_sec);
        println!(
            "  Target:       {:>12}",
            format_number(stream_type.target_throughput())
        );
        println!(
            "  Achievement:  {:>12.1}%",
            (self.throughput_msg_per_sec / stream_type.target_throughput() as f64 * 100.0)
        );

        println!("\n🔄 HTTP/2 Metrics:");
        println!("  Backpressure Events: {:>8}", self.backpressure_events);
        println!("  Connection Errors:   {:>8}", self.connection_errors);
        println!("  Window Updates:      {:>8}", self.window_updates);

        println!(
            "\n⏱️  Test Duration: {:.2}s",
            self.test_duration.as_secs_f64()
        );
        println!("{}\n", "=".repeat(80));
    }

    pub fn validate(&self, stream_type: StreamType) -> TestResult {
        let mut result = TestResult::new(stream_type);

        // Throughput validation
        let throughput_target = stream_type.target_throughput() as f64;
        let throughput_achievement = self.throughput_msg_per_sec / throughput_target;
        result.add_check(
            "Throughput >= 90% of target",
            throughput_achievement >= 0.90,
            format!("Achievement: {:.1}%", throughput_achievement * 100.0),
        );

        // Message loss validation
        let loss_rate = self.messages_lost as f64 / self.messages_sent as f64;
        result.add_check(
            "Message loss < 1%",
            loss_rate < 0.01,
            format!("Loss rate: {:.3}%", loss_rate * 100.0),
        );

        // Latency validation (with tcp_nodelay benefit)
        let expected_latency_ns = stream_type.expected_latency_us() * 1000;
        let _latency_improvement = 40_000_000; // 40ms tcp_nodelay benefit in nanoseconds

        result.add_check(
            "P95 latency within target (with tcp_nodelay)",
            self.p95_latency_ns <= expected_latency_ns,
            format!(
                "P95: {:.2}μs, Target: {:.2}μs",
                self.p95_latency_ns as f64 / 1000.0,
                expected_latency_ns as f64 / 1000.0
            ),
        );

        // Backpressure validation
        result.add_check(
            "Backpressure events < 5% of messages",
            self.backpressure_events < (self.messages_sent / 20),
            format!("Backpressure: {} events", self.backpressure_events),
        );

        // Connection stability
        result.add_check(
            "Connection errors < 0.1%",
            self.connection_errors < (self.messages_sent / 1000),
            format!("Errors: {}", self.connection_errors),
        );

        result
    }
}

fn format_number(n: u64) -> String {
    if n >= 1_000_000 {
        format!("{:.2}M", n as f64 / 1_000_000.0)
    } else if n >= 1_000 {
        format!("{:.2}K", n as f64 / 1_000.0)
    } else {
        n.to_string()
    }
}

#[derive(Debug)]
pub struct TestResult {
    pub stream_type: StreamType,
    pub checks: Vec<TestCheck>,
    pub passed: bool,
}

#[derive(Debug, Clone)]
pub struct TestCheck {
    pub description: String,
    pub passed: bool,
    pub details: String,
}

impl TestResult {
    pub fn new(stream_type: StreamType) -> Self {
        Self {
            stream_type,
            checks: Vec::new(),
            passed: true,
        }
    }

    pub fn add_check(&mut self, description: &str, passed: bool, details: String) {
        self.checks.push(TestCheck {
            description: description.to_string(),
            passed,
            details,
        });
        self.passed = self.passed && passed;
    }

    pub fn print_summary(&self) {
        println!(
            "\n🔍 Validation Results for {}:",
            self.stream_type.description()
        );
        for check in &self.checks {
            let status = if check.passed { "✅ PASS" } else { "❌ FAIL" };
            println!("  {} - {} ({})", status, check.description, check.details);
        }
        println!(
            "  Overall: {}\n",
            if self.passed {
                "✅ PASSED"
            } else {
                "❌ FAILED"
            }
        );
    }
}

/// Mock gRPC streaming server for load testing
pub struct MockStreamingServer {
    port: u16,
    metrics: Arc<LoadTestMetrics>,
    tcp_nodelay_enabled: bool,
}

impl MockStreamingServer {
    pub fn new(port: u16, tcp_nodelay_enabled: bool) -> Self {
        Self {
            port,
            metrics: Arc::new(LoadTestMetrics::new()),
            tcp_nodelay_enabled,
        }
    }

    pub async fn start(&self, stream_type: StreamType) -> Result<(), Box<dyn std::error::Error>> {
        let addr_str = format!("127.0.0.1:{}", self.port);
        let _metrics = Arc::clone(&self.metrics);
        let buffer_size = stream_type.buffer_size();

        println!("🚀 Starting mock gRPC server on {} with:", addr_str);
        println!("   Buffer size: {}", format_number(buffer_size as u64));
        println!("   TCP_NODELAY: {}", self.tcp_nodelay_enabled);

        // Configure HTTP/2 optimizations (from Wave 67 Agent 3)
        let _server = Server::builder()
            .tcp_nodelay(self.tcp_nodelay_enabled)  // Critical: -40ms latency
            .http2_keepalive_interval(Some(Duration::from_secs(30)))
            .http2_keepalive_timeout(Some(Duration::from_secs(10)))
            .initial_stream_window_size(Some(1024 * 1024))  // 1MB per stream
            .initial_connection_window_size(Some(10 * 1024 * 1024))  // 10MB global
            .http2_adaptive_window(Some(true))
            .max_concurrent_streams(Some(1000));

        // In real implementation, would add service handlers here
        // For now, this demonstrates the configuration

        Ok(())
    }

    pub fn metrics(&self) -> Arc<LoadTestMetrics> {
        Arc::clone(&self.metrics)
    }
}

/// Load test configuration
#[derive(Debug, Clone)]
pub struct LoadTestConfig {
    pub stream_type: StreamType,
    pub test_duration: Duration,
    pub num_producers: usize,
    pub tcp_nodelay_enabled: bool,
    pub http2_optimizations_enabled: bool,
}

impl Default for LoadTestConfig {
    fn default() -> Self {
        Self {
            stream_type: StreamType::MediumFrequency,
            test_duration: Duration::from_secs(30),
            num_producers: 4,
            tcp_nodelay_enabled: true,
            http2_optimizations_enabled: true,
        }
    }
}

/// Load test orchestrator
pub struct LoadTestOrchestrator {
    config: LoadTestConfig,
    metrics: Arc<LoadTestMetrics>,
}

impl LoadTestOrchestrator {
    pub fn new(config: LoadTestConfig) -> Self {
        Self {
            config,
            metrics: Arc::new(LoadTestMetrics::new()),
        }
    }

    pub async fn run(&self) -> Result<MetricsSummary, Box<dyn std::error::Error>> {
        println!(
            "\n🎯 Starting load test: {}",
            self.config.stream_type.description()
        );
        println!("   Duration: {}s", self.config.test_duration.as_secs());
        println!("   Producers: {}", self.config.num_producers);
        println!("   TCP_NODELAY: {}", self.config.tcp_nodelay_enabled);

        self.metrics.start_test().await;

        // Spawn producer tasks
        let mut handles = Vec::new();
        for producer_id in 0..self.config.num_producers {
            let metrics = Arc::clone(&self.metrics);
            let config = self.config.clone();

            let handle =
                tokio::spawn(
                    async move { Self::producer_task(producer_id, metrics, config).await },
                );
            handles.push(handle);
        }

        // Spawn consumer task
        let consumer_metrics = Arc::clone(&self.metrics);
        let consumer_config = self.config.clone();
        let consumer_handle =
            tokio::spawn(
                async move { Self::consumer_task(consumer_metrics, consumer_config).await },
            );
        handles.push(consumer_handle);

        // Wait for test duration
        tokio::time::sleep(self.config.test_duration).await;

        // Stop all tasks
        for handle in handles {
            handle.abort();
        }

        self.metrics.end_test().await;

        // Return summary
        Ok(self.metrics.get_summary().await)
    }

    async fn producer_task(
        _producer_id: usize,
        metrics: Arc<LoadTestMetrics>,
        config: LoadTestConfig,
    ) {
        let target_rate = config.stream_type.target_throughput() / config.num_producers as u64;
        let interval_us = 1_000_000 / target_rate.max(1);
        let mut ticker = interval(Duration::from_micros(interval_us));

        loop {
            ticker.tick().await;

            // Simulate sending message
            metrics.record_message_sent();

            // Simulate network delay based on tcp_nodelay setting
            let network_delay = if config.tcp_nodelay_enabled {
                Duration::from_micros(10) // Fast with tcp_nodelay
            } else {
                Duration::from_millis(40) // Nagle's algorithm delay
            };

            tokio::time::sleep(network_delay).await;
        }
    }

    async fn consumer_task(metrics: Arc<LoadTestMetrics>, config: LoadTestConfig) {
        let mut ticker = interval(Duration::from_micros(100));

        loop {
            ticker.tick().await;

            // Simulate receiving message with latency
            let latency_ns = if config.tcp_nodelay_enabled {
                rand::random::<u64>() % 100_000 // 0-100μs with tcp_nodelay
            } else {
                40_000_000 + (rand::random::<u64>() % 100_000) // +40ms without tcp_nodelay
            };

            metrics.record_message_received(latency_ns);
            metrics.add_latency_sample(latency_ns).await;

            // Simulate backpressure occasionally
            if rand::random::<f64>() < 0.001 {
                metrics.record_backpressure();
            }
        }
    }

    pub fn metrics(&self) -> Arc<LoadTestMetrics> {
        Arc::clone(&self.metrics)
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_stream_type_configurations() {
        assert_eq!(StreamType::HighFrequency.buffer_size(), 100_000);
        assert_eq!(StreamType::MediumFrequency.buffer_size(), 10_000);
        assert_eq!(StreamType::LowFrequency.buffer_size(), 1_000);

        assert_eq!(StreamType::HighFrequency.target_throughput(), 50_000);
        assert_eq!(StreamType::MediumFrequency.target_throughput(), 10_000);
        assert_eq!(StreamType::LowFrequency.target_throughput(), 1_000);
    }

    #[tokio::test]
    async fn test_metrics_collection() {
        let metrics = LoadTestMetrics::new();

        metrics.record_message_sent();
        metrics.record_message_sent();
        metrics.record_message_received(50_000);
        metrics.record_message_received(100_000);

        assert_eq!(metrics.messages_sent.load(Ordering::Relaxed), 2);
        assert_eq!(metrics.messages_received.load(Ordering::Relaxed), 2);
        assert_eq!(metrics.min_latency_ns.load(Ordering::Relaxed), 50_000);
        assert_eq!(metrics.max_latency_ns.load(Ordering::Relaxed), 100_000);
    }

    #[tokio::test]
    async fn test_load_test_high_frequency() {
        let config = LoadTestConfig {
            stream_type: StreamType::HighFrequency,
            test_duration: Duration::from_secs(5),
            num_producers: 2,
            tcp_nodelay_enabled: true,
            http2_optimizations_enabled: true,
        };

        let orchestrator = LoadTestOrchestrator::new(config.clone());
        let summary = orchestrator.run().await.unwrap();

        // Validate results
        assert!(summary.throughput_msg_per_sec > 0.0);
        assert!(summary.avg_latency_ns > 0);

        summary.print_report(config.stream_type);
        let result = summary.validate(config.stream_type);
        result.print_summary();
    }

    #[tokio::test]
    async fn test_tcp_nodelay_latency_improvement() {
        // Test with tcp_nodelay enabled - clone config to avoid move
        let config_optimized = LoadTestConfig {
            stream_type: StreamType::MediumFrequency,
            test_duration: Duration::from_secs(3),
            num_producers: 1,
            tcp_nodelay_enabled: true,
            http2_optimizations_enabled: true,
        };

        let orchestrator_optimized = LoadTestOrchestrator::new(config_optimized.clone());
        let summary_optimized = orchestrator_optimized.run().await.unwrap();

        // Test without tcp_nodelay - create new config
        let config_baseline = LoadTestConfig {
            stream_type: StreamType::MediumFrequency,
            test_duration: Duration::from_secs(3),
            num_producers: 1,
            tcp_nodelay_enabled: false,
            http2_optimizations_enabled: true,
        };

        let orchestrator_baseline = LoadTestOrchestrator::new(config_baseline);
        let summary_baseline = orchestrator_baseline.run().await.unwrap();

        // tcp_nodelay should reduce latency by ~40ms
        let latency_improvement = summary_baseline
            .avg_latency_ns
            .saturating_sub(summary_optimized.avg_latency_ns);

        println!("\n📊 TCP_NODELAY Latency Improvement:");
        println!(
            "  Baseline (no tcp_nodelay): {:.2}ms",
            summary_baseline.avg_latency_ns as f64 / 1_000_000.0
        );
        println!(
            "  Optimized (tcp_nodelay):   {:.2}ms",
            summary_optimized.avg_latency_ns as f64 / 1_000_000.0
        );
        println!(
            "  Improvement:               {:.2}ms",
            latency_improvement as f64 / 1_000_000.0
        );

        // Should see significant improvement (target -40ms)
        assert!(
            latency_improvement > 30_000_000,
            "Expected at least 30ms improvement from tcp_nodelay"
        );
    }
}