// Gated behind the `slow` feature (WOR-982): this suite spins up a local
// wiremock HTTP server, so it is excluded from the default fast gate. Run with
// `cargo test -p mcptest-core --features slow`.
#![cfg(feature = "slow")]

//! Integration tests for the streamable HTTP transport.
//!
//! Exercises the end-to-end wire path against a `wiremock` server:
//! happy-path POST + JSON response, session-id propagation, SSE
//! notification routing, retry on 503, auth header injection, and
//! per-request timeout.

use std::collections::HashMap;
use std::time::Duration;

use mcptest_core::transport::{
    StreamableHttp, StreamableHttpConfig, STREAMABLE_HTTP_SESSION_HEADER,
};
use serde_json::{json, Value};
use wiremock::matchers::{header, header_exists, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

const POST_PATH: &str = "/mcp";

async fn start_server() -> MockServer {
    MockServer::start().await
}

fn cfg_for(url: String) -> StreamableHttpConfig {
    StreamableHttpConfig {
        url,
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..Default::default()
    }
}

fn request_msg(id: i64, method: &str) -> mcptest_core::protocol::JsonRpcMessage {
    use mcptest_core::protocol::{JsonRpcId, JsonRpcMessage};
    JsonRpcMessage::request(JsonRpcId::Number(id), method, json!({}))
}

#[tokio::test]
async fn happy_path_post_returns_json_response_and_routes_to_inbound() {
    let server = start_server().await;
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"ok": true}
                })),
        )
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let (mut transport, handle) = StreamableHttp::connect(cfg_for(url)).expect("connect");

    transport
        .outbound_tx
        .send(request_msg(1, "ping"))
        .expect("send");
    let envelope = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("response in time")
        .expect("envelope");
    assert_eq!(envelope.result, Some(json!({"ok": true})));

    handle.shutdown().await;
}

#[tokio::test]
async fn session_id_from_first_response_is_sent_on_followups() {
    let server = start_server().await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .insert_header(STREAMABLE_HTTP_SESSION_HEADER, "sess-xyz")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"first": true}
                })),
        )
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header(STREAMABLE_HTTP_SESSION_HEADER, "sess-xyz"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 2,
                    "result": {"second": true}
                })),
        )
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let (mut transport, handle) = StreamableHttp::connect(cfg_for(url)).expect("connect");

    transport.outbound_tx.send(request_msg(1, "first")).unwrap();
    let _ = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("first response")
        .expect("envelope");

    transport
        .outbound_tx
        .send(request_msg(2, "second"))
        .unwrap();
    let second = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("second response")
        .expect("envelope");
    assert_eq!(second.result, Some(json!({"second": true})));

    handle.shutdown().await;
}

#[tokio::test]
async fn sse_response_to_post_routes_every_envelope_through_inbound() {
    let server = start_server().await;

    // SSE body with two events: a response to the POST plus a server
    // initiated notification.
    let sse_body = concat!(
        "event: message\n",
        "data: {\"jsonrpc\":\"2.0\",\"id\":1,\"result\":{\"ok\":true}}\n",
        "\n",
        "event: message\n",
        "data: {\"jsonrpc\":\"2.0\",\"method\":\"progress\",\"params\":{\"pct\":50}}\n",
        "\n",
    );

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(ResponseTemplate::new(200).set_body_raw(sse_body, "text/event-stream"))
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let (mut transport, handle) = StreamableHttp::connect(cfg_for(url)).expect("connect");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();

    let m1 = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("response")
        .expect("envelope");
    let m2 = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("notification")
        .expect("envelope");
    assert!(m1.result.is_some(), "first event should be a response");
    assert_eq!(m2.method.as_deref(), Some("progress"));

    handle.shutdown().await;
}

#[tokio::test]
async fn transient_5xx_is_retried_then_succeeds() {
    let server = start_server().await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(ResponseTemplate::new(503))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"recovered": true}
                })),
        )
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let mut cfg = cfg_for(url);
    cfg.request_timeout = Duration::from_secs(2);
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();

    let env = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("response after retry")
        .expect("envelope");
    assert_eq!(env.result, Some(json!({"recovered": true})));

    handle.shutdown().await;
}

#[tokio::test]
async fn auth_headers_and_static_headers_are_injected_on_every_post() {
    let server = start_server().await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header("authorization", "Bearer my-token"))
        .and(header("x-api-key", "k123"))
        .and(header_exists("content-type"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"authed": true}
                })),
        )
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let mut headers = HashMap::new();
    headers.insert("X-API-Key".to_string(), "k123".to_string());
    let cfg = StreamableHttpConfig {
        url,
        headers,
        bearer_token: Some("my-token".into()),
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..StreamableHttpConfig::default()
    };
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();

    let env = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("response")
        .expect("envelope");
    assert_eq!(env.result, Some(json!({"authed": true})));

    handle.shutdown().await;
}

#[tokio::test]
async fn slow_server_eventually_fails_per_request_timeout() {
    let server = start_server().await;

    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"ok": true}
                }))
                .set_delay(Duration::from_secs(2)),
        )
        .mount(&server)
        .await;

    let url = format!("{}{}", server.uri(), POST_PATH);
    let mut cfg = cfg_for(url);
    cfg.request_timeout = Duration::from_millis(150);
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect");
    transport.outbound_tx.send(request_msg(1, "slow")).unwrap();

    // We expect the inbound channel to close (writer task gives up
    // after exhausting retries) before any envelope arrives.
    let result = tokio::time::timeout(Duration::from_secs(5), transport.inbound_rx.recv()).await;
    match result {
        Ok(Some(_)) => panic!("expected no successful response on timeout"),
        Ok(None) => {}      // channel closed: writer gave up.
        Err(_elapsed) => {} // also acceptable: nothing arrived in window.
    }

    handle.shutdown().await;
}

#[tokio::test]
async fn mtls_identity_composes_with_bearer_and_still_posts() {
    // mTLS is applied at client-build time; this test proves that a
    // configured client identity does not break the request path and
    // composes with a bearer token. The mock does not require a client
    // cert (wiremock has no client-cert matcher), so success here means
    // the identity loaded and the client built and sent the POST.
    let server = start_server().await;
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header("authorization", "Bearer my-token"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"mtls": true}
                })),
        )
        .mount(&server)
        .await;

    let cert = rcgen::generate_simple_self_signed(vec!["mcptest.invalid".to_string()])
        .expect("self-signed cert");
    let dir = tempfile::tempdir().unwrap();
    let cert_path = dir.path().join("client.pem");
    let key_path = dir.path().join("client.key");
    std::fs::write(&cert_path, cert.cert.pem()).unwrap();
    std::fs::write(&key_path, cert.signing_key.serialize_pem()).unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        let mut perms = std::fs::metadata(&key_path).unwrap().permissions();
        perms.set_mode(0o600);
        std::fs::set_permissions(&key_path, perms).unwrap();
    }

    let url = format!("{}{}", server.uri(), POST_PATH);
    let cfg = StreamableHttpConfig {
        url,
        bearer_token: Some("my-token".into()),
        mtls: Some(mcptest_core::auth::MtlsConfig {
            client_cert_path: Some(cert_path),
            client_key_path: Some(key_path),
            ..Default::default()
        }),
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..StreamableHttpConfig::default()
    };
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect with mtls identity");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();
    let env = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("response")
        .expect("envelope");
    assert_eq!(env.result, Some(json!({"mtls": true})));
    handle.shutdown().await;
}

#[tokio::test]
async fn sigv4_signs_every_post_with_an_authorization_header() {
    use wiremock::matchers::header_regex;

    let server = start_server().await;
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        // The SigV4 Authorization header is well-formed and present.
        .and(header_regex(
            "authorization",
            r"^AWS4-HMAC-SHA256 Credential=.+/execute-api/aws4_request",
        ))
        .and(header_exists("x-amz-date"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"signed": true}
                })),
        )
        .mount(&server)
        .await;

    // Set credentials in the explicit env vars the config names. Use a
    // process-unique name so parallel tests do not clash.
    let pid = std::process::id();
    let ak = format!("MCPTEST_SIGV4_AK_{pid}");
    let sk = format!("MCPTEST_SIGV4_SK_{pid}");
    std::env::set_var(&ak, "AKIDEXAMPLE");
    std::env::set_var(&sk, "wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY");

    let url = format!("{}{}", server.uri(), POST_PATH);
    let cfg = StreamableHttpConfig {
        url,
        sigv4: Some(mcptest_core::auth::SigV4Config {
            region: "us-east-1".into(),
            service: "execute-api".into(),
            credentials: mcptest_core::auth::SigV4Credentials::Explicit {
                access_key_id_env: ak.clone(),
                secret_access_key_env: sk.clone(),
                session_token_env: None,
            },
            signature_lifetime: None,
        }),
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..StreamableHttpConfig::default()
    };
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect with sigv4");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();
    let env = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("response")
        .expect("envelope");
    std::env::remove_var(&ak);
    std::env::remove_var(&sk);
    assert_eq!(env.result, Some(json!({"signed": true})));
    handle.shutdown().await;
}

#[tokio::test]
async fn web_bot_auth_signs_every_post_with_signature_headers() {
    use ed25519_dalek::pkcs8::EncodePrivateKey;
    use wiremock::matchers::header_regex;

    let server = start_server().await;
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .and(header_regex("signature-input", r#"^sig1=\("@authority""#))
        .and(header_exists("signature"))
        .and(header_exists("signature-agent"))
        .respond_with(
            ResponseTemplate::new(200)
                .insert_header("content-type", "application/json")
                .set_body_json(json!({
                    "jsonrpc": "2.0",
                    "id": 1,
                    "result": {"botauth": true}
                })),
        )
        .mount(&server)
        .await;

    // A throwaway Ed25519 key in PKCS#8 PEM, supplied via env.
    let key = ed25519_dalek::SigningKey::from_bytes(&[9u8; 32]);
    let pem = key
        .to_pkcs8_pem(ed25519_dalek::pkcs8::spki::der::pem::LineEnding::LF)
        .expect("pem")
        .to_string();
    let env_name = format!("MCPTEST_WBA_INT_KEY_{}", std::process::id());
    std::env::set_var(&env_name, pem);

    let url = format!("{}{}", server.uri(), POST_PATH);
    let cfg = StreamableHttpConfig {
        url,
        web_bot_auth: Some(mcptest_core::auth::WebBotAuthConfig {
            key_path: None,
            key_env: Some(env_name.clone()),
            directory_url: "https://mcptest.sh/.well-known/http-message-signatures-directory"
                .into(),
            signature_agent: "https://mcptest.sh".into(),
            algorithm: mcptest_core::auth::SignatureAlgorithm::Ed25519,
            components: vec![
                "@authority".into(),
                "@target-uri".into(),
                "content-digest".into(),
            ],
            expires_in_seconds: 30,
        }),
        request_timeout: Duration::from_secs(3),
        connect_timeout: Duration::from_secs(2),
        ..StreamableHttpConfig::default()
    };
    let (mut transport, handle) = StreamableHttp::connect(cfg).expect("connect with web bot auth");
    transport.outbound_tx.send(request_msg(1, "go")).unwrap();
    let env = tokio::time::timeout(Duration::from_secs(2), transport.inbound_rx.recv())
        .await
        .expect("response")
        .expect("envelope");
    std::env::remove_var(&env_name);
    assert_eq!(env.result, Some(json!({"botauth": true})));
    handle.shutdown().await;
}

// Confirms the public re-exports are intact so the runner crate's
// `use mcptest_core::transport::{StreamableHttp, StreamableHttpConfig};`
// keeps compiling. Static so the test costs nothing at runtime.
#[test]
fn public_types_are_accessible_from_the_crate_root() {
    let _ = StreamableHttpConfig::default();
    let _ = std::mem::size_of::<StreamableHttp>();
    let _ = std::mem::size_of::<Value>();
}
