//! 队列管理系统
//! 提供高级的队列监控、任务优先级管理和批量操作功能

use anyhow::{Result, anyhow};
use std::collections::{HashMap, VecDeque, BTreeMap};
use std::sync::Arc;
use std::time::{Duration};
use tokio::sync::{RwLock, broadcast};
use tracing::{info, warn, error, debug};
use serde::{Serialize, Deserialize};
use chrono::{DateTime, Utc};

use crate::data::models::comfyui::{ExecutionModel, ExecutionStatus, ParameterValues, WorkflowModel, TemplateModel};
use crate::data::repositories::comfyui_repository::ComfyUIRepository;
use crate::business::services::comfyui_manager::ComfyUIManager;

/// 任务优先级
#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
pub enum TaskPriority {
    /// 低优先级
    Low = 1,
    /// 普通优先级
    Normal = 2,
    /// 高优先级
    High = 3,
    /// 紧急优先级
    Urgent = 4,
}

impl Default for TaskPriority {
    fn default() -> Self {
        TaskPriority::Normal
    }
}

impl std::fmt::Display for TaskPriority {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            TaskPriority::Low => write!(f, "low"),
            TaskPriority::Normal => write!(f, "normal"),
            TaskPriority::High => write!(f, "high"),
            TaskPriority::Urgent => write!(f, "urgent"),
        }
    }
}

/// 队列任务
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueTask {
    /// 任务 ID
    pub id: String,
    /// 执行 ID（如果已分配）
    pub execution_id: Option<String>,
    /// 工作流 ID（可选）
    pub workflow_id: Option<String>,
    /// 模板 ID（可选）
    pub template_id: Option<String>,
    /// 任务类型
    pub task_type: TaskType,
    /// 优先级
    pub priority: TaskPriority,
    /// 参数
    pub parameters: Option<ParameterValues>,
    /// 创建时间
    pub created_at: DateTime<Utc>,
    /// 预计执行时间（秒）
    pub estimated_duration: Option<u64>,
    /// 重试次数
    pub retry_count: u32,
    /// 最大重试次数
    pub max_retries: u32,
    /// 任务状态
    pub status: QueueTaskStatus,
    /// 标签
    pub tags: Vec<String>,
    /// 用户 ID（可选）
    pub user_id: Option<String>,
}

/// 任务类型
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum TaskType {
    /// 工作流执行
    WorkflowExecution,
    /// 模板执行
    TemplateExecution,
    /// 批量执行
    BatchExecution,
}

/// 队列任务状态
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
pub enum QueueTaskStatus {
    /// 等待中
    Pending,
    /// 已调度
    Scheduled,
    /// 执行中
    Running,
    /// 已完成
    Completed,
    /// 失败
    Failed,
    /// 已取消
    Cancelled,
    /// 暂停
    Paused,
}

/// 队列统计信息
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct QueueStatistics {
    /// 总任务数
    pub total_tasks: u64,
    /// 等待中的任务数
    pub pending_tasks: u32,
    /// 运行中的任务数
    pub running_tasks: u32,
    /// 已完成的任务数
    pub completed_tasks: u64,
    /// 失败的任务数
    pub failed_tasks: u64,
    /// 平均等待时间（秒）
    pub average_wait_time: f64,
    /// 平均执行时间（秒）
    pub average_execution_time: f64,
    /// 队列吞吐量（任务/分钟）
    pub throughput: f64,
    /// 按优先级分组的统计
    pub by_priority: HashMap<String, u32>,
    /// 按类型分组的统计
    pub by_type: HashMap<String, u32>,
}

/// 队列配置
#[derive(Debug, Clone)]
pub struct QueueConfig {
    /// 最大队列大小
    pub max_queue_size: usize,
    /// 最大并发执行数
    pub max_concurrent_executions: u32,
    /// 任务超时时间
    pub task_timeout: Duration,
    /// 队列检查间隔
    pub queue_check_interval: Duration,
    /// 是否启用优先级调度
    pub enable_priority_scheduling: bool,
    /// 是否启用自动重试
    pub enable_auto_retry: bool,
    /// 默认重试次数
    pub default_max_retries: u32,
}

impl Default for QueueConfig {
    fn default() -> Self {
        Self {
            max_queue_size: 1000,
            max_concurrent_executions: 4,
            task_timeout: Duration::from_secs(600), // 10分钟
            queue_check_interval: Duration::from_secs(1),
            enable_priority_scheduling: true,
            enable_auto_retry: true,
            default_max_retries: 3,
        }
    }
}

/// 队列管理器
pub struct QueueManager {
    /// 数据仓库
    repository: Arc<ComfyUIRepository>,
    /// 配置
    config: QueueConfig,
    /// 任务队列（按优先级排序）
    task_queue: Arc<RwLock<BTreeMap<TaskPriority, VecDeque<QueueTask>>>>,
    /// 运行中的任务
    running_tasks: Arc<RwLock<HashMap<String, QueueTask>>>,
    /// 任务历史
    task_history: Arc<RwLock<VecDeque<QueueTask>>>,
    /// 队列统计
    statistics: Arc<RwLock<QueueStatistics>>,
    /// 任务调度器句柄
    scheduler_handle: Arc<RwLock<Option<tokio::task::JoinHandle<()>>>>,
    /// 事件发送器
    event_sender: broadcast::Sender<QueueEvent>,
    /// 是否正在运行
    is_running: Arc<RwLock<bool>>,
}

/// 队列事件
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum QueueEvent {
    /// 任务已添加
    TaskAdded {
        task_id: String,
        priority: TaskPriority,
        timestamp: String,
    },
    /// 任务开始执行
    TaskStarted {
        task_id: String,
        execution_id: String,
        timestamp: String,
    },
    /// 任务完成
    TaskCompleted {
        task_id: String,
        execution_id: String,
        execution_time: u64,
        timestamp: String,
    },
    /// 任务失败
    TaskFailed {
        task_id: String,
        execution_id: Option<String>,
        error: String,
        timestamp: String,
    },
    /// 任务取消
    TaskCancelled {
        task_id: String,
        timestamp: String,
    },
    /// 队列状态更新
    QueueStatusUpdated {
        pending_count: u32,
        running_count: u32,
        timestamp: String,
    },
}

impl QueueManager {
    /// 创建新的队列管理器
    pub fn new(
        repository: Arc<ComfyUIRepository>,
        config: Option<QueueConfig>,
    ) -> Self {
        let config = config.unwrap_or_default();
        let (event_sender, _) = broadcast::channel(1000);

        Self {
            repository,
            config,
            task_queue: Arc::new(RwLock::new(BTreeMap::new())),
            running_tasks: Arc::new(RwLock::new(HashMap::new())),
            task_history: Arc::new(RwLock::new(VecDeque::new())),
            statistics: Arc::new(RwLock::new(QueueStatistics {
                total_tasks: 0,
                pending_tasks: 0,
                running_tasks: 0,
                completed_tasks: 0,
                failed_tasks: 0,
                average_wait_time: 0.0,
                average_execution_time: 0.0,
                throughput: 0.0,
                by_priority: HashMap::new(),
                by_type: HashMap::new(),
            })),
            scheduler_handle: Arc::new(RwLock::new(None)),
            event_sender,
            is_running: Arc::new(RwLock::new(false)),
        }
    }

    /// 启动队列管理器
    pub async fn start(&self) -> Result<()> {
        info!("启动队列管理器");

        // 检查是否已经在运行
        {
            let running = self.is_running.read().await;
            if *running {
                return Ok(());
            }
        }

        // 启动任务调度器
        let scheduler_task = self.spawn_scheduler_task().await?;
        
        {
            let mut handle = self.scheduler_handle.write().await;
            *handle = Some(scheduler_task);
        }

        {
            let mut running = self.is_running.write().await;
            *running = true;
        }

        info!("队列管理器已启动");
        Ok(())
    }

    /// 停止队列管理器
    pub async fn stop(&self) -> Result<()> {
        info!("停止队列管理器");

        let handle = {
            let mut handle_guard = self.scheduler_handle.write().await;
            handle_guard.take()
        };

        if let Some(handle) = handle {
            handle.abort();
        }

        {
            let mut running = self.is_running.write().await;
            *running = false;
        }

        info!("队列管理器已停止");
        Ok(())
    }

    /// 添加任务到队列
    pub async fn add_task(&self, mut task: QueueTask) -> Result<String> {
        // 检查队列大小
        let current_size = self.get_queue_size().await;
        if current_size >= self.config.max_queue_size {
            return Err(anyhow!("队列已满，无法添加新任务"));
        }

        // 设置默认值
        task.id = uuid::Uuid::new_v4().to_string();
        task.created_at = Utc::now();
        task.status = QueueTaskStatus::Pending;
        task.retry_count = 0;
        if task.max_retries == 0 {
            task.max_retries = self.config.default_max_retries;
        }

        let task_id = task.id.clone();
        let priority = task.priority;

        // 添加到队列
        {
            let mut queue = self.task_queue.write().await;
            queue.entry(priority).or_insert_with(VecDeque::new).push_back(task);
        }

        // 更新统计
        self.update_statistics().await;

        // 发送事件
        let _ = self.event_sender.send(QueueEvent::TaskAdded {
            task_id: task_id.clone(),
            priority,
            timestamp: Utc::now().to_rfc3339(),
        });

        info!("任务已添加到队列: {} (优先级: {})", task_id, priority);
        Ok(task_id)
    }

    /// 取消任务
    pub async fn cancel_task(&self, task_id: &str) -> Result<()> {
        // 先检查运行中的任务
        {
            let mut running = self.running_tasks.write().await;
            if let Some(mut task) = running.remove(task_id) {
                task.status = QueueTaskStatus::Cancelled;
                
                // 如果有执行 ID，尝试取消执行
                if let Some(execution_id) = &task.execution_id {
                    // TODO: 调用执行引擎取消执行
                    debug!("取消执行: {}", execution_id);
                }

                // 添加到历史
                {
                    let mut history = self.task_history.write().await;
                    history.push_back(task);
                    if history.len() > 1000 {
                        history.pop_front();
                    }
                }

                // 发送事件
                let _ = self.event_sender.send(QueueEvent::TaskCancelled {
                    task_id: task_id.to_string(),
                    timestamp: Utc::now().to_rfc3339(),
                });

                info!("运行中的任务已取消: {}", task_id);
                return Ok(());
            }
        }

        // 检查队列中的任务
        {
            let mut queue = self.task_queue.write().await;
            for (_, task_list) in queue.iter_mut() {
                if let Some(pos) = task_list.iter().position(|t| t.id == task_id) {
                    let mut task = task_list.remove(pos).unwrap();
                    task.status = QueueTaskStatus::Cancelled;

                    // 添加到历史
                    {
                        let mut history = self.task_history.write().await;
                        history.push_back(task);
                        if history.len() > 1000 {
                            history.pop_front();
                        }
                    }

                    // 发送事件
                    let _ = self.event_sender.send(QueueEvent::TaskCancelled {
                        task_id: task_id.to_string(),
                        timestamp: Utc::now().to_rfc3339(),
                    });

                    info!("队列中的任务已取消: {}", task_id);
                    return Ok(());
                }
            }
        }

        Err(anyhow!("任务不存在: {}", task_id))
    }

    /// 获取队列大小
    pub async fn get_queue_size(&self) -> usize {
        let queue = self.task_queue.read().await;
        queue.values().map(|v| v.len()).sum()
    }

    /// 获取队列统计
    pub async fn get_statistics(&self) -> QueueStatistics {
        self.statistics.read().await.clone()
    }

    /// 订阅队列事件
    pub fn subscribe_events(&self) -> broadcast::Receiver<QueueEvent> {
        self.event_sender.subscribe()
    }

    /// 检查是否正在运行
    pub async fn is_running(&self) -> bool {
        *self.is_running.read().await
    }

    /// 提交工作流执行任务
    pub async fn submit_workflow_execution(
        &self,
        execution_id: String,
        workflow: WorkflowModel,
        parameters: HashMap<String, serde_json::Value>,
        priority: Option<i32>,
    ) -> Result<String> {
        let task_priority = match priority {
            Some(p) if p >= 4 => TaskPriority::Urgent,
            Some(p) if p >= 3 => TaskPriority::High,
            Some(p) if p >= 2 => TaskPriority::Normal,
            _ => TaskPriority::Low,
        };

        let task = QueueTask {
            id: execution_id.clone(),
            task_type: TaskType::WorkflowExecution,
            priority: task_priority,
            workflow_id: Some(workflow.id.clone()),
            template_id: None,
            parameters: Some(parameters),
            execution_id: Some(execution_id.clone()),
            status: QueueTaskStatus::Pending,
            created_at: Utc::now(),
            estimated_duration: None,
            retry_count: 0,
            max_retries: self.config.default_max_retries,
            tags: Vec::new(),
            user_id: None,
        };

        self.add_task(task).await
    }

    /// 提交模板执行任务
    pub async fn submit_template_execution(
        &self,
        execution_id: String,
        template: TemplateModel,
        workflow: WorkflowModel,
        parameters: HashMap<String, serde_json::Value>,
        priority: Option<i32>,
    ) -> Result<String> {
        let task_priority = match priority {
            Some(p) if p >= 4 => TaskPriority::Urgent,
            Some(p) if p >= 3 => TaskPriority::High,
            Some(p) if p >= 2 => TaskPriority::Normal,
            _ => TaskPriority::Low,
        };

        let task = QueueTask {
            id: execution_id.clone(),
            task_type: TaskType::TemplateExecution,
            priority: task_priority,
            workflow_id: Some(workflow.id.clone()),
            template_id: Some(template.id.clone()),
            parameters: Some(parameters),
            execution_id: Some(execution_id.clone()),
            status: QueueTaskStatus::Pending,
            created_at: Utc::now(),
            estimated_duration: None,
            retry_count: 0,
            max_retries: self.config.default_max_retries,
            tags: Vec::new(),
            user_id: None,
        };

        self.add_task(task).await
    }

    /// 批量提交任务
    pub async fn submit_batch_tasks(&self, tasks: Vec<QueueTask>) -> Result<Vec<String>> {
        let mut task_ids = Vec::new();

        for task in tasks {
            match self.add_task(task).await {
                Ok(task_id) => task_ids.push(task_id),
                Err(e) => {
                    error!("批量提交任务失败: {}", e);
                    return Err(e);
                }
            }
        }

        info!("批量提交了 {} 个任务", task_ids.len());
        Ok(task_ids)
    }

    /// 批量取消任务
    pub async fn cancel_batch_tasks(&self, task_ids: Vec<String>) -> Result<Vec<String>> {
        let mut cancelled_ids = Vec::new();

        for task_id in task_ids {
            match self.cancel_task(&task_id).await {
                Ok(_) => cancelled_ids.push(task_id),
                Err(e) => {
                    warn!("取消任务失败 {}: {}", task_id, e);
                }
            }
        }

        info!("批量取消了 {} 个任务", cancelled_ids.len());
        Ok(cancelled_ids)
    }

    /// 暂停队列
    pub async fn pause_queue(&self) -> Result<String> {
        // 这里可以实现暂停逻辑
        info!("队列已暂停");
        Ok("队列已暂停".to_string())
    }

    /// 恢复队列
    pub async fn resume_queue(&self) -> Result<String> {
        // 这里可以实现恢复逻辑
        info!("队列已恢复");
        Ok("队列已恢复".to_string())
    }

    /// 清空队列
    pub async fn clear_queue(&self) -> Result<String> {
        let mut queue = self.task_queue.write().await;
        let mut total_cleared = 0;

        for (_, task_list) in queue.iter_mut() {
            total_cleared += task_list.len();
            task_list.clear();
        }

        self.update_statistics().await;

        info!("清空了 {} 个队列任务", total_cleared);
        Ok(format!("清空了 {} 个队列任务", total_cleared))
    }

    /// 生成调度器任务
    async fn spawn_scheduler_task(&self) -> Result<tokio::task::JoinHandle<()>> {
        let task_queue = Arc::clone(&self.task_queue);
        let running_tasks = Arc::clone(&self.running_tasks);
        let task_history = Arc::clone(&self.task_history);
        let statistics = Arc::clone(&self.statistics);
        // TODO: 需要添加 comfyui_manager 字段或使用其他方式获取 ComfyUI 管理器
        // let comfyui_manager = Arc::clone(&self.comfyui_manager);
        let repository = Arc::clone(&self.repository);
        let event_sender = self.event_sender.clone();
        let config = self.config.clone();
        let is_running = Arc::clone(&self.is_running);

        let handle = tokio::spawn(async move {
            info!("队列调度器已启动");

            while *is_running.read().await {
                // 检查并调度任务
                // TODO: 暂时跳过任务调度，需要重新设计架构
                // if let Err(e) = Self::schedule_tasks(...).await {
                //     error!("任务调度失败: {}", e);
                // }

                // 检查运行中的任务状态
                if let Err(e) = Self::check_running_tasks(
                    &running_tasks,
                    &task_history,
                    &statistics,
                    &repository,
                    &event_sender,
                    &config,
                ).await {
                    error!("检查运行任务失败: {}", e);
                }

                // 更新统计信息
                Self::update_statistics_internal(&task_queue, &running_tasks, &task_history, &statistics).await;

                // 等待下次检查
                tokio::time::sleep(config.queue_check_interval).await;
            }

            info!("队列调度器已停止");
        });

        Ok(handle)
    }

    /// 调度任务
    async fn schedule_tasks(
        task_queue: &Arc<RwLock<BTreeMap<TaskPriority, VecDeque<QueueTask>>>>,
        running_tasks: &Arc<RwLock<HashMap<String, QueueTask>>>,
        task_history: &Arc<RwLock<VecDeque<QueueTask>>>,
        statistics: &Arc<RwLock<QueueStatistics>>,
        comfyui_manager: &Arc<ComfyUIManager>,
        repository: &Arc<ComfyUIRepository>,
        event_sender: &broadcast::Sender<QueueEvent>,
        config: &QueueConfig,
    ) -> Result<()> {
        // 检查当前运行的任务数
        let current_running = running_tasks.read().await.len() as u32;
        if current_running >= config.max_concurrent_executions {
            return Ok(());
        }

        // 可以启动的任务数
        let available_slots = config.max_concurrent_executions - current_running;

        // 按优先级从高到低调度任务
        let mut tasks_to_schedule = Vec::new();
        {
            let mut queue = task_queue.write().await;

            // 从最高优先级开始
            for priority in [TaskPriority::Urgent, TaskPriority::High, TaskPriority::Normal, TaskPriority::Low] {
                if tasks_to_schedule.len() >= available_slots as usize {
                    break;
                }

                if let Some(task_list) = queue.get_mut(&priority) {
                    while let Some(task) = task_list.pop_front() {
                        tasks_to_schedule.push(task);
                        if tasks_to_schedule.len() >= available_slots as usize {
                            break;
                        }
                    }
                }
            }
        }

        // 启动调度的任务
        for mut task in tasks_to_schedule {
            if let Err(e) = Self::start_task(
                &mut task,
                running_tasks,
                comfyui_manager,
                repository,
                event_sender,
            ).await {
                error!("启动任务失败 {}: {}", task.id, e);

                // 任务启动失败，标记为失败并添加到历史
                task.status = QueueTaskStatus::Failed;
                {
                    let mut history = task_history.write().await;
                    history.push_back(task.clone());
                    if history.len() > 1000 {
                        history.pop_front();
                    }
                }

                // 发送失败事件
                let _ = event_sender.send(QueueEvent::TaskFailed {
                    task_id: task.id.clone(),
                    execution_id: task.execution_id.clone(),
                    error: e.to_string(),
                    timestamp: Utc::now().to_rfc3339(),
                });
            }
        }

        Ok(())
    }

    /// 启动单个任务
    async fn start_task(
        task: &mut QueueTask,
        running_tasks: &Arc<RwLock<HashMap<String, QueueTask>>>,
        comfyui_manager: &Arc<ComfyUIManager>,
        repository: &Arc<ComfyUIRepository>,
        event_sender: &broadcast::Sender<QueueEvent>,
    ) -> Result<()> {
        debug!("启动任务: {}", task.id);

        // 检查 ComfyUI 连接状态
        if !comfyui_manager.is_connected().await {
            return Err(anyhow!("ComfyUI 未连接"));
        }

        // 创建执行记录
        let execution = match &task.task_type {
            TaskType::WorkflowExecution => {
                if let Some(workflow_id) = &task.workflow_id {
                    // 获取工作流
                    let workflow = repository.get_workflow(workflow_id)?
                        .ok_or_else(|| anyhow!("工作流不存在: {}", workflow_id))?;

                    // TODO: 使用执行引擎执行工作流
                    // 这里需要集成执行引擎
                    ExecutionModel::for_workflow(
                        workflow_id.clone(),
                        uuid::Uuid::new_v4().to_string(), // 临时 prompt_id
                        task.parameters.clone(),
                    )
                } else {
                    return Err(anyhow!("工作流任务缺少 workflow_id"));
                }
            }
            TaskType::TemplateExecution => {
                if let Some(template_id) = &task.template_id {
                    // TODO: 使用执行引擎执行模板
                    ExecutionModel::for_template(
                        template_id.clone(),
                        uuid::Uuid::new_v4().to_string(), // 临时 prompt_id
                        task.parameters.clone().unwrap_or_default(),
                    )
                } else {
                    return Err(anyhow!("模板任务缺少 template_id"));
                }
            }
            TaskType::BatchExecution => {
                return Err(anyhow!("批量执行暂未实现"));
            }
        };

        // 保存执行记录
        repository.create_execution(&execution)?;

        // 更新任务状态
        task.execution_id = Some(execution.id.clone());
        task.status = QueueTaskStatus::Running;

        // 添加到运行中的任务
        {
            let mut running = running_tasks.write().await;
            running.insert(task.id.clone(), task.clone());
        }

        // 发送任务开始事件
        let _ = event_sender.send(QueueEvent::TaskStarted {
            task_id: task.id.clone(),
            execution_id: execution.id.clone(),
            timestamp: Utc::now().to_rfc3339(),
        });

        info!("任务已启动: {} -> {}", task.id, execution.id);
        Ok(())
    }

    /// 检查运行中的任务状态
    async fn check_running_tasks(
        running_tasks: &Arc<RwLock<HashMap<String, QueueTask>>>,
        task_history: &Arc<RwLock<VecDeque<QueueTask>>>,
        statistics: &Arc<RwLock<QueueStatistics>>,
        repository: &Arc<ComfyUIRepository>,
        event_sender: &broadcast::Sender<QueueEvent>,
        config: &QueueConfig,
    ) -> Result<()> {
        let mut completed_tasks = Vec::new();
        let mut failed_tasks = Vec::new();
        let mut timed_out_tasks = Vec::new();

        // 检查每个运行中的任务
        {
            let running = running_tasks.read().await;
            for (task_id, task) in running.iter() {
                if let Some(execution_id) = &task.execution_id {
                    // 检查执行状态
                    match repository.get_execution(execution_id) {
                        Ok(Some(execution)) => {
                            match execution.status {
                                ExecutionStatus::Completed => {
                                    completed_tasks.push((task_id.clone(), task.clone(), execution));
                                }
                                ExecutionStatus::Failed => {
                                    failed_tasks.push((task_id.clone(), task.clone(), execution));
                                }
                                ExecutionStatus::Cancelled => {
                                    failed_tasks.push((task_id.clone(), task.clone(), execution));
                                }
                                _ => {
                                    // 检查是否超时
                                    let elapsed = Utc::now().signed_duration_since(task.created_at);
                                    if elapsed.to_std().unwrap_or_default() > config.task_timeout {
                                        timed_out_tasks.push((task_id.clone(), task.clone()));
                                    }
                                }
                            }
                        }
                        Ok(None) => {
                            warn!("执行记录不存在: {}", execution_id);
                            failed_tasks.push((task_id.clone(), task.clone(), ExecutionModel::new(execution_id.clone())));
                        }
                        Err(e) => {
                            error!("检查执行状态失败 {}: {}", execution_id, e);
                        }
                    }
                }
            }
        }

        // 处理完成的任务
        for (task_id, mut task, execution) in completed_tasks {
            task.status = QueueTaskStatus::Completed;

            // 从运行中移除
            {
                let mut running = running_tasks.write().await;
                running.remove(&task_id);
            }

            // 添加到历史
            {
                let mut history = task_history.write().await;
                history.push_back(task.clone());
                if history.len() > 1000 {
                    history.pop_front();
                }
            }

            // 发送完成事件
            let execution_time = execution.execution_time.unwrap_or(0);
            let _ = event_sender.send(QueueEvent::TaskCompleted {
                task_id: task_id.clone(),
                execution_id: execution.id,
                execution_time,
                timestamp: Utc::now().to_rfc3339(),
            });

            info!("任务已完成: {}", task_id);
        }

        // 处理失败的任务
        for (task_id, mut task, execution) in failed_tasks {
            task.retry_count += 1;

            // 检查是否需要重试
            if config.enable_auto_retry && task.retry_count <= task.max_retries {
                // 重新加入队列
                task.status = QueueTaskStatus::Pending;
                task.execution_id = None;

                // TODO: 重新加入队列的逻辑
                warn!("任务将重试: {} (第 {} 次)", task_id, task.retry_count);
            } else {
                // 标记为失败
                task.status = QueueTaskStatus::Failed;

                // 从运行中移除
                {
                    let mut running = running_tasks.write().await;
                    running.remove(&task_id);
                }

                // 添加到历史
                {
                    let mut history = task_history.write().await;
                    history.push_back(task.clone());
                    if history.len() > 1000 {
                        history.pop_front();
                    }
                }

                // 发送失败事件
                let error_msg = execution.error_message.unwrap_or_else(|| "执行失败".to_string());
                let _ = event_sender.send(QueueEvent::TaskFailed {
                    task_id: task_id.clone(),
                    execution_id: Some(execution.id),
                    error: error_msg,
                    timestamp: Utc::now().to_rfc3339(),
                });

                error!("任务失败: {}", task_id);
            }
        }

        // 处理超时的任务
        for (task_id, mut task) in timed_out_tasks {
            warn!("任务超时: {}", task_id);

            // TODO: 取消执行
            if let Some(execution_id) = &task.execution_id {
                debug!("取消超时任务的执行: {}", execution_id);
            }

            task.status = QueueTaskStatus::Failed;

            // 从运行中移除
            {
                let mut running = running_tasks.write().await;
                running.remove(&task_id);
            }

            // 添加到历史
            {
                let mut history = task_history.write().await;
                history.push_back(task.clone());
                if history.len() > 1000 {
                    history.pop_front();
                }
            }

            // 发送失败事件
            let _ = event_sender.send(QueueEvent::TaskFailed {
                task_id: task_id.clone(),
                execution_id: task.execution_id.clone(),
                error: "任务超时".to_string(),
                timestamp: Utc::now().to_rfc3339(),
            });
        }

        Ok(())
    }

    /// 更新统计信息
    async fn update_statistics(&self) {
        Self::update_statistics_internal(
            &self.task_queue,
            &self.running_tasks,
            &self.task_history,
            &self.statistics,
        ).await;
    }

    /// 内部统计更新方法
    async fn update_statistics_internal(
        task_queue: &Arc<RwLock<BTreeMap<TaskPriority, VecDeque<QueueTask>>>>,
        running_tasks: &Arc<RwLock<HashMap<String, QueueTask>>>,
        task_history: &Arc<RwLock<VecDeque<QueueTask>>>,
        statistics: &Arc<RwLock<QueueStatistics>>,
    ) {
        let queue = task_queue.read().await;
        let running = running_tasks.read().await;
        let history = task_history.read().await;

        let pending_count = queue.values().map(|v| v.len()).sum::<usize>() as u32;
        let running_count = running.len() as u32;

        let completed_count = history.iter()
            .filter(|t| t.status == QueueTaskStatus::Completed)
            .count() as u64;

        let failed_count = history.iter()
            .filter(|t| t.status == QueueTaskStatus::Failed)
            .count() as u64;

        // 计算平均等待时间
        let total_wait_time: i64 = history.iter()
            .filter(|t| t.status == QueueTaskStatus::Completed)
            .map(|t| {
                // 这里需要更精确的等待时间计算
                // 暂时使用创建时间到现在的时间差
                Utc::now().signed_duration_since(t.created_at).num_seconds()
            })
            .sum();

        let average_wait_time = if completed_count > 0 {
            total_wait_time as f64 / completed_count as f64
        } else {
            0.0
        };

        // 按优先级统计
        let mut by_priority = HashMap::new();
        for (priority, tasks) in queue.iter() {
            by_priority.insert(priority.to_string(), tasks.len() as u32);
        }

        // 按类型统计
        let mut by_type = HashMap::new();
        for tasks in queue.values() {
            for task in tasks {
                let type_name = match task.task_type {
                    TaskType::WorkflowExecution => "workflow",
                    TaskType::TemplateExecution => "template",
                    TaskType::BatchExecution => "batch",
                };
                *by_type.entry(type_name.to_string()).or_insert(0) += 1;
            }
        }

        // 更新统计
        {
            let mut stats = statistics.write().await;
            stats.pending_tasks = pending_count;
            stats.running_tasks = running_count;
            stats.completed_tasks = completed_count;
            stats.failed_tasks = failed_count;
            stats.total_tasks = completed_count + failed_count + pending_count as u64 + running_count as u64;
            stats.average_wait_time = average_wait_time;
            stats.by_priority = by_priority;
            stats.by_type = by_type;
            // TODO: 计算吞吐量和平均执行时间
        }
    }
}
