//! Plugin capability dispatch
//!
//! Wires plugin capabilities (tools, webhooks, hooks, channels) into the gateway's
//! WebSocket and HTTP dispatch systems.
//!
//! # Webhook Routing
//!
//! Plugin webhooks are served under `/plugins/<plugin-id>/<path>` to ensure isolation.
//! This is a breaking change from the Node gateway which allowed arbitrary paths.
//!
//! # Hook Dispatch
//!
//! Hooks are dispatched to all registered hook plugins in registration order.
//! Modifiable hooks (before_agent_start, message_sending, before_tool_call,
//! tool_result_persist) can return modified payloads; other hooks are read-only.
//!
//! # Tool Dispatch
//!
//! Tools are dispatched by name. Plugin tools are namespaced as `<plugin-id>_<tool-name>`
//! to avoid collisions with built-in tools.

use std::collections::HashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};

use serde::{Deserialize, Serialize};
use thiserror::Error;

use super::bindings::{
    BindingError, HookEvent, PluginRegistry, ToolContext, ToolDefinition, ToolResult,
    WebhookRequest, WebhookResponse,
};

/// Dispatch errors
#[derive(Error, Debug)]
pub enum DispatchError {
    #[error("Plugin not found: {0}")]
    PluginNotFound(String),

    #[error("Tool not found: {0}")]
    ToolNotFound(String),

    #[error("Webhook path not found: {0}")]
    WebhookPathNotFound(String),

    #[error("Hook dispatch failed: {0}")]
    HookDispatchFailed(String),

    #[error("Binding error: {0}")]
    BindingError(#[from] BindingError),
}

/// List of hooks that allow payload modification
pub const MODIFIABLE_HOOKS: &[&str] = &[
    "before_agent_start",
    "message_sending",
    "before_tool_call",
    "tool_result_persist",
];

/// Check if a hook allows payload modification
pub fn is_modifiable_hook(hook_name: &str) -> bool {
    MODIFIABLE_HOOKS.contains(&hook_name)
}

/// Tool dispatcher that routes tool invocations to plugins
pub struct ToolDispatcher {
    registry: Arc<PluginRegistry>,
    /// Cache of tool name -> plugin ID mapping
    tool_map: parking_lot::RwLock<HashMap<String, String>>,
}

impl ToolDispatcher {
    /// Create a new tool dispatcher
    pub fn new(registry: Arc<PluginRegistry>) -> Self {
        Self {
            registry,
            tool_map: parking_lot::RwLock::new(HashMap::new()),
        }
    }

    /// Refresh the tool mapping from the registry
    pub fn refresh_tool_map(&self) -> Result<(), DispatchError> {
        let mut map = self.tool_map.write();
        map.clear();

        for (plugin_id, instance) in self.registry.get_tools() {
            let definitions = instance.get_definitions()?;
            for def in definitions {
                // Namespace tool names to avoid collisions
                let namespaced_name = format!("{}_{}", plugin_id, def.name);
                if let Some(existing) = map.insert(namespaced_name.clone(), plugin_id.clone()) {
                    tracing::warn!(
                        tool = %namespaced_name,
                        new_plugin = %plugin_id,
                        existing_plugin = %existing,
                        "tool name collision: overwriting existing mapping"
                    );
                }
                // Also store the original name for plugins that want shorter names
                if let Some(existing) = map.insert(def.name.clone(), plugin_id.clone()) {
                    tracing::warn!(
                        tool = %def.name,
                        new_plugin = %plugin_id,
                        existing_plugin = %existing,
                        "tool name collision: overwriting existing mapping"
                    );
                }
            }
        }

        Ok(())
    }

    /// List all available tools from plugins
    pub fn list_tools(&self) -> Result<Vec<ToolDefinition>, DispatchError> {
        let mut tools = Vec::new();

        for (plugin_id, instance) in self.registry.get_tools() {
            let definitions = instance.get_definitions()?;
            for mut def in definitions {
                // Namespace the tool name
                def.name = format!("{}_{}", plugin_id, def.name);
                tools.push(def);
            }
        }

        Ok(tools)
    }

    /// Invoke a tool by name
    pub fn invoke(
        &self,
        tool_name: &str,
        params: &str,
        ctx: ToolContext,
    ) -> Result<ToolResult, DispatchError> {
        // Look up the plugin that provides this tool
        let plugin_id = {
            let map = self.tool_map.read();
            map.get(tool_name).cloned()
        };

        let plugin_id = match plugin_id {
            Some(id) => id,
            None => {
                // Try to find by checking each plugin's definitions
                let mut found_id = None;
                for (id, instance) in self.registry.get_tools() {
                    if let Ok(defs) = instance.get_definitions() {
                        for def in defs {
                            if def.name == tool_name || format!("{}_{}", id, def.name) == tool_name
                            {
                                found_id = Some(id.clone());
                                break;
                            }
                        }
                    }
                    if found_id.is_some() {
                        break;
                    }
                }
                found_id.ok_or_else(|| DispatchError::ToolNotFound(tool_name.to_string()))?
            }
        };

        // Get the plugin instance
        let instance = self
            .registry
            .get_tool(&plugin_id)
            .ok_or_else(|| DispatchError::PluginNotFound(plugin_id.clone()))?;

        // Extract the actual tool name (remove plugin prefix if present)
        let actual_name = tool_name
            .strip_prefix(&format!("{}_", plugin_id))
            .unwrap_or(tool_name);

        // Invoke the tool
        instance
            .invoke(actual_name, params, ctx)
            .map_err(Into::into)
    }
}

/// Webhook dispatcher that routes HTTP requests to plugins
pub struct WebhookDispatcher {
    registry: Arc<PluginRegistry>,
    /// Cache of path -> plugin ID mapping
    path_map: parking_lot::RwLock<HashMap<String, String>>,
    /// Last time the path map was refreshed
    last_refresh: parking_lot::RwLock<Option<Instant>>,
    /// Minimum interval between refreshes
    refresh_interval: Duration,
}

impl WebhookDispatcher {
    /// Create a new webhook dispatcher
    pub fn new(registry: Arc<PluginRegistry>) -> Self {
        Self {
            registry,
            path_map: parking_lot::RwLock::new(HashMap::new()),
            last_refresh: parking_lot::RwLock::new(None),
            refresh_interval: Duration::from_secs(5),
        }
    }

    /// Refresh the path mapping from the registry
    pub fn refresh_path_map(&self) -> Result<(), DispatchError> {
        let mut map = self.path_map.write();
        map.clear();

        for (plugin_id, instance) in self.registry.get_webhooks() {
            let paths = instance.get_paths()?;
            for path in paths {
                if !plugin_webhook_path_is_well_formed(&path) {
                    tracing::warn!(
                        plugin_id = %plugin_id,
                        path = %path,
                        "refusing to register malformed plugin webhook path; \
                         path must start with '/' and contain no '..' segments"
                    );
                    continue;
                }
                // Namespace paths under /plugins/<plugin-id>/
                let full_path = format!("/plugins/{}{}", plugin_id, path);
                map.insert(full_path, plugin_id.clone());
            }
        }

        let mut last_refresh = self.last_refresh.write();
        *last_refresh = Some(Instant::now());
        Ok(())
    }

    /// Refresh the path map if the cache is stale.
    pub fn refresh_path_map_if_stale(&self) -> Result<(), DispatchError> {
        let should_refresh = {
            let last_refresh = self.last_refresh.read();
            match *last_refresh {
                Some(ts) => ts.elapsed() >= self.refresh_interval,
                None => true,
            }
        };

        if should_refresh {
            self.refresh_path_map()?;
        }

        Ok(())
    }

    /// List all registered webhook paths
    pub fn list_paths(&self) -> Result<Vec<String>, DispatchError> {
        let mut paths = Vec::new();

        for (plugin_id, instance) in self.registry.get_webhooks() {
            let plugin_paths = instance.get_paths()?;
            for path in plugin_paths {
                if !plugin_webhook_path_is_well_formed(&path) {
                    continue;
                }
                paths.push(format!("/plugins/{}{}", plugin_id, path));
            }
        }

        Ok(paths)
    }

    /// Handle a webhook request
    ///
    /// The path should be the full path including the `/plugins/<plugin-id>/` prefix.
    pub fn handle(
        &self,
        path: &str,
        request: WebhookRequest,
    ) -> Result<WebhookResponse, DispatchError> {
        // Look up the plugin that handles this path
        let plugin_id = {
            let map = self.path_map.read();

            // Try exact match first
            if let Some(id) = map.get(path) {
                Some(id.clone())
            } else {
                // Try boundary-aware prefix match for paths with
                // parameters. SECURITY: bare `starts_with` lets a
                // shorter-named plugin hijack a longer-named plugin's
                // routes (e.g. plugin "a" registers `/plugins/a` and
                // also receives `/plugins/abcd/foo`). Require that the
                // byte immediately following the registered prefix is
                // `/`, OR the registered prefix itself ends with `/`.
                map.iter()
                    .find(|(p, _)| {
                        let p = p.as_str();
                        path.starts_with(p)
                            && (path.len() == p.len()
                                || p.ends_with('/')
                                || path.as_bytes().get(p.len()) == Some(&b'/'))
                    })
                    .map(|(_, id)| id.clone())
            }
        };

        let plugin_id = match plugin_id {
            Some(id) => id,
            None => {
                // Try to extract plugin ID from path and check directly
                if let Some(id) = extract_plugin_id_from_path(path) {
                    if self.registry.get_webhook(&id).is_some() {
                        id
                    } else {
                        return Err(DispatchError::WebhookPathNotFound(path.to_string()));
                    }
                } else {
                    return Err(DispatchError::WebhookPathNotFound(path.to_string()));
                }
            }
        };

        // Get the plugin instance
        let instance = self
            .registry
            .get_webhook(&plugin_id)
            .ok_or_else(|| DispatchError::PluginNotFound(plugin_id.clone()))?;

        // Strip the /plugins/<plugin-id> prefix from the path for the plugin
        let plugin_path = strip_plugin_prefix(path, &plugin_id);

        // Create a modified request with the stripped path
        let plugin_request = WebhookRequest {
            path: plugin_path,
            ..request
        };

        // Handle the request
        instance.handle(plugin_request).map_err(Into::into)
    }
}

/// Validate that a plugin-supplied webhook path is well-formed before
/// it is joined onto the `/plugins/<id>` namespace prefix.
///
/// SECURITY: refuses any path that does not start with `/` or contains
/// `..` segments. A path that does not start with `/` would let a
/// plugin escape the `/plugins/<id>/...` namespace via string
/// concatenation: `format!("/plugins/{}{}", "a", "bcd/foo")` =
/// `/plugins/abcd/foo`, which the prefix-match step would then route
/// to plugin `a` instead of plugin `abcd`. A `..` segment in the path
/// would similarly walk up out of the plugin namespace at request
/// time. The boundary-aware prefix-match in `handle` defends against
/// the same hijack class from the other side, but rejecting malformed
/// paths at registration keeps the path-map free of these footguns in
/// the first place and matches the `validate_plugin_id` discipline
/// already applied to plugin IDs.
pub(crate) fn plugin_webhook_path_is_well_formed(path: &str) -> bool {
    if !path.starts_with('/') {
        return false;
    }
    for segment in path.split('/') {
        if segment == ".." || segment == "." {
            return false;
        }
    }
    true
}

/// Extract plugin ID from a path like /plugins/<plugin-id>/...
fn extract_plugin_id_from_path(path: &str) -> Option<String> {
    let path = path.strip_prefix("/plugins/")?;
    let end = path.find('/').unwrap_or(path.len());
    Some(path[..end].to_string())
}

/// Strip the /plugins/<plugin-id> prefix from a path
fn strip_plugin_prefix(path: &str, plugin_id: &str) -> String {
    let prefix = format!("/plugins/{}", plugin_id);
    path.strip_prefix(&prefix).unwrap_or(path).to_string()
}

/// Hook dispatcher that routes hook events to plugins
pub struct HookDispatcher {
    registry: Arc<PluginRegistry>,
}

/// Result of dispatching a hook to all handlers
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct HookDispatchResult {
    /// Whether any handler handled the event
    pub handled: bool,
    /// Whether any handler requested cancellation (only for cancellable hooks)
    pub cancelled: bool,
    /// The final payload after all modifications (only for modifiable hooks)
    pub final_payload: Option<String>,
    /// Number of handlers that processed the event
    pub handler_count: usize,
}

impl HookDispatcher {
    /// Create a new hook dispatcher
    pub fn new(registry: Arc<PluginRegistry>) -> Self {
        Self { registry }
    }

    /// Dispatch a hook event to all registered handlers
    ///
    /// Hooks are called in registration order. For modifiable hooks, each handler
    /// receives the payload as modified by the previous handler.
    pub fn dispatch(
        &self,
        hook_name: &str,
        payload: &str,
    ) -> Result<HookDispatchResult, DispatchError> {
        let is_modifiable = is_modifiable_hook(hook_name);
        let mut current_payload = payload.to_string();
        let mut handled = false;
        let mut cancelled = false;
        let mut handler_count = 0;

        // Get all hook plugins
        let hooks = self.registry.get_hooks();

        for (plugin_id, instance) in hooks {
            // Check if this plugin handles this hook
            let registered_hooks = match instance.get_hooks() {
                Ok(hooks) => hooks,
                Err(e) => {
                    tracing::warn!(
                        plugin_id = %plugin_id,
                        error = %e,
                        "Failed to get hooks from plugin"
                    );
                    continue;
                }
            };

            if !registered_hooks.contains(&hook_name.to_string()) {
                continue;
            }

            // Create the event
            let event = HookEvent {
                hook_name: hook_name.to_string(),
                payload: current_payload.clone(),
            };

            // Call the handler
            let result = match instance.handle(event) {
                Ok(result) => result,
                Err(e) => {
                    tracing::warn!(
                        plugin_id = %plugin_id,
                        hook = %hook_name,
                        error = %e,
                        "Hook handler failed"
                    );
                    continue;
                }
            };

            handler_count += 1;

            if result.handled {
                handled = true;
            }

            if result.cancel {
                cancelled = true;
            }

            // Update payload if modified (only for modifiable hooks)
            if is_modifiable {
                if let Some(modified) = result.modified_payload {
                    current_payload = modified;
                }
            }
        }

        Ok(HookDispatchResult {
            handled,
            cancelled,
            final_payload: if is_modifiable {
                Some(current_payload)
            } else {
                None
            },
            handler_count,
        })
    }

    /// List all registered hooks across all plugins
    pub fn list_hooks(&self) -> Result<Vec<(String, Vec<String>)>, DispatchError> {
        let mut result = Vec::new();

        for (plugin_id, instance) in self.registry.get_hooks() {
            let hooks = instance.get_hooks()?;
            result.push((plugin_id, hooks));
        }

        Ok(result)
    }
}

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

    #[test]
    fn test_modifiable_hooks() {
        assert!(is_modifiable_hook("before_agent_start"));
        assert!(is_modifiable_hook("message_sending"));
        assert!(is_modifiable_hook("before_tool_call"));
        assert!(is_modifiable_hook("tool_result_persist"));

        assert!(!is_modifiable_hook("agent_end"));
        assert!(!is_modifiable_hook("session_start"));
        assert!(!is_modifiable_hook("gateway_start"));
    }

    /// Round-8 plugin-sandbox-HIGH 1 regression: a plugin's
    /// `get_paths()` export is operator-untrusted (the plugin author
    /// supplies the WASM that returns it). Malformed paths must be
    /// rejected before they enter the path-map; otherwise a path of
    /// `""` or `"bcd/foo"` lets plugin `a` hijack `/plugins/abcd/...`
    /// routes intended for plugin `abcd` via prefix matching.
    #[test]
    fn plugin_webhook_path_is_well_formed_rejects_path_traversal_and_no_leading_slash() {
        // Good shapes
        assert!(plugin_webhook_path_is_well_formed("/webhook"));
        assert!(plugin_webhook_path_is_well_formed("/some/deep/path"));
        assert!(plugin_webhook_path_is_well_formed("/"));
        assert!(plugin_webhook_path_is_well_formed("/x"));

        // Missing leading slash → string concat would escape the namespace
        assert!(!plugin_webhook_path_is_well_formed(""));
        assert!(!plugin_webhook_path_is_well_formed("bcd/foo"));
        assert!(!plugin_webhook_path_is_well_formed("webhook"));

        // Dot segments → request-time traversal
        assert!(!plugin_webhook_path_is_well_formed("/../sibling"));
        assert!(!plugin_webhook_path_is_well_formed("/./current"));
        assert!(!plugin_webhook_path_is_well_formed("/foo/../bar"));
    }

    /// Sanity: the boundary-aware prefix-match in `handle` must NOT
    /// let `/plugins/a` match `/plugins/abcd/foo`. The same shape would
    /// have hijacked routes pre-fix; this test pins the boundary at
    /// the dispatcher's prefix-match step rather than at the
    /// registration validator, so even a hypothetical mis-registration
    /// (e.g. via a future code path that bypasses
    /// `plugin_webhook_path_is_well_formed`) still cannot hijack.
    #[test]
    fn webhook_prefix_match_requires_segment_boundary() {
        use std::collections::HashMap;
        let mut map: HashMap<String, String> = HashMap::new();
        map.insert("/plugins/a".to_string(), "a".to_string());

        let path = "/plugins/abcd/foo";
        // Naive `starts_with` would match (regression risk); the
        // boundary-aware check should NOT.
        let hit = map.iter().find(|(p, _)| {
            let p = p.as_str();
            path.starts_with(p)
                && (path.len() == p.len()
                    || p.ends_with('/')
                    || path.as_bytes().get(p.len()) == Some(&b'/'))
        });
        assert!(
            hit.is_none(),
            "/plugins/a must NOT prefix-match /plugins/abcd/foo (boundary-aware)"
        );

        // Sanity: a legitimate child path WITH a `/` boundary should
        // match the same prefix.
        let legit = "/plugins/a/foo";
        let hit = map.iter().find(|(p, _)| {
            let p = p.as_str();
            legit.starts_with(p)
                && (legit.len() == p.len()
                    || p.ends_with('/')
                    || legit.as_bytes().get(p.len()) == Some(&b'/'))
        });
        assert!(hit.is_some(), "/plugins/a/foo must prefix-match /plugins/a");
    }

    #[test]
    fn test_extract_plugin_id_from_path() {
        assert_eq!(
            extract_plugin_id_from_path("/plugins/msteams/webhook"),
            Some("msteams".to_string())
        );
        assert_eq!(
            extract_plugin_id_from_path("/plugins/my-plugin/callback/123"),
            Some("my-plugin".to_string())
        );
        assert_eq!(extract_plugin_id_from_path("/other/path"), None);
        assert_eq!(
            extract_plugin_id_from_path("/plugins/"),
            Some("".to_string())
        );
    }

    #[test]
    fn test_strip_plugin_prefix() {
        assert_eq!(
            strip_plugin_prefix("/plugins/msteams/webhook", "msteams"),
            "/webhook"
        );
        assert_eq!(
            strip_plugin_prefix("/plugins/my-plugin/callback/123", "my-plugin"),
            "/callback/123"
        );
        assert_eq!(strip_plugin_prefix("/other/path", "msteams"), "/other/path");
    }

    #[test]
    fn test_hook_dispatch_result() {
        let result = HookDispatchResult {
            handled: true,
            cancelled: false,
            final_payload: Some(r#"{"modified": true}"#.to_string()),
            handler_count: 2,
        };
        assert!(result.handled);
        assert!(!result.cancelled);
        assert_eq!(result.handler_count, 2);
    }

    #[test]
    fn test_tool_dispatcher_creation() {
        let registry = Arc::new(PluginRegistry::new());
        let dispatcher = ToolDispatcher::new(registry);
        let tools = dispatcher.list_tools().unwrap();
        assert!(tools.is_empty());
    }

    #[test]
    fn test_webhook_dispatcher_creation() {
        let registry = Arc::new(PluginRegistry::new());
        let dispatcher = WebhookDispatcher::new(registry);
        let paths = dispatcher.list_paths().unwrap();
        assert!(paths.is_empty());
    }

    #[test]
    fn test_hook_dispatcher_creation() {
        let registry = Arc::new(PluginRegistry::new());
        let dispatcher = HookDispatcher::new(registry);
        let hooks = dispatcher.list_hooks().unwrap();
        assert!(hooks.is_empty());
    }
}
