//! Legacy SSE transport (MCP 2024-11-05 spec).
//!
//! Why this lives alongside the streamable HTTP transport: many
//! production MCP servers still ship the older two-endpoint shape, and
//! compatibility testing requires first-class support. The wire shape:
//!
//! 1. The client opens a long-lived SSE GET against an SSE endpoint.
//! 2. The server's first SSE event is `event: endpoint` with `data:`
//!    set to the POST URL (often relative). The client posts every
//!    subsequent JSON-RPC envelope to that URL.
//! 3. Server-pushed JSON-RPC messages arrive as `event: message` SSE
//!    events on the original GET stream.
//!
//! The transport pins the protocol negotiation to `2024-11-05` only;
//! the version-negotiation logic itself lives in the protocol
//! [`Client`](crate::protocol::Client), so this module just propagates
//! whatever it sees. Acceptance criteria in the  description.

use std::collections::HashMap;
use std::sync::{Arc, Mutex};
use std::time::Duration;

use bytes::Bytes;
use futures_util::StreamExt;
use reqwest::header::{HeaderMap, HeaderName, HeaderValue, ACCEPT, CONTENT_TYPE};
use reqwest::{Client as HttpClient, Method};
use serde_json::Value;
use tokio::sync::{mpsc, oneshot};
use tokio::task::JoinHandle;
use tokio::time::sleep;

use crate::protocol::{JsonRpcMessage, ProtocolVersion, Transport};
use crate::transport::header_capture::HeaderCapture;
use crate::transport::sse_parser::SseParser;

/// MCP protocol version this transport speaks. The 2024-11-05 spec is
/// the only revision that uses the GET-SSE plus POST shape.
pub const SUPPORTED_VERSION: ProtocolVersion = ProtocolVersion::V2024_11_05;

/// Reconnect backoff schedule, mirrors the streamable HTTP transport.
pub const BACKOFF_STEPS_MS: &[u64] = &[50, 100, 200, 400];

/// Construction parameters for [`LegacySse::connect`].
#[derive(Debug, Clone)]
pub struct LegacySseConfig {
    /// SSE GET endpoint. The transport opens a long-lived connection
    /// here and waits for the initial `endpoint` event to learn where
    /// to POST.
    pub sse_url: String,
    /// Static headers injected on both the GET and every POST.
    pub headers: HashMap<String, String>,
    /// Optional bearer token shortcut. When set, sends
    /// `Authorization: Bearer <token>` on every request.
    pub bearer_token: Option<String>,
    /// Per-request timeout (POST). Default 30s.
    pub request_timeout: Duration,
    /// TCP and TLS connect timeout. Default 30s.
    pub connect_timeout: Duration,
    /// HTTP / HTTPS proxy policy. Default empty preserves
    /// reqwest's env-var pickup.
    pub proxy: crate::network::ProxyConfig,
    /// Optional mTLS client identity. Applied at client-build
    /// time as a `reqwest::Identity`; composes with `bearer_token` and
    /// static `headers`.
    pub mtls: Option<crate::auth::MtlsConfig>,
    /// Optional AWS SigV4 request signer. When set, every POST
    /// and the SSE GET is signed per request.
    pub sigv4: Option<crate::auth::SigV4Config>,
    /// Optional Web Bot Auth request signer. When set, every
    /// POST and the SSE GET carries RFC 9421 `Signature` headers.
    pub web_bot_auth: Option<crate::auth::WebBotAuthConfig>,
}

impl Default for LegacySseConfig {
    fn default() -> Self {
        Self {
            sse_url: String::new(),
            headers: HashMap::new(),
            bearer_token: None,
            request_timeout: Duration::from_secs(30),
            connect_timeout: Duration::from_secs(30),
            proxy: crate::network::ProxyConfig::default(),
            mtls: None,
            sigv4: None,
            web_bot_auth: None,
        }
    }
}

/// Errors raised while constructing a [`LegacySse`] transport.
#[derive(Debug, thiserror::Error)]
#[non_exhaustive]
pub enum LegacySseError {
    /// The SSE URL was empty or could not be parsed.
    #[error("invalid SSE URL `{0}`")]
    InvalidUrl(String),
    /// A configured header name or value was not valid for `reqwest`.
    #[error("invalid header `{name}`: {reason}")]
    InvalidHeader {
        /// Header name as supplied by the caller.
        name: String,
        /// Underlying reason from the HTTP stack.
        reason: String,
    },
    /// `reqwest` rejected the client construction.
    #[error("reqwest client construction failed: {0}")]
    Reqwest(#[from] reqwest::Error),
    /// Proxy URL or combination rejected at builder time.
    #[error(transparent)]
    Proxy(#[from] crate::network::ProxyError),
    /// A client-auth scheme (mTLS) failed to build at connect time.
    #[error("client auth setup failed: {0}")]
    Auth(#[from] crate::auth::AuthError),
    /// The initial `endpoint` SSE event never arrived in time.
    #[error("server never announced POST endpoint within {timeout_ms}ms")]
    EndpointTimeout {
        /// Timeout in milliseconds.
        timeout_ms: u64,
    },
}

/// Live legacy SSE transport handle.
pub struct LegacySse {
    writer: JoinHandle<()>,
    reader: JoinHandle<()>,
    header_capture: HeaderCapture,
}

#[derive(Default)]
struct SharedState {
    post_endpoint: Mutex<Option<String>>,
    last_event_id: Mutex<Option<String>>,
    /// Per-request response header capture (WOR-1287). The writer task
    /// records the POST acknowledgment headers; the executor takes them
    /// via the clone exposed on [`LegacySse::header_capture`].
    header_capture: HeaderCapture,
}

impl SharedState {
    fn get_endpoint(&self) -> Option<String> {
        self.post_endpoint.lock().ok().and_then(|g| g.clone())
    }

    fn set_endpoint(&self, value: String) {
        if let Ok(mut guard) = self.post_endpoint.lock() {
            *guard = Some(value);
        }
    }

    fn get_last_id(&self) -> Option<String> {
        self.last_event_id.lock().ok().and_then(|g| g.clone())
    }

    fn set_last_id(&self, value: String) {
        if let Ok(mut guard) = self.last_event_id.lock() {
            *guard = Some(value);
        }
    }
}

impl LegacySse {
    /// Open the transport. Spawns the reader (SSE GET) and writer (POST
    /// queue) tasks, then waits up to `connect_timeout` for the server
    /// to announce its POST endpoint before returning.
    pub async fn connect(config: LegacySseConfig) -> Result<(Transport, Self), LegacySseError> {
        if config.sse_url.trim().is_empty() {
            return Err(LegacySseError::InvalidUrl(config.sse_url));
        }
        let mut base_headers = HeaderMap::new();
        for (name, value) in &config.headers {
            let hname = HeaderName::from_bytes(name.as_bytes()).map_err(|e| {
                LegacySseError::InvalidHeader {
                    name: name.clone(),
                    reason: e.to_string(),
                }
            })?;
            let hval = HeaderValue::from_str(value).map_err(|e| LegacySseError::InvalidHeader {
                name: name.clone(),
                reason: e.to_string(),
            })?;
            base_headers.insert(hname, hval);
        }
        if let Some(token) = &config.bearer_token {
            let v = HeaderValue::from_str(&format!("Bearer {token}")).map_err(|e| {
                LegacySseError::InvalidHeader {
                    name: "Authorization".into(),
                    reason: e.to_string(),
                }
            })?;
            base_headers.insert(reqwest::header::AUTHORIZATION, v);
        }
        let mut builder = HttpClient::builder()
            .connect_timeout(config.connect_timeout)
            .timeout(config.request_timeout)
            .default_headers(base_headers);
        builder = config.proxy.apply(builder)?;
        builder = apply_mtls(builder, config.mtls.as_ref())?;
        let http = builder.build()?;

        let (outbound_tx, outbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (inbound_tx, inbound_rx) = mpsc::unbounded_channel::<JsonRpcMessage>();
        let (endpoint_tx, endpoint_rx) = oneshot::channel::<()>();

        let shared = Arc::new(SharedState::default());
        let signer = Arc::new(crate::auth::RequestSigner::from_configs(
            config.sigv4.clone(),
            config.web_bot_auth.clone(),
        ));

        let reader = spawn_reader(
            http.clone(),
            config.sse_url.clone(),
            Arc::clone(&shared),
            Arc::clone(&signer),
            inbound_tx,
            Some(endpoint_tx),
        );
        let writer = spawn_writer(
            http,
            config.sse_url.clone(),
            Arc::clone(&shared),
            signer,
            outbound_rx,
        );

        // Wait for the initial `endpoint` event. The reader fires the
        // oneshot the first time it parses one.
        let timeout =
            Duration::from_millis(config.connect_timeout.as_millis().min(u64::MAX as u128) as u64);
        match tokio::time::timeout(timeout, endpoint_rx).await {
            Ok(Ok(())) => {}
            Ok(Err(_dropped)) => {
                reader.abort();
                writer.abort();
                return Err(LegacySseError::EndpointTimeout {
                    timeout_ms: timeout.as_millis() as u64,
                });
            }
            Err(_elapsed) => {
                reader.abort();
                writer.abort();
                return Err(LegacySseError::EndpointTimeout {
                    timeout_ms: timeout.as_millis() as u64,
                });
            }
        }

        let transport = Transport::new(outbound_tx, inbound_rx);
        let header_capture = shared.header_capture.clone();
        Ok((
            transport,
            Self {
                writer,
                reader,
                header_capture,
            },
        ))
    }

    /// The MCP protocol version this transport supports.
    pub fn supported_version() -> ProtocolVersion {
        SUPPORTED_VERSION
    }

    /// Clone of the per-request response header capture channel
    /// (WOR-1287). For this transport the recorded headers are the POST
    /// acknowledgment's; see the writer task for the rationale.
    pub fn header_capture(&self) -> HeaderCapture {
        self.header_capture.clone()
    }

    /// Abort both tasks. Idempotent.
    pub async fn shutdown(self) {
        self.writer.abort();
        self.reader.abort();
        let _ = self.writer.await;
        let _ = self.reader.await;
    }
}

/// Apply an optional mTLS client identity to the reqwest builder.
///
/// A no-op when `mtls` is absent or disabled. Otherwise it attaches the
/// client identity and, when a CA bundle is configured, an extra root
/// certificate. Mirrors the streamable HTTP transport.
fn apply_mtls(
    builder: reqwest::ClientBuilder,
    mtls: Option<&crate::auth::MtlsConfig>,
) -> Result<reqwest::ClientBuilder, LegacySseError> {
    let Some(cfg) = mtls else {
        return Ok(builder);
    };
    if !cfg.is_enabled() {
        return Ok(builder);
    }
    let mut builder = builder.identity(crate::auth::identity(cfg)?);
    if let Some(root) = crate::auth::root_certificate(cfg)? {
        builder = builder.add_root_certificate(root);
    }
    Ok(builder)
}

/// Drain outbound envelopes and POST each to the announced endpoint.
fn spawn_writer(
    http: HttpClient,
    sse_url: String,
    shared: Arc<SharedState>,
    signer: Arc<Option<crate::auth::RequestSigner>>,
    mut outbound_rx: mpsc::UnboundedReceiver<JsonRpcMessage>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        while let Some(msg) = outbound_rx.recv().await {
            let Some(post_url) = shared.get_endpoint() else {
                // Should never happen because connect waits for the
                // endpoint before returning, but stay defensive.
                continue;
            };
            let absolute = absolutize(&sse_url, &post_url);
            // Serialize the body up front so the signer commits to the
            // exact bytes we send.
            let body = serde_json::to_vec(&msg).unwrap_or_default();
            let mut req = http
                .request(Method::POST, &absolute)
                .header(CONTENT_TYPE, "application/json")
                .body(body.clone());
            match sign_request(req, signer.as_ref().as_ref(), "POST", &absolute, &body) {
                Ok(signed) => req = signed,
                Err(()) => {
                    // Signing failed (fatal config error); drop the
                    // message rather than sending unsigned.
                    continue;
                }
            }
            // WOR-1287: for legacy SSE the per-call HTTP response is the
            // POST acknowledgment. The JSON-RPC result arrives later on
            // the long-lived GET stream, whose headers belong to the
            // connection, not to any one call, so the ack headers are the
            // only honest per-request capture this transport can offer.
            match req.send().await {
                Ok(resp) => {
                    if let Some(id) = msg.id.as_ref().and_then(|i| i.as_i64()) {
                        shared.header_capture.record(id, resp.headers());
                    }
                }
                Err(_e) => {
                    // Same contract as before the capture landed: a failed
                    // POST surfaces to the caller as a request timeout in
                    // the protocol client, so the error is dropped here.
                }
            }
        }
    })
}

/// Attach per-request signature headers (SigV4 / Web Bot Auth) to the
/// builder. A no-op when no signer is configured. Returns `Err(())` and
/// logs (without secrets) when signing fails.
fn sign_request(
    mut req: reqwest::RequestBuilder,
    signer: Option<&crate::auth::RequestSigner>,
    method: &str,
    url: &str,
    body: &[u8],
) -> Result<reqwest::RequestBuilder, ()> {
    let Some(signer) = signer else {
        return Ok(req);
    };
    let headers = signer.sign(method, url, body).map_err(|e| {
        tracing::error!(
            target: "mcptest_core::transport::legacy_sse",
            error = %e,
            "request signing failed"
        );
    })?;
    for h in headers {
        req = req.header(h.name, h.value);
    }
    Ok(req)
}

/// Open the SSE GET, parse events, fire the `endpoint` oneshot when the
/// first `endpoint` event arrives, and push every `message` event into
/// the inbound channel.
fn spawn_reader(
    http: HttpClient,
    sse_url: String,
    shared: Arc<SharedState>,
    signer: Arc<Option<crate::auth::RequestSigner>>,
    inbound_tx: mpsc::UnboundedSender<JsonRpcMessage>,
    mut endpoint_signal: Option<oneshot::Sender<()>>,
) -> JoinHandle<()> {
    tokio::spawn(async move {
        let mut attempt = 0usize;
        loop {
            if inbound_tx.is_closed() {
                return;
            }
            let last_id = shared.get_last_id();
            match run_sse_get(
                &http,
                &sse_url,
                &shared,
                signer.as_ref().as_ref(),
                &inbound_tx,
                last_id.as_deref(),
                &mut endpoint_signal,
            )
            .await
            {
                Ok(()) => {
                    attempt = 0;
                }
                Err(SseGetError::Signing) => {
                    // Fatal: a signing failure will recur. Stop the loop.
                    return;
                }
                Err(SseGetError::Reqwest(_e)) => {
                    if attempt >= BACKOFF_STEPS_MS.len() {
                        return;
                    }
                    sleep(Duration::from_millis(BACKOFF_STEPS_MS[attempt])).await;
                    attempt += 1;
                }
            }
        }
    })
}

/// Failure modes of one SSE GET: a transient wire error the reader backs
/// off on, or a fatal signing error that will recur and so stops the loop.
enum SseGetError {
    /// Wire failure (network, non-2xx). Retryable.
    Reqwest(reqwest::Error),
    /// Per-request signing failed. Fatal.
    Signing,
}

impl From<reqwest::Error> for SseGetError {
    fn from(e: reqwest::Error) -> Self {
        SseGetError::Reqwest(e)
    }
}

async fn run_sse_get(
    http: &HttpClient,
    sse_url: &str,
    shared: &SharedState,
    signer: Option<&crate::auth::RequestSigner>,
    inbound_tx: &mpsc::UnboundedSender<JsonRpcMessage>,
    last_event_id: Option<&str>,
    endpoint_signal: &mut Option<oneshot::Sender<()>>,
) -> Result<(), SseGetError> {
    let mut req = http
        .request(Method::GET, sse_url)
        .header(ACCEPT, "text/event-stream");
    if let Some(id) = last_event_id {
        req = req.header("Last-Event-Id", id);
    }
    req = sign_request(req, signer, "GET", sse_url, &[]).map_err(|()| SseGetError::Signing)?;
    let resp = req.send().await?;
    if !resp.status().is_success() {
        return Err(resp.error_for_status().unwrap_err().into());
    }
    let mut stream = resp.bytes_stream();
    let mut parser = SseParser::new();
    while let Some(chunk) = stream.next().await {
        let bytes: Bytes = chunk?;
        parser.feed(&bytes);
        for ev in parser.take_ready() {
            if let Some(id) = &ev.id {
                shared.set_last_id(id.clone());
            }
            match ev.event.as_str() {
                "endpoint" if !ev.data.is_empty() => {
                    shared.set_endpoint(ev.data.clone());
                    if let Some(tx) = endpoint_signal.take() {
                        let _ = tx.send(());
                    }
                }
                "message" | "" => {
                    if ev.data.is_empty() {
                        continue;
                    }
                    dispatch_json_bytes(ev.data.as_bytes(), inbound_tx);
                }
                _ => {
                    // Unknown event types and empty-data endpoint
                    // events are ignored per the spec.
                }
            }
        }
    }
    Ok(())
}

/// Resolve a possibly-relative POST endpoint against the SSE base URL.
///
/// Most servers emit relative paths (e.g. `/messages/abc`). We use
/// `url::Url::join` so the resulting URL inherits scheme, host, and
/// port. Falls back to the input on parse failure so the writer's
/// reqwest call surfaces the error.
fn absolutize(base: &str, target: &str) -> String {
    if target.starts_with("http://") || target.starts_with("https://") {
        return target.to_owned();
    }
    let Ok(parsed) = url::Url::parse(base) else {
        return target.to_owned();
    };
    match parsed.join(target) {
        Ok(joined) => joined.to_string(),
        Err(_) => target.to_owned(),
    }
}

/// Dispatch JSON bytes that may carry a single envelope or an array.
fn dispatch_json_bytes(body: &[u8], inbound_tx: &mpsc::UnboundedSender<JsonRpcMessage>) {
    let Ok(json) = serde_json::from_slice::<Value>(body) else {
        return;
    };
    match json {
        Value::Array(items) => {
            for item in items {
                if let Ok(msg) = serde_json::from_value::<JsonRpcMessage>(item) {
                    let _ = inbound_tx.send(msg);
                }
            }
        }
        other => {
            if let Ok(msg) = serde_json::from_value::<JsonRpcMessage>(other) {
                let _ = inbound_tx.send(msg);
            }
        }
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn supported_version_is_2024_11_05() {
        assert_eq!(SUPPORTED_VERSION, ProtocolVersion::V2024_11_05);
        assert_eq!(LegacySse::supported_version(), ProtocolVersion::V2024_11_05);
    }

    #[test]
    fn absolutize_returns_absolute_url_unchanged() {
        assert_eq!(
            absolutize("http://x.example/sse", "http://y.example/post"),
            "http://y.example/post"
        );
    }

    #[test]
    fn absolutize_resolves_relative_path_against_base() {
        assert_eq!(
            absolutize("http://x.example/mcp/sse", "/messages/abc"),
            "http://x.example/messages/abc"
        );
    }

    #[tokio::test]
    async fn connect_with_empty_url_errors() {
        let cfg = LegacySseConfig {
            sse_url: "".into(),
            ..Default::default()
        };
        match LegacySse::connect(cfg).await {
            Err(LegacySseError::InvalidUrl(_)) => {}
            Err(other) => panic!("expected InvalidUrl, got {other}"),
            Ok(_) => panic!("expected error for empty URL"),
        }
    }

    #[test]
    fn shared_state_round_trips_endpoint_and_last_id() {
        let s = SharedState::default();
        assert!(s.get_endpoint().is_none());
        assert!(s.get_last_id().is_none());
        s.set_endpoint("/messages/x".into());
        s.set_last_id("ev-1".into());
        assert_eq!(s.get_endpoint().as_deref(), Some("/messages/x"));
        assert_eq!(s.get_last_id().as_deref(), Some("ev-1"));
    }
}
