//! End-to-end JSON-RPC over real NATS via testcontainers.
//! Gated on `KLIEO_RUN_INTEGRATION=1` to keep `cargo test` fast.
//! Mirror of `crates/klieo-bus-nats/tests/bus_nats_smoke.rs` Plan #9 T8.

#![cfg(test)]

use bytes::Bytes;
use klieo_a2a::{A2aServer, EchoHandler, JsonRpcRequest, JsonRpcResponse};
use klieo_auth_common::AllowAnonymous;
use klieo_bus_nats::{NatsBus, NatsBusConfig};
use klieo_core::{DurableName, Headers};
use serde_json::json;
use std::sync::Arc;
use std::time::Duration;
use testcontainers::{core::WaitFor, runners::AsyncRunner, GenericImage, ImageExt};
use tokio_stream::StreamExt;

fn integration_enabled() -> bool {
    std::env::var("KLIEO_RUN_INTEGRATION").as_deref() == Ok("1")
}

async fn boot_bus() -> (
    NatsBus,
    testcontainers::ContainerAsync<GenericImage>,
    String,
) {
    let container = GenericImage::new("nats", "2.10")
        .with_exposed_port(4222_u16.into())
        .with_wait_for(WaitFor::message_on_stderr("Server is ready"))
        .with_cmd(vec!["-js"])
        .start()
        .await
        .expect("failed to start nats container");
    let host = container.get_host().await.expect("failed to get nats host");
    let port = container
        .get_host_port_ipv4(4222)
        .await
        .expect("failed to get nats port");
    let url = format!("nats://{host}:{port}");
    let cfg = NatsBusConfig {
        url: url.clone(),
        app_prefix: "test".into(),
        ..NatsBusConfig::default()
    };
    let bus = NatsBus::connect(cfg).await.expect("connect to nats");
    (bus, container, url)
}

#[tokio::test]
async fn send_message_round_trip_over_nats() {
    if !integration_enabled() {
        eprintln!("skipping (set KLIEO_RUN_INTEGRATION=1 to run)");
        return;
    }
    let (bus, _container, _url) = boot_bus().await;

    let server = A2aServer::new(
        "test".into(),
        "agent-x".into(),
        Arc::new(EchoHandler::default()),
        Arc::new(AllowAnonymous),
        bus.pubsub.clone(),
    );
    tokio::spawn(async move { server.run().await.unwrap() });
    tokio::time::sleep(Duration::from_millis(200)).await;

    let reply = "test.reply.nats";
    let mut reply_stream = bus
        .pubsub
        .subscribe(reply, DurableName::new("nats-test-reply"))
        .await
        .unwrap();

    let req = JsonRpcRequest {
        jsonrpc: "2.0".into(),
        id: json!(1),
        method: "SendMessage".into(),
        params: json!({
            "message": {
                "messageId": "m-1",
                "contextId": "c-1",
                "role": "user",
                "parts": [{"type": "text", "content": "hello over nats"}]
            }
        }),
    };
    let mut headers = Headers::new();
    headers.insert("Reply-To".into(), reply.into());
    headers.insert("A2A-Version".into(), "1.0".into());
    bus.pubsub
        .publish(
            "test.a2a.agent-x.rpc",
            Bytes::from(serde_json::to_vec(&req).unwrap()),
            headers,
        )
        .await
        .unwrap();

    let msg = tokio::time::timeout(Duration::from_secs(5), reply_stream.next())
        .await
        .expect("reply timeout")
        .expect("ended")
        .expect("bus error");
    let resp: JsonRpcResponse = serde_json::from_slice(&msg.payload).unwrap();
    assert!(resp.error.is_none(), "{:?}", resp.error);
}
