mecha10-core 4.2.0

//! Unit tests for Context module
//!
//! Tests core messaging functionality including:
//! - Subscribe/publish operations
//! - Consumer group behavior
//! - Error conditions
//! - Message serialization/deserialization
//! - Bounded vs unbounded channels
//!
//! These tests require a running Redis instance on localhost:6379
//! Run with: cargo test --package mecha10-core --test context_unit_tests

#![allow(
    clippy::print_stdout,
    clippy::print_stderr,
    reason = "diagnostic output printed during test runs (visible with --nocapture)"
)]

use mecha10_core::prelude::*;
use serde::{Deserialize, Serialize};
use std::time::Duration;
use tokio::time::timeout;

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
struct TestMessage {
    id: u32,
    data: String,
}

impl Message for TestMessage {}

#[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
struct SimpleCounter {
    count: u32,
}

impl Message for SimpleCounter {}

/// Helper to setup test environment
async fn setup() -> Result<Context> {
    let _ = tracing_subscriber::fmt()
        .with_env_filter("debug")
        .with_test_writer()
        .try_init();

    Context::new("test_context_unit").await
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_basic_publish_subscribe() -> Result<()> {
    let ctx = setup().await?;

    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/basic_pubsub").await?;

    let msg = TestMessage {
        id: 1,
        data: "Hello, World!".to_string(),
    };
    ctx.publish_raw("/test/basic_pubsub", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, msg.id);
            assert_eq!(received.data, msg.data);
        }
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_multiple_messages_in_sequence() -> Result<()> {
    let ctx = setup().await?;

    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/sequence").await?;

    for i in 0..5 {
        let msg = TestMessage {
            id: i,
            data: format!("Message {}", i),
        };
        ctx.publish_raw("/test/sequence", &msg).await?;
    }

    for i in 0..5 {
        match timeout(Duration::from_secs(2), rx.recv()).await {
            Ok(Some(received)) => {
                assert_eq!(received.id, i);
                assert_eq!(received.data, format!("Message {}", i));
            }
            Ok(None) => panic!("Channel closed at message {}", i),
            Err(_) => panic!("Timeout at message {}", i),
        }
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_publish_before_subscribe() -> Result<()> {
    let ctx = setup().await?;

    let msg = TestMessage {
        id: 100,
        data: "Early bird".to_string(),
    };
    ctx.publish_raw("/test/early_pub", &msg).await?;

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

    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/early_pub").await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, msg.id);
            assert_eq!(received.data, msg.data);
        }
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for message - Redis may not have persisted it"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_single_consumer_group_behavior() -> Result<()> {
    let ctx1 = Context::new("test_consumer_1").await?;
    let ctx2 = Context::new("test_consumer_1").await?;

    let mut rx1 = ctx1.subscribe_raw::<TestMessage>("/test/consumer_group").await?;
    let mut rx2 = ctx2.subscribe_raw::<TestMessage>("/test/consumer_group").await?;

    let publisher = Context::new("test_publisher").await?;
    for i in 0..10 {
        let msg = TestMessage {
            id: i,
            data: format!("Message {}", i),
        };
        publisher.publish_raw("/test/consumer_group", &msg).await?;
    }

    let mut received1 = Vec::new();
    let mut received2 = Vec::new();

    for _ in 0..10 {
        tokio::select! {
            msg = rx1.recv() => {
                if let Some(m) = msg {
                    received1.push(m);
                }
            }
            msg = rx2.recv() => {
                if let Some(m) = msg {
                    received2.push(m);
                }
            }
            _ = tokio::time::sleep(Duration::from_millis(100)) => {
                break;
            }
        }
    }

    let total_received = received1.len() + received2.len();
    println!("rx1 received: {}, rx2 received: {}", received1.len(), received2.len());

    assert!(
        total_received >= 5,
        "Expected at least 5 messages total, got {}",
        total_received
    );

    let mut all_ids: Vec<u32> = received1.iter().map(|m| m.id).collect();
    all_ids.extend(received2.iter().map(|m| m.id));
    all_ids.sort();
    all_ids.dedup();

    assert_eq!(
        all_ids.len(),
        total_received,
        "Detected duplicate messages across consumers"
    );

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_different_consumer_groups() -> Result<()> {
    let ctx1 = Context::new("test_consumer_a").await?;
    let ctx2 = Context::new("test_consumer_b").await?;

    let mut rx1 = ctx1.subscribe_raw::<TestMessage>("/test/different_groups").await?;
    let mut rx2 = ctx2.subscribe_raw::<TestMessage>("/test/different_groups").await?;

    let publisher = Context::new("test_group_publisher").await?;
    for i in 0..5 {
        let msg = TestMessage {
            id: i,
            data: format!("Broadcast {}", i),
        };
        publisher.publish_raw("/test/different_groups", &msg).await?;
    }

    let mut received1 = Vec::new();
    let mut received2 = Vec::new();

    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline && (received1.len() < 5 || received2.len() < 5) {
        tokio::select! {
            msg = rx1.recv() => {
                if let Some(m) = msg {
                    received1.push(m);
                }
            }
            msg = rx2.recv() => {
                if let Some(m) = msg {
                    received2.push(m);
                }
            }
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
    }

    println!("rx1 received: {}, rx2 received: {}", received1.len(), received2.len());

    assert!(
        received1.len() >= 3,
        "rx1 should receive at least 3 messages, got {}",
        received1.len()
    );
    assert!(
        received2.len() >= 3,
        "rx2 should receive at least 3 messages, got {}",
        received2.len()
    );

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_complex_message_serialization() -> Result<()> {
    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    struct ComplexMessage {
        id: u64,
        name: String,
        values: Vec<f64>,
        metadata: std::collections::HashMap<String, String>,
    }
    impl Message for ComplexMessage {}

    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<ComplexMessage>("/test/complex").await?;

    let mut metadata = std::collections::HashMap::new();
    metadata.insert("key1".to_string(), "value1".to_string());
    metadata.insert("key2".to_string(), "value2".to_string());

    let msg = ComplexMessage {
        id: 42,
        name: "Complex Test".to_string(),
        values: vec![1.1, 2.2, 3.3, 4.4],
        metadata,
    };

    ctx.publish_raw("/test/complex", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, msg.id);
            assert_eq!(received.name, msg.name);
            assert_eq!(received.values, msg.values);
            assert_eq!(received.metadata, msg.metadata);
        }
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_empty_message() -> Result<()> {
    #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
    struct EmptyMessage {}
    impl Message for EmptyMessage {}

    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<EmptyMessage>("/test/empty").await?;

    let msg = EmptyMessage {};
    ctx.publish_raw("/test/empty", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(_received)) => {}
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_subscribe_to_same_topic_twice() -> Result<()> {
    let ctx = setup().await?;

    let mut rx1 = ctx.subscribe_raw::<TestMessage>("/test/double_sub").await?;
    let mut rx2 = ctx.subscribe_raw::<TestMessage>("/test/double_sub").await?;

    let msg = TestMessage {
        id: 999,
        data: "Double sub test".to_string(),
    };
    ctx.publish_raw("/test/double_sub", &msg).await?;

    let recv1 = timeout(Duration::from_secs(2), rx1.recv()).await;
    let recv2 = timeout(Duration::from_secs(2), rx2.recv()).await;

    assert!(
        recv1.is_ok() || recv2.is_ok(),
        "At least one receiver should get the message"
    );

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_publish_to_nonexistent_topic() -> Result<()> {
    let ctx = setup().await?;

    let msg = TestMessage {
        id: 404,
        data: "Nobody listening".to_string(),
    };

    let result = ctx.publish_raw("/test/nonexistent", &msg).await;
    assert!(result.is_ok(), "Publishing to nonexistent topic should succeed");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_context_with_instance() -> Result<()> {
    let ctx = Context::new("camera_node").await?.with_instance("left");

    assert_eq!(ctx.instance(), Some("left"));

    let msg = TestMessage {
        id: 1,
        data: "Scoped message".to_string(),
    };

    let result = ctx.publish_raw("/test/scoped", &msg).await;
    assert!(result.is_ok());

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_receiver_drop_stops_forwarding() -> Result<()> {
    let ctx = setup().await?;

    {
        let mut rx = ctx.subscribe_raw::<TestMessage>("/test/drop").await?;

        let msg = TestMessage {
            id: 1,
            data: "First".to_string(),
        };
        ctx.publish_raw("/test/drop", &msg).await?;

        match timeout(Duration::from_secs(2), rx.recv()).await {
            Ok(Some(received)) => {
                assert_eq!(received.id, 1);
            }
            _ => panic!("Should receive first message"),
        }
    }

    let msg2 = TestMessage {
        id: 2,
        data: "Second".to_string(),
    };
    ctx.publish_raw("/test/drop", &msg2).await?;

    let mut rx2 = ctx.subscribe_raw::<TestMessage>("/test/drop").await?;

    match timeout(Duration::from_secs(2), rx2.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, 2);
        }
        _ => {
            println!("Second receiver didn't get message (expected with auto-ack)");
        }
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_concurrent_publishers() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<SimpleCounter>("/test/concurrent").await?;

    let mut handles = Vec::new();
    for i in 0..5 {
        let ctx_clone = Context::new(&format!("publisher_{}", i)).await?;
        let handle = tokio::spawn(async move {
            for j in 0..10 {
                let msg = SimpleCounter { count: i * 10 + j };
                ctx_clone.publish_raw("/test/concurrent", &msg).await.unwrap();
                tokio::time::sleep(Duration::from_millis(5)).await;
            }
        });
        handles.push(handle);
    }

    let mut received = Vec::new();
    let deadline = tokio::time::Instant::now() + Duration::from_secs(5);

    while tokio::time::Instant::now() < deadline && received.len() < 50 {
        match timeout(Duration::from_millis(100), rx.recv()).await {
            Ok(Some(msg)) => {
                received.push(msg.count);
            }
            Ok(None) => break,
            Err(_) => break,
        }
    }

    for handle in handles {
        let _ = handle.await;
    }

    println!("Received {} messages from concurrent publishers", received.len());

    assert!(
        received.len() >= 40,
        "Expected at least 40 messages, got {}",
        received.len()
    );

    received.sort();
    received.dedup();
    assert_eq!(received.len(), received.len(), "Should have no duplicates");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_high_throughput_messaging() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<SimpleCounter>("/test/throughput").await?;

    let start = std::time::Instant::now();
    for i in 0..1000 {
        let msg = SimpleCounter { count: i };
        ctx.publish_raw("/test/throughput", &msg).await?;
    }
    let publish_duration = start.elapsed();

    println!("Published 1000 messages in {:?}", publish_duration);

    let mut received = 0;
    let deadline = tokio::time::Instant::now() + Duration::from_secs(10);

    while tokio::time::Instant::now() < deadline && received < 1000 {
        match timeout(Duration::from_millis(10), rx.recv()).await {
            Ok(Some(_)) => {
                received += 1;
            }
            Ok(None) => break,
            Err(_) => {
                if received > 0 {
                    break;
                }
            }
        }
    }

    println!("Received {} out of 1000 messages", received);

    assert!(received >= 900, "Expected at least 900 messages, got {}", received);

    Ok(())
}

#[tokio::test]
async fn test_context_creation_with_invalid_redis() -> Result<()> {
    std::env::set_var("MECHA10_REDIS_URL", "redis://localhost:9999");

    let result = timeout(Duration::from_secs(5), Context::new("test_invalid_redis"))
        .await
        .expect("Context::new should fail fast, not hang, against an unreachable Redis port");

    std::env::remove_var("MECHA10_REDIS_URL");

    assert!(result.is_err(), "Should fail to connect to invalid Redis");

    if let Err(e) = result {
        let error_msg = format!("{}", e);
        assert!(
            error_msg.contains("Redis") || error_msg.contains("connect") || error_msg.contains("connection"),
            "Error should mention Redis or connection, got: {}",
            error_msg
        );
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_publish_with_invalid_topic_path() -> Result<()> {
    let ctx = setup().await?;

    let invalid_topics = vec!["", " ", "/", "no_slash"];

    for topic_path in invalid_topics {
        let msg = TestMessage {
            id: 1,
            data: "test".to_string(),
        };

        let result = ctx.publish_raw(topic_path, &msg).await;

        println!("Topic '{}': {:?}", topic_path, result);
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_subscribe_timeout_no_messages() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/timeout").await?;

    let result = timeout(Duration::from_millis(500), rx.recv()).await;

    assert!(result.is_err(), "Should timeout when no messages published");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_type_mismatch_handling() -> Result<()> {
    let ctx = setup().await?;

    let msg = TestMessage {
        id: 42,
        data: "Type test".to_string(),
    };
    ctx.publish_raw("/test/type_mismatch", &msg).await?;

    let mut rx = ctx.subscribe_raw::<SimpleCounter>("/test/type_mismatch").await?;

    match timeout(Duration::from_secs(1), rx.recv()).await {
        Ok(Some(_)) => {
            println!("Surprisingly deserialized different type");
        }
        Ok(None) => {
            println!("Channel closed on type mismatch");
        }
        Err(_) => {
            println!("Timeout on type mismatch - message may be stuck");
        }
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_multiple_instances_same_base_node() -> Result<()> {
    let ctx_left = Context::new("camera").await?.with_instance("left");
    let ctx_right = Context::new("camera").await?.with_instance("right");
    let ctx_center = Context::new("camera").await?.with_instance("center");

    let mut rx_left = ctx_left.subscribe_raw::<TestMessage>("/test/instances").await?;
    let mut rx_right = ctx_right.subscribe_raw::<TestMessage>("/test/instances").await?;
    let mut rx_center = ctx_center.subscribe_raw::<TestMessage>("/test/instances").await?;

    let publisher = Context::new("publisher").await?;
    for i in 0..9 {
        let msg = TestMessage {
            id: i,
            data: format!("Message {}", i),
        };
        publisher.publish_raw("/test/instances", &msg).await?;
    }

    let mut left_count = 0;
    let mut right_count = 0;
    let mut center_count = 0;

    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline && (left_count + right_count + center_count) < 9 {
        tokio::select! {
            msg = rx_left.recv() => {
                if msg.is_some() {
                    left_count += 1;
                }
            }
            msg = rx_right.recv() => {
                if msg.is_some() {
                    right_count += 1;
                }
            }
            msg = rx_center.recv() => {
                if msg.is_some() {
                    center_count += 1;
                }
            }
            _ = tokio::time::sleep(Duration::from_millis(50)) => {}
        }
    }

    println!(
        "Distribution: left={}, right={}, center={}",
        left_count, right_count, center_count
    );

    let total = left_count + right_count + center_count;
    assert!(total >= 6, "Should receive at least 6 out of 9 messages, got {}", total);

    let instances_with_messages = [left_count > 0, right_count > 0, center_count > 0]
        .iter()
        .filter(|&&x| x)
        .count();
    assert!(
        instances_with_messages >= 2,
        "At least 2 instances should receive messages"
    );

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_publish_to_scoped() -> Result<()> {
    let ctx = Context::new("test_scoped_pub").await?.with_instance("instance1");

    let scoped_topic = "/test/scoped/instance1";
    let mut rx = ctx.subscribe_raw::<TestMessage>(scoped_topic).await?;

    let msg = TestMessage {
        id: 42,
        data: "Scoped message".to_string(),
    };

    ctx.publish_raw("/test/scoped/instance1", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, 42);
        }
        _ => panic!("Should receive scoped message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_publish_to_instance_explicit() -> Result<()> {
    let ctx = setup().await?;

    let instance_topic = "/test/instance_pub/robot_1";
    let mut rx = ctx.subscribe_raw::<TestMessage>(instance_topic).await?;

    let msg = TestMessage {
        id: 99,
        data: "Instance message".to_string(),
    };

    ctx.publish_raw("/test/instance_pub/robot_1", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, 99);
        }
        _ => panic!("Should receive instance message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_discover_topics_basic() -> Result<()> {
    let ctx = setup().await?;

    let topics = vec![
        "/discovery/sensor/camera/front",
        "/discovery/sensor/camera/back",
        "/discovery/sensor/lidar",
    ];

    let msg = TestMessage {
        id: 1,
        data: "Discovery test".to_string(),
    };

    for topic in &topics {
        ctx.publish_raw(topic, &msg).await?;
    }

    tokio::time::sleep(Duration::from_millis(100)).await;

    let discovered = ctx.discover_topics("/discovery/sensor/camera/*").await?;

    println!("Discovered topics: {:?}", discovered);

    assert!(discovered.len() >= 2, "Should discover at least 2 camera topics");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_shutdown_flag() -> Result<()> {
    let ctx = setup().await?;

    assert!(!ctx.is_shutdown().await);

    ctx.shutdown().await;

    assert!(ctx.is_shutdown().await);

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_wait_for_shutdown_with_timeout() -> Result<()> {
    let ctx = setup().await?;

    let ctx_clone = ctx.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(100)).await;
        ctx_clone.shutdown().await;
    });

    match timeout(Duration::from_secs(1), ctx.wait_for_shutdown()).await {
        Ok(_) => {
            assert!(ctx.is_shutdown().await);
        }
        Err(_) => panic!("Shutdown wait timed out"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_context_clone_shares_state() -> Result<()> {
    let ctx1 = setup().await?;
    let ctx2 = ctx1.clone();

    ctx2.shutdown().await;

    assert!(ctx1.is_shutdown().await);

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_large_message_transfer() -> Result<()> {
    let ctx = setup().await?;

    let large_data = "x".repeat(1024 * 1024);
    let msg = TestMessage {
        id: 1,
        data: large_data.clone(),
    };

    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/large_message").await?;

    ctx.publish_raw("/test/large_message", &msg).await?;

    match timeout(Duration::from_secs(5), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.id, 1);
            assert_eq!(received.data.len(), large_data.len());
        }
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for large message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_unicode_message_data() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/unicode").await?;

    let msg = TestMessage {
        id: 1,
        data: "Hello 世界 🌍 Здравствуй мир".to_string(),
    };

    ctx.publish_raw("/test/unicode", &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(received)) => {
            assert_eq!(received.data, msg.data);
        }
        _ => panic!("Should receive unicode message"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_special_characters_in_topic() -> Result<()> {
    let ctx = setup().await?;

    let topic = "/test/special_chars/topic-1.0";
    let mut rx = ctx.subscribe_raw::<TestMessage>(topic).await?;

    let msg = TestMessage {
        id: 1,
        data: "Special topic".to_string(),
    };

    ctx.publish_raw(topic, &msg).await?;

    match timeout(Duration::from_secs(2), rx.recv()).await {
        Ok(Some(_)) => {}
        _ => panic!("Should handle special characters in topic"),
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_rapid_subscribe_unsubscribe() -> Result<()> {
    let ctx = setup().await?;

    for i in 0..10 {
        let topic = format!("/test/rapid_sub/{}", i);
        let rx = ctx.subscribe_raw::<TestMessage>(&topic).await?;
        drop(rx);
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_subscribe_to_same_topic_different_types() -> Result<()> {
    let ctx = setup().await?;

    let mut rx1 = ctx.subscribe_raw::<TestMessage>("/test/multi_type").await?;
    let mut rx2 = ctx.subscribe_raw::<SimpleCounter>("/test/multi_type").await?;

    let msg = TestMessage {
        id: 1,
        data: "test".to_string(),
    };
    ctx.publish_raw("/test/multi_type", &msg).await?;

    match timeout(Duration::from_millis(500), rx1.recv()).await {
        Ok(Some(_)) => {}
        _ => panic!("First subscriber should receive"),
    }

    match timeout(Duration::from_millis(200), rx2.recv()).await {
        Ok(Some(_)) => {
            println!("Warning: Type mismatch deserialized successfully");
        }
        Ok(None) => {}
        Err(_) => {}
    }

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_node_id_retrieval() -> Result<()> {
    let ctx = Context::new("test_node_id_retrieval").await?;
    assert_eq!(ctx.node_id(), "test_node_id_retrieval");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_node_id_with_instance() -> Result<()> {
    let ctx = Context::new("test_node").await?.with_instance("instance_1");

    assert_eq!(ctx.node_id(), "test_node/instance_1");
    assert_eq!(ctx.instance(), Some("instance_1"));

    Ok(())
}

/// `LAB-4131`: node-runner sets `NODE_INSTANCE_ID` (not a per-node custom env var) on a
/// multi-instance node's process environment, so `Context::new` must pick it up the same way
/// it already picks up `NODE_NAME` - no per-node opt-in or explicit `.with_instance(..)` call
/// required.
#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
#[serial_test::serial(node_instance_id_env)]
async fn test_node_id_picks_up_node_instance_id_env_var() -> Result<()> {
    std::env::set_var("NODE_INSTANCE_ID", "real");
    let ctx = Context::new("camera").await;
    std::env::remove_var("NODE_INSTANCE_ID");
    let ctx = ctx?;

    assert_eq!(ctx.node_id(), "camera/real");
    assert_eq!(ctx.instance(), Some("real"));

    Ok(())
}

/// An empty `NODE_INSTANCE_ID` (e.g. set but blank in a process environment) must behave like
/// it's unset, not produce a trailing-slash `node_id`.
#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
#[serial_test::serial(node_instance_id_env)]
async fn test_node_id_ignores_empty_node_instance_id_env_var() -> Result<()> {
    std::env::set_var("NODE_INSTANCE_ID", "");
    let ctx = Context::new("camera").await;
    std::env::remove_var("NODE_INSTANCE_ID");
    let ctx = ctx?;

    assert_eq!(ctx.node_id(), "camera");
    assert_eq!(ctx.instance(), None);

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_message_ordering_preserved() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<SimpleCounter>("/test/ordering").await?;

    for i in 0..20 {
        let msg = SimpleCounter { count: i };
        ctx.publish_raw("/test/ordering", &msg).await?;
    }

    let mut prev_count = None;
    let mut received = 0;

    let deadline = tokio::time::Instant::now() + Duration::from_secs(3);
    while tokio::time::Instant::now() < deadline && received < 20 {
        match timeout(Duration::from_millis(100), rx.recv()).await {
            Ok(Some(msg)) => {
                if let Some(prev) = prev_count {
                    assert!(
                        msg.count > prev,
                        "Messages should be in order: got {} after {}",
                        msg.count,
                        prev
                    );
                }
                prev_count = Some(msg.count);
                received += 1;
            }
            Ok(None) => break,
            Err(_) => break,
        }
    }

    assert!(received >= 15, "Should receive most messages in order");

    Ok(())
}

#[tokio::test]
#[ignore = "Requires Redis infrastructure"]
async fn test_receiver_try_recv_integration() -> Result<()> {
    let ctx = setup().await?;
    let mut rx = ctx.subscribe_raw::<TestMessage>("/test/try_recv").await?;

    assert!(rx.try_recv().is_err());

    let msg = TestMessage {
        id: 42,
        data: "Try recv test".to_string(),
    };
    ctx.publish_raw("/test/try_recv", &msg).await?;

    tokio::time::sleep(Duration::from_millis(100)).await;

    let received = rx.try_recv();
    assert!(received.is_ok(), "Should have message available");

    Ok(())
}