//! Core parallel tool executor implementation

use crate::tools::base::{ConcurrencyMode, Tool};
use crate::tools::permission::{PermissionCache, SharedPermissionHandler, ToolContext};
use crate::tools::types::{ToolCall, ToolResult};
use dashmap::DashMap;
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio::sync::{Mutex, RwLock, Semaphore};
use tokio::time::timeout;
use tokio_util::sync::CancellationToken;

use super::super::config::{ExecutorStats, ParallelExecutorConfig, ToolExecutionResult};

/// Enhanced parallel tool executor
pub struct ParallelToolExecutor {
    /// Registered tools
    pub(super) tools: DashMap<String, Arc<dyn Tool>>,
    /// Configuration
    pub(super) config: ParallelExecutorConfig,
    /// Global concurrency semaphore
    pub(super) global_semaphore: Arc<Semaphore>,
    /// Per-type semaphores for ExclusiveByType mode
    pub(super) type_semaphores: DashMap<String, Arc<Semaphore>>,
    /// Sequential execution lock (for Sequential mode)
    pub(super) sequential_lock: Arc<Mutex<()>>,
    /// Limited concurrency semaphores (tool name -> semaphore)
    pub(super) limited_semaphores: DashMap<String, Arc<Semaphore>>,
    /// Permission handler
    pub(super) permission_handler: RwLock<Option<SharedPermissionHandler>>,
    /// Permission cache
    pub(super) permission_cache: Arc<PermissionCache>,
    /// Tool context for permission checking
    pub(super) tool_context: RwLock<ToolContext>,
    /// Cancellation token
    pub(super) cancellation_token: CancellationToken,
    /// Execution statistics
    pub(super) stats: Arc<RwLock<ExecutorStats>>,
}

impl ParallelToolExecutor {
    /// Create a new parallel executor with default configuration
    pub fn new() -> Self {
        Self::with_config(ParallelExecutorConfig::default())
    }

    /// Create a new parallel executor with custom configuration
    pub fn with_config(config: ParallelExecutorConfig) -> Self {
        let global_semaphore = Arc::new(Semaphore::new(config.max_global_concurrency));

        Self {
            tools: DashMap::new(),
            config,
            global_semaphore,
            type_semaphores: DashMap::new(),
            sequential_lock: Arc::new(Mutex::new(())),
            limited_semaphores: DashMap::new(),
            permission_handler: RwLock::new(None),
            permission_cache: Arc::new(PermissionCache::new()),
            tool_context: RwLock::new(ToolContext::default()),
            cancellation_token: CancellationToken::new(),
            stats: Arc::new(RwLock::new(ExecutorStats::default())),
        }
    }

    /// Register a tool
    pub fn register_tool(&self, tool: Arc<dyn Tool>) {
        let name = tool.name().to_string();

        match tool.concurrency_mode() {
            ConcurrencyMode::Limited(max) => {
                self.limited_semaphores
                    .insert(name.clone(), Arc::new(Semaphore::new(max)));
            }
            ConcurrencyMode::ExclusiveByType => {
                self.type_semaphores
                    .insert(name.clone(), Arc::new(Semaphore::new(1)));
            }
            _ => {}
        }

        self.tools.insert(name, tool);
    }

    /// Register multiple tools
    pub fn register_tools(&self, tools: Vec<Arc<dyn Tool>>) {
        for tool in tools {
            self.register_tool(tool);
        }
    }

    /// Set the permission handler
    pub async fn set_permission_handler(&self, handler: SharedPermissionHandler) {
        *self.permission_handler.write().await = Some(handler);
    }

    /// Set the tool context
    pub async fn set_tool_context(&self, context: ToolContext) {
        *self.tool_context.write().await = context;
    }

    /// Get a child cancellation token
    pub fn child_token(&self) -> CancellationToken {
        self.cancellation_token.child_token()
    }

    /// Cancel all running executions
    pub fn cancel_all(&self) {
        self.cancellation_token.cancel();
    }

    /// Execute a single tool call
    pub async fn execute_tool(&self, call: &ToolCall) -> ToolExecutionResult {
        let mut wait_time = Duration::ZERO;
        let mut permission_checked = false;

        // Check for cancellation
        if self.cancellation_token.is_cancelled() {
            return ToolExecutionResult {
                result: ToolResult::error(&call.id, &call.name, "Execution cancelled"),
                wait_time,
                execution_time: Duration::ZERO,
                permission_checked,
            };
        }

        // Get the tool
        let tool = match self.tools.get(&call.name) {
            Some(tool) => tool.clone(),
            None => {
                return ToolExecutionResult {
                    result: ToolResult::error(
                        &call.id,
                        &call.name,
                        format!("Tool '{}' not found", call.name),
                    ),
                    wait_time,
                    execution_time: Duration::ZERO,
                    permission_checked,
                };
            }
        };

        // Check permissions if enabled
        if self.config.check_permissions {
            permission_checked = true;
            if let Some(result) = self.check_permission(call, &tool).await {
                return result;
            }
        }

        // Acquire necessary semaphores
        let wait_start = Instant::now();
        let _permits = match self.acquire_permits(&tool).await {
            Ok(permits) => {
                wait_time = wait_start.elapsed();
                permits
            }
            Err(e) => {
                return ToolExecutionResult {
                    result: ToolResult::error(&call.id, &call.name, e.to_string()),
                    wait_time: wait_start.elapsed(),
                    execution_time: Duration::ZERO,
                    permission_checked,
                };
            }
        };

        // Check cancellation again after waiting
        if self.cancellation_token.is_cancelled() {
            self.increment_cancellations().await;
            return ToolExecutionResult {
                result: ToolResult::error(&call.id, &call.name, "Execution cancelled"),
                wait_time,
                execution_time: Duration::ZERO,
                permission_checked,
            };
        }

        // Execute the tool with timeout
        let execution_start = Instant::now();
        let execution_timeout = tool
            .max_execution_duration()
            .unwrap_or(self.config.default_timeout);

        let result = tokio::select! {
            _ = self.cancellation_token.cancelled() => {
                self.increment_cancellations().await;
                ToolResult::error(&call.id, &call.name, "Execution cancelled")
            }
            result = timeout(execution_timeout, tool.execute_with_timing(call)) => {
                match result {
                    Ok(r) => r,
                    Err(_) => {
                        self.increment_timeouts().await;
                        ToolResult::error(
                            &call.id,
                            &call.name,
                            format!("Tool execution timed out after {:?}", execution_timeout),
                        )
                    }
                }
            }
        };

        let execution_time = execution_start.elapsed();

        // Update statistics
        {
            let mut stats = self.stats.write().await;
            stats.total_executions += 1;
            if result.success {
                stats.successful_executions += 1;
            } else {
                stats.failed_executions += 1;
            }
            stats.total_execution_time += execution_time;
            stats.total_wait_time += wait_time;
        }

        ToolExecutionResult {
            result,
            wait_time,
            execution_time,
            permission_checked,
        }
    }

    async fn increment_cancellations(&self) {
        let mut stats = self.stats.write().await;
        stats.cancellations += 1;
    }

    async fn increment_timeouts(&self) {
        let mut stats = self.stats.write().await;
        stats.timeouts += 1;
    }

    /// Execute multiple tool calls with appropriate concurrency
    pub async fn execute_tools(&self, calls: &[ToolCall]) -> Vec<ToolExecutionResult> {
        if calls.is_empty() {
            return Vec::new();
        }

        let (parallel_calls, sequential_calls) = self.partition_by_concurrency(calls);
        let mut results = Vec::with_capacity(calls.len());

        // Execute parallel-capable tools concurrently
        if !parallel_calls.is_empty() {
            let handles: Vec<_> = parallel_calls
                .into_iter()
                .map(|call| {
                    // call is already owned from into_iter(), no clone needed
                    let self_ref = self;
                    async move { self_ref.execute_tool(&call).await }
                })
                .collect();

            let parallel_results = futures::future::join_all(handles).await;
            results.extend(parallel_results);
        }

        // Execute sequential tools one by one
        for call in sequential_calls {
            let result = self.execute_tool(&call).await;
            results.push(result);
        }

        self.reorder_results(calls, results)
    }

    /// Get execution statistics
    pub async fn get_stats(&self) -> ExecutorStats {
        self.stats.read().await.clone()
    }

    /// Reset statistics
    pub async fn reset_stats(&self) {
        *self.stats.write().await = ExecutorStats::default();
    }

    /// Clear permission cache
    pub async fn clear_permission_cache(&self) {
        self.permission_cache.clear().await;
    }

    /// Get tool by name
    pub fn get_tool(&self, name: &str) -> Option<Arc<dyn Tool>> {
        self.tools.get(name).map(|t| t.clone())
    }

    /// Get all tool names
    pub fn tool_names(&self) -> Vec<String> {
        self.tools.iter().map(|e| e.key().clone()).collect()
    }
}

impl Default for ParallelToolExecutor {
    fn default() -> Self {
        Self::new()
    }
}
