mecha10-core 4.2.1

//! Comprehensive unit tests for stream.rs module
//!
//! Tests stream combinator operations including:
//! - StreamBuilder methods: filter, map, throttle, debounce, latest
//! - BatchBuilder: batch collection with size and timeout
//! - ReceiverExt trait: stream combinators
//! - Zip: combining two streams
//! - Window: time-based message collection
//! - Combinator chaining and edge cases
//!
//! These tests do NOT require Redis - they test stream logic in isolation.
//! They complement the existing basic tests in stream.rs by providing
//! comprehensive coverage of all stream operations and edge cases.

use mecha10_core::context::Receiver;
use mecha10_core::messages::Message;
use mecha10_core::stream::ReceiverExt;
use std::time::Duration;
use tokio::sync::mpsc;
use tokio::time::{sleep, timeout, Instant};

#[derive(Debug, Clone, PartialEq)]
struct TestData {
    value: i32,
    flag: bool,
}

impl Message for TestData {}

#[tokio::test]
async fn test_filter_none_pass() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|&x: &i32| x < 0);

    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let result = timeout(Duration::from_millis(100), filtered.recv()).await;

    match result {
        Ok(None) => {}
        Err(_) => {}
        Ok(Some(_)) => panic!("Should not receive any messages"),
    }
}

#[tokio::test]
async fn test_filter_all_pass() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|_: &i32| true);

    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut count = 0;
    while filtered.recv().await.is_some() {
        count += 1;
    }

    assert_eq!(count, 5);
}

#[tokio::test]
async fn test_filter_complex_predicate() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|d: &TestData| d.flag && d.value > 10);

    tx.send(TestData { value: 5, flag: true }).unwrap();
    tx.send(TestData { value: 15, flag: false }).unwrap();
    tx.send(TestData { value: 15, flag: true }).unwrap();
    tx.send(TestData { value: 20, flag: true }).unwrap();
    drop(tx);

    let mut count = 0;
    while filtered.recv().await.is_some() {
        count += 1;
    }

    assert_eq!(count, 2);
}

#[tokio::test]
async fn test_map_identity() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut mapped = receiver.map(|x: i32| x);

    for i in 0..3 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut results = Vec::new();
    while let Some(val) = mapped.recv().await {
        results.push(val);
    }

    assert_eq!(results, vec![0, 1, 2]);
}

#[tokio::test]
async fn test_map_transform_same_type() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut mapped = receiver.map(|x: i32| x * 2);

    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut results = Vec::new();
    while let Some(val) = mapped.recv().await {
        results.push(val);
    }

    assert_eq!(results, vec![0, 2, 4, 6, 8]);
}

#[tokio::test]
async fn test_map_modify_struct() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut mapped = receiver.map(|mut d: TestData| {
        d.value *= 2;
        d.flag = !d.flag;
        d
    });

    tx.send(TestData { value: 5, flag: true }).unwrap();
    tx.send(TestData { value: 10, flag: false }).unwrap();
    drop(tx);

    let d1 = mapped.recv().await.unwrap();
    assert_eq!(d1.value, 10);
    assert!(!d1.flag);

    let d2 = mapped.recv().await.unwrap();
    assert_eq!(d2.value, 20);
    assert!(d2.flag);
}

#[tokio::test]
async fn test_throttle_basic() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut throttled = receiver.throttle(Duration::from_millis(100)).build();

    for i in 0..5 {
        tx.send(i).unwrap();
        sleep(Duration::from_millis(120)).await;
    }
    drop(tx);

    let mut count = 0;
    while let Ok(Some(_)) = timeout(Duration::from_millis(50), throttled.recv()).await {
        count += 1;
    }

    assert_eq!(
        count, 5,
        "Messages spaced beyond the throttle window should all pass through"
    );
}

#[tokio::test]
async fn test_throttle_first_message_immediate() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut throttled = receiver.throttle(Duration::from_millis(100)).build();

    let start = Instant::now();

    tx.send(1).unwrap();

    let val = timeout(Duration::from_millis(50), throttled.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(val, 1);
    assert!(
        start.elapsed() < Duration::from_millis(50),
        "First message should be immediate"
    );
}

#[tokio::test]
async fn test_debounce_basic() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut debounced = receiver.debounce(Duration::from_millis(100));

    let start = Instant::now();

    tx.send(1).unwrap();

    let val = timeout(Duration::from_millis(150), debounced.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(val, 1);
    assert!(
        start.elapsed() >= Duration::from_millis(100),
        "First message should wait out the full quiet period before being emitted"
    );

    tx.send(2).unwrap();
    sleep(Duration::from_millis(30)).await;
    tx.send(3).unwrap();
    sleep(Duration::from_millis(30)).await;
    tx.send(4).unwrap();

    let val = timeout(Duration::from_millis(200), debounced.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(
        val, 4,
        "Only the latest coalesced message should be flushed after the window"
    );
}

#[tokio::test]
async fn test_debounce_rate_limiting() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut debounced = receiver.debounce(Duration::from_millis(100));

    tx.send(1).unwrap();
    let val = debounced.recv().await.unwrap();
    assert_eq!(val, 1);

    for i in 2..6 {
        tx.send(i).unwrap();
        sleep(Duration::from_millis(10)).await;
    }

    let val = timeout(Duration::from_millis(200), debounced.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(val, 5, "Rapid sends should coalesce down to the latest message");

    sleep(Duration::from_millis(150)).await;

    tx.send(10).unwrap();

    let val = timeout(Duration::from_millis(150), debounced.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(val, 10);
}

#[tokio::test]
async fn test_latest_drops_old() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut latest = receiver.latest().build();

    for i in 0..10 {
        tx.send(i).unwrap();
    }

    sleep(Duration::from_millis(50)).await;

    let val = timeout(Duration::from_millis(100), latest.recv())
        .await
        .unwrap()
        .unwrap();

    assert!(val >= 8, "Should get latest message, got {}", val);
}

#[tokio::test]
async fn test_latest_single_message() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut latest = receiver.latest().build();

    tx.send(42).unwrap();

    let val = latest.recv().await.unwrap();
    assert_eq!(val, 42);
}

#[tokio::test]
async fn test_chain_filter_map() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut processed = receiver.filter(|&x: &i32| x % 2 == 0).map(|x: i32| x * 2);

    for i in 0..10 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut results = Vec::new();
    while let Some(val) = processed.recv().await {
        results.push(val);
    }

    assert_eq!(results, vec![0, 4, 8, 12, 16]);
}

#[tokio::test]
async fn test_chain_map_filter() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut processed = receiver.map(|x: i32| x * 2).filter(|&x| x > 5);

    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut results = Vec::new();
    while let Some(val) = processed.recv().await {
        results.push(val);
    }

    assert_eq!(results, vec![6, 8]);
}

#[tokio::test]
async fn test_chain_filter_throttle() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut processed = receiver
        .filter(|&x: &i32| x < 100)
        .throttle(Duration::from_millis(50))
        .build();

    for i in 0..10 {
        tx.send(i).unwrap();
    }

    sleep(Duration::from_millis(200)).await;
    drop(tx);

    let mut count = 0;
    while let Ok(Some(_)) = timeout(Duration::from_millis(30), processed.recv()).await {
        count += 1;
    }

    assert!(count <= 5, "Should be throttled, got {}", count);
}

#[tokio::test]
async fn test_batch_by_size() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut batched = receiver.batch(3, Duration::from_secs(10)).build();

    for i in 0..3 {
        tx.send(i).unwrap();
    }

    let batch = timeout(Duration::from_millis(100), batched.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(batch, vec![0, 1, 2]);
}

#[tokio::test]
async fn test_batch_by_timeout() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut batched = receiver.batch(10, Duration::from_millis(100)).build();

    tx.send(1).unwrap();
    tx.send(2).unwrap();

    let batch = timeout(Duration::from_millis(150), batched.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(batch, vec![1, 2]);
}

#[tokio::test]
async fn test_batch_multiple_batches() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut batched = receiver.batch(3, Duration::from_secs(1)).build();

    for i in 0..7 {
        tx.send(i).unwrap();
    }

    let batch1 = batched.recv().await.unwrap();
    assert_eq!(batch1, vec![0, 1, 2]);

    let batch2 = batched.recv().await.unwrap();
    assert_eq!(batch2, vec![3, 4, 5]);

    drop(tx);

    let batch3 = batched.recv().await.unwrap();
    assert_eq!(batch3, vec![6]);
}

#[tokio::test]
async fn test_batch_size_one() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut batched = receiver.batch(1, Duration::from_secs(10)).build();

    tx.send(1).unwrap();
    tx.send(2).unwrap();
    tx.send(3).unwrap();

    let batch1 = batched.recv().await.unwrap();
    assert_eq!(batch1, vec![1]);

    let batch2 = batched.recv().await.unwrap();
    assert_eq!(batch2, vec![2]);

    let batch3 = batched.recv().await.unwrap();
    assert_eq!(batch3, vec![3]);
}

#[tokio::test]
async fn test_batch_with_filter() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut batched = receiver
        .filter(|&x: &i32| x % 2 == 0)
        .batch(2, Duration::from_secs(1))
        .build();

    for i in 0..10 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let batch1 = batched.recv().await.unwrap();
    assert_eq!(batch1, vec![0, 2]);

    let batch2 = batched.recv().await.unwrap();
    assert_eq!(batch2, vec![4, 6]);

    let batch3 = batched.recv().await.unwrap();
    assert_eq!(batch3, vec![8]);
}

#[tokio::test]
async fn test_window_basic() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut windowed = receiver.window(Duration::from_millis(100));

    for i in 0..3 {
        tx.send(i).unwrap();
    }

    let batch = timeout(Duration::from_millis(150), windowed.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(batch.len(), 3);
    assert_eq!(batch, vec![0, 1, 2]);
}

#[tokio::test]
async fn test_window_multiple_windows() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut windowed = receiver.window(Duration::from_millis(100));

    tx.send(1).unwrap();
    tx.send(2).unwrap();

    let batch1 = timeout(Duration::from_millis(150), windowed.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(batch1, vec![1, 2]);

    tx.send(10).unwrap();
    tx.send(11).unwrap();
    tx.send(12).unwrap();

    let batch2 = timeout(Duration::from_millis(150), windowed.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(batch2, vec![10, 11, 12]);
}

#[tokio::test]
async fn test_window_empty_window() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut windowed = receiver.window(Duration::from_millis(50));

    tx.send(1).unwrap();

    let batch1 = timeout(Duration::from_millis(70), windowed.recv())
        .await
        .unwrap()
        .unwrap();
    assert!(batch1.contains(&1));

    let batch2 = timeout(Duration::from_millis(70), windowed.recv())
        .await
        .unwrap()
        .unwrap();
    assert_eq!(batch2.len(), 0, "Empty window should still emit");
}

#[tokio::test]
async fn test_window_stream_ends() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut windowed = receiver.window(Duration::from_millis(200));

    tx.send(1).unwrap();
    tx.send(2).unwrap();
    drop(tx);

    let batch = timeout(Duration::from_millis(50), windowed.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(batch, vec![1, 2]);
}

#[tokio::test]
async fn test_window_with_map() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut windowed_sum = receiver
        .window(Duration::from_millis(100))
        .map(|batch: Vec<i32>| batch.iter().sum::<i32>());

    for i in 1..=5 {
        tx.send(i).unwrap();
    }

    let sum = timeout(Duration::from_millis(150), windowed_sum.recv())
        .await
        .unwrap()
        .unwrap();

    assert_eq!(sum, 15);
}

#[tokio::test]
async fn test_zip_one_stream_empty() {
    let (tx1, rx1) = mpsc::unbounded_channel();
    let (_tx2, rx2) = mpsc::unbounded_channel::<i32>();

    let receiver1 = Receiver::from(rx1);
    let receiver2 = Receiver::from(rx2);

    let mut zipped = receiver1.zip(receiver2);

    tx1.send(1).unwrap();
    tx1.send(2).unwrap();
    drop(tx1);

    let result = timeout(Duration::from_millis(100), zipped.recv()).await;

    match result {
        Ok(None) => {}
        Err(_) => {}
        Ok(Some(_)) => panic!("Should not get pairs when one stream is empty"),
    }
}

#[tokio::test]
async fn test_zip_synchronization() {
    let (tx1, rx1) = mpsc::unbounded_channel();
    let (tx2, rx2) = mpsc::unbounded_channel();

    let receiver1 = Receiver::from(rx1);
    let receiver2 = Receiver::from(rx2);

    let mut zipped = receiver1.zip(receiver2);

    tx1.send(1).unwrap();

    let recv_task = tokio::spawn(async move { zipped.recv().await });

    sleep(Duration::from_millis(50)).await;

    assert!(!recv_task.is_finished());

    tx2.send(10).unwrap();

    let result = timeout(Duration::from_millis(100), recv_task).await.unwrap().unwrap();

    assert_eq!(result, Some((1, 10)));
}

#[tokio::test]
async fn test_zip_with_different_types() {
    let (tx1, rx1) = mpsc::unbounded_channel();
    let (tx2, rx2) = mpsc::unbounded_channel();

    let receiver1 = Receiver::from(rx1);
    let receiver2 = Receiver::from(rx2);

    let mut zipped = receiver1.zip(receiver2);

    tx1.send(1_i32).unwrap();
    tx2.send("hello".to_string()).unwrap();
    tx1.send(2).unwrap();
    tx2.send("world".to_string()).unwrap();
    drop(tx1);
    drop(tx2);

    let (n1, s1) = zipped.recv().await.unwrap();
    assert_eq!(n1, 1);
    assert_eq!(s1, "hello");

    let (n2, s2) = zipped.recv().await.unwrap();
    assert_eq!(n2, 2);
    assert_eq!(s2, "world");
}

#[tokio::test]
async fn test_receiver_dropped_early() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|&x: &i32| x > 0);

    for i in 0..5 {
        tx.send(i).unwrap();
    }

    let val = filtered.recv().await.unwrap();
    assert!(val > 0);

    drop(filtered);

    for i in 10..15 {
        let result = tx.send(i);
        assert!(result.is_ok(), "Sender should not error");
    }
}

#[tokio::test]
async fn test_sender_dropped_early() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut mapped = receiver.map(|x: i32| x * 2);

    tx.send(1).unwrap();
    tx.send(2).unwrap();
    drop(tx);

    let val1 = mapped.recv().await.unwrap();
    assert_eq!(val1, 2);

    let val2 = mapped.recv().await.unwrap();
    assert_eq!(val2, 4);

    let val3 = mapped.recv().await;
    assert!(val3.is_none());
}

#[tokio::test]
async fn test_empty_stream() {
    let (_tx, rx) = mpsc::unbounded_channel::<i32>();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|&x| x > 0);

    let result = timeout(Duration::from_millis(100), filtered.recv()).await;
    assert!(result.is_err(), "Should timeout on empty stream");
}

#[tokio::test]
async fn test_zero_duration_throttle() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut throttled = receiver.throttle(Duration::from_millis(0)).build();

    for i in 0..5 {
        tx.send(i).unwrap();
    }
    drop(tx);

    let mut count = 0;
    while throttled.recv().await.is_some() {
        count += 1;
    }

    assert!(count >= 3, "Zero throttle should pass most messages, got {}", count);
}

#[tokio::test]
async fn test_complex_chain() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut processed = receiver
        .filter(|&x: &i32| x % 2 == 0)
        .map(|x: i32| x * 2)
        .throttle(Duration::from_millis(20))
        .build()
        .map(|x: i32| x + 1);

    for i in 0..10 {
        tx.send(i).unwrap();
    }

    sleep(Duration::from_millis(100)).await;
    drop(tx);

    let mut results = Vec::new();
    while let Ok(Some(val)) = timeout(Duration::from_millis(30), processed.recv()).await {
        results.push(val);
    }

    assert!(!results.is_empty(), "Should get some results");

    for val in &results {
        assert_eq!(val % 2, 1, "All results should be odd");
    }
}

#[tokio::test]
async fn test_cloneable_messages() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut filtered = receiver.filter(|d: &TestData| d.value > 2).map(|mut d: TestData| {
        d.value *= 2;
        d
    });

    tx.send(TestData { value: 1, flag: true }).unwrap();
    tx.send(TestData { value: 3, flag: false }).unwrap();
    tx.send(TestData { value: 5, flag: true }).unwrap();
    drop(tx);

    let d1 = filtered.recv().await.unwrap();
    assert_eq!(d1.value, 6);

    let d2 = filtered.recv().await.unwrap();
    assert_eq!(d2.value, 10);
}

#[tokio::test]
async fn test_concurrent_processing() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut processed = receiver.map(|x: i32| x * 2);

    let sender_handle = tokio::spawn(async move {
        for i in 0..10 {
            tx.send(i).unwrap();
            sleep(Duration::from_millis(10)).await;
        }
    });

    let mut results = Vec::new();
    for _ in 0..10 {
        if let Some(val) = timeout(Duration::from_millis(100), processed.recv())
            .await
            .ok()
            .flatten()
        {
            results.push(val);
        }
    }

    sender_handle.await.unwrap();

    assert_eq!(results.len(), 10);
    assert_eq!(results[0], 0);
    assert_eq!(results[9], 18);
}

#[tokio::test]
async fn test_backpressure_simulation() {
    let (tx, rx) = mpsc::unbounded_channel();
    let receiver = Receiver::from(rx);

    let mut throttled = receiver.throttle(Duration::from_millis(50)).build();

    for i in 0..100 {
        tx.send(i).unwrap();
    }

    sleep(Duration::from_millis(200)).await;

    let mut count = 0;
    while timeout(Duration::from_millis(30), throttled.recv()).await.is_ok() {
        count += 1;
        if count > 20 {
            break;
        }
    }

    assert!(count < 20, "Throttle should create backpressure effect, got {}", count);
}