//! HTTP Listener — handles plain HTTP requests on the same TCP port.
//!
//! This module implements a minimal HTTP/1.1 request parser and response writer.
//! It is used for:
//!   - Webhook delivery: `/_hook/{conn_id}/{path}` → HTTP_INBOUND to PLP client
//!   - Health check: `/health` → 200 OK
//!   - Readiness: `/ready` → 200 OK
//!
//! All other paths return 404. The HTTP handling is intentionally minimal —
//! Pulse is not an HTTP server, it just needs to accept enough HTTP for
//! webhooks and infrastructure probes.

use crate::webhook::WebhookRegistry;
use std::collections::HashMap;
use std::sync::Arc;
use tokio::io::{AsyncBufReadExt, AsyncReadExt, AsyncWriteExt, BufReader};
use tokio::net::TcpStream;
use tracing::{debug, info, warn};

/// Parsed HTTP request (minimal — just what we need for webhooks).
#[derive(Debug)]
struct HttpRequest {
    method: String,
    path: String,
    headers: HashMap<String, String>,
    body: Option<String>,
}

/// Handle a plain HTTP request (non-WebSocket) on the TCP port.
///
/// This is called by the gateway when protocol detection determines the
/// incoming connection is plain HTTP (not a WebSocket upgrade).
pub async fn handle_http_request(
    mut stream: TcpStream,
    webhook_registry: Arc<WebhookRegistry>,
) {
    match parse_request(&mut stream).await {
        Ok(req) => {
            let response = route_request(&req, &webhook_registry).await;
            if let Err(e) = write_response(&mut stream, &response).await {
                warn!("[HTTP] failed to write response: {}", e);
            }
        }
        Err(e) => {
            warn!("[HTTP] failed to parse request: {}", e);
            let _ = write_response(
                &mut stream,
                &HttpResponse {
                    status: 400,
                    status_text: "Bad Request".to_string(),
                    headers: HashMap::new(),
                    body: format!("{{\"error\":\"bad request: {}\"}}", e),
                },
            )
            .await;
        }
    }
}

struct HttpResponse {
    status: u16,
    status_text: String,
    headers: HashMap<String, String>,
    body: String,
}

impl HttpResponse {
    fn json(status: u16, body: &str) -> Self {
        let status_text = match status {
            200 => "OK",
            400 => "Bad Request",
            404 => "Not Found",
            503 => "Service Unavailable",
            504 => "Gateway Timeout",
            _ => "Unknown",
        };
        let mut headers = HashMap::new();
        headers.insert("Content-Type".to_string(), "application/json".to_string());
        headers.insert("Access-Control-Allow-Origin".to_string(), "*".to_string());
        headers.insert(
            "Access-Control-Allow-Methods".to_string(),
            "GET, POST, PUT, DELETE, OPTIONS".to_string(),
        );
        headers.insert(
            "Access-Control-Allow-Headers".to_string(),
            "Content-Type, Authorization, X-Webhook-Signature".to_string(),
        );
        Self {
            status,
            status_text: status_text.to_string(),
            headers,
            body: body.to_string(),
        }
    }

    fn cors_preflight() -> Self {
        let mut headers = HashMap::new();
        headers.insert("Access-Control-Allow-Origin".to_string(), "*".to_string());
        headers.insert(
            "Access-Control-Allow-Methods".to_string(),
            "GET, POST, PUT, DELETE, OPTIONS".to_string(),
        );
        headers.insert(
            "Access-Control-Allow-Headers".to_string(),
            "Content-Type, Authorization, X-Webhook-Signature, Stripe-Signature".to_string(),
        );
        headers.insert("Access-Control-Max-Age".to_string(), "86400".to_string());
        Self {
            status: 204,
            status_text: "No Content".to_string(),
            headers,
            body: String::new(),
        }
    }
}

async fn route_request(req: &HttpRequest, webhook_registry: &WebhookRegistry) -> HttpResponse {
    debug!("[HTTP] {} {}", req.method, req.path);

    // CORS preflight
    if req.method == "OPTIONS" {
        return HttpResponse::cors_preflight();
    }

    // Health check
    if req.path == "/health" || req.path == "/healthz" {
        return HttpResponse::json(200, r#"{"status":"ok","service":"pulse"}"#);
    }

    // Readiness check
    if req.path == "/ready" || req.path == "/readyz" {
        return HttpResponse::json(200, r#"{"status":"ready","service":"pulse"}"#);
    }

    // Webhook delivery: /_hook/{conn_id}/{path...}
    if req.path.starts_with("/_hook/") {
        let hook_key = req.path.trim_start_matches("/_hook/");
        if hook_key.is_empty() || !hook_key.contains('/') {
            return HttpResponse::json(404, r#"{"error":"invalid webhook path"}"#);
        }

        match webhook_registry
            .deliver(
                hook_key,
                &req.method,
                req.headers.clone(),
                req.body.clone(),
            )
            .await
        {
            Some(resp) => {
                let body = if let Some(ref err) = resp.error {
                    format!(r#"{{"error":"{}"}}"#, err)
                } else if let Some(ref b) = resp.body {
                    // Body is base64 in the wire format — decode it.
                    match base64::Engine::decode(
                        &base64::engine::general_purpose::STANDARD,
                        b,
                    ) {
                        Ok(decoded) => String::from_utf8_lossy(&decoded).to_string(),
                        Err(_) => b.clone(),
                    }
                } else {
                    String::new()
                };

                let mut http_resp = HttpResponse::json(resp.status, &body);
                // Merge response headers from client.
                for (k, v) in &resp.headers {
                    http_resp.headers.insert(k.clone(), v.clone());
                }
                http_resp
            }
            None => HttpResponse::json(404, r#"{"error":"no webhook registered for this path"}"#),
        }
    } else {
        HttpResponse::json(
            404,
            r#"{"error":"not found","hint":"This is a Pulse server. Connect via PLP (QUIC or WebSocket)."}"#,
        )
    }
}

/// Parse a minimal HTTP/1.1 request from a TCP stream.
async fn parse_request(stream: &mut TcpStream) -> Result<HttpRequest, String> {
    let mut reader = BufReader::new(stream);
    let mut request_line = String::new();
    reader
        .read_line(&mut request_line)
        .await
        .map_err(|e| format!("read request line: {}", e))?;

    let parts: Vec<&str> = request_line.trim().split_whitespace().collect();
    if parts.len() < 2 {
        return Err("invalid request line".to_string());
    }
    let method = parts[0].to_string();
    let path = parts[1].to_string();

    let mut headers = HashMap::new();
    let mut content_length: usize = 0;

    loop {
        let mut line = String::new();
        reader
            .read_line(&mut line)
            .await
            .map_err(|e| format!("read header: {}", e))?;
        let line = line.trim().to_string();
        if line.is_empty() {
            break;
        }
        if let Some((key, value)) = line.split_once(':') {
            let key = key.trim().to_lowercase();
            let value = value.trim().to_string();
            if key == "content-length" {
                content_length = value.parse().unwrap_or(0);
            }
            headers.insert(key, value);
        }
    }

    let body = if content_length > 0 {
        // Cap at 1 MB for webhook bodies.
        let capped = content_length.min(1024 * 1024);
        let mut body_buf = vec![0u8; capped];
        reader
            .read_exact(&mut body_buf)
            .await
            .map_err(|e| format!("read body: {}", e))?;
        Some(String::from_utf8_lossy(&body_buf).to_string())
    } else {
        None
    };

    Ok(HttpRequest {
        method,
        path,
        headers,
        body,
    })
}

/// Write an HTTP response to a TCP stream.
async fn write_response(
    stream: &mut TcpStream,
    resp: &HttpResponse,
) -> Result<(), std::io::Error> {
    let mut out = format!("HTTP/1.1 {} {}\r\n", resp.status, resp.status_text);
    out.push_str(&format!("Content-Length: {}\r\n", resp.body.len()));
    out.push_str("Connection: close\r\n");
    out.push_str("Server: pulse\r\n");
    for (k, v) in &resp.headers {
        out.push_str(&format!("{}: {}\r\n", k, v));
    }
    out.push_str("\r\n");
    out.push_str(&resp.body);

    stream.write_all(out.as_bytes()).await?;
    stream.flush().await?;
    Ok(())
}

/// Detect whether the first bytes of a TCP stream look like an HTTP request.
///
/// Returns `true` if the first bytes are a known HTTP method (GET, POST, PUT, etc.).
/// We use `peek()` so the bytes remain in the buffer for the actual parser.
pub async fn is_http_request(stream: &TcpStream) -> bool {
    let mut peek_buf = [0u8; 8];
    match stream.peek(&mut peek_buf).await {
        Ok(n) if n >= 3 => {
            let prefix = &peek_buf[..n];
            // Check for common HTTP method prefixes.
            prefix.starts_with(b"GET ")
                || prefix.starts_with(b"POST")
                || prefix.starts_with(b"PUT ")
                || prefix.starts_with(b"DELE")
                || prefix.starts_with(b"PATC")
                || prefix.starts_with(b"HEAD")
                || prefix.starts_with(b"OPTI")
        }
        _ => false,
    }
}

/// Check if the peeked HTTP request contains a WebSocket upgrade header.
///
/// We read more bytes to look for "Upgrade: websocket" (case-insensitive).
/// This is a heuristic — the full header parsing happens in the actual handler.
pub async fn is_websocket_upgrade(stream: &TcpStream) -> bool {
    // Peek enough to see the Upgrade header (typical HTTP request headers are < 4 KB).
    let mut peek_buf = [0u8; 4096];
    match stream.peek(&mut peek_buf).await {
        Ok(n) if n > 0 => {
            let s = String::from_utf8_lossy(&peek_buf[..n]).to_lowercase();
            s.contains("upgrade: websocket")
        }
        _ => false,
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::AsyncWriteExt;
    use tokio::net::TcpListener;

    async fn send_request(addr: std::net::SocketAddr, request: &str) -> String {
        let mut stream = TcpStream::connect(addr).await.unwrap();
        stream.write_all(request.as_bytes()).await.unwrap();
        stream.flush().await.unwrap();

        let mut response = Vec::new();
        stream.read_to_end(&mut response).await.unwrap();
        String::from_utf8_lossy(&response).to_string()
    }

    #[tokio::test]
    async fn health_endpoint_returns_200() {
        let registry = Arc::new(WebhookRegistry::new());
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let reg = Arc::clone(&registry);
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            handle_http_request(stream, reg).await;
        });

        let resp = send_request(addr, "GET /health HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
        server.await.unwrap();

        assert!(resp.contains("200 OK"));
        assert!(resp.contains("\"status\":\"ok\""));
    }

    #[tokio::test]
    async fn unknown_path_returns_404() {
        let registry = Arc::new(WebhookRegistry::new());
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let reg = Arc::clone(&registry);
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            handle_http_request(stream, reg).await;
        });

        let resp = send_request(addr, "GET /nonexistent HTTP/1.1\r\nHost: localhost\r\n\r\n").await;
        server.await.unwrap();

        assert!(resp.contains("404"));
        assert!(resp.contains("not found"));
    }

    #[tokio::test]
    async fn missing_webhook_returns_404() {
        let registry = Arc::new(WebhookRegistry::new());
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let reg = Arc::clone(&registry);
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            handle_http_request(stream, reg).await;
        });

        let resp = send_request(
            addr,
            "POST /_hook/conn1/stripe HTTP/1.1\r\nHost: localhost\r\n\r\n",
        )
        .await;
        server.await.unwrap();

        assert!(resp.contains("404"));
        assert!(resp.contains("no webhook registered"));
    }

    #[tokio::test]
    async fn webhook_delivery_end_to_end() {
        let registry = Arc::new(WebhookRegistry::new());
        let (tx, mut rx) = tokio::sync::mpsc::channel(16);

        // Register webhook.
        registry.register("conn1", "/stripe", tx).await;

        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let reg = Arc::clone(&registry);
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            handle_http_request(stream, reg).await;
        });

        // Simulate client responding to webhook.
        let client = tokio::spawn(async move {
            let (inbound, resp_tx) = rx.recv().await.unwrap();
            assert_eq!(inbound.method, "POST");
            assert!(inbound.body.as_ref().unwrap().contains("payment_intent"));

            resp_tx
                .send(crate::http_bridge::HttpResponsePayload {
                    request_id: inbound.request_id,
                    status: 200,
                    headers: HashMap::new(),
                    body: None,
                    error: None,
                })
                .unwrap();
        });

        let resp = send_request(
            addr,
            "POST /_hook/conn1/stripe HTTP/1.1\r\nHost: localhost\r\nContent-Length: 42\r\nStripe-Signature: t=123,v1=abc\r\n\r\n{\"type\":\"payment_intent.succeeded\",\"id\":1}",
        )
        .await;

        server.await.unwrap();
        client.await.unwrap();

        assert!(resp.contains("200 OK"));
    }

    #[tokio::test]
    async fn cors_preflight_returns_204() {
        let registry = Arc::new(WebhookRegistry::new());
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap();

        let reg = Arc::clone(&registry);
        let server = tokio::spawn(async move {
            let (stream, _) = listener.accept().await.unwrap();
            handle_http_request(stream, reg).await;
        });

        let resp = send_request(
            addr,
            "OPTIONS /_hook/conn1/stripe HTTP/1.1\r\nHost: localhost\r\n\r\n",
        )
        .await;
        server.await.unwrap();

        assert!(resp.contains("204"));
        assert!(resp.contains("Access-Control-Allow-Origin"));
    }
}
