mecha10-core 0.6.2

//! 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

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

// Test message types
#[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> {
    // Initialize logging for tests
    let _ = tracing_subscriber::fmt()
        .with_env_filter("debug")
        .with_test_writer()
        .try_init();

    // Create context for test node
    Context::new("test_context_unit").await
}

// =============================================================================
// Basic Subscribe/Publish Tests
// =============================================================================

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

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

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

    // Receive the message
    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 to be running
async fn test_multiple_messages_in_sequence() -> Result<()> {
    let ctx = setup().await?;

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

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

    // Receive all 5 messages in order
    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 to be running
async fn test_publish_before_subscribe() -> Result<()> {
    let ctx = setup().await?;

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

    // Small delay to ensure Redis processes the message
    tokio::time::sleep(Duration::from_millis(50)).await;

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

    // Should still receive the message (Redis Streams persistence)
    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(())
}

// =============================================================================
// Consumer Group Tests
// =============================================================================

#[tokio::test]
#[ignore] // Requires Redis to be running
async fn test_single_consumer_group_behavior() -> Result<()> {
    let ctx1 = Context::new("test_consumer_1").await?;
    let ctx2 = Context::new("test_consumer_1").await?; // Same node ID = same consumer group

    // Both subscribe to same topic (same consumer group)
    let mut rx1 = ctx1.subscribe_raw::<TestMessage>("/test/consumer_group").await?;
    let mut rx2 = ctx2.subscribe_raw::<TestMessage>("/test/consumer_group").await?;

    // Publish 10 messages
    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?;
    }

    // Collect messages from both receivers
    let mut received1 = Vec::new();
    let mut received2 = Vec::new();

    // Try to receive 10 messages total (distributed between rx1 and rx2)
    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());

    // With same consumer group, messages should be distributed
    // Total should be approximately 10 (allowing for Redis timing)
    assert!(
        total_received >= 5,
        "Expected at least 5 messages total, got {}",
        total_received
    );

    // Verify no duplicates (each consumer should get unique messages)
    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();

    // If we got duplicates, the dedup would reduce the length
    assert_eq!(
        all_ids.len(),
        total_received,
        "Detected duplicate messages across consumers"
    );

    Ok(())
}

#[tokio::test]
#[ignore] // Requires Redis to be running
async fn test_different_consumer_groups() -> Result<()> {
    let ctx1 = Context::new("test_consumer_a").await?;
    let ctx2 = Context::new("test_consumer_b").await?; // Different node ID = different consumer group

    // Both subscribe to same topic (different consumer groups)
    let mut rx1 = ctx1.subscribe_raw::<TestMessage>("/test/different_groups").await?;
    let mut rx2 = ctx2.subscribe_raw::<TestMessage>("/test/different_groups").await?;

    // Publish 5 messages
    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?;
    }

    // Both consumers should receive all messages (different consumer groups = broadcast)
    let mut received1 = Vec::new();
    let mut received2 = Vec::new();

    // Collect messages with timeout
    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());

    // Both should receive messages (not necessarily all due to timing)
    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(())
}

// =============================================================================
// Message Serialization/Deserialization Tests
// =============================================================================

#[tokio::test]
#[ignore] // Requires Redis to be running
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?;

    // Create complex message
    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,
    };

    // Publish and receive
    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 to be running
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)) => {
            // Successfully received empty message - test passes
        }
        Ok(None) => panic!("Channel closed unexpectedly"),
        Err(_) => panic!("Timeout waiting for message"),
    }

    Ok(())
}

// =============================================================================
// Error Condition Tests
// =============================================================================

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

    // Subscribe to same topic twice - should succeed
    let mut rx1 = ctx.subscribe_raw::<TestMessage>("/test/double_sub").await?;
    let mut rx2 = ctx.subscribe_raw::<TestMessage>("/test/double_sub").await?;

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

    // Both receivers should get the message (same consumer group behavior)
    let recv1 = timeout(Duration::from_secs(2), rx1.recv()).await;
    let recv2 = timeout(Duration::from_secs(2), rx2.recv()).await;

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

    Ok(())
}

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

    // Publishing to a topic that has no subscribers should still succeed
    // (Redis Streams persists messages even without consumers)
    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 to be running
async fn test_context_with_instance() -> Result<()> {
    let ctx = Context::new("camera_node").await?.with_instance("left");

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

    // Verify instance-scoped publishing works
    let msg = TestMessage {
        id: 1,
        data: "Scoped message".to_string(),
    };

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

    Ok(())
}

// =============================================================================
// Channel Behavior Tests
// =============================================================================

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

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

        // Publish and receive one message
        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"),
        }

        // rx is dropped here
    }

    // Publish another message after receiver is dropped
    let msg2 = TestMessage {
        id: 2,
        data: "Second".to_string(),
    };
    ctx.publish_raw("/test/drop", &msg2).await?;

    // Create new receiver - should get the second message (Redis persistence)
    let mut rx2 = ctx.subscribe_raw::<TestMessage>("/test/drop").await?;

    match timeout(Duration::from_secs(2), rx2.recv()).await {
        Ok(Some(received)) => {
            // Should receive the second message (first was acknowledged)
            assert_eq!(received.id, 2);
        }
        _ => {
            // This is acceptable - message may have been auto-acknowledged
            println!("Second receiver didn't get message (expected with auto-ack)");
        }
    }

    Ok(())
}

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

    // Spawn 5 concurrent publishers
    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);
    }

    // Collect messages
    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,
        }
    }

    // Wait for publishers to complete
    for handle in handles {
        let _ = handle.await;
    }

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

    // Should receive most/all of the 50 messages
    assert!(
        received.len() >= 40,
        "Expected at least 40 messages, got {}",
        received.len()
    );

    // Verify no duplicate counts (all should be unique 0-49)
    received.sort();
    received.dedup();
    assert_eq!(received.len(), received.len(), "Should have no duplicates");

    Ok(())
}

// =============================================================================
// Performance and Stress Tests
// =============================================================================

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

    // Publish 1000 messages as fast as possible
    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);

    // Try to receive all messages
    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);

    // Should receive most messages (allowing for some loss in high throughput)
    assert!(received >= 900, "Expected at least 900 messages, got {}", received);

    Ok(())
}

// =============================================================================
// Error Condition Tests - Redis Connection Issues
// =============================================================================

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

    // Attempt to create context - should fail with connection error
    let result = Context::new("test_invalid_redis").await;

    // Restore default
    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 to be running
async fn test_publish_with_invalid_topic_path() -> Result<()> {
    let ctx = setup().await?;

    // Try various invalid topic paths
    let invalid_topics = vec![
        "",         // Empty
        " ",        // Whitespace
        "/",        // Just root
        "no_slash", // Missing leading slash
    ];

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

        // Some paths may work (Redis is permissive), but we're testing the behavior
        let result = ctx.publish_raw(topic_path, &msg).await;

        // Document the behavior - Redis may accept these
        println!("Topic '{}': {:?}", topic_path, result);
    }

    // Test is informational about topic path handling
    Ok(())
}

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

    // Don't publish anything, just wait for timeout
    let result = timeout(Duration::from_millis(500), rx.recv()).await;

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

    Ok(())
}

// =============================================================================
// Type Safety Tests
// =============================================================================

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

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

    // Try to receive as SimpleCounter (different type)
    // This will fail at deserialization
    let mut rx = ctx.subscribe_raw::<SimpleCounter>("/test/type_mismatch").await?;

    // The receiver may get an error or no message due to deserialization failure
    match timeout(Duration::from_secs(1), rx.recv()).await {
        Ok(Some(_)) => {
            // If deserialization somehow succeeded, that's surprising but okay
            println!("Surprisingly deserialized different type");
        }
        Ok(None) => {
            // Channel closed - acceptable
            println!("Channel closed on type mismatch");
        }
        Err(_) => {
            // Timeout - also acceptable behavior
            println!("Timeout on type mismatch - message may be stuck");
        }
    }

    // Test passes regardless - we're documenting behavior
    Ok(())
}

// =============================================================================
// Node Instance Tests
// =============================================================================

#[tokio::test]
#[ignore] // Requires Redis to be running
async fn test_multiple_instances_same_base_node() -> Result<()> {
    // Create three instances of the same node
    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");

    // Subscribe all three to same base topic
    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?;

    // Publish messages
    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?;
    }

    // Collect from all three
    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
    );

    // Messages should be distributed across the three instances
    let total = left_count + right_count + center_count;
    assert!(total >= 6, "Should receive at least 6 out of 9 messages, got {}", total);

    // At least two instances should get messages (distribution)
    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(())
}

// =============================================================================
// Scoped Publishing Tests
// =============================================================================

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

    // Subscribe to the scoped topic directly
    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(),
    };

    // Publish using raw scoped path
    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 to be running
async fn test_publish_to_instance_explicit() -> Result<()> {
    let ctx = setup().await?;

    // Subscribe to instance-specific topic
    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(),
    };

    // Publish to specific instance
    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(())
}

// =============================================================================
// Topic Discovery Tests
// =============================================================================

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

    // Publish to multiple topics to populate Redis
    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?;
    }

    // Small delay for Redis to process
    tokio::time::sleep(Duration::from_millis(100)).await;

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

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

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

    Ok(())
}

// =============================================================================
// Shutdown and Lifecycle Tests
// =============================================================================

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

    // Should not be shutdown initially
    assert!(!ctx.is_shutdown().await);

    // Request shutdown
    ctx.shutdown().await;

    // Should be shutdown now
    assert!(ctx.is_shutdown().await);

    Ok(())
}

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

    // Spawn task to shutdown after delay
    let ctx_clone = ctx.clone();
    tokio::spawn(async move {
        tokio::time::sleep(Duration::from_millis(100)).await;
        ctx_clone.shutdown().await;
    });

    // Wait for shutdown (with safety timeout)
    match timeout(Duration::from_secs(1), ctx.wait_for_shutdown()).await {
        Ok(_) => {
            // Shutdown completed successfully
            assert!(ctx.is_shutdown().await);
        }
        Err(_) => panic!("Shutdown wait timed out"),
    }

    Ok(())
}

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

    // Shutdown via clone
    ctx2.shutdown().await;

    // Original should see the shutdown
    assert!(ctx1.is_shutdown().await);

    Ok(())
}

// =============================================================================
// Large Message Tests
// =============================================================================

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

    // Create a large message (1MB of data)
    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(())
}

// =============================================================================
// UTF-8 and Special Characters Tests
// =============================================================================

#[tokio::test]
#[ignore] // Requires Redis to be running
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 to be running
async fn test_special_characters_in_topic() -> Result<()> {
    let ctx = setup().await?;

    // Topic with underscores, dots, and dashes
    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(_)) => {
            // Successfully sent and received with special chars
        }
        _ => panic!("Should handle special characters in topic"),
    }

    Ok(())
}

// =============================================================================
// Edge Case Tests
// =============================================================================

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

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

    // Should not crash or leak resources
    Ok(())
}

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

    // Subscribe with two different types to same topic
    let mut rx1 = ctx.subscribe_raw::<TestMessage>("/test/multi_type").await?;
    let mut rx2 = ctx.subscribe_raw::<SimpleCounter>("/test/multi_type").await?;

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

    // First subscriber should get it
    match timeout(Duration::from_millis(500), rx1.recv()).await {
        Ok(Some(_)) => {
            // Got the message as TestMessage
        }
        _ => panic!("First subscriber should receive"),
    }

    // Second subscriber will fail deserialization or timeout
    // This is expected behavior
    match timeout(Duration::from_millis(200), rx2.recv()).await {
        Ok(Some(_)) => {
            // Unexpectedly deserialized (shouldn't happen)
            println!("Warning: Type mismatch deserialized successfully");
        }
        Ok(None) => {
            // Channel closed
        }
        Err(_) => {
            // Timeout - expected for type mismatch
        }
    }

    Ok(())
}

// =============================================================================
// Context Node ID Tests
// =============================================================================

#[tokio::test]
#[ignore] // Requires Redis to be running
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 to be running
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(())
}

// =============================================================================
// Message Ordering Tests
// =============================================================================

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

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

    // Receive and verify order
    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)) => {
                // Verify increasing order
                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(())
}

// =============================================================================
// Receiver Try Recv Tests
// =============================================================================

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

    // Initially empty
    assert!(rx.try_recv().is_err());

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

    // Wait for message to arrive
    tokio::time::sleep(Duration::from_millis(100)).await;

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

    Ok(())
}