//! # Comprehensive Streaming Edge Case Tests
//!
//! This module contains extensive edge case testing for streaming market data,
//! covering backpressure, error handling, windowing, joins, and late data handling.
//!
//! ## Test Coverage
//!
//! - Stream backpressure (slow consumer, buffer overflow, flow control)
//! - Stream error handling (network errors, malformed data, reconnection)
//! - Stream windowing (time-based, count-based, session windows)
//! - Stream joins (inner, left, outer joins on event time)
//! - Late data handling (watermarks, allowed lateness, side outputs)
//! - Memory leak detection (long-running streams)
//! - Throughput measurements (events/sec)

#![allow(unused_crate_dependencies)]

use chrono::{DateTime, Duration as ChronoDuration, Utc};
use common::market_data::{MarketDataEvent, QuoteEvent, TradeEvent};
use common::{OrderSide, Price, Quantity, Symbol};
use data::error::DataError;
use rust_decimal::Decimal;
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering};
use std::sync::Arc;
use tokio::sync::{mpsc, Mutex, RwLock};
use tokio::time::{sleep, timeout, Duration, Instant};

// ============================================================================
// Test Utilities
// ============================================================================

/// Generate a test trade event
fn create_trade(symbol: &str, price: f64, quantity: f64, timestamp: DateTime<Utc>) -> TradeEvent {
    TradeEvent {
        symbol: Symbol::from(symbol),
        price: Price::from_decimal(Decimal::from_f64_retain(price).unwrap()),
        quantity: Quantity::new(quantity).unwrap(),
        timestamp,
        trade_id: format!("trade_{}", timestamp.timestamp_nanos_opt().unwrap_or(0)),
        side: OrderSide::Buy,
    }
}

/// Generate a test quote event
fn create_quote(
    symbol: &str,
    bid: f64,
    ask: f64,
    timestamp: DateTime<Utc>,
) -> QuoteEvent {
    QuoteEvent {
        symbol: Symbol::from(symbol),
        bid_price: Price::from_decimal(Decimal::from_f64_retain(bid).unwrap()),
        ask_price: Price::from_decimal(Decimal::from_f64_retain(ask).unwrap()),
        bid_quantity: Quantity::new(100.0).unwrap(),
        ask_quantity: Quantity::new(100.0).unwrap(),
        timestamp,
    }
}

/// Backpressure controller for stream flow control
struct BackpressureController {
    buffer_size: usize,
    high_water_mark: usize,
    low_water_mark: usize,
    current_size: Arc<AtomicUsize>,
    is_overloaded: Arc<AtomicBool>,
    messages_dropped: Arc<AtomicU64>,
}

impl BackpressureController {
    fn new(buffer_size: usize) -> Self {
        Self {
            buffer_size,
            high_water_mark: (buffer_size as f64 * 0.8) as usize,
            low_water_mark: (buffer_size as f64 * 0.2) as usize,
            current_size: Arc::new(AtomicUsize::new(0)),
            is_overloaded: Arc::new(AtomicBool::new(false)),
            messages_dropped: Arc::new(AtomicU64::new(0)),
        }
    }

    fn should_drop_message(&self, queue_size: usize) -> bool {
        self.current_size.store(queue_size, Ordering::Relaxed);

        if queue_size > self.high_water_mark {
            self.is_overloaded.store(true, Ordering::Relaxed);
            // Drop 10% of messages when overloaded (deterministic for testing)
            if queue_size % 10 == 0 {
                self.messages_dropped.fetch_add(1, Ordering::Relaxed);
                return true;
            }
        } else if queue_size < self.low_water_mark {
            self.is_overloaded.store(false, Ordering::Relaxed);
        }

        false
    }

    fn is_overloaded(&self) -> bool {
        self.is_overloaded.load(Ordering::Relaxed)
    }

    fn get_dropped_count(&self) -> u64 {
        self.messages_dropped.load(Ordering::Relaxed)
    }
}

/// Time-based window for stream aggregation
struct TimeWindow<T> {
    window_duration: ChronoDuration,
    events: VecDeque<(DateTime<Utc>, T)>,
}

impl<T: Clone> TimeWindow<T> {
    fn new(window_duration: ChronoDuration) -> Self {
        Self {
            window_duration,
            events: VecDeque::new(),
        }
    }

    fn add_event(&mut self, timestamp: DateTime<Utc>, event: T) {
        self.events.push_back((timestamp, event));
        self.evict_old_events(timestamp);
    }

    fn evict_old_events(&mut self, current_time: DateTime<Utc>) {
        let cutoff = current_time - self.window_duration;
        while let Some((ts, _)) = self.events.front() {
            if *ts < cutoff {
                self.events.pop_front();
            } else {
                break;
            }
        }
    }

    fn get_events(&self) -> Vec<T> {
        self.events.iter().map(|(_, e)| e.clone()).collect()
    }

    fn count(&self) -> usize {
        self.events.len()
    }
}

/// Count-based window for stream aggregation
struct CountWindow<T> {
    max_count: usize,
    events: VecDeque<T>,
}

impl<T: Clone> CountWindow<T> {
    fn new(max_count: usize) -> Self {
        Self {
            max_count,
            events: VecDeque::with_capacity(max_count),
        }
    }

    fn add_event(&mut self, event: T) {
        if self.events.len() >= self.max_count {
            self.events.pop_front();
        }
        self.events.push_back(event);
    }

    fn get_events(&self) -> Vec<T> {
        self.events.iter().cloned().collect()
    }

    fn is_full(&self) -> bool {
        self.events.len() >= self.max_count
    }
}

/// Stream join coordinator for correlating events across streams
struct StreamJoinCoordinator {
    trade_buffer: HashMap<Symbol, VecDeque<TradeEvent>>,
    quote_buffer: HashMap<Symbol, VecDeque<QuoteEvent>>,
    max_buffer_per_symbol: usize,
    time_tolerance: ChronoDuration,
}

impl StreamJoinCoordinator {
    fn new(max_buffer_per_symbol: usize, time_tolerance: ChronoDuration) -> Self {
        Self {
            trade_buffer: HashMap::new(),
            quote_buffer: HashMap::new(),
            max_buffer_per_symbol,
            time_tolerance,
        }
    }

    fn add_trade(&mut self, trade: TradeEvent) {
        let buffer = self.trade_buffer.entry(trade.symbol.clone()).or_insert_with(|| {
            VecDeque::with_capacity(self.max_buffer_per_symbol)
        });

        if buffer.len() >= self.max_buffer_per_symbol {
            buffer.pop_front();
        }
        buffer.push_back(trade);
    }

    fn add_quote(&mut self, quote: QuoteEvent) {
        let buffer = self.quote_buffer.entry(quote.symbol.clone()).or_insert_with(|| {
            VecDeque::with_capacity(self.max_buffer_per_symbol)
        });

        if buffer.len() >= self.max_buffer_per_symbol {
            buffer.pop_front();
        }
        buffer.push_back(quote);
    }

    fn inner_join(&self, symbol: &Symbol) -> Vec<(TradeEvent, QuoteEvent)> {
        let trades = self.trade_buffer.get(symbol);
        let quotes = self.quote_buffer.get(symbol);

        if trades.is_none() || quotes.is_none() {
            return Vec::new();
        }

        let trades = trades.unwrap();
        let quotes = quotes.unwrap();
        let mut results = Vec::new();

        for trade in trades {
            for quote in quotes {
                let time_diff = (trade.timestamp - quote.timestamp).abs();
                if time_diff <= self.time_tolerance {
                    results.push((trade.clone(), quote.clone()));
                    break; // Take first matching quote
                }
            }
        }

        results
    }

    fn left_join(&self, symbol: &Symbol) -> Vec<(TradeEvent, Option<QuoteEvent>)> {
        let trades = self.trade_buffer.get(symbol);
        if trades.is_none() {
            return Vec::new();
        }

        let trades = trades.unwrap();
        let quotes = self.quote_buffer.get(symbol);
        let mut results = Vec::new();

        for trade in trades {
            if let Some(quotes) = quotes {
                let mut matched = false;
                for quote in quotes {
                    let time_diff = (trade.timestamp - quote.timestamp).abs();
                    if time_diff <= self.time_tolerance {
                        results.push((trade.clone(), Some(quote.clone())));
                        matched = true;
                        break;
                    }
                }
                if !matched {
                    results.push((trade.clone(), None));
                }
            } else {
                results.push((trade.clone(), None));
            }
        }

        results
    }
}

/// Watermark manager for handling late data
struct WatermarkManager {
    current_watermark: Arc<RwLock<DateTime<Utc>>>,
    allowed_lateness: ChronoDuration,
    late_events: Arc<Mutex<Vec<MarketDataEvent>>>,
}

impl WatermarkManager {
    fn new(allowed_lateness: ChronoDuration) -> Self {
        Self {
            current_watermark: Arc::new(RwLock::new(Utc::now())),
            allowed_lateness,
            late_events: Arc::new(Mutex::new(Vec::new())),
        }
    }

    async fn update_watermark(&self, timestamp: DateTime<Utc>) {
        let mut watermark = self.current_watermark.write().await;
        if timestamp > *watermark {
            *watermark = timestamp;
        }
    }

    async fn is_late(&self, timestamp: DateTime<Utc>) -> bool {
        let watermark = self.current_watermark.read().await;
        let cutoff = *watermark - self.allowed_lateness;
        timestamp < cutoff
    }

    async fn process_event(&self, event: MarketDataEvent) -> bool {
        let timestamp = match event.timestamp() {
            Some(ts) => ts,
            None => return false, // Invalid event without timestamp
        };

        if self.is_late(timestamp).await {
            let mut late = self.late_events.lock().await;
            late.push(event);
            return false; // Event is late, moved to side output
        }

        self.update_watermark(timestamp).await;
        true // Event is on time
    }

    async fn get_late_events(&self) -> Vec<MarketDataEvent> {
        let late = self.late_events.lock().await;
        late.clone()
    }
}

// ============================================================================
// Backpressure Tests
// ============================================================================

#[tokio::test]
async fn test_backpressure_slow_consumer() {
    // Consumer processes events slower than producer
    let (tx, mut rx) = mpsc::channel::<TradeEvent>(100);
    let controller = Arc::new(BackpressureController::new(100));

    // Producer: 1000 events/sec
    let producer_handle = tokio::spawn(async move {
        for i in 0..500 {
            let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, Utc::now());
            if tx.send(trade).await.is_err() {
                break;
            }
            sleep(Duration::from_micros(1000)).await; // 1ms = 1000/sec
        }
    });

    // Consumer: 100 events/sec (10x slower)
    let consumer_controller = controller.clone();
    let consumer_handle = tokio::spawn(async move {
        let mut processed = 0;
        while let Some(_event) = rx.recv().await {
            processed += 1;
            sleep(Duration::from_micros(10000)).await; // 10ms = 100/sec

            if processed >= 100 {
                break; // Process 100 events
            }
        }
        processed
    });

    // Wait for both tasks
    let _ = producer_handle.await;
    let processed = consumer_handle.await.unwrap();

    // Consumer should process exactly 100 events
    assert_eq!(processed, 100);
    println!("✓ Backpressure test: Processed {} events with slow consumer", processed);
}

#[tokio::test]
async fn test_backpressure_buffer_overflow() {
    // Test buffer overflow and message dropping
    let controller = BackpressureController::new(1000);
    let mut dropped_count = 0;

    // Simulate 2000 messages (exceeds buffer)
    for i in 0..2000 {
        if controller.should_drop_message(i) {
            dropped_count += 1;
        }
    }

    // Should have dropped some messages when queue size exceeded high water mark
    assert!(dropped_count > 0, "Expected some messages to be dropped");
    assert!(controller.is_overloaded(), "Controller should be overloaded");

    println!("✓ Buffer overflow test: Dropped {} messages", dropped_count);
}

#[tokio::test]
async fn test_backpressure_flow_control() {
    // Test flow control with dynamic rate adjustment
    let controller = BackpressureController::new(100);

    // Phase 1: Low load (should not drop)
    for i in 0..20 {
        assert!(!controller.should_drop_message(i));
    }
    assert!(!controller.is_overloaded());

    // Phase 2: High load (should start dropping)
    for i in 80..95 {
        let _ = controller.should_drop_message(i);
    }
    assert!(controller.is_overloaded());

    // Phase 3: Load decreases (should stop dropping)
    for i in (15..25).rev() {
        let _ = controller.should_drop_message(i);
    }
    assert!(!controller.is_overloaded());

    println!("✓ Flow control test: Dynamic rate adjustment working");
}

#[tokio::test]
async fn test_backpressure_burst_traffic() {
    // Test handling of burst traffic (1000+ events/ms)
    let (tx, mut rx) = mpsc::channel::<TradeEvent>(10000);
    let start = Instant::now();

    // Producer: Send 5000 events as fast as possible
    let producer_handle = tokio::spawn(async move {
        for i in 0..5000 {
            let trade = create_trade("SPY", 400.0, 100.0, Utc::now());
            if tx.send(trade).await.is_err() {
                break;
            }
        }
        start.elapsed()
    });

    // Consumer: Process all events
    let consumer_handle = tokio::spawn(async move {
        let mut count = 0;
        while let Some(_event) = rx.recv().await {
            count += 1;
            if count >= 5000 {
                break;
            }
        }
        count
    });

    let producer_time = producer_handle.await.unwrap();
    let count = consumer_handle.await.unwrap();

    assert_eq!(count, 5000);
    let events_per_ms = count as f64 / producer_time.as_millis() as f64;
    println!("✓ Burst traffic test: {} events/ms", events_per_ms);
}

// ============================================================================
// Error Handling Tests
// ============================================================================

#[tokio::test]
async fn test_stream_network_error_recovery() {
    // Simulate network disconnection and reconnection
    let (tx, mut rx) = mpsc::channel::<Result<TradeEvent, DataError>>(100);
    let reconnect_count = Arc::new(AtomicUsize::new(0));

    let producer_reconnect = reconnect_count.clone();
    let producer_handle = tokio::spawn(async move {
        // Send 10 events successfully
        for i in 0..10 {
            let trade = create_trade("AAPL", 150.0, 100.0, Utc::now());
            let _ = tx.send(Ok(trade)).await;
        }

        // Simulate network error
        let _ = tx.send(Err(DataError::Connection("Network timeout".to_string()))).await;
        producer_reconnect.fetch_add(1, Ordering::Relaxed);

        // Reconnect and send more events
        sleep(Duration::from_millis(100)).await;
        for i in 0..10 {
            let trade = create_trade("AAPL", 151.0, 100.0, Utc::now());
            let _ = tx.send(Ok(trade)).await;
        }
    });

    // Consumer with error recovery
    let mut success_count = 0;
    let mut error_count = 0;

    while let Some(result) = rx.recv().await {
        match result {
            Ok(_) => success_count += 1,
            Err(_) => {
                error_count += 1;
                // Simulate reconnection logic
                sleep(Duration::from_millis(50)).await;
            }
        }

        if success_count >= 20 {
            break;
        }
    }

    let _ = producer_handle.await;

    assert_eq!(success_count, 20);
    assert_eq!(error_count, 1);
    assert_eq!(reconnect_count.load(Ordering::Relaxed), 1);
    println!("✓ Network error recovery: {} reconnections, {} events processed",
             error_count, success_count);
}

#[tokio::test]
async fn test_stream_malformed_data_handling() {
    // Test handling of malformed/invalid data
    let (tx, mut rx) = mpsc::channel::<Result<TradeEvent, DataError>>(100);

    let producer_handle = tokio::spawn(async move {
        // Send valid events
        for _ in 0..5 {
            let trade = create_trade("AAPL", 150.0, 100.0, Utc::now());
            let _ = tx.send(Ok(trade)).await;
        }

        // Send malformed data error
        let _ = tx.send(Err(DataError::Parse {
            message: "Invalid price format".to_string(),
        })).await;

        // Continue with valid events
        for _ in 0..5 {
            let trade = create_trade("AAPL", 151.0, 100.0, Utc::now());
            let _ = tx.send(Ok(trade)).await;
        }
    });

    let mut valid_count = 0;
    let mut invalid_count = 0;

    while let Some(result) = rx.recv().await {
        match result {
            Ok(_) => valid_count += 1,
            Err(DataError::Parse { .. }) => {
                invalid_count += 1;
                // Skip malformed event and continue
            }
            Err(_) => {}
        }

        if valid_count >= 10 {
            break;
        }
    }

    let _ = producer_handle.await;

    assert_eq!(valid_count, 10);
    assert_eq!(invalid_count, 1);
    println!("✓ Malformed data handling: {} valid, {} invalid", valid_count, invalid_count);
}

#[tokio::test]
async fn test_stream_very_large_messages() {
    // Test handling of messages >1MB
    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(10);

    let producer_handle = tokio::spawn(async move {
        // Send a 2MB message
        let large_message = vec![0u8; 2 * 1024 * 1024];
        let _ = tx.send(large_message).await;
    });

    let result = timeout(Duration::from_secs(5), rx.recv()).await;
    assert!(result.is_ok(), "Should receive large message within timeout");

    if let Ok(Some(msg)) = result {
        assert_eq!(msg.len(), 2 * 1024 * 1024);
        println!("✓ Large message test: Received {}MB message", msg.len() / (1024 * 1024));
    }

    let _ = producer_handle.await;
}

// ============================================================================
// Windowing Tests
// ============================================================================

#[tokio::test]
async fn test_time_based_windowing() {
    // Test time-based window (5-second tumbling window)
    let mut window = TimeWindow::<TradeEvent>::new(ChronoDuration::seconds(5));
    let base_time = Utc::now();

    // Add events within 5-second window
    for i in 0..10 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 500);
        let trade = create_trade("AAPL", 150.0, 100.0, timestamp);
        window.add_event(timestamp, trade);
    }

    assert_eq!(window.count(), 10, "Window should contain all events");

    // Add event 6 seconds later (outside window)
    let late_timestamp = base_time + ChronoDuration::seconds(6);
    let late_trade = create_trade("AAPL", 151.0, 100.0, late_timestamp);
    window.add_event(late_timestamp, late_trade);

    // Old events should be evicted
    assert!(window.count() <= 3, "Old events should be evicted");
    println!("✓ Time-based window: {} events remaining after eviction", window.count());
}

#[tokio::test]
async fn test_count_based_windowing() {
    // Test count-based window (sliding window of 100 events)
    let mut window = CountWindow::<TradeEvent>::new(100);

    // Add 150 events
    for i in 0..150 {
        let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, Utc::now());
        window.add_event(trade);
    }

    // Window should contain only last 100 events
    assert_eq!(window.get_events().len(), 100);
    assert!(window.is_full());
    println!("✓ Count-based window: Maintained {} events max", window.get_events().len());
}

#[tokio::test]
async fn test_session_windowing() {
    // Test session window (gap-based windowing with 1-second inactivity gap)
    let session_gap = ChronoDuration::seconds(1);
    let mut sessions: Vec<Vec<TradeEvent>> = Vec::new();
    let mut current_session: Vec<TradeEvent> = Vec::new();
    let mut last_timestamp: Option<DateTime<Utc>> = None;

    // Generate events with gaps
    let base_time = Utc::now();
    let event_times = vec![
        0,    // Session 1
        100,
        200,
        2000, // Session 2 (1.8s gap)
        2100,
        2200,
        4000, // Session 3 (1.8s gap)
        4100,
    ];

    for (i, offset_ms) in event_times.iter().enumerate() {
        let timestamp = base_time + ChronoDuration::milliseconds(*offset_ms);
        let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, timestamp);

        if let Some(last_ts) = last_timestamp {
            if timestamp - last_ts > session_gap {
                // Start new session
                sessions.push(current_session.clone());
                current_session.clear();
            }
        }

        current_session.push(trade);
        last_timestamp = Some(timestamp);
    }

    // Add final session
    if !current_session.is_empty() {
        sessions.push(current_session);
    }

    assert_eq!(sessions.len(), 3, "Should have 3 sessions");
    assert_eq!(sessions[0].len(), 3, "Session 1 should have 3 events");
    assert_eq!(sessions[1].len(), 3, "Session 2 should have 3 events");
    assert_eq!(sessions[2].len(), 2, "Session 3 should have 2 events");
    println!("✓ Session window: {} sessions detected", sessions.len());
}

// ============================================================================
// Stream Join Tests
// ============================================================================

#[tokio::test]
async fn test_stream_inner_join() {
    // Test inner join between trade and quote streams
    let mut coordinator = StreamJoinCoordinator::new(100, ChronoDuration::milliseconds(100));
    let base_time = Utc::now();

    // Add trades
    for i in 0..10 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 10);
        let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, timestamp);
        coordinator.add_trade(trade);
    }

    // Add matching quotes (within 100ms tolerance)
    for i in 0..10 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 10 + 5);
        let quote = create_quote("AAPL", 149.0 + i as f64, 151.0 + i as f64, timestamp);
        coordinator.add_quote(quote);
    }

    let joined = coordinator.inner_join(&Symbol::from("AAPL"));
    assert_eq!(joined.len(), 10, "All trades should match with quotes");
    println!("✓ Inner join: {} matched pairs", joined.len());
}

#[tokio::test]
async fn test_stream_left_join() {
    // Test left join (all trades, some without matching quotes)
    let mut coordinator = StreamJoinCoordinator::new(100, ChronoDuration::milliseconds(50));
    let base_time = Utc::now();

    // Add 10 trades
    for i in 0..10 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 10);
        let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, timestamp);
        coordinator.add_trade(trade);
    }

    // Add only 5 matching quotes
    for i in 0..5 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 10 + 5);
        let quote = create_quote("AAPL", 149.0 + i as f64, 151.0 + i as f64, timestamp);
        coordinator.add_quote(quote);
    }

    let joined = coordinator.left_join(&Symbol::from("AAPL"));
    assert_eq!(joined.len(), 10, "All trades should be in result");

    let matched_count = joined.iter().filter(|(_, q)| q.is_some()).count();
    assert_eq!(matched_count, 5, "Only 5 trades should have matching quotes");
    println!("✓ Left join: {} total, {} matched", joined.len(), matched_count);
}

#[tokio::test]
async fn test_stream_join_different_symbols() {
    // Test join with multiple symbols
    let mut coordinator = StreamJoinCoordinator::new(100, ChronoDuration::milliseconds(100));
    let base_time = Utc::now();

    // Add trades for AAPL and SPY
    for symbol in &["AAPL", "SPY"] {
        for i in 0..5 {
            let timestamp = base_time + ChronoDuration::milliseconds(i * 10);
            let trade = create_trade(symbol, 150.0 + i as f64, 100.0, timestamp);
            coordinator.add_trade(trade);
        }
    }

    // Add quotes only for AAPL
    for i in 0..5 {
        let timestamp = base_time + ChronoDuration::milliseconds(i * 10 + 5);
        let quote = create_quote("AAPL", 149.0 + i as f64, 151.0 + i as f64, timestamp);
        coordinator.add_quote(quote);
    }

    let aapl_joined = coordinator.inner_join(&Symbol::from("AAPL"));
    let spy_joined = coordinator.inner_join(&Symbol::from("SPY"));

    assert_eq!(aapl_joined.len(), 5, "AAPL trades should match");
    assert_eq!(spy_joined.len(), 0, "SPY trades should not match");
    println!("✓ Multi-symbol join: AAPL={}, SPY={}", aapl_joined.len(), spy_joined.len());
}

// ============================================================================
// Late Data Handling Tests
// ============================================================================

#[tokio::test]
async fn test_watermark_late_data_detection() {
    // Test watermark-based late data detection
    let manager = WatermarkManager::new(ChronoDuration::seconds(5));
    let base_time = Utc::now();

    // Process events in order
    for i in 0..10 {
        let timestamp = base_time + ChronoDuration::seconds(i);
        let trade = create_trade("AAPL", 150.0, 100.0, timestamp);
        let event = MarketDataEvent::Trade(trade);

        let is_on_time = manager.process_event(event).await;
        assert!(is_on_time, "Sequential events should be on time");
    }

    // Send a late event (before watermark - allowed lateness)
    let late_timestamp = base_time + ChronoDuration::seconds(3);
    let late_trade = create_trade("AAPL", 149.0, 100.0, late_timestamp);
    let late_event = MarketDataEvent::Trade(late_trade);

    let is_on_time = manager.process_event(late_event).await;
    assert!(!is_on_time, "Late event should be detected");

    let late_events = manager.get_late_events().await;
    assert_eq!(late_events.len(), 1, "Should have one late event");
    println!("✓ Watermark test: Detected {} late events", late_events.len());
}

#[tokio::test]
async fn test_allowed_lateness_handling() {
    // Test allowed lateness window
    let manager = WatermarkManager::new(ChronoDuration::seconds(5));
    let base_time = Utc::now();

    // Advance watermark
    manager.update_watermark(base_time + ChronoDuration::seconds(10)).await;

    // Event within allowed lateness (6 seconds old, allowed 5)
    let event1_time = base_time + ChronoDuration::seconds(6);
    let trade1 = create_trade("AAPL", 150.0, 100.0, event1_time);
    let is_late1 = manager.is_late(event1_time).await;
    assert!(!is_late1, "Event within allowed lateness should not be late");

    // Event outside allowed lateness (4 seconds old, allowed 5)
    let event2_time = base_time + ChronoDuration::seconds(4);
    let is_late2 = manager.is_late(event2_time).await;
    assert!(is_late2, "Event outside allowed lateness should be late");

    println!("✓ Allowed lateness: Within window={}, Outside window={}",
             !is_late1, is_late2);
}

#[tokio::test]
async fn test_side_output_for_late_events() {
    // Test side output stream for late events
    let manager = Arc::new(WatermarkManager::new(ChronoDuration::seconds(2)));
    let base_time = Utc::now();

    // Process 20 events with some late ones
    let mut on_time_count = 0;

    for i in 0..20 {
        let timestamp = if i % 5 == 0 {
            // Every 5th event is late
            base_time + ChronoDuration::seconds(i / 2)
        } else {
            base_time + ChronoDuration::seconds(i)
        };

        let trade = create_trade("AAPL", 150.0, 100.0, timestamp);
        let event = MarketDataEvent::Trade(trade);

        if manager.process_event(event).await {
            on_time_count += 1;
        }
    }

    let late_events = manager.get_late_events().await;
    assert_eq!(on_time_count + late_events.len(), 20, "All events should be accounted for");
    println!("✓ Side output: {} on-time, {} late", on_time_count, late_events.len());
}

// ============================================================================
// Performance & Memory Tests
// ============================================================================

#[tokio::test]
async fn test_throughput_measurement() {
    // Measure streaming throughput
    let (tx, mut rx) = mpsc::channel::<TradeEvent>(10000);
    let start = Instant::now();
    let event_count = 10000;

    // Producer
    let producer_handle = tokio::spawn(async move {
        for i in 0..event_count {
            let trade = create_trade("AAPL", 150.0, 100.0, Utc::now());
            if tx.send(trade).await.is_err() {
                break;
            }
        }
    });

    // Consumer
    let consumer_handle = tokio::spawn(async move {
        let mut count = 0;
        while let Some(_event) = rx.recv().await {
            count += 1;
            if count >= event_count {
                break;
            }
        }
        count
    });

    let _ = producer_handle.await;
    let processed = consumer_handle.await.unwrap();
    let elapsed = start.elapsed();

    let events_per_sec = processed as f64 / elapsed.as_secs_f64();
    assert_eq!(processed, event_count);
    assert!(events_per_sec > 1000.0, "Should process >1000 events/sec");

    println!("✓ Throughput: {:.0} events/sec ({} events in {:?})",
             events_per_sec, processed, elapsed);
}

#[tokio::test]
#[ignore] // Long-running test
async fn test_memory_leak_detection() {
    // Test for memory leaks in long-running stream
    let (tx, mut rx) = mpsc::channel::<TradeEvent>(1000);

    // Producer: Send events for 10 seconds
    let producer_handle = tokio::spawn(async move {
        let start = Instant::now();
        let mut count = 0;

        while start.elapsed() < Duration::from_secs(10) {
            let trade = create_trade("AAPL", 150.0, 100.0, Utc::now());
            if tx.send(trade).await.is_ok() {
                count += 1;
            }
            sleep(Duration::from_micros(100)).await;
        }
        count
    });

    // Consumer: Process events and track memory
    let consumer_handle = tokio::spawn(async move {
        let mut count = 0;
        let mut max_buffer = 0;

        while let Some(_event) = rx.recv().await {
            count += 1;

            // Track buffer size (approximation)
            let buffer_size = rx.len();
            if buffer_size > max_buffer {
                max_buffer = buffer_size;
            }

            sleep(Duration::from_micros(150)).await;
        }

        (count, max_buffer)
    });

    let sent = producer_handle.await.unwrap();
    let (received, max_buffer) = consumer_handle.await.unwrap();

    // Verify reasonable buffer size (no runaway growth)
    assert!(max_buffer < 500, "Buffer should not grow unbounded");
    println!("✓ Memory leak test: {} events, max buffer size={}", received, max_buffer);
}

#[tokio::test]
async fn test_stream_cleanup_on_cancellation() {
    // Test proper cleanup when stream is cancelled
    let (tx, mut rx) = mpsc::channel::<TradeEvent>(100);
    let cleanup_flag = Arc::new(AtomicBool::new(false));

    let producer_cleanup = cleanup_flag.clone();
    let producer_handle = tokio::spawn(async move {
        for i in 0..1000 {
            let trade = create_trade("AAPL", 150.0, 100.0, Utc::now());
            if tx.send(trade).await.is_err() {
                producer_cleanup.store(true, Ordering::Relaxed);
                break;
            }
            sleep(Duration::from_micros(100)).await;
        }
    });

    // Consumer: Cancel after 50 events
    let mut count = 0;
    while let Some(_event) = rx.recv().await {
        count += 1;
        if count >= 50 {
            drop(rx); // Cancel stream
            break;
        }
    }

    sleep(Duration::from_millis(100)).await;
    let _ = producer_handle.await;

    assert!(cleanup_flag.load(Ordering::Relaxed), "Producer should detect cancellation");
    println!("✓ Cleanup test: Stream cancelled after {} events", count);
}

#[tokio::test]
async fn test_out_of_order_event_handling() {
    // Test handling of out-of-order events
    let mut events = Vec::new();
    let base_time = Utc::now();

    // Generate out-of-order timestamps
    let timestamps = vec![0, 5, 2, 8, 3, 10, 1, 7, 4, 9];

    for (i, &offset) in timestamps.iter().enumerate() {
        let timestamp = base_time + ChronoDuration::seconds(offset);
        let trade = create_trade("AAPL", 150.0 + i as f64, 100.0, timestamp);
        events.push(trade);
    }

    // Sort by event time
    events.sort_by_key(|e| e.timestamp);

    // Verify sorted order
    for i in 1..events.len() {
        assert!(events[i].timestamp >= events[i-1].timestamp,
                "Events should be sorted by timestamp");
    }

    println!("✓ Out-of-order handling: Sorted {} events", events.len());
}

#[tokio::test]
async fn test_duplicate_event_deduplication() {
    // Test deduplication of duplicate events
    let mut seen_ids = std::collections::HashSet::new();
    let mut unique_count = 0;
    let mut duplicate_count = 0;

    // Generate events with some duplicates
    for i in 0..20 {
        let trade_id = if i % 3 == 0 {
            format!("trade_{}", i / 3) // Duplicate every 3rd event
        } else {
            format!("trade_{}", i)
        };

        if seen_ids.insert(trade_id) {
            unique_count += 1;
        } else {
            duplicate_count += 1;
        }
    }

    assert_eq!(unique_count + duplicate_count, 20);
    assert!(duplicate_count > 0, "Should have detected duplicates");
    println!("✓ Deduplication: {} unique, {} duplicates", unique_count, duplicate_count);
}