//! Rama HTTP server for MCP Streamable HTTP transport
//!
//! Provides a production-grade HTTP server with:
//! - POST/GET/DELETE endpoints per MCP spec
//! - SSE streaming for server→client messages
//! - Session management and cleanup
//! - Graceful shutdown support

use std::sync::Arc;
use std::time::Duration;

use rama::Layer;
use rama::http::body::util::BodyExt;
use rama::http::layer::trace::TraceLayer;
use rama::http::{Body, Method, Request, Response, StatusCode};
use rama::rt::Executor;
use rama::service::service_fn;
use rama_http_backend::server::HttpServer;
use tokio::task::JoinHandle;
use tokio_stream::StreamExt;

use crate::server::core::Server as McpServer;

use super::handlers::{handle_delete, handle_get, handle_post};
use super::session::HttpSessionManager;
use super::types::{HttpTransportConfig, HttpTransportError};

/// HTTP Transport for MCP
///
/// Runs a Rama HTTP server that implements MCP Streamable HTTP protocol
pub struct HttpTransport {
    /// MCP server instance
    mcp_server: Arc<McpServer>,

    /// HTTP transport configuration
    config: Arc<HttpTransportConfig>,

    /// Session manager
    session_manager: Arc<HttpSessionManager>,

    /// Cleanup task handle
    cleanup_handle: Option<JoinHandle<()>>,
}

impl HttpTransport {
    /// Create new HTTP transport
    ///
    /// # Arguments
    ///
    /// * `mcp_server` - MCP server instance
    /// * `config` - HTTP transport configuration
    pub fn new(mcp_server: McpServer, config: HttpTransportConfig) -> Self {
        let config = Arc::new(config);
        let session_manager = Arc::new(HttpSessionManager::new((*config).clone()));

        Self {
            mcp_server: Arc::new(mcp_server),
            config,
            session_manager,
            cleanup_handle: None,
        }
    }

    /// Start HTTP server and session cleanup background task
    ///
    /// Runs until shutdown signal received
    pub async fn run(mut self) -> Result<(), HttpTransportError> {
        tracing::info!(
            bind_addr = %self.config.bind_addr,
            endpoint = %self.config.endpoint,
            "Starting MCP HTTP transport"
        );

        // 1. Spawn session cleanup task
        self.cleanup_handle = Some(self.spawn_cleanup_task());

        // 2. Build handler service
        let mcp_server = Arc::clone(&self.mcp_server);
        let session_manager = Arc::clone(&self.session_manager);
        let config = Arc::clone(&self.config);

        let service = service_fn(move |req: Request| {
            let mcp_server = Arc::clone(&mcp_server);
            let session_manager = Arc::clone(&session_manager);
            let config = Arc::clone(&config);

            async move {
                // Make service infallible by catching all errors
                match handle_request(req, mcp_server, session_manager, config).await {
                    Ok(response) => Ok::<Response, std::convert::Infallible>(response),
                    Err(e) => {
                        tracing::error!(error = ?e, "Request handling error");
                        Ok(error_response(
                            StatusCode::INTERNAL_SERVER_ERROR,
                            e.to_string(),
                        ))
                    }
                }
            }
        });

        // 3. Start HTTP server
        let bind_addr = self.config.bind_addr.clone();

        HttpServer::auto(Executor::new())
            .listen(bind_addr, TraceLayer::new_for_http().into_layer(service))
            .await
            .map_err(|e| {
                tracing::error!(error = ?e, "HTTP server error");
                HttpTransportError::Http(e.to_string())
            })?;

        tracing::info!("MCP HTTP transport stopped");

        Ok(())
    }

    /// Spawn session cleanup background task
    fn spawn_cleanup_task(&self) -> JoinHandle<()> {
        let session_manager = Arc::clone(&self.session_manager);
        let interval = Duration::from_secs(60); // Cleanup every minute

        tokio::spawn(async move {
            let mut interval_timer = tokio::time::interval(interval);

            loop {
                interval_timer.tick().await;

                tracing::debug!("Running session cleanup");
                session_manager.cleanup_expired().await;
            }
        })
    }
}

impl Drop for HttpTransport {
    fn drop(&mut self) {
        if let Some(handle) = &self.cleanup_handle {
            handle.abort();
        }
    }
}

/// Main request handler - routes to appropriate endpoint
async fn handle_request(
    req: Request,
    mcp_server: Arc<McpServer>,
    session_manager: Arc<HttpSessionManager>,
    config: Arc<HttpTransportConfig>,
) -> Result<Response, HttpTransportError> {
    let method = req.method().clone();
    let path = req.uri().path().to_string();

    // Check if this is the MCP endpoint
    if path != config.endpoint {
        // Not MCP endpoint - check for health endpoint
        if path == "/health" && method == Method::GET {
            return Ok(Response::builder()
                .status(StatusCode::OK)
                .body(Body::from("OK"))
                .unwrap());
        }

        // Unknown endpoint
        return Ok(error_response(
            StatusCode::NOT_FOUND,
            "Endpoint not found".to_string(),
        ));
    }

    // Route to appropriate MCP handler
    match method {
        Method::OPTIONS => {
            // Handle CORS preflight
            handle_options_request(req, config).await
        }
        Method::POST => handle_post_request(req, session_manager, mcp_server, config).await,
        Method::GET => handle_get_request(req, session_manager, config).await,
        Method::DELETE => handle_delete_request(req, session_manager, config).await,
        _ => Ok(error_response(
            StatusCode::METHOD_NOT_ALLOWED,
            format!("Method {} not allowed", method),
        )),
    }
}

/// Handle POST request
async fn handle_post_request(
    req: Request,
    session_manager: Arc<HttpSessionManager>,
    mcp_server: Arc<McpServer>,
    config: Arc<HttpTransportConfig>,
) -> Result<Response, HttpTransportError> {
    // Extract headers and origin
    let headers = extract_headers(&req);
    let origin = headers
        .iter()
        .find(|(name, _)| name.to_lowercase() == "origin")
        .map(|(_, value)| value.clone());

    // Read body
    let (_parts, body) = req.into_parts();
    let body_bytes = match body.collect().await {
        Ok(collected) => collected.to_bytes().to_vec(),
        Err(e) => {
            return Ok(error_response(
                StatusCode::BAD_REQUEST,
                format!("Failed to read body: {}", e),
            ));
        }
    };

    // Call handler
    match handle_post(body_bytes, headers, session_manager, mcp_server, config).await {
        Ok((response_body, response_headers, content_type)) => {
            // Per MCP spec: notifications (empty body) return 202 Accepted
            let status = if response_body.is_empty() {
                StatusCode::ACCEPTED
            } else {
                StatusCode::OK
            };

            let mut builder = Response::builder().status(status);

            // Add CORS headers
            builder = add_cors_headers(builder, origin);

            // Add response headers (but not for empty notifications)
            if !response_body.is_empty() {
                for (name, value) in response_headers {
                    builder = builder.header(name, value);
                }
                // Set content type only for non-empty responses
                builder = builder.header("Content-Type", content_type);
            }

            Ok(builder.body(Body::from(response_body)).unwrap())
        }
        Err(e) => {
            tracing::error!(error = ?e, "POST handler error");
            Ok(error_response(
                StatusCode::INTERNAL_SERVER_ERROR,
                e.to_string(),
            ))
        }
    }
}

/// Handle GET request (SSE)
async fn handle_get_request(
    req: Request,
    session_manager: Arc<HttpSessionManager>,
    config: Arc<HttpTransportConfig>,
) -> Result<Response, HttpTransportError> {
    // Extract headers and origin
    let headers = extract_headers(&req);
    let origin = headers
        .iter()
        .find(|(name, _)| name.to_lowercase() == "origin")
        .map(|(_, value)| value.clone());

    // Call handler
    match handle_get(headers, session_manager, config).await {
        Ok(stream) => {
            // Convert SSE event stream to bytes stream
            let body_stream = stream.map(|result| result.map(rama::bytes::Bytes::from));

            let body = Body::from_stream(body_stream);

            let mut builder = Response::builder()
                .status(StatusCode::OK)
                .header("Content-Type", "text/event-stream")
                .header("Cache-Control", "no-cache")
                .header("Connection", "keep-alive");

            // Add CORS headers
            builder = add_cors_headers(builder, origin);

            Ok(builder.body(body).unwrap())
        }
        Err(e) => {
            tracing::error!(error = ?e, "GET handler error");
            Ok(match e {
                HttpTransportError::MissingSessionId => error_response(
                    StatusCode::METHOD_NOT_ALLOWED,
                    "SSE not available without session. Send POST initialize first.".to_string(),
                ),
                HttpTransportError::SessionNotFound(id) => {
                    error_response(StatusCode::NOT_FOUND, format!("Session not found: {}", id))
                }
                HttpTransportError::TooManyConnections(limit) => error_response(
                    StatusCode::TOO_MANY_REQUESTS,
                    format!("Too many SSE connections (limit: {})", limit),
                ),
                _ => error_response(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()),
            })
        }
    }
}

/// Handle DELETE request
async fn handle_delete_request(
    req: Request,
    session_manager: Arc<HttpSessionManager>,
    config: Arc<HttpTransportConfig>,
) -> Result<Response, HttpTransportError> {
    // Extract headers and origin
    let headers = extract_headers(&req);
    let origin = headers
        .iter()
        .find(|(name, _)| name.to_lowercase() == "origin")
        .map(|(_, value)| value.clone());

    // Call handler
    match handle_delete(headers, session_manager, config).await {
        Ok(()) => {
            let mut builder = Response::builder().status(StatusCode::NO_CONTENT);

            // Add CORS headers
            builder = add_cors_headers(builder, origin);

            Ok(builder.body(Body::empty()).unwrap())
        }
        Err(e) => {
            tracing::error!(error = ?e, "DELETE handler error");
            Ok(match e {
                HttpTransportError::MissingSessionId => error_response(
                    StatusCode::BAD_REQUEST,
                    "Missing Mcp-Session-Id header".to_string(),
                ),
                HttpTransportError::SessionNotFound(id) => {
                    error_response(StatusCode::NOT_FOUND, format!("Session not found: {}", id))
                }
                _ => error_response(StatusCode::INTERNAL_SERVER_ERROR, e.to_string()),
            })
        }
    }
}

/// Extract headers from request
fn extract_headers(req: &Request) -> Vec<(String, String)> {
    req.headers()
        .iter()
        .map(|(name, value)| {
            (
                name.to_string(),
                value.to_str().unwrap_or_default().to_string(),
            )
        })
        .collect()
}

/// Handle OPTIONS preflight request
async fn handle_options_request(
    req: Request,
    config: Arc<HttpTransportConfig>,
) -> Result<Response, HttpTransportError> {
    let headers = extract_headers(&req);
    let origin = headers
        .iter()
        .find(|(name, _)| name.to_lowercase() == "origin")
        .map(|(_, value)| value.as_str());

    // Validate origin
    super::security::validate_origin(origin, &config)?;

    // Return CORS headers
    let mut builder = Response::builder()
        .status(StatusCode::NO_CONTENT)
        .header("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS")
        .header(
            "Access-Control-Allow-Headers",
            "Content-Type, Mcp-Session-Id, Mcp-Protocol-Version, Last-Event-ID, Accept",
        )
        .header("Access-Control-Max-Age", "86400"); // 24 hours

    // Add origin header if present
    if let Some(origin_value) = origin {
        builder = builder.header("Access-Control-Allow-Origin", origin_value);
        builder = builder.header("Access-Control-Allow-Credentials", "true");
    }

    Ok(builder.body(Body::empty()).unwrap())
}

/// Add CORS headers to response
fn add_cors_headers(
    mut builder: rama::http::response::Builder,
    origin: Option<String>,
) -> rama::http::response::Builder {
    if let Some(origin_value) = origin {
        builder = builder.header("Access-Control-Allow-Origin", origin_value);
        builder = builder.header("Access-Control-Allow-Credentials", "true");
        builder = builder.header(
            "Access-Control-Expose-Headers",
            "Mcp-Session-Id, Mcp-Protocol-Version",
        );
    }
    builder
}

/// Create error response
fn error_response(status: StatusCode, message: String) -> Response {
    Response::builder()
        .status(status)
        .header("Content-Type", "application/json")
        .body(Body::from(
            serde_json::json!({
                "error": message
            })
            .to_string(),
        ))
        .unwrap()
}

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

    #[test]
    fn test_http_transport_creation() {
        let server = McpServer::new("test-server", "1.0.0");
        let config = HttpTransportConfig::default();

        let transport = HttpTransport::new(server, config);

        assert!(transport.cleanup_handle.is_none());
    }

    #[test]
    fn test_extract_headers() {
        let req = Request::builder()
            .header("Content-Type", "application/json")
            .header("Mcp-Session-Id", "session-123")
            .body(Body::empty())
            .unwrap();

        let headers = extract_headers(&req);

        assert!(headers.contains(&("content-type".to_string(), "application/json".to_string())));
        assert!(headers.contains(&("mcp-session-id".to_string(), "session-123".to_string())));
    }
}
