//! HTTP endpoint handlers for MCP Streamable HTTP transport
//!
//! Implements three endpoints per MCP spec 2025-03-26:
//! - POST /mcp - Client sends JSON-RPC requests
//! - GET /mcp - Client opens SSE stream for server→client messages
//! - DELETE /mcp - Client terminates session

use std::sync::Arc;

use serde_json::Value;

use crate::protocol::types::JsonRpcRequest;
use crate::server::core::Server;

use super::security::validate_origin;
use super::session::HttpSessionManager;
use super::sse::create_sse_stream;
use super::types::{HttpTransportConfig, HttpTransportError};

/// HTTP request context extracted from headers
pub struct RequestContext {
    /// Mcp-Session-Id header value
    pub session_id: Option<String>,

    /// Origin header value
    pub origin: Option<String>,

    /// Last-Event-ID header value (for SSE resumability)
    pub last_event_id: Option<String>,

    /// Accept header value
    #[allow(dead_code)]
    pub accept: Option<String>,
}

impl RequestContext {
    /// Extract request context from HTTP headers
    pub fn from_headers(headers: &[(String, String)]) -> Self {
        let mut session_id = None;
        let mut origin = None;
        let mut last_event_id = None;
        let mut accept = None;

        for (name, value) in headers {
            let name_lower = name.to_lowercase();
            match name_lower.as_str() {
                "mcp-session-id" => session_id = Some(value.clone()),
                "origin" => origin = Some(value.clone()),
                "last-event-id" => last_event_id = Some(value.clone()),
                "accept" => accept = Some(value.clone()),
                _ => {}
            }
        }

        Self {
            session_id,
            origin,
            last_event_id,
            accept,
        }
    }
}

/// Handle POST /mcp - Client sends JSON-RPC request
///
/// Per MCP spec:
/// - Client MAY include Mcp-Session-Id header (if resuming session)
/// - Client MUST include Accept header (application/json or text/event-stream)
/// - Server returns Mcp-Session-Id header on initialize response
/// - Server returns JSON-RPC response OR SSE stream (if Accept: text/event-stream)
///
/// # Arguments
///
/// * `body` - Request body (JSON-RPC message or batch)
/// * `headers` - HTTP headers
/// * `session_manager` - HTTP session manager
/// * `mcp_server` - MCP server instance
/// * `config` - HTTP transport configuration
///
/// # Returns
///
/// Returns (response_body, response_headers, content_type)
pub async fn handle_post(
    body: Vec<u8>,
    headers: Vec<(String, String)>,
    session_manager: Arc<HttpSessionManager>,
    mcp_server: Arc<Server>,
    config: Arc<HttpTransportConfig>,
) -> Result<(Vec<u8>, Vec<(String, String)>, String), HttpTransportError> {
    // 1. Extract request context
    let ctx = RequestContext::from_headers(&headers);

    // 2. Validate origin per MCP security requirements
    validate_origin(ctx.origin.as_deref(), &config)?;

    // 3. Get or create HTTP session
    let http_session = if let Some(session_id) = &ctx.session_id {
        // Resume existing session
        session_manager
            .get_session(session_id)
            .ok_or_else(|| HttpTransportError::SessionNotFound(session_id.clone()))?
    } else {
        // Create new session
        session_manager.create_session()
    };

    // 4. Update activity timestamp
    http_session.touch();

    // 5. Parse JSON-RPC message(s)
    let json_value: Value = serde_json::from_slice(&body)
        .map_err(|e| HttpTransportError::InvalidJsonRpc(e.to_string()))?;

    // 6. Determine if batch or single request
    let is_batch = json_value.is_array();

    // 7. Route to MCP server
    // TODO: This requires exposing Server.handle_request() publicly
    // For now, we'll process the request manually
    let response = if is_batch {
        // Batch request
        let requests: Vec<JsonRpcRequest> = serde_json::from_value(json_value)
            .map_err(|e| HttpTransportError::InvalidJsonRpc(e.to_string()))?;

        // Process each request
        let mut responses = Vec::new();
        for request in requests {
            // Use HTTP session ID as MCP session ID
            let response = mcp_server.handle_request(&http_session.id, request).await;
            responses.push(response);
        }

        serde_json::to_vec(&responses)
            .map_err(HttpTransportError::Json)?
    } else {
        // Single request/notification
        let request: JsonRpcRequest = serde_json::from_value(json_value)
            .map_err(|e| HttpTransportError::InvalidJsonRpc(e.to_string()))?;

        // Check if this is a notification (no ID)
        let is_notification = request.id.is_none();

        // Use HTTP session ID as MCP session ID
        let response = mcp_server.handle_request(&http_session.id, request).await;

        // For notifications, return empty body (will use 202 Accepted)
        if is_notification {
            Vec::new()
        } else {
            serde_json::to_vec(&response)
                .map_err(HttpTransportError::Json)?
        }
    };

    // 8. Build response headers
    // Always include Mcp-Session-Id header
    let response_headers = vec![(
        "Mcp-Session-Id".to_string(),
        http_session.id.clone(),
    )];

    // 9. Determine response content type
    // Per MCP spec: single request/response is always application/json
    // SSE streams are only used via GET endpoint or long-polling (not implemented yet)
    let content_type = "application/json".to_string();

    Ok((response, response_headers, content_type))
}

/// Handle GET /mcp - Client opens SSE stream
///
/// Per MCP spec:
/// - Client MUST include Mcp-Session-Id header
/// - Client MAY include Last-Event-ID header (for resumability)
/// - Server returns text/event-stream with server→client messages
///
/// # Arguments
///
/// * `headers` - HTTP headers
/// * `session_manager` - HTTP session manager
/// * `config` - HTTP transport configuration
///
/// # Returns
///
/// Returns SSE stream
pub async fn handle_get(
    headers: Vec<(String, String)>,
    session_manager: Arc<HttpSessionManager>,
    config: Arc<HttpTransportConfig>,
) -> Result<impl futures::Stream<Item = Result<String, HttpTransportError>> + Send, HttpTransportError> {
    // 1. Extract request context
    let ctx = RequestContext::from_headers(&headers);

    // 2. Validate origin
    validate_origin(ctx.origin.as_deref(), &config)?;

    // 3. Require Mcp-Session-Id header
    let session_id = ctx
        .session_id
        .ok_or(HttpTransportError::MissingSessionId)?;

    // 4. Get session
    let http_session = session_manager
        .get_session(&session_id)
        .ok_or_else(|| HttpTransportError::SessionNotFound(session_id.clone()))?;

    // 5. Check connection limit
    let connections = http_session.inc_connections();
    if connections > config.max_sse_connections as u64 {
        http_session.dec_connections();
        return Err(HttpTransportError::TooManyConnections(
            config.max_sse_connections,
        ));
    }

    // 6. Update activity
    http_session.touch();

    // 7. Create SSE stream (with resumability support)
    let last_event_id_for_log = ctx.last_event_id.clone();
    let stream = create_sse_stream(http_session.clone(), ctx.last_event_id);

    // 8. Decrement connection count on stream drop
    // Note: This is handled by the stream itself or the HTTP layer

    tracing::info!(
        session_id = %session_id,
        connections = connections,
        last_event_id = ?last_event_id_for_log.as_deref(),
        "SSE stream opened"
    );

    Ok(stream)
}

/// Handle DELETE /mcp - Client terminates session
///
/// Per MCP spec:
/// - Client MUST include Mcp-Session-Id header
/// - Server removes session and closes all SSE streams
/// - Server returns 204 No Content
///
/// # Arguments
///
/// * `headers` - HTTP headers
/// * `session_manager` - HTTP session manager
/// * `config` - HTTP transport configuration
///
/// # Returns
///
/// Ok(()) on success
pub async fn handle_delete(
    headers: Vec<(String, String)>,
    session_manager: Arc<HttpSessionManager>,
    config: Arc<HttpTransportConfig>,
) -> Result<(), HttpTransportError> {
    // 1. Extract request context
    let ctx = RequestContext::from_headers(&headers);

    // 2. Validate origin
    validate_origin(ctx.origin.as_deref(), &config)?;

    // 3. Require Mcp-Session-Id header
    let session_id = ctx
        .session_id
        .ok_or(HttpTransportError::MissingSessionId)?;

    // 4. Remove session
    let removed = session_manager.remove_session(&session_id);

    if removed {
        tracing::info!(
            session_id = %session_id,
            "Session terminated by client"
        );
        Ok(())
    } else {
        Err(HttpTransportError::SessionNotFound(session_id))
    }
}

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

    #[test]
    fn test_request_context_from_headers() {
        let headers = vec![
            ("Mcp-Session-Id".to_string(), "session-123".to_string()),
            ("Origin".to_string(), "http://localhost:3000".to_string()),
            ("Last-Event-ID".to_string(), "event-42".to_string()),
            ("Accept".to_string(), "text/event-stream".to_string()),
        ];

        let ctx = RequestContext::from_headers(&headers);

        assert_eq!(ctx.session_id, Some("session-123".to_string()));
        assert_eq!(ctx.origin, Some("http://localhost:3000".to_string()));
        assert_eq!(ctx.last_event_id, Some("event-42".to_string()));
        assert_eq!(ctx.accept, Some("text/event-stream".to_string()));
    }

    #[test]
    fn test_request_context_case_insensitive() {
        let headers = vec![
            ("mcp-session-id".to_string(), "session-123".to_string()),
            ("ORIGIN".to_string(), "http://localhost".to_string()),
        ];

        let ctx = RequestContext::from_headers(&headers);

        assert_eq!(ctx.session_id, Some("session-123".to_string()));
        assert_eq!(ctx.origin, Some("http://localhost".to_string()));
    }

    #[test]
    fn test_request_context_missing_headers() {
        let headers = vec![];
        let ctx = RequestContext::from_headers(&headers);

        assert!(ctx.session_id.is_none());
        assert!(ctx.origin.is_none());
        assert!(ctx.last_event_id.is_none());
        assert!(ctx.accept.is_none());
    }

    #[tokio::test]
    async fn test_handle_delete_missing_session_id() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        let headers = vec![];

        let result = handle_delete(headers, session_manager, config).await;
        assert!(matches!(result, Err(HttpTransportError::MissingSessionId)));
    }

    #[tokio::test]
    async fn test_handle_delete_session_not_found() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        let headers = vec![("Mcp-Session-Id".to_string(), "nonexistent".to_string())];

        let result = handle_delete(headers, session_manager, config).await;
        assert!(matches!(
            result,
            Err(HttpTransportError::SessionNotFound(_))
        ));
    }

    #[tokio::test]
    async fn test_handle_delete_success() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        // Create session
        let session = session_manager.create_session();
        let session_id = session.id.clone();

        let headers = vec![("Mcp-Session-Id".to_string(), session_id.clone())];

        let result = handle_delete(headers, session_manager.clone(), config).await;
        assert!(result.is_ok());

        // Verify session is removed
        assert!(session_manager.get_session(&session_id).is_none());
    }

    #[tokio::test]
    async fn test_handle_get_missing_session_id() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        let headers = vec![];

        let result = handle_get(headers, session_manager, config).await;
        assert!(matches!(result, Err(HttpTransportError::MissingSessionId)));
    }

    #[tokio::test]
    async fn test_handle_get_session_not_found() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        let headers = vec![("Mcp-Session-Id".to_string(), "nonexistent".to_string())];

        let result = handle_get(headers, session_manager, config).await;
        assert!(matches!(
            result,
            Err(HttpTransportError::SessionNotFound(_))
        ));
    }

    #[tokio::test]
    async fn test_handle_get_success() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        // Create session
        let session = session_manager.create_session();
        let session_id = session.id.clone();

        let headers = vec![("Mcp-Session-Id".to_string(), session_id.clone())];

        let result = handle_get(headers, session_manager, config).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_get_with_resumability() {
        let config = Arc::new(HttpTransportConfig::default());
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        // Create session
        let session = session_manager.create_session();
        let session_id = session.id.clone();

        let headers = vec![
            ("Mcp-Session-Id".to_string(), session_id.clone()),
            ("Last-Event-ID".to_string(), "event-10".to_string()),
        ];

        let result = handle_get(headers, session_manager, config).await;
        assert!(result.is_ok());
    }

    #[tokio::test]
    async fn test_handle_get_connection_limit() {
        let mut config = HttpTransportConfig::default();
        config.max_sse_connections = 1;
        let config = Arc::new(config);
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        // Create session
        let session = session_manager.create_session();
        let session_id = session.id.clone();

        // First connection succeeds
        let headers1 = vec![("Mcp-Session-Id".to_string(), session_id.clone())];
        let _stream1 = handle_get(headers1, session_manager.clone(), config.clone())
            .await
            .unwrap();

        // Second connection should fail (limit exceeded)
        let headers2 = vec![("Mcp-Session-Id".to_string(), session_id.clone())];
        let result = handle_get(headers2, session_manager, config).await;
        assert!(matches!(
            result,
            Err(HttpTransportError::TooManyConnections(_))
        ));
    }
}
