//! 服务端连接管理器抽象
//!
//! 定义服务端连接管理的标准接口，支持用户自定义实现
//! 默认实现使用 ConnectionManager

use crate::common::MessageParser;
use crate::common::error::Result;
use crate::common::protocol::Frame;
use crate::transport::connection::Connection;
use async_trait::async_trait;
use std::sync::Arc;
use std::time::Duration;
use tokio::sync::Mutex;

/// 连接信息（Trait 版本，用于跨异步边界传递）
#[derive(Clone)]
pub struct ConnectionInfo {
    /// 连接 ID（唯一标识符）
    pub connection_id: String,
    /// 用户 ID（如果已认证）
    pub user_id: Option<String>,
    /// 创建时间戳（Unix 时间戳，秒）
    pub created_at: u64,
    /// 最后活跃时间戳（Unix 时间戳，秒）
    pub last_active: u64,
    /// 连接元数据
    pub metadata: std::collections::HashMap<String, String>,
    /// 设备信息（如果已提供）
    pub device_info: Option<crate::common::device::DeviceInfo>,
    /// 序列化格式（由客户端协商决定）
    pub serialization_format: crate::common::protocol::SerializationFormat,
    /// 压缩算法（由客户端协商决定）
    pub compression: crate::common::compression::CompressionAlgorithm,
    /// 加密方式（由客户端协商决定）
    pub encryption: crate::common::encryption::EncryptionAlgorithm,
    /// 是否已验证（如果启用认证，只有已验证的连接才能收发消息）
    pub authenticated: bool,
    /// 认证时间戳（Unix 时间戳，秒，如果已验证）
    pub authenticated_at: Option<u64>,
    /// 协商是否已完成（CONNECT 和 CONNECT_ACK 完成）
    pub negotiation_completed: bool,
    /// 协商是否已确认（客户端收到 CONNECT_ACK 后发送确认，服务端收到后标记）
    /// 确认后，消息必须严格按照协商好的方式处理，不再容错
    pub negotiation_confirmed: bool,
    /// 缓存的 MessageParser（协商完成后创建，避免每次消息处理都创建新的 parser）
    pub cached_parser: Option<std::sync::Arc<crate::common::MessageParser>>,
    /// 缓存的 MessagePipeline（协商完成后创建，如果配置了中间件或处理器）
    /// 可选，默认不创建以保持性能
    pub cached_pipeline: Option<std::sync::Arc<crate::common::message::pipeline::MessagePipeline>>,
}

/// 连接管理器抽象 trait
///
/// 实现此 trait 以提供自定义的连接管理逻辑
/// 例如：基于 Redis 的分布式连接管理、基于数据库的持久化等
#[async_trait]
pub trait ConnectionManagerTrait: Send + Sync + std::any::Any {
    /// 获取 Any 引用，用于类型向下转换
    fn as_any(&self) -> &dyn std::any::Any;
    /// 添加连接
    async fn add_connection(
        &self,
        connection_id: String,
        connection: Arc<Mutex<Box<dyn Connection>>>,
        user_id: Option<String>,
    ) -> Result<()>;

    /// 移除连接
    async fn remove_connection(&self, connection_id: &str) -> Result<()>;

    /// 获取连接
    async fn get_connection(
        &self,
        connection_id: &str,
    ) -> Option<(Arc<Mutex<Box<dyn Connection>>>, ConnectionInfo)>;

    /// 获取用户的所有连接 ID
    async fn get_user_connections(&self, user_id: &str) -> Vec<String>;

    /// 绑定用户到连接
    async fn bind_user(&self, connection_id: &str, user_id: String) -> Result<()>;

    /// 更新连接的最后活跃时间
    async fn update_connection_active(&self, connection_id: &str) -> Result<()>;

    /// 设置连接为已验证状态（认证通过后调用）
    async fn set_connection_authenticated(
        &self,
        connection_id: &str,
        user_id: Option<String>,
    ) -> Result<()>;

    /// 获取所有连接 ID
    async fn list_connections(&self) -> Vec<String>;

    /// 获取连接总数
    async fn connection_count(&self) -> usize;

    /// 连接总数的同步快照（用于同步 stats 路径，禁止阻塞/网络）。
    ///
    /// 实现必须以缓存/原子计数等廉价方式返回，分布式实现可返回最近一次快照。
    fn connection_count_snapshot(&self) -> usize;

    /// 绑定用户数的同步快照（约束同 [`connection_count_snapshot`]）。
    fn user_count_snapshot(&self) -> usize;

    /// 清理超时连接
    async fn cleanup_timeout_connections(&self, timeout: Duration) -> Vec<String>;

    // ========== 底层发送方法（字节数组）==========

    /// 向指定连接发送数据（字节数组）
    async fn send_to_connection(&self, connection_id: &str, data: &[u8]) -> Result<()>;

    /// 向指定用户的所有连接发送数据（字节数组）
    async fn send_to_user(&self, user_id: &str, data: &[u8]) -> Result<()>;

    /// 广播消息到所有连接（字节数组）
    async fn broadcast(&self, data: &[u8]) -> Result<()>;

    /// 广播消息到所有连接，排除指定连接（字节数组）
    async fn broadcast_except(&self, data: &[u8], exclude_connection_id: &str) -> Result<()>;

    // ========== Frame 级别发送方法（需要 MessageParser）==========

    /// 向指定连接发送 Frame（自动序列化）
    ///
    /// # 参数
    /// - `connection_id`: 连接 ID
    /// - `frame`: 要发送的 Frame
    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则从连接的协商信息创建）
    ///
    /// # 返回
    /// 发送成功返回 `Ok(())`，失败返回错误
    ///
    /// # 注意
    /// 如果 parser 为 None，将从连接的 ConnectionInfo 中获取协商后的序列化格式和压缩算法创建 parser
    async fn send_frame_to(
        &self,
        connection_id: &str,
        frame: &Frame,
        parser: Option<&MessageParser>,
    ) -> Result<()> {
        let connection_info = self.get_connection(connection_id).await;

        if let Some((_, info)) = &connection_info {
            if !info.authenticated {
                let is_system_command = frame.command.as_ref().and_then(|cmd| {
                    if let Some(crate::common::protocol::flare::core::commands::command::Type::System(sys_cmd)) = &cmd.r#type {
                        Some(sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::ConnectAck as i32
                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Ping as i32
                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Pong as i32
                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Error as i32
                            || sys_cmd.r#type == crate::common::protocol::flare::core::commands::system_command::Type::Close as i32)
                    } else {
                        None
                    }
                }).unwrap_or(false);

                if !is_system_command {
                    return Err(crate::common::error::FlareError::authentication_failed(
                        format!("连接 {} 未验证，无法发送消息", connection_id),
                    ));
                }
            }
        } else {
            return Err(crate::common::error::FlareError::connection_failed(
                format!("连接 {} 不存在", connection_id),
            ));
        }

        let data = if let Some(p) = parser {
            p.serialize(frame)?
        } else if let Some((_, info)) = connection_info {
            let connection_parser =
                MessageParser::new(info.serialization_format, info.compression, info.encryption);
            connection_parser.serialize(frame)?
        } else {
            use crate::common::message::parser::PRE_NEGOTIATION_PARSER;
            PRE_NEGOTIATION_PARSER.serialize(frame)?
        };

        self.send_to_connection(connection_id, &data).await?;
        self.update_connection_active(connection_id).await?;
        Ok(())
    }

    /// 向指定用户的所有连接发送 Frame（自动序列化）
    ///
    /// # 参数
    /// - `user_id`: 用户 ID
    /// - `frame`: 要发送的 Frame
    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
    ///
    /// # 返回
    /// 发送成功返回 `Ok(())`，失败返回错误
    async fn send_frame_to_user(
        &self,
        user_id: &str,
        frame: &Frame,
        parser: Option<&MessageParser>,
    ) -> Result<()> {
        let connection_ids = self.get_user_connections(user_id).await;
        for conn_id in connection_ids {
            // 为每个连接使用其协商的格式（如果 parser 为 None）
            let _ = self.send_frame_to(&conn_id, frame, parser).await;
        }
        Ok(())
    }

    /// 广播 Frame 到所有连接（自动序列化）
    ///
    /// # 参数
    /// - `frame`: 要广播的 Frame
    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
    ///
    /// # 返回
    /// 广播成功返回 `Ok(())`，失败返回错误
    async fn broadcast_frame(&self, frame: &Frame, parser: Option<&MessageParser>) -> Result<()> {
        let connection_ids = self.list_connections().await;
        for conn_id in connection_ids {
            // 为每个连接使用其协商的格式（如果 parser 为 None）
            let _ = self.send_frame_to(&conn_id, frame, parser).await;
        }
        Ok(())
    }

    /// 广播 Frame 到所有连接，排除指定连接（自动序列化）
    ///
    /// # 参数
    /// - `frame`: 要广播的 Frame
    /// - `exclude_connection_id`: 要排除的连接 ID
    /// - `parser`: 消息解析器，用于序列化 Frame（如果为 None，则为每个连接使用其协商的格式）
    ///
    /// # 返回
    /// 广播成功返回 `Ok(())`，失败返回错误
    async fn broadcast_frame_except(
        &self,
        frame: &Frame,
        exclude_connection_id: &str,
        parser: Option<&MessageParser>,
    ) -> Result<()> {
        let connection_ids = self.list_connections().await;
        for conn_id in connection_ids {
            if conn_id != exclude_connection_id {
                // 为每个连接使用其协商的格式（如果 parser 为 None）
                let _ = self.send_frame_to(&conn_id, frame, parser).await;
            }
        }
        Ok(())
    }
}

/// 连接统计信息
#[derive(Debug, Clone)]
pub struct ConnectionStats {
    /// 总连接数
    pub total_connections: usize,
    /// 总用户数
    pub total_users: usize,
}
