//! Server-Sent Events (SSE) stream implementation for MCP HTTP transport
//!
//! Per MCP spec 2025-03-26, SSE is used for server→client push:
//! - Server can send requests (e.g., sampling/createMessage)
//! - Server can send notifications (e.g., notifications/resources/list_changed)
//! - Supports resumability via Last-Event-ID header

use serde::Serialize;
use std::sync::Arc;
use tokio_stream::StreamExt;
use tokio_stream::wrappers::BroadcastStream;

use super::session::HttpSession;
use super::types::{HttpTransportError, SseEvent};

/// Format event as SSE wire format
///
/// Per SSE spec: https://html.spec.whatwg.org/multipage/server-sent-events.html
/// ```text
/// id: unique-event-id
/// event: message
/// data: {"jsonrpc":"2.0",...}
///
/// ```
pub fn format_sse_event(event: &SseEvent) -> String {
    let mut result = String::new();

    if let Some(id) = &event.id {
        result.push_str(&format!("id: {}\n", id));
    }

    if let Some(event_type) = &event.event {
        result.push_str(&format!("event: {}\n", event_type));
    }

    // SSE data field - can be multiline
    for line in event.data.lines() {
        result.push_str(&format!("data: {}\n", line));
    }

    // Double newline marks end of event
    result.push('\n');

    result
}

/// Create SSE event stream from broadcast channel
///
/// Supports resumability: if last_event_id is provided, replays missed events
/// from session buffer before streaming new events
pub fn create_sse_stream(
    session: Arc<HttpSession>,
    last_event_id: Option<String>,
) -> impl futures::Stream<Item = Result<String, HttpTransportError>> + Send {
    async_stream::stream! {
        // 1. Replay missed events if resuming
        if let Some(last_id) = last_event_id {
            tracing::debug!(
                session_id = %session.id,
                last_event_id = %last_id,
                "Replaying missed SSE events"
            );

            let missed_events = session.get_events_after(&last_id);

            for event in missed_events {
                yield Ok(format_sse_event(&event));
            }
        }

        // 2. Subscribe to new events
        let rx = session.sse_tx.subscribe();
        let mut stream = BroadcastStream::new(rx);

        // 3. Stream new events as they arrive
        while let Some(result) = stream.next().await {
            match result {
                Ok(event) => {
                    yield Ok(format_sse_event(&event));
                }
                Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(skipped)) => {
                    // Client fell behind, some events were missed
                    tracing::warn!(
                        session_id = %session.id,
                        skipped = skipped,
                        "SSE client lagged, events skipped"
                    );
                    // Continue streaming - client should reconnect with Last-Event-ID
                    continue;
                }
            }
        }
    }
}

/// Send JSON-RPC message via SSE to a session
///
/// Serializes message to JSON, wraps in SSE event, broadcasts to all connected
/// SSE streams, and buffers for resumability
pub fn send_jsonrpc_via_sse<T: Serialize>(
    session: &HttpSession,
    message: &T,
) -> Result<(), HttpTransportError> {
    // 1. Serialize to JSON
    let data = serde_json::to_string(message).map_err(HttpTransportError::Json)?;

    // 2. Generate event ID for resumability
    let event_id = session.next_event_id();

    // 3. Create SSE event
    let event = SseEvent::new(data)
        .with_id(event_id.clone())
        .with_event("message".to_string());

    // 4. Parse sequence number from event ID
    let seq = event_id
        .split('-')
        .next_back()
        .and_then(|s| s.parse::<u64>().ok())
        .unwrap_or(0);

    // 5. Buffer for resumability
    session.buffer_event(seq, event.clone());

    // 6. Broadcast to all connected SSE streams
    // Ignore send errors - means no active subscribers
    let _ = session.sse_tx.send(event);

    // 7. Update activity timestamp
    session.touch();

    Ok(())
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::protocol::types::{JsonRpcRequest, JsonRpcResponse};

    #[test]
    fn test_format_sse_event_minimal() {
        let event = SseEvent::new("test data".to_string());
        let formatted = format_sse_event(&event);

        assert_eq!(formatted, "data: test data\n\n");
    }

    #[test]
    fn test_format_sse_event_with_id() {
        let event = SseEvent::new("data".to_string()).with_id("event-42".to_string());

        let formatted = format_sse_event(&event);

        assert!(formatted.contains("id: event-42\n"));
        assert!(formatted.contains("data: data\n"));
        assert!(formatted.ends_with("\n\n"));
    }

    #[test]
    fn test_format_sse_event_with_type() {
        let event = SseEvent::new("data".to_string())
            .with_id("1".to_string())
            .with_event("message".to_string());

        let formatted = format_sse_event(&event);

        assert!(formatted.contains("id: 1\n"));
        assert!(formatted.contains("event: message\n"));
        assert!(formatted.contains("data: data\n"));
    }

    #[test]
    fn test_format_sse_event_multiline() {
        let event = SseEvent::new("line1\nline2\nline3".to_string());
        let formatted = format_sse_event(&event);

        assert!(formatted.contains("data: line1\n"));
        assert!(formatted.contains("data: line2\n"));
        assert!(formatted.contains("data: line3\n"));
    }

    #[test]
    fn test_send_jsonrpc_via_sse() {
        let session = Arc::new(crate::transport::http::session::HttpSession::new(10));

        let request = JsonRpcRequest {
            jsonrpc: "2.0".to_string(),
            id: Some(serde_json::json!(1)),
            method: "test/method".to_string(),
            params: None,
        };

        let result = send_jsonrpc_via_sse(&session, &request);
        assert!(result.is_ok());

        // Verify event was buffered
        let buffer = session.event_buffer.read();
        assert_eq!(buffer.len(), 1);
    }

    #[tokio::test]
    async fn test_create_sse_stream() {
        use tokio_stream::StreamExt;

        let session = Arc::new(crate::transport::http::session::HttpSession::new(10));

        // Start stream
        let mut stream = Box::pin(create_sse_stream(Arc::clone(&session), None));

        // Send event
        let event = SseEvent::new("test".to_string()).with_id("1".to_string());
        let _ = session.sse_tx.send(event);

        // Receive event
        if let Some(Ok(formatted)) = stream.next().await {
            assert!(formatted.contains("data: test\n"));
        } else {
            panic!("Expected event");
        }
    }

    #[tokio::test]
    async fn test_sse_stream_with_replay() {
        use tokio_stream::StreamExt;

        let session = Arc::new(crate::transport::http::session::HttpSession::new(10));

        // Buffer some events
        session.buffer_event(
            0,
            SseEvent::new("event0".to_string()).with_id(format!("{}-0", session.id)),
        );
        session.buffer_event(
            1,
            SseEvent::new("event1".to_string()).with_id(format!("{}-1", session.id)),
        );
        session.buffer_event(
            2,
            SseEvent::new("event2".to_string()).with_id(format!("{}-2", session.id)),
        );

        // Resume from event 0
        let last_event_id = format!("{}-0", session.id);
        let mut stream = Box::pin(create_sse_stream(Arc::clone(&session), Some(last_event_id)));

        // Should replay events 1 and 2
        let evt1 = stream.next().await.unwrap().unwrap();
        assert!(evt1.contains("event1"));

        let evt2 = stream.next().await.unwrap().unwrap();
        assert!(evt2.contains("event2"));
    }
}
