use futures::StreamExt;
use mimirs_bridge::{bridge_nats_to_zenoh, bridge_zenoh_to_nats};
use std::sync::Arc;
use testcontainers::runners::AsyncRunner;
use testcontainers_modules::nats::Nats;
use tokio::time::{Duration, sleep};
use tracing::info;

#[tokio::test(flavor = "multi_thread")]
async fn test_bridge_integration_zenoh_to_nats() {
    // 1. Start NATS container
    let nats_node = Nats::default().start().await.unwrap();
    let nats_url = format!(
        "nats://127.0.0.1:{}",
        nats_node.get_host_port_ipv4(4222).await.unwrap()
    );

    // 2. Initialize Zenoh and NATS clients
    let zenoh_session = Arc::new(zenoh::open(zenoh::Config::default()).await.unwrap());
    let nats_client = Arc::new(async_nats::connect(&nats_url).await.unwrap());

    // 3. Start Bridge
    let _z_to_n_handle = tokio::spawn(bridge_zenoh_to_nats(
        zenoh_session.clone(),
        nats_client.clone(),
        "mimirs/signal/**".to_string(),
    ));

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

    // 4. Subscribe to NATS to verify forwarding
    let mut nats_sub = nats_client
        .subscribe("MIMIRS.SIGNAL.TEST_AGENT".to_string())
        .await
        .unwrap();

    // 5. Put something to Zenoh
    let test_payload = b"Odin is watching";
    zenoh_session
        .put("mimirs/signal/test_agent", test_payload)
        .await
        .unwrap();

    // 6. Verify NATS received it
    match tokio::time::timeout(Duration::from_secs(5), nats_sub.next()).await {
        Ok(Some(msg)) => {
            assert_eq!(msg.payload, test_payload.as_slice());
        }
        _ => panic!("Bridge failed to forward Zenoh -> NATS"),
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn test_bridge_nats_to_zenoh() {
    // 1. Start NATS container
    let nats_node = Nats::default().start().await.unwrap();
    let nats_url = format!(
        "nats://127.0.0.1:{}",
        nats_node.get_host_port_ipv4(4222).await.unwrap()
    );

    // 2. Initialize Zenoh and NATS clients
    let zenoh_session = Arc::new(zenoh::open(zenoh::Config::default()).await.unwrap());
    let nats_client = Arc::new(async_nats::connect(&nats_url).await.unwrap());

    // 3. Start Bridge
    let _n_to_z_handle = tokio::spawn(bridge_nats_to_zenoh(
        zenoh_session.clone(),
        nats_client.clone(),
        "MIMIRS.SIGNAL.>".to_string(),
    ));

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

    // 4. Subscribe to Zenoh to verify forwarding
    let subscriber = zenoh_session
        .declare_subscriber("mimirs/signal/asgard")
        .await
        .unwrap();

    // 5. Publish something to NATS
    let test_payload = b"Heimdall at the gate";
    nats_client
        .publish(
            "MIMIRS.SIGNAL.ASGARD".to_string(),
            test_payload.as_slice().into(),
        )
        .await
        .unwrap();

    // 6. Verify Zenoh received it
    match tokio::time::timeout(Duration::from_secs(5), subscriber.recv_async()).await {
        Ok(Ok(sample)) => {
            assert_eq!(sample.payload().to_bytes(), test_payload.as_slice());
        }
        _ => panic!("Bridge failed to forward NATS -> Zenoh"),
    }
}

#[tokio::test(flavor = "multi_thread")]
async fn test_bridge_bidirectional_journey() {
    // 1. Start NATS container
    let nats_node = Nats::default().start().await.unwrap();
    let nats_url = format!(
        "nats://127.0.0.1:{}",
        nats_node.get_host_port_ipv4(4222).await.unwrap()
    );

    // 2. Initialize Zenoh and NATS clients
    let zenoh_session = Arc::new(zenoh::open(zenoh::Config::default()).await.unwrap());
    let nats_client = Arc::new(async_nats::connect(&nats_url).await.unwrap());

    // 3. Start Full Bridge (Both directions)
    tokio::spawn(bridge_zenoh_to_nats(
        zenoh_session.clone(),
        nats_client.clone(),
        "mimirs/signal/**".to_string(),
    ));
    tokio::spawn(bridge_nats_to_zenoh(
        zenoh_session.clone(),
        nats_client.clone(),
        "MIMIRS.SIGNAL.>".to_string(),
    ));

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

    // JOURNEY 1: Zenoh -> NATS
    let mut nats_sub = nats_client
        .subscribe("MIMIRS.SIGNAL.AGENT_ALPHA".to_string())
        .await
        .unwrap();
    zenoh_session
        .put("mimirs/signal/agent_alpha", "Msg From Midgard")
        .await
        .unwrap();

    match tokio::time::timeout(Duration::from_secs(5), nats_sub.next()).await {
        Ok(Some(msg)) => {
            assert_eq!(msg.payload, "Msg From Midgard");
            info!("Journey 1 (Zenoh -> NATS) Successful");
        }
        _ => panic!("Journey 1 failed: Zenoh -> NATS"),
    }

    // JOURNEY 2: NATS -> Zenoh
    let zenoh_sub = zenoh_session
        .declare_subscriber("mimirs/signal/agent_beta")
        .await
        .unwrap();
    nats_client
        .publish(
            "MIMIRS.SIGNAL.AGENT_BETA".to_string(),
            "Msg From Asgard".into(),
        )
        .await
        .unwrap();

    match tokio::time::timeout(Duration::from_secs(5), zenoh_sub.recv_async()).await {
        Ok(Ok(sample)) => {
            assert_eq!(sample.payload().to_bytes(), "Msg From Asgard".as_bytes());
            info!("Journey 2 (NATS -> Zenoh) Successful");
        }
        _ => panic!("Journey 2 failed: NATS -> Zenoh"),
    }
}
