// 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 legacy SSE transport.
//!
//! Exercises the GET-SSE + POST handshake against a wiremock server:
//! initial `endpoint` event handshake, server-side notification
//! delivery, `Last-Event-Id` resumption, and version-only behavior.

use std::time::Duration;

use mcptest_core::protocol::ProtocolVersion;
use mcptest_core::transport::{LegacySse, LegacySseConfig, LEGACY_SSE_SUPPORTED_VERSION};
use wiremock::matchers::{header, method, path};
use wiremock::{Mock, MockServer, ResponseTemplate};

const SSE_PATH: &str = "/sse";
const POST_PATH: &str = "/messages";

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

fn cfg_for(sse_url: String) -> LegacySseConfig {
    LegacySseConfig {
        sse_url,
        request_timeout: Duration::from_secs(2),
        connect_timeout: Duration::from_secs(2),
        ..Default::default()
    }
}

/// Compose an SSE stream that announces the POST endpoint and (optionally)
/// pushes a single JSON-RPC notification afterwards.
fn sse_handshake_body(endpoint: &str, extra: &str) -> String {
    format!("event: endpoint\ndata: {endpoint}\n\n{extra}",)
}

#[tokio::test]
async fn endpoint_event_announces_post_url_and_handshake_completes() {
    let server = start_server().await;
    let body = sse_handshake_body(POST_PATH, "");

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

    let sse_url = format!("{}{}", server.uri(), SSE_PATH);
    let (transport, handle) = LegacySse::connect(cfg_for(sse_url))
        .await
        .expect("connect succeeds");
    drop(transport);
    handle.shutdown().await;
}

#[tokio::test]
async fn server_pushed_notification_arrives_on_inbound() {
    let server = start_server().await;
    let extra = concat!(
        "event: message\n",
        "data: {\"jsonrpc\":\"2.0\",\"method\":\"progress\",\"params\":{\"pct\":42}}\n",
        "\n",
    );
    let body = sse_handshake_body(POST_PATH, extra);

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

    // We also mount a POST handler so the writer task's eventual
    // outbound call does not 404. The body is irrelevant here.
    Mock::given(method("POST"))
        .and(path(POST_PATH))
        .respond_with(ResponseTemplate::new(200))
        .mount(&server)
        .await;

    let sse_url = format!("{}{}", server.uri(), SSE_PATH);
    let (mut transport, handle) = LegacySse::connect(cfg_for(sse_url))
        .await
        .expect("connect succeeds");

    let envelope = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("notification arrives")
        .expect("envelope present");
    assert_eq!(envelope.method.as_deref(), Some("progress"));

    handle.shutdown().await;
}

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

    // First connection: announces endpoint, sends one message with an
    // id, then the body ends and the connection closes.
    let first_body = format!(
        "event: endpoint\ndata: {POST_PATH}\n\n\
         id: ev-1\n\
         event: message\n\
         data: {{\"jsonrpc\":\"2.0\",\"method\":\"first\",\"params\":{{}}}}\n\
         \n",
    );

    Mock::given(method("GET"))
        .and(path(SSE_PATH))
        .respond_with(ResponseTemplate::new(200).set_body_raw(first_body, "text/event-stream"))
        .up_to_n_times(1)
        .mount(&server)
        .await;

    // Second connection: must carry the Last-Event-Id from the first.
    let second_body = format!(
        "event: endpoint\ndata: {POST_PATH}\n\n\
         event: message\n\
         data: {{\"jsonrpc\":\"2.0\",\"method\":\"resumed\",\"params\":{{}}}}\n\
         \n",
    );
    Mock::given(method("GET"))
        .and(path(SSE_PATH))
        .and(header("last-event-id", "ev-1"))
        .respond_with(ResponseTemplate::new(200).set_body_raw(second_body, "text/event-stream"))
        .mount(&server)
        .await;

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

    let sse_url = format!("{}{}", server.uri(), SSE_PATH);
    let (mut transport, handle) = LegacySse::connect(cfg_for(sse_url))
        .await
        .expect("connect succeeds");

    let first = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("first envelope")
        .expect("envelope");
    assert_eq!(first.method.as_deref(), Some("first"));

    // After the first connection drops, the reader reconnects with
    // Last-Event-Id. The second message should arrive shortly.
    let second = tokio::time::timeout(Duration::from_secs(3), transport.inbound_rx.recv())
        .await
        .expect("second envelope")
        .expect("envelope");
    assert_eq!(second.method.as_deref(), Some("resumed"));

    handle.shutdown().await;
}

#[tokio::test]
async fn version_only_negotiation_pins_to_2024_11_05() {
    // The transport carries no version state itself; it only advertises
    // the supported version for the protocol Client to consult. This
    // is exactly the contract  asks for.
    assert_eq!(LegacySse::supported_version(), ProtocolVersion::V2024_11_05);
    assert_eq!(LEGACY_SSE_SUPPORTED_VERSION, ProtocolVersion::V2024_11_05);
}

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

    // Send SSE bytes that have no `endpoint` event. The connect call
    // should fail after the configured timeout.
    let body = "event: message\ndata: {}\n\n";

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

    let sse_url = format!("{}{}", server.uri(), SSE_PATH);
    let mut cfg = cfg_for(sse_url);
    cfg.connect_timeout = Duration::from_millis(200);
    match LegacySse::connect(cfg).await {
        Err(mcptest_core::transport::LegacySseError::EndpointTimeout { .. }) => {}
        Err(other) => panic!("expected EndpointTimeout, got {other}"),
        Ok(_) => panic!("expected error when server never announces endpoint"),
    }
}
