//! HTTP Bridge — handles PLP HTTP_INBOUND / HTTP_OUTBOUND / HTTP_RESPONSE packets.
//!
//! **Outbound proxy:** The client sends `HttpOutbound` over PLP, the relay makes
//! the HTTP call via `reqwest`, and returns the result as `HttpResponse`.
//!
//! **Inbound webhooks:** External HTTP requests to `/_hook/{conn_id}/...` are
//! delivered to the connected client as `HttpInbound` packets. The client
//! responds with `HttpResponse`, which the relay sends back as an HTTP reply.

use bytes::Bytes;
use pulse_core::plp::{Header, Packet, PacketFlags, PacketType, PLP_VERSION};
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{debug, info, warn};

// ── Outbound: Client asks relay to make an HTTP call ───────────────────────

/// JSON payload for an HttpOutbound packet (client → relay).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpOutboundRequest {
    /// Unique request id for correlating the response.
    pub request_id: String,
    /// HTTP method (GET, POST, PUT, DELETE, PATCH, etc.)
    pub method: String,
    /// Full URL to call.
    pub url: String,
    /// Optional HTTP headers.
    #[serde(default)]
    pub headers: HashMap<String, String>,
    /// Optional body (base64-encoded for binary safety).
    #[serde(default)]
    pub body: Option<String>,
}

/// JSON payload for an HttpResponse packet (relay → client, or client → relay for webhooks).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpResponsePayload {
    /// The request id this response correlates to.
    pub request_id: String,
    /// HTTP status code.
    pub status: u16,
    /// Response headers.
    #[serde(default)]
    pub headers: HashMap<String, String>,
    /// Response body (base64-encoded).
    #[serde(default)]
    pub body: Option<String>,
    /// Error message if the request failed entirely (network error, timeout, etc.)
    #[serde(skip_serializing_if = "Option::is_none")]
    pub error: Option<String>,
}

/// JSON payload for an HttpInbound packet (relay → client, delivering a webhook).
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HttpInboundRequest {
    /// Unique request id for correlating the client's response.
    pub request_id: String,
    /// HTTP method of the incoming request.
    pub method: String,
    /// Path of the webhook (e.g. "/payments/stripe").
    pub path: String,
    /// HTTP headers from the incoming request.
    #[serde(default)]
    pub headers: HashMap<String, String>,
    /// Request body (base64-encoded).
    #[serde(default)]
    pub body: Option<String>,
}

/// Security configuration for the outbound HTTP proxy.
#[derive(Debug, Clone)]
pub struct HttpProxyConfig {
    /// Maximum allowed outbound domains (empty = allow all).
    pub allowed_domains: Vec<String>,
    /// Maximum response body size in bytes.
    pub max_response_bytes: usize,
    /// Request timeout in seconds.
    pub timeout_secs: u64,
    /// Rate limit: max outbound requests per minute per connection. (0 = unlimited).
    pub rate_limit_rpm: u32,
    /// Optional shared cache for GET requests (URL -> (Response, Expiry)).
    pub cache: Option<Arc<RwLock<HashMap<String, (HttpResponsePayload, std::time::Instant)>>>>,
    /// Cache TTL in seconds.
    pub cache_ttl_secs: u64,
    /// Rate limit state per session
    pub rate_limits: Option<Arc<RwLock<HashMap<String, Vec<std::time::Instant>>>>>,
}

impl Default for HttpProxyConfig {
    fn default() -> Self {
        Self {
            allowed_domains: Vec::new(), // allow all by default in dev
            max_response_bytes: 10 * 1024 * 1024, // 10 MB
            timeout_secs: 30,
            rate_limit_rpm: 60, // 1 per sec default
            cache: Some(Arc::new(RwLock::new(HashMap::new()))),
            cache_ttl_secs: 60, // Cache for 1 min by default
            rate_limits: Some(Arc::new(RwLock::new(HashMap::new()))),
        }
    }
}

/// Execute an outbound HTTP request on behalf of the client.
pub async fn handle_http_outbound(
    req: HttpOutboundRequest,
    config: &HttpProxyConfig,
    session_id: &str,
) -> HttpResponsePayload {
    debug!(
        "[HTTP Bridge] outbound {} {} (id={}, session={})",
        req.method, req.url, req.request_id, session_id
    );

    // ── Security: Rate limiting ──
    if config.rate_limit_rpm > 0 {
        if let Some(rate_limits) = &config.rate_limits {
            let mut state = rate_limits.write().await;
            let now = std::time::Instant::now();
            let window_start = now - std::time::Duration::from_secs(60);

            let timestamps = state.entry(session_id.to_string()).or_default();
            // Prune old timestamps
            timestamps.retain(|&t| t > window_start);

            if timestamps.len() >= config.rate_limit_rpm as usize {
                warn!("[HTTP Bridge] rate limit exceeded for session {}", session_id);
                return HttpResponsePayload {
                    request_id: req.request_id,
                    status: 429,
                    headers: HashMap::new(),
                    body: None,
                    error: Some("Rate limit exceeded (Too Many Requests)".to_string()),
                };
            }
            timestamps.push(now);

            // Housekeeping: prevent memory leak of expired sessions
            if state.len() > 10_000 {
                state.retain(|_, v| !v.is_empty());
            }
        }
    }

    // ── Security: domain allowlist ──
    if !config.allowed_domains.is_empty() {
        let allowed = match url::Url::parse(&req.url) {
            Ok(parsed) => {
                if let Some(host) = parsed.host_str() {
                    config.allowed_domains.iter().any(|d| host.ends_with(d))
                } else {
                    false
                }
            }
            Err(_) => false,
        };
        if !allowed {
            return HttpResponsePayload {
                request_id: req.request_id,
                status: 403,
                headers: HashMap::new(),
                body: None,
                error: Some("domain not in allowlist".to_string()),
            };
        }
    }

    // ── Security: block private IPs ──
    if let Ok(parsed) = url::Url::parse(&req.url) {
        if let Some(host) = parsed.host_str() {
            if is_private_host(host) {
                return HttpResponsePayload {
                    request_id: req.request_id,
                    status: 403,
                    headers: HashMap::new(),
                    body: None,
                    error: Some("requests to private IPs are blocked".to_string()),
                };
            }
        }
    }

    // ── Caching: check cache for GET requests ──
    let is_get = req.method.to_uppercase() == "GET";
    if is_get {
        if let Some(cache) = &config.cache {
            let cache_lock = cache.read().await;
            if let Some((cached_resp, expiry)) = cache_lock.get(&req.url) {
                if std::time::Instant::now() < *expiry {
                    debug!("[HTTP Bridge] outbound cache hit for {}", req.url);
                    let mut resp = cached_resp.clone();
                    resp.request_id = req.request_id; // Set correct request_id for the current caller
                    return resp;
                }
            }
        }
    }

    // ── Build and send the request ──
    let client = reqwest::Client::builder()
        .timeout(std::time::Duration::from_secs(config.timeout_secs))
        .build()
        .unwrap_or_default();

    let method = match req.method.to_uppercase().as_str() {
        "GET" => reqwest::Method::GET,
        "POST" => reqwest::Method::POST,
        "PUT" => reqwest::Method::PUT,
        "DELETE" => reqwest::Method::DELETE,
        "PATCH" => reqwest::Method::PATCH,
        "HEAD" => reqwest::Method::HEAD,
        "OPTIONS" => reqwest::Method::OPTIONS,
        _ => {
            return HttpResponsePayload {
                request_id: req.request_id,
                status: 400,
                headers: HashMap::new(),
                body: None,
                error: Some(format!("unsupported HTTP method: {}", req.method)),
            };
        }
    };

    let mut builder = client.request(method, &req.url);
    for (k, v) in &req.headers {
        builder = builder.header(k, v);
    }

    if let Some(ref body_b64) = req.body {
        match base64_decode(body_b64) {
            Ok(bytes) => {
                builder = builder.body(bytes);
            }
            Err(_) => {
                // Try as raw string (not base64)
                builder = builder.body(body_b64.clone());
            }
        }
    }

    match builder.send().await {
        Ok(resp) => {
            let status = resp.status().as_u16();
            let mut headers = HashMap::new();
            for (k, v) in resp.headers().iter() {
                if let Ok(val) = v.to_str() {
                    headers.insert(k.to_string(), val.to_string());
                }
            }

            match resp.bytes().await {
                Ok(body) => {
                    if body.len() > config.max_response_bytes {
                        return HttpResponsePayload {
                            request_id: req.request_id,
                            status: 413,
                            headers,
                            body: None,
                            error: Some("response body exceeds size limit".to_string()),
                        };
                    }
                    let payload = HttpResponsePayload {
                        request_id: req.request_id,
                        status,
                        headers,
                        body: Some(base64_encode(&body)),
                        error: None,
                    };

                    // ── Caching: Store successful GET responses ──
                    if is_get && status >= 200 && status < 300 {
                        if let Some(cache) = &config.cache {
                            let mut cache_lock = cache.write().await;
                            // Naive cache bounds (prevent unbounded growth)
                            if cache_lock.len() > 1000 {
                                cache_lock.clear(); // simplest form of LRU/eviction
                            }
                            let expiry = std::time::Instant::now()
                                + std::time::Duration::from_secs(config.cache_ttl_secs);
                            cache_lock.insert(req.url.clone(), (payload.clone(), expiry));
                        }
                    }

                    payload
                }
                Err(e) => HttpResponsePayload {
                    request_id: req.request_id,
                    status,
                    headers,
                    body: None,
                    error: Some(format!("failed to read response body: {}", e)),
                },
            }
        }
        Err(e) => {
            warn!("[HTTP Bridge] outbound request failed: {}", e);
            HttpResponsePayload {
                request_id: req.request_id,
                status: 502,
                headers: HashMap::new(),
                body: None,
                error: Some(format!("request failed: {}", e)),
            }
        }
    }
}

/// Encode an HttpResponsePayload into a PLP packet.
pub fn encode_http_response(resp: &HttpResponsePayload, request_seq: u32) -> Packet {
    let payload = serde_json::to_vec(resp).unwrap_or_default();
    Packet {
        header: Header {
            version: PLP_VERSION,
            packet_type: PacketType::HttpResponse,
            flags: PacketFlags::RELIABLE,
            stream_id: 0,
            sequence_number: request_seq,
            payload_length: payload.len() as u64,
        },
        payload: Bytes::from(payload),
    }
}

/// Check if a hostname resolves to a private/loopback IP (SSRF protection).
fn is_private_host(host: &str) -> bool {
    // Simple heuristic — block obvious private ranges
    let private_prefixes = [
        "127.", "10.", "172.16.", "172.17.", "172.18.", "172.19.", "172.20.", "172.21.",
        "172.22.", "172.23.", "172.24.", "172.25.", "172.26.", "172.27.", "172.28.",
        "172.29.", "172.30.", "172.31.", "192.168.", "0.", "169.254.",
    ];
    let lower = host.to_lowercase();
    if lower == "localhost" || lower == "::1" {
        return true;
    }
    private_prefixes.iter().any(|p| lower.starts_with(p))
}

fn base64_encode(data: &[u8]) -> String {
    use base64::Engine;
    base64::engine::general_purpose::STANDARD.encode(data)
}

fn base64_decode(s: &str) -> Result<Vec<u8>, base64::DecodeError> {
    use base64::Engine;
    base64::engine::general_purpose::STANDARD.decode(s)
}

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

    #[test]
    fn private_ip_detection() {
        assert!(is_private_host("127.0.0.1"));
        assert!(is_private_host("localhost"));
        assert!(is_private_host("10.0.0.1"));
        assert!(is_private_host("192.168.1.1"));
        assert!(is_private_host("172.16.0.1"));
        assert!(is_private_host("169.254.1.1"));
        assert!(!is_private_host("8.8.8.8"));
        assert!(!is_private_host("api.stripe.com"));
        assert!(!is_private_host("example.com"));
    }

    #[tokio::test]
    async fn blocked_domain_returns_403() {
        let config = HttpProxyConfig {
            allowed_domains: vec!["api.stripe.com".to_string()],
            ..Default::default()
        };
        let req = HttpOutboundRequest {
            request_id: "r1".to_string(),
            method: "GET".to_string(),
            url: "https://evil.example.com/steal".to_string(),
            headers: HashMap::new(),
            body: None,
        };
        let resp = handle_http_outbound(req, &config, "test_session").await;
        assert_eq!(resp.status, 403);
        assert!(resp.error.unwrap().contains("allowlist"));
    }

    #[tokio::test]
    async fn private_ip_returns_403() {
        let config = HttpProxyConfig::default();
        let req = HttpOutboundRequest {
            request_id: "r2".to_string(),
            method: "GET".to_string(),
            url: "http://127.0.0.1:8080/admin".to_string(),
            headers: HashMap::new(),
            body: None,
        };
        let resp = handle_http_outbound(req, &config, "test_session").await;
        assert_eq!(resp.status, 403);
        assert!(resp.error.unwrap().contains("private"));
    }

    #[tokio::test]
    async fn invalid_method_returns_400() {
        let config = HttpProxyConfig::default();
        let req = HttpOutboundRequest {
            request_id: "r3".to_string(),
            method: "FOOBAR".to_string(),
            url: "https://example.com".to_string(),
            headers: HashMap::new(),
            body: None,
        };
        let resp = handle_http_outbound(req, &config, "test_session").await;
        assert_eq!(resp.status, 400);
        assert!(resp.error.unwrap().contains("unsupported"));
    }

    #[test]
    fn http_response_packet_roundtrip() {
        let resp = HttpResponsePayload {
            request_id: "r4".to_string(),
            status: 200,
            headers: HashMap::new(),
            body: Some(base64_encode(b"hello")),
            error: None,
        };
        let pkt = encode_http_response(&resp, 99);
        assert_eq!(pkt.header.packet_type, PacketType::HttpResponse);

        let mut buf = Vec::new();
        pkt.encode(&mut buf);
        let decoded = Packet::decode(&mut buf.as_slice()).unwrap();
        assert_eq!(decoded.header.packet_type, PacketType::HttpResponse);

        let decoded_resp: HttpResponsePayload =
            serde_json::from_slice(&decoded.payload).unwrap();
        assert_eq!(decoded_resp.request_id, "r4");
        assert_eq!(decoded_resp.status, 200);
        assert_eq!(
            base64_decode(decoded_resp.body.as_ref().unwrap()).unwrap(),
            b"hello"
        );
    }
}
