rustytorch/crates/production/rtx-streaming/tests/integration_tests.rs

//! # Comprehensive Integration Tests
//!
//! Production-grade integration tests for the rtx-streaming crate covering
//! all major components: real-time inference, stream processing, message queues,
//! adaptive processing, state management, edge computing, and monitoring.

use rtx_streaming::stream_processing::{StreamProcessor, StreamProcessorConfig};
use rtx_streaming::*;
use std::{
    collections::HashMap,
    sync::Arc,
    time::{Duration, Instant, SystemTime},
};
use tokio::{
    sync::RwLock,
    time::{sleep, timeout},
};
use uuid::Uuid;

/// Test suite for comprehensive streaming functionality
#[cfg(test)]
mod integration_tests {
    use super::*;

    /// Integration test 1: End-to-end streaming pipeline with sub-millisecond latency
    #[tokio::test]
    async fn test_end_to_end_streaming_pipeline() {
        // Setup comprehensive streaming configuration
        let config = create_test_streaming_config();

        // Initialize all major components
        let _streaming_server = StreamingServer::new(config.clone())
            .await
            .expect("Failed to create streaming server");

        let realtime_pipeline = RealtimePipeline::new(PipelineConfig::default())
            .expect("Failed to create realtime pipeline");

        let monitoring_system = MonitoringSystem::new(MonitoringConfig::default())
            .await
            .expect("Failed to create monitoring system");

        // Start the complete pipeline
        let start_time = Instant::now();

        // Create test events for processing
        let test_events = create_test_events(100);

        // Process events through the complete pipeline
        for event in test_events {
            let pipeline_start = Instant::now();

            // Process through realtime pipeline
            let _pipeline_results = realtime_pipeline
                .process_event(event.clone())
                .expect("Pipeline processing failed");

            // Record monitoring metrics
            monitoring_system
                .record_stream_metrics(&[event])
                .await
                .expect("Failed to record metrics");

            // Verify sub-millisecond processing
            let processing_latency = pipeline_start.elapsed();
            assert!(
                processing_latency < Duration::from_millis(1),
                "Processing latency {} exceeds 1ms requirement",
                processing_latency.as_micros()
            );
        }

        let total_time = start_time.elapsed();
        println!(
            "End-to-end pipeline processed 100 events in {:?}",
            total_time
        );

        // Verify overall performance meets requirements
        assert!(
            total_time < Duration::from_millis(100),
            "Total processing time too high"
        );
    }

    /// Integration test 2: High-throughput concurrent streaming with backpressure
    #[tokio::test]
    async fn test_high_throughput_concurrent_streaming() {
        let config = create_test_streaming_config();
        let mut streaming_server = StreamingServer::new(config).await.unwrap();
        let adaptive_processor = Arc::new(
            AdaptiveProcessor::new(AdaptiveProcessingConfig::default())
                .await
                .unwrap(),
        );

        // Start systems
        streaming_server
            .start()
            .await
            .expect("Failed to start streaming server");
        adaptive_processor
            .start()
            .await
            .expect("Failed to start adaptive processor");

        let start_time = Instant::now();
        let mut handles = Vec::new();

        // Create 1000 concurrent streams
        for stream_id in 0..1000 {
            let _server = streaming_server.clone();
            let processor = adaptive_processor.clone();

            let handle = tokio::spawn(async move {
                let events = create_test_events(10);

                // Test concurrent processing with backpressure handling
                for event in events {
                    let result = timeout(
                        Duration::from_millis(10),
                        processor
                            .process_adaptive_batch(vec![event], &format!("stream_{}", stream_id)),
                    )
                    .await;

                    match result {
                        Ok(Ok(_)) => {}  // Success
                        Ok(Err(_)) => {} // Expected backpressure
                        Err(_) => panic!("Processing timeout"),
                    }
                }

                stream_id
            });

            handles.push(handle);
        }

        // Wait for all concurrent streams to complete
        let mut completed_streams = 0;
        for handle in handles {
            match handle.await {
                Ok(_) => completed_streams += 1,
                Err(e) => println!("Stream failed: {}", e),
            }
        }

        let elapsed = start_time.elapsed();
        let throughput = (completed_streams as f64 * 10.0) / elapsed.as_secs_f64();

        println!(
            "Processed {} streams with throughput: {} events/sec",
            completed_streams, throughput
        );

        // Verify high throughput achievement
        assert!(
            throughput > 1000.0,
            "Throughput {} below requirement",
            throughput
        );
        assert!(
            completed_streams >= 950,
            "Too many streams failed: {}",
            completed_streams
        );

        // Stop systems
        adaptive_processor
            .stop()
            .await
            .expect("Failed to stop adaptive processor");
        // streaming_server.shutdown().await.expect("Failed to shutdown server");
    }

    /// Integration test 3: Stream processing with windowing and watermarks
    #[tokio::test]
    async fn test_stream_processing_windowing_watermarks() {
        let stream_processor = StreamProcessor::new(StreamProcessorConfig::default())
            .await
            .expect("Failed to create stream processor");

        // Create time-series events for windowing
        let mut events = Vec::new();
        let base_time = SystemTime::now();

        for i in 0..50 {
            let event_time = base_time + Duration::from_millis(i * 100);
            let event_id = Uuid::new_v4();
            events.push(StreamEvent {
                event_id,
                event_time,
                processing_time: SystemTime::now(),
                stream_id: "windowing_test".to_string(),
                data: EventData::TimeSeries {
                    values: vec![i as f64],
                    timestamp: event_time,
                },
                metadata: HashMap::new(),
                watermark: if i % 10 == 9 {
                    Some(Watermark {
                        timestamp: event_time,
                        stream_id: "windowing_test".to_string(),
                    })
                } else {
                    None
                },
                id: event_id.to_string(),
                timestamp: event_time,
                partition_key: Some("windowing_test".to_string()),
                sequence_number: i,
            });
        }

        // Process events with windowing
        let start_time = Instant::now();
        let windowed_results = stream_processor
            .process_windowed_events(events)
            .await
            .expect("Windowed processing failed");
        let processing_time = start_time.elapsed();

        // Verify windowing performance
        assert!(
            processing_time < Duration::from_millis(50),
            "Windowing too slow"
        );
        assert!(!windowed_results.is_empty(), "No windowed results produced");

        println!("Windowed processing completed in {:?}", processing_time);
    }

    /// Integration test 4: Message queue integration with Kafka/Redis simulation
    #[tokio::test]
    async fn test_message_queue_integration() {
        let mq_manager = MessageQueueManager::new(MessageQueueConfig::default())
            .expect("Failed to create message queue manager");

        // Test message production
        let mut sent_messages = Vec::new();
        for i in 0..100 {
            let message = OutgoingMessage {
                message_id: Uuid::new_v4().to_string(),
                topic: "test_topic".to_string(),
                key: Some(format!("key_{}", i).as_bytes().to_vec()),
                value: format!("test_message_{}", i).as_bytes().to_vec(),
                headers: HashMap::new(),
                timestamp: SystemTime::now(),
            };

            let start_time = Instant::now();
            match mq_manager.send_message(message.clone()).await {
                Ok(_) => {
                    let send_latency = start_time.elapsed();
                    assert!(
                        send_latency < Duration::from_millis(10),
                        "Send latency too high"
                    );
                    sent_messages.push(message);
                }
                Err(e) => {
                    // In real test with actual message queues, this might fail
                    // For now, we accept simulated behavior
                    println!("Message send simulation: {:?}", e);
                }
            }
        }

        // Test message consumption simulation
        let receive_request = ReceiveRequest {
            consumer_id: "test_consumer".to_string(),
            timeout_ms: 1000,
            max_messages: 10,
        };

        let start_time = Instant::now();
        let _received_messages = mq_manager
            .receive_messages(receive_request)
            .expect("Failed to receive messages");
        let receive_time = start_time.elapsed();

        println!("Message queue operations completed in {:?}", receive_time);
        assert!(
            receive_time < Duration::from_millis(100),
            "Message queue operations too slow"
        );
    }

    /// Integration test 5: State management with exactly-once processing
    #[tokio::test]
    async fn test_state_management_exactly_once() {
        let mut state_config = StateManagementConfig::default();
        state_config.exactly_once_config.enabled = true;

        let state_manager = StreamStateManager::new(state_config)
            .await
            .expect("Failed to create state manager");

        // Test state storage and retrieval
        let test_key = "test_state_key";
        let test_value = b"test_state_value";
        let stream_id = "exactly_once_test";

        // Store state multiple times (should be idempotent)
        for i in 0..5 {
            let start_time = Instant::now();
            let store_result = state_manager
                .store_state(test_key, test_value, stream_id)
                .await
                .expect("Failed to store state");
            let store_latency = start_time.elapsed();

            assert!(
                store_latency < Duration::from_millis(10),
                "State store too slow"
            );
            assert!(store_result.success, "State store should succeed");

            if i > 0 {
                // Should detect duplicates for exactly-once processing
                println!(
                    "Store result {}: duplicate_detected={}",
                    i, store_result.duplicate_detected
                );
            }
        }

        // Retrieve and verify state
        let start_time = Instant::now();
        let retrieved_value = state_manager
            .retrieve_state(test_key)
            .await
            .expect("Failed to retrieve state");
        let retrieve_latency = start_time.elapsed();

        assert!(
            retrieve_latency < Duration::from_millis(5),
            "State retrieve too slow"
        );
        assert_eq!(
            retrieved_value,
            Some(test_value.to_vec()),
            "Retrieved value mismatch"
        );

        // Test exactly-once event processing
        let test_events = create_duplicate_test_events();
        let start_time = Instant::now();
        let _processed_events = state_manager
            .process_exactly_once(test_events)
            .await
            .expect("Exactly-once processing failed");
        let processing_time = start_time.elapsed();

        assert!(
            processing_time < Duration::from_millis(50),
            "Exactly-once processing too slow"
        );

        // Should have fewer processed events due to deduplication
        println!("Exactly-once processing completed in {:?}", processing_time);
    }

    /// Integration test 6: Edge computing with resource constraints
    #[tokio::test]
    async fn test_edge_computing_resource_constraints() {
        let mut edge_config = EdgeComputingConfig::default();
        edge_config
            .inference_config
            .resource_constraints
            .max_latency_ms = 5; // Strict constraint
        edge_config
            .inference_config
            .resource_constraints
            .max_memory_bytes = 10 * 1024 * 1024; // 10MB

        let edge_manager = EdgeComputingManager::new(edge_config)
            .await
            .expect("Failed to create edge manager");

        edge_manager
            .start()
            .await
            .expect("Failed to start edge manager");

        // Test lightweight inference under constraints
        let test_events = create_test_events(10);
        let start_time = Instant::now();

        match timeout(
            Duration::from_millis(100),
            edge_manager.lightweight_inference(test_events, "test_model"),
        )
        .await
        {
            Ok(Ok(_results)) => {
                let inference_time = start_time.elapsed();
                assert!(
                    inference_time < Duration::from_millis(50),
                    "Edge inference too slow"
                );
                println!("Edge inference completed in {:?}", inference_time);
            }
            Ok(Err(e)) => {
                // Resource constraints might prevent processing
                println!("Edge inference failed due to constraints: {:?}", e);
            }
            Err(_) => {
                panic!("Edge inference timeout");
            }
        }

        // Test device failure handling
        let start_time = Instant::now();
        let failure_result = edge_manager
            .handle_device_failure(DeviceFailureType::ConnectivityFailure)
            .await;
        let failure_handling_time = start_time.elapsed();

        assert!(failure_result.is_ok(), "Device failure handling failed");
        assert!(
            failure_handling_time < Duration::from_millis(20),
            "Failure handling too slow"
        );

        edge_manager
            .stop()
            .await
            .expect("Failed to stop edge manager");
    }

    /// Integration test 7: Monitoring and observability system
    #[tokio::test]
    async fn test_monitoring_observability() {
        let mut monitoring_config = MonitoringConfig::default();
        monitoring_config.metrics_config.collection_interval_ms = 100;
        monitoring_config.tracing_config.enabled = true;

        let monitoring_system = MonitoringSystem::new(monitoring_config)
            .await
            .expect("Failed to create monitoring system");

        monitoring_system
            .start()
            .await
            .expect("Failed to start monitoring");

        // Test metrics recording
        let test_events = create_test_events(50);
        let start_time = Instant::now();

        for chunk in test_events.chunks(10) {
            monitoring_system
                .record_stream_metrics(chunk)
                .await
                .expect("Failed to record metrics");
        }

        let metrics_time = start_time.elapsed();
        assert!(
            metrics_time < Duration::from_millis(100),
            "Metrics recording too slow"
        );

        // Test distributed tracing
        let start_time = Instant::now();
        let trace_context = monitoring_system
            .create_trace("integration_test")
            .await
            .expect("Failed to create trace");
        let trace_time = start_time.elapsed();

        assert!(
            trace_time < Duration::from_millis(10),
            "Trace creation too slow"
        );
        assert!(!trace_context.trace_id.is_empty(), "Invalid trace context");

        // Test health checking
        let start_time = Instant::now();
        let _health_status = monitoring_system
            .check_system_health()
            .await
            .expect("Failed to check health");
        let health_time = start_time.elapsed();

        assert!(
            health_time < Duration::from_millis(50),
            "Health check too slow"
        );

        // Test metrics export
        let start_time = Instant::now();
        let metrics_export = monitoring_system
            .export_metrics(ExportFormat::Json)
            .await
            .expect("Failed to export metrics");
        let export_time = start_time.elapsed();

        assert!(
            export_time < Duration::from_millis(100),
            "Metrics export too slow"
        );
        assert!(!metrics_export.is_empty(), "Empty metrics export");

        monitoring_system
            .stop()
            .await
            .expect("Failed to stop monitoring");

        println!("Monitoring system tests completed successfully");
    }

    /// Integration test 8: Full system stress test
    #[tokio::test]
    async fn test_full_system_stress_test() {
        // Create comprehensive system configuration
        let streaming_config = create_test_streaming_config();
        let pipeline_config = PipelineConfig::default();
        let adaptive_config = AdaptiveProcessingConfig::default();
        let state_config = StateManagementConfig::default();
        let monitoring_config = MonitoringConfig::default();

        // Initialize all systems
        let mut streaming_server = StreamingServer::new(streaming_config).await.unwrap();
        let pipeline = Arc::new(RealtimePipeline::new(pipeline_config).unwrap());
        let adaptive_processor = Arc::new(AdaptiveProcessor::new(adaptive_config).await.unwrap());
        let state_manager = Arc::new(StreamStateManager::new(state_config).await.unwrap());
        let monitoring = Arc::new(MonitoringSystem::new(monitoring_config).await.unwrap());

        // Start all systems
        streaming_server
            .start()
            .await
            .expect("Failed to start streaming server");
        pipeline.start().expect("Failed to start pipeline");
        adaptive_processor
            .start()
            .await
            .expect("Failed to start adaptive processor");
        monitoring
            .start()
            .await
            .expect("Failed to start monitoring");

        println!("All systems started for stress test");

        let stress_start = Instant::now();
        let mut handles = Vec::new();

        // Create 100 concurrent stress test workers
        for worker_id in 0..100 {
            let pipeline = pipeline.clone();
            let adaptive_processor = adaptive_processor.clone();
            let state_manager = state_manager.clone();
            let monitoring = monitoring.clone();

            let handle = tokio::spawn(async move {
                let mut worker_stats = WorkerStats::default();

                for iteration in 0..10 {
                    let events = create_test_events(5);
                    let iteration_start = Instant::now();

                    // Process through pipeline
                    for event in &events {
                        if let Ok(_) = pipeline.process_event(event.clone()) {
                            worker_stats.successful_pipeline_operations += 1;
                        } else {
                            worker_stats.failed_operations += 1;
                        }
                    }

                    // Process through adaptive system
                    let stream_id = format!("stress_worker_{}", worker_id);
                    if let Ok(_) = adaptive_processor
                        .process_adaptive_batch(events.clone(), &stream_id)
                        .await
                    {
                        worker_stats.successful_adaptive_operations += 1;
                    } else {
                        worker_stats.failed_operations += 1;
                    }

                    // State operations
                    let state_key = format!("stress_key_{}_{}", worker_id, iteration);
                    let state_value = format!("value_{}", iteration).as_bytes().to_vec();
                    if let Ok(_) = state_manager
                        .store_state(&state_key, &state_value, &stream_id)
                        .await
                    {
                        worker_stats.successful_state_operations += 1;
                    } else {
                        worker_stats.failed_operations += 1;
                    }

                    // Record metrics
                    if let Ok(_) = monitoring.record_stream_metrics(&events).await {
                        worker_stats.successful_monitoring_operations += 1;
                    } else {
                        worker_stats.failed_operations += 1;
                    }

                    worker_stats.total_iterations += 1;
                    worker_stats.total_processing_time += iteration_start.elapsed();

                    // Small delay to prevent overwhelming
                    sleep(Duration::from_millis(10)).await;
                }

                worker_stats
            });

            handles.push(handle);
        }

        // Collect results from all workers
        let mut total_stats = WorkerStats::default();
        for handle in handles {
            match handle.await {
                Ok(worker_stats) => {
                    total_stats.combine(worker_stats);
                }
                Err(e) => {
                    println!("Worker failed: {:?}", e);
                    total_stats.failed_operations += 1;
                }
            }
        }

        let stress_duration = stress_start.elapsed();

        // Calculate and verify stress test results
        let total_operations = total_stats.successful_pipeline_operations
            + total_stats.successful_adaptive_operations
            + total_stats.successful_state_operations
            + total_stats.successful_monitoring_operations
            + total_stats.failed_operations;

        let success_rate = if total_operations > 0 {
            ((total_operations - total_stats.failed_operations) as f64 / total_operations as f64)
                * 100.0
        } else {
            0.0
        };

        let operations_per_second = total_operations as f64 / stress_duration.as_secs_f64();

        println!("Stress Test Results:");
        println!("  Duration: {:?}", stress_duration);
        println!("  Total Operations: {}", total_operations);
        println!("  Failed Operations: {}", total_stats.failed_operations);
        println!("  Success Rate: {:.2}%", success_rate);
        println!("  Operations/Second: {:.2}", operations_per_second);
        println!(
            "  Pipeline Operations: {}",
            total_stats.successful_pipeline_operations
        );
        println!(
            "  Adaptive Operations: {}",
            total_stats.successful_adaptive_operations
        );
        println!(
            "  State Operations: {}",
            total_stats.successful_state_operations
        );
        println!(
            "  Monitoring Operations: {}",
            total_stats.successful_monitoring_operations
        );

        // Verify stress test requirements
        assert!(
            success_rate >= 95.0,
            "Success rate {} below 95% requirement",
            success_rate
        );
        assert!(
            operations_per_second >= 100.0,
            "Operations/sec {} below requirement",
            operations_per_second
        );
        assert!(
            stress_duration < Duration::from_secs(60),
            "Stress test took too long"
        );

        // Stop all systems
        monitoring.stop().await.expect("Failed to stop monitoring");
        adaptive_processor
            .stop()
            .await
            .expect("Failed to stop adaptive processor");
        // pipeline.stop().await.expect("Failed to stop pipeline"); // stop() method not available
        // streaming_server.shutdown().await.expect("Failed to shutdown server");

        println!("Full system stress test completed successfully!");
    }

    // Helper functions for test setup and data creation
    fn create_test_streaming_config() -> StreamingConfig {
        StreamingConfig {
            max_connections: 1000,
            target_latency: Duration::from_micros(900), // <1ms
            connection_pool_size: 100,
            backpressure_threshold: 0.8,
            websocket_addr: "127.0.0.1:8080".to_string(),
            grpc_addr: "127.0.0.1:50051".to_string(),
            memory_pool_size: 1024 * 1024 * 100, // 100MB
            metrics_enabled: true,
        }
    }

    fn create_test_events(count: usize) -> Vec<StreamEvent> {
        (0..count)
            .map(|i| {
                let event_id = Uuid::new_v4();
                let event_time = SystemTime::now();
                StreamEvent {
                    event_id,
                    event_time,
                    processing_time: SystemTime::now(),
                    stream_id: format!("test_stream_{}", i),
                    data: EventData::Text {
                        content: format!("test_content_{}", i),
                        tokens: Some(vec![format!("token_{}", i)]),
                    },
                    metadata: {
                        let mut metadata = HashMap::new();
                        metadata.insert("test_key".to_string(), format!("test_value_{}", i));
                        metadata
                    },
                    watermark: None,
                    id: event_id.to_string(),
                    timestamp: event_time,
                    partition_key: Some(format!("partition_{}", i % 4)),
                    sequence_number: i as u64,
                }
            })
            .collect()
    }

    fn create_duplicate_test_events() -> Vec<StreamEvent> {
        let event_id = Uuid::new_v4();
        let event_time = SystemTime::now();
        let base_event = StreamEvent {
            event_id,
            event_time,
            processing_time: SystemTime::now(),
            stream_id: "duplicate_test".to_string(),
            data: EventData::Text {
                content: "duplicate_content".to_string(),
                tokens: None,
            },
            metadata: HashMap::new(),
            watermark: None,
            id: event_id.to_string(),
            timestamp: event_time,
            partition_key: Some("duplicate_partition".to_string()),
            sequence_number: 0,
        };

        // Create duplicates with same event ID
        vec![base_event.clone(), base_event.clone(), base_event.clone()]
    }

    #[derive(Debug, Default, Clone)]
    struct WorkerStats {
        successful_pipeline_operations: u64,
        successful_adaptive_operations: u64,
        successful_state_operations: u64,
        successful_monitoring_operations: u64,
        failed_operations: u64,
        total_iterations: u64,
        total_processing_time: Duration,
    }

    impl WorkerStats {
        fn combine(&mut self, other: WorkerStats) {
            self.successful_pipeline_operations += other.successful_pipeline_operations;
            self.successful_adaptive_operations += other.successful_adaptive_operations;
            self.successful_state_operations += other.successful_state_operations;
            self.successful_monitoring_operations += other.successful_monitoring_operations;
            self.failed_operations += other.failed_operations;
            self.total_iterations += other.total_iterations;
            self.total_processing_time += other.total_processing_time;
        }
    }
}