//! Webhook Registry — maps inbound HTTP webhook paths to connected PLP clients.
//!
//! When a client registers a webhook via `conn.service('webhooks', 'register', { path: '/stripe' })`,
//! an entry is created here. When an HTTP request arrives at `/_hook/{conn_id}/{path}`,
//! it is delivered to the connected client as an `HTTP_INBOUND` packet, and the
//! client's `HTTP_RESPONSE` is sent back as the HTTP reply.

use crate::http_bridge::{HttpInboundRequest, HttpResponsePayload};
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::{oneshot, RwLock};
use tracing::{debug, info, warn};

/// A registered webhook endpoint.
#[derive(Debug, Clone)]
pub struct WebhookRegistration {
    /// The user-specified path (e.g. "/payments/stripe").
    pub path: String,
    /// Connection identifier this webhook belongs to.
    pub conn_id: String,
    /// Channel to deliver inbound HTTP requests to the PLP session handler.
    /// The session handler will send the request as an HTTP_INBOUND packet.
    pub inbound_tx: tokio::sync::mpsc::Sender<(HttpInboundRequest, oneshot::Sender<HttpResponsePayload>)>,
}

/// Global webhook registry — shared across all connections and the HTTP listener.
pub struct WebhookRegistry {
    /// Key: "{conn_id}/{path}" → registration
    registrations: RwLock<HashMap<String, WebhookRegistration>>,
    /// Counter for generating unique request IDs.
    next_id: std::sync::atomic::AtomicU64,
}

impl WebhookRegistry {
    pub fn new() -> Self {
        Self {
            registrations: RwLock::new(HashMap::new()),
            next_id: std::sync::atomic::AtomicU64::new(1),
        }
    }

    /// Register a webhook path for a connection.
    /// Returns the full external URL path like `/_hook/{conn_id}/{user_path}`.
    pub async fn register(
        &self,
        conn_id: &str,
        user_path: &str,
        inbound_tx: tokio::sync::mpsc::Sender<(HttpInboundRequest, oneshot::Sender<HttpResponsePayload>)>,
    ) -> String {
        let clean_path = user_path.trim_start_matches('/');
        let key = format!("{}/{}", conn_id, clean_path);
        let hook_url = format!("/_hook/{}/{}", conn_id, clean_path);

        let reg = WebhookRegistration {
            path: clean_path.to_string(),
            conn_id: conn_id.to_string(),
            inbound_tx,
        };

        self.registrations.write().await.insert(key, reg);
        info!(
            "[Webhook] registered: {} for conn={}",
            hook_url, conn_id
        );
        hook_url
    }

    /// Unregister all webhooks for a connection (called on disconnect).
    pub async fn unregister_conn(&self, conn_id: &str) {
        let mut regs = self.registrations.write().await;
        let keys_to_remove: Vec<String> = regs
            .keys()
            .filter(|k| k.starts_with(&format!("{}/", conn_id)))
            .cloned()
            .collect();
        for key in &keys_to_remove {
            regs.remove(key);
        }
        if !keys_to_remove.is_empty() {
            info!(
                "[Webhook] unregistered {} hooks for conn={}",
                keys_to_remove.len(),
                conn_id
            );
        }
    }

    /// List all webhooks for a connection.
    pub async fn list_for_conn(&self, conn_id: &str) -> Vec<String> {
        let regs = self.registrations.read().await;
        regs.values()
            .filter(|r| r.conn_id == conn_id)
            .map(|r| format!("/_hook/{}/{}", r.conn_id, r.path))
            .collect()
    }

    /// Deliver an inbound HTTP request to the registered webhook handler.
    /// Returns `None` if no webhook is registered for the path.
    /// Returns `Some(response)` after the PLP client responds (or times out).
    pub async fn deliver(
        &self,
        hook_path: &str,  // e.g. "conn123/payments/stripe"
        method: &str,
        headers: HashMap<String, String>,
        body: Option<String>,
    ) -> Option<HttpResponsePayload> {
        let regs = self.registrations.read().await;
        let reg = regs.get(hook_path)?;

        let request_id = format!(
            "wh-{}",
            self.next_id
                .fetch_add(1, std::sync::atomic::Ordering::Relaxed)
        );

        let inbound = HttpInboundRequest {
            request_id: request_id.clone(),
            method: method.to_string(),
            path: format!("/{}", reg.path),
            headers,
            body,
        };

        let (resp_tx, resp_rx) = oneshot::channel();

        // Send to the PLP session handler.
        if reg.inbound_tx.send((inbound, resp_tx)).await.is_err() {
            warn!("[Webhook] client disconnected for {}", hook_path);
            return Some(HttpResponsePayload {
                request_id,
                status: 503,
                headers: HashMap::new(),
                body: None,
                error: Some("webhook client is disconnected".to_string()),
            });
        }

        // Wait for the client's response (with timeout).
        match tokio::time::timeout(std::time::Duration::from_secs(30), resp_rx).await {
            Ok(Ok(resp)) => Some(resp),
            Ok(Err(_)) => {
                // Channel dropped — client disconnected mid-request.
                Some(HttpResponsePayload {
                    request_id,
                    status: 503,
                    headers: HashMap::new(),
                    body: None,
                    error: Some("webhook client disconnected".to_string()),
                })
            }
            Err(_) => {
                // Timeout.
                warn!("[Webhook] client response timeout for {}", hook_path);
                Some(HttpResponsePayload {
                    request_id,
                    status: 504,
                    headers: HashMap::new(),
                    body: None,
                    error: Some("webhook client response timed out (30s)".to_string()),
                })
            }
        }
    }

    /// Generate a unique connection ID for webhook URL generation.
    pub fn generate_conn_id() -> String {
        use std::time::{SystemTime, UNIX_EPOCH};
        let ts = SystemTime::now()
            .duration_since(UNIX_EPOCH)
            .unwrap_or_default()
            .as_millis();
        // Short but unguessable enough for webhook URLs.
        format!("{:x}", ts ^ 0xDEAD_BEEF_CAFE)
    }
}

/// Encode an HttpInboundRequest into a PLP HTTP_INBOUND packet.
pub fn encode_http_inbound(req: &HttpInboundRequest, seq: u32) -> Packet {
    let payload = serde_json::to_vec(req).unwrap_or_default();
    Packet {
        header: Header {
            version: PLP_VERSION,
            packet_type: PacketType::HttpInbound,
            flags: PacketFlags::RELIABLE,
            stream_id: 0,
            sequence_number: seq,
            payload_length: payload.len() as u64,
        },
        payload: bytes::Bytes::from(payload),
    }
}

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

    #[tokio::test]
    async fn register_and_list_webhooks() {
        let registry = WebhookRegistry::new();
        let (tx, _rx) = tokio::sync::mpsc::channel(16);

        let url = registry.register("conn1", "/stripe", tx.clone()).await;
        assert_eq!(url, "/_hook/conn1/stripe");

        let urls = registry.list_for_conn("conn1").await;
        assert_eq!(urls.len(), 1);
        assert!(urls[0].contains("stripe"));
    }

    #[tokio::test]
    async fn unregister_removes_all_for_conn() {
        let registry = WebhookRegistry::new();
        let (tx, _rx) = tokio::sync::mpsc::channel(16);

        registry.register("conn1", "/stripe", tx.clone()).await;
        registry.register("conn1", "/github", tx.clone()).await;
        registry.register("conn2", "/twilio", tx.clone()).await;

        registry.unregister_conn("conn1").await;

        assert!(registry.list_for_conn("conn1").await.is_empty());
        assert_eq!(registry.list_for_conn("conn2").await.len(), 1);
    }

    #[tokio::test]
    async fn deliver_to_missing_returns_none() {
        let registry = WebhookRegistry::new();
        let result = registry
            .deliver("nonexistent/path", "POST", HashMap::new(), None)
            .await;
        assert!(result.is_none());
    }

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

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

        // Simulate a client responding.
        let deliver_handle = tokio::spawn({
            let registry_ref = &registry as *const WebhookRegistry;
            // Safety: we join the task before the test ends.
            let registry_ref = unsafe { &*registry_ref };
            async move {
                registry_ref
                    .deliver("conn1/stripe", "POST", HashMap::new(), Some("body".into()))
                    .await
            }
        });

        // Receive the inbound request on the PLP side.
        let (inbound, resp_tx) = rx.recv().await.unwrap();
        assert_eq!(inbound.method, "POST");
        assert_eq!(inbound.path, "/stripe");

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

        let result = deliver_handle.await.unwrap().unwrap();
        assert_eq!(result.status, 200);
    }

    #[test]
    fn conn_id_is_hex() {
        let id = WebhookRegistry::generate_conn_id();
        assert!(!id.is_empty());
        // Should be valid hex
        assert!(id.chars().all(|c| c.is_ascii_hexdigit()));
    }
}
