flare-core 1.0.3

//! 混合客户端接口
//!
//! 支持单个协议或多协议竞速
//! 统一管理连接状态、心跳、消息路由等功能

use crate::client::config::ClientConfig;
use crate::client::transports::{Client, ClientCore};
use crate::common::config_types::TransportProtocol;
use crate::common::error::{FlareError, Result};
use crate::common::platform::{MonotonicInstant, monotonic_now, timeout as platform_timeout};
use crate::common::protocol::Frame;
use crate::transport::events::ArcObserver;
use async_trait::async_trait;
use futures_util::future::select_all;
use std::sync::{Arc, Mutex as StdMutex};
use std::time::Duration;
use tokio::sync::Mutex;

#[cfg(feature = "quic")]
use crate::client::transports::quic::QUICClient;
#[cfg(feature = "tcp")]
use crate::client::transports::tcp::TCPClient;
#[cfg(feature = "websocket")]
use crate::client::transports::websocket::WebSocketClient;

/// 混合客户端
///
/// 支持单个协议连接或多协议竞速
/// 统一管理连接状态、心跳、消息路由等功能
pub struct HybridClient {
    /// 内部客户端（根据配置动态选择）
    inner: Arc<Mutex<Box<dyn Client>>>,
    /// 使用的协议
    active_protocol: TransportProtocol,
    /// 客户端核心功能（统一管理连接状态、心跳、消息路由）
    core: ClientCore,
}

/// 连接结果类型别名
type ConnectionResult = Result<Box<dyn Client>>;

/// 连接任务结果
type ConnectionTaskResult = (TransportProtocol, usize, ConnectionResult, Duration);
type RaceInflight = Arc<StdMutex<Vec<tokio::task::JoinHandle<ConnectionTaskResult>>>>;

/// 成功的连接信息
type SuccessfulConnection = (usize, TransportProtocol, Box<dyn Client>, Duration);

/// 失败的连接信息
type FailedConnection = (usize, TransportProtocol, FlareError, Duration);

impl HybridClient {
    /// 创建新的混合客户端
    ///
    /// # 参数
    /// - `config`: 客户端配置
    ///
    /// # 返回
    /// 混合客户端实例
    pub fn new(config: ClientConfig) -> Result<Self> {
        let core = ClientCore::new(&config);
        let protocols = config.get_protocols();

        if protocols.len() == 1 {
            Self::create_single_protocol(config, protocols[0], core)
        } else {
            Self::create_race_mode_placeholder(config, protocols[0], core)
        }
    }

    /// 创建单个协议客户端
    fn create_single_protocol(
        config: ClientConfig,
        protocol: TransportProtocol,
        core: ClientCore,
    ) -> Result<Self> {
        let mut single_config = config;
        single_config.transport = protocol;
        single_config.transports = None;
        // 与竞速模式一致：优先使用 with_protocol_url 注册的地址，避免裸 host:port 缺少 scheme。
        single_config.server_url = single_config.get_protocol_url(&protocol);

        let client = Self::create_protocol_client(single_config, core.clone())?;

        Ok(Self {
            inner: Arc::new(Mutex::new(client)),
            active_protocol: protocol,
            core,
        })
    }

    /// 创建竞速模式的占位符客户端
    ///
    /// 返回一个占位符，实际连接在 connect_with_race 时完成
    fn create_race_mode_placeholder(
        config: ClientConfig,
        default_protocol: TransportProtocol,
        core: ClientCore,
    ) -> Result<Self> {
        let default_config = ClientConfig {
            server_url: config.get_protocol_url(&default_protocol),
            transport: default_protocol,
            ..config
        };

        let client = Self::create_protocol_client(default_config, core.clone())?;

        Ok(Self {
            inner: Arc::new(Mutex::new(client)),
            active_protocol: default_protocol,
            core,
        })
    }

    /// 创建协议客户端（统一创建逻辑）
    fn create_protocol_client(config: ClientConfig, core: ClientCore) -> Result<Box<dyn Client>> {
        match config.transport {
            TransportProtocol::WebSocket => {
                #[cfg(feature = "websocket")]
                {
                    Ok(Box::new(WebSocketClient::with_core(config, core)))
                }
                #[cfg(not(feature = "websocket"))]
                {
                    let _ = (config, core);
                    Err(Self::transport_feature_disabled(
                        TransportProtocol::WebSocket,
                    ))
                }
            }
            TransportProtocol::QUIC => {
                #[cfg(feature = "quic")]
                {
                    QUICClient::with_core(config, core).map(|c| Box::new(c) as Box<dyn Client>)
                }
                #[cfg(not(feature = "quic"))]
                {
                    let _ = (config, core);
                    Err(Self::transport_feature_disabled(TransportProtocol::QUIC))
                }
            }
            TransportProtocol::TCP => {
                #[cfg(feature = "tcp")]
                {
                    Ok(Box::new(TCPClient::with_core(config, core)))
                }
                #[cfg(not(feature = "tcp"))]
                {
                    let _ = (config, core);
                    Err(Self::transport_feature_disabled(TransportProtocol::TCP))
                }
            }
        }
    }

    #[allow(dead_code)]
    fn transport_feature_disabled(protocol: TransportProtocol) -> FlareError {
        FlareError::operation_not_supported(format!("{protocol:?} transport feature is disabled"))
    }

    /// 协议竞速连接
    ///
    /// 并行连接多个协议，选择最快成功的连接
    /// 如果多个连接几乎同时成功（时间差 < 100ms），则使用优先级最高的
    /// 协议列表的顺序就是优先级顺序（index 越小优先级越高）
    async fn race_connect(
        config: ClientConfig,
        shared_core: ClientCore,
        inflight: RaceInflight,
    ) -> Result<(Box<dyn Client>, TransportProtocol)> {
        let protocols = config.get_protocols();
        let race_start = monotonic_now();

        Self::spawn_connection_tasks(config, &protocols, &shared_core, Arc::clone(&inflight));

        let (first_success, successful_clients, errors) =
            Self::wait_for_connections(inflight).await;

        // 处理连接结果（选择最快协议，然后发送 CONNECT）
        Self::process_race_results(
            first_success,
            successful_clients,
            errors,
            shared_core,
            race_start,
        )
        .await
    }

    /// 为所有协议创建连接任务
    fn spawn_connection_tasks(
        config: ClientConfig,
        protocols: &[TransportProtocol],
        shared_core: &ClientCore,
        inflight: RaceInflight,
    ) {
        for (index, protocol) in protocols.iter().enumerate() {
            let protocol_url = config.get_protocol_url(protocol);
            tracing::debug!(
                "协议竞速: [{}] {:?} 使用地址: {}",
                index,
                protocol,
                protocol_url
            );

            let protocol_config = ClientConfig {
                server_url: protocol_url.clone(),
                transport: *protocol,
                transports: None,
                ..config.clone()
            };

            // 为每个协议创建独立的 core 副本，但共享竞速关键状态
            let mut protocol_core = ClientCore::new(&protocol_config);
            protocol_core.share_race_state_from(shared_core);

            let protocol_clone = *protocol;
            let protocol_index = index;

            let handle = tokio::spawn(async move {
                let start_time = monotonic_now();
                tracing::debug!(
                    "开始建立网络连接: {:?} (优先级: {}, 地址: {})",
                    protocol_clone,
                    protocol_index,
                    protocol_url
                );

                // 仅建立网络连接，不发送 CONNECT（用于协议竞速）
                let network_result = Self::establish_protocol_network(
                    protocol_clone,
                    protocol_config,
                    protocol_core,
                    protocol_index,
                )
                .await;

                let elapsed = start_time.elapsed();

                // 将网络连接结果转换为 ConnectionResult 格式
                let client_result = network_result.map(|(client, _)| client);

                (protocol_clone, protocol_index, client_result, elapsed)
            });

            if let Ok(mut guard) = inflight.lock() {
                guard.push(handle);
            }
        }
    }

    /// 仅建立网络连接（不发送 CONNECT，用于协议竞速）
    async fn establish_protocol_network(
        protocol: TransportProtocol,
        config: ClientConfig,
        core: ClientCore,
        priority: usize,
    ) -> Result<(Box<dyn Client>, Duration)> {
        match protocol {
            TransportProtocol::WebSocket => {
                #[cfg(feature = "websocket")]
                {
                    let (client, elapsed) =
                        Self::establish_websocket_network(config, core, priority).await?;
                    Ok((Box::new(client), elapsed))
                }
                #[cfg(not(feature = "websocket"))]
                {
                    let _ = (config, core, priority);
                    Err(Self::transport_feature_disabled(
                        TransportProtocol::WebSocket,
                    ))
                }
            }
            TransportProtocol::QUIC => {
                #[cfg(feature = "quic")]
                {
                    let (client, elapsed) =
                        Self::establish_quic_network(config, core, priority).await?;
                    Ok((Box::new(client), elapsed))
                }
                #[cfg(not(feature = "quic"))]
                {
                    let _ = (config, core, priority);
                    Err(Self::transport_feature_disabled(TransportProtocol::QUIC))
                }
            }
            TransportProtocol::TCP => {
                #[cfg(feature = "tcp")]
                {
                    let (client, elapsed) =
                        Self::establish_tcp_network(config, core, priority).await?;
                    Ok((Box::new(client), elapsed))
                }
                #[cfg(not(feature = "tcp"))]
                {
                    let _ = (config, core, priority);
                    Err(Self::transport_feature_disabled(TransportProtocol::TCP))
                }
            }
        }
    }

    /// 仅建立 WebSocket 网络连接（不发送 CONNECT 消息）
    ///
    /// 用于协议竞速：先建立网络连接，选择最快协议，然后再发送 CONNECT
    #[cfg(feature = "websocket")]
    async fn establish_websocket_network(
        config: ClientConfig,
        core: ClientCore,
        priority: usize,
    ) -> Result<(WebSocketClient, Duration)> {
        let start_time = monotonic_now();
        let mut client = WebSocketClient::with_core(config, core);

        // 仅建立网络连接，不发送 CONNECT
        // establish_network_connection 会保存连接到 client.connection
        let _connection_arc = client.establish_network_connection().await?;
        let elapsed = start_time.elapsed();

        tracing::debug!(
            "WebSocket 网络连接建立成功 (优先级: {}, 耗时: {:?})",
            priority,
            elapsed
        );
        Ok((client, elapsed))
    }

    /// 仅建立 QUIC 网络连接（不发送 CONNECT 消息）
    ///
    /// 用于协议竞速：先建立网络连接，选择最快协议，然后再发送 CONNECT
    ///
    /// 优化：将 endpoint 创建移到计时开始之前，只测量网络连接建立时间
    /// 这样可以更公平地比较 QUIC 和 WebSocket 的网络连接速度
    #[cfg(feature = "quic")]
    async fn establish_quic_network(
        config: ClientConfig,
        core: ClientCore,
        priority: usize,
    ) -> Result<(QUICClient, Duration)> {
        // 在计时开始之前创建 endpoint（排除 endpoint 创建时间）
        // 这样可以更公平地比较网络连接时间，而不是包含 endpoint 创建时间
        let (endpoint, client_config) = match QUICClient::create_quic_endpoint_with_tls(&config.tls)
        {
            Ok(ep) => ep,
            Err(e) => {
                tracing::warn!(
                    "QUIC endpoint 创建失败 (优先级: {}, 地址: {}): {}",
                    priority,
                    config.server_url,
                    e
                );
                return Err(e);
            }
        };

        // 现在开始计时（只测量网络连接建立时间）
        let start_time = monotonic_now();

        let mut client = match QUICClient::with_core_and_endpoint(
            config.clone(),
            core,
            Some((endpoint, client_config)),
        ) {
            Ok(client) => client,
            Err(e) => {
                let elapsed = start_time.elapsed();
                tracing::warn!(
                    "QUIC 客户端创建失败 (优先级: {}, 耗时: {:?}, 地址: {}): {}",
                    priority,
                    elapsed,
                    config.server_url,
                    e
                );
                return Err(e);
            }
        };

        // 仅建立网络连接，不发送 CONNECT
        let _connection_arc = client.establish_network_connection().await?;
        let elapsed = start_time.elapsed();

        tracing::debug!(
            "QUIC 网络连接建立成功 (优先级: {}, 耗时: {:?})",
            priority,
            elapsed
        );
        Ok((client, elapsed))
    }

    /// 仅建立 TCP 网络连接（不发送 CONNECT 消息）
    #[cfg(feature = "tcp")]
    async fn establish_tcp_network(
        config: ClientConfig,
        core: ClientCore,
        priority: usize,
    ) -> Result<(TCPClient, Duration)> {
        let start_time = monotonic_now();
        let mut client = TCPClient::with_core(config, core);
        let _connection_arc = client.establish_network_connection().await?;
        let elapsed = start_time.elapsed();
        tracing::debug!(
            "TCP 网络连接建立成功 (优先级: {}, 耗时: {:?})",
            priority,
            elapsed
        );
        Ok((client, elapsed))
    }

    /// 竞速超时或失败时关闭仍在进行/已建立但未选中的连接。
    async fn cleanup_inflight(inflight: &RaceInflight) {
        let handles = inflight
            .lock()
            .ok()
            .map(|mut guard| std::mem::take(&mut *guard))
            .unwrap_or_default();
        if !handles.is_empty() {
            Self::cleanup_remaining_race_handles(handles).await;
        }
    }

    /// 等待所有连接任务完成，收集结果
    async fn wait_for_connections(
        inflight: RaceInflight,
    ) -> (
        Option<SuccessfulConnection>,
        Vec<SuccessfulConnection>,
        Vec<FailedConnection>,
    ) {
        const TIME_THRESHOLD: Duration = Duration::from_millis(100);

        let mut first_success: Option<SuccessfulConnection> = None;
        let mut successful_clients = Vec::new();
        let mut errors = Vec::new();

        loop {
            let batch = inflight
                .lock()
                .ok()
                .map(|mut guard| std::mem::take(&mut *guard))
                .unwrap_or_default();
            if batch.is_empty() {
                break;
            }

            let total_protocols = batch.len();
            tracing::debug!("协议竞速 select 批次: {} 个任务", total_protocols);

            let (result, _index, remaining) = select_all(batch).await;
            if let Ok(mut guard) = inflight.lock() {
                guard.extend(remaining);
            }

            match result {
                Ok((protocol, protocol_index, client_result, elapsed)) => {
                    match client_result {
                        Ok(client) => {
                            if first_success.is_none() {
                                // 第一个成功的连接，立即返回
                                first_success = Some((protocol_index, protocol, client, elapsed));
                                let ms = elapsed.as_secs_f64() * 1000.0;
                                tracing::info!(
                                    "🏆 第一个成功的连接: {:?} (优先级: {}, 耗时: {:.3}ms)",
                                    protocol,
                                    protocol_index,
                                    ms
                                );
                                tracing::debug!("第一个连接成功，立即返回，不再等待其他连接");
                                break;
                            } else if let Some((_, _, _, first_elapsed)) = &first_success {
                                // 检查是否在时间阈值内（几乎同时成功）
                                if elapsed <= *first_elapsed + TIME_THRESHOLD {
                                    successful_clients.push((
                                        protocol_index,
                                        protocol,
                                        client,
                                        elapsed,
                                    ));
                                    let ms = elapsed.as_secs_f64() * 1000.0;
                                    tracing::debug!(
                                        "⚡ 几乎同时成功的连接: {:?} (优先级: {}, 耗时: {:.3}ms)",
                                        protocol,
                                        protocol_index,
                                        ms
                                    );
                                } else {
                                    // 连接太慢，关闭它
                                    let ms = elapsed.as_secs_f64() * 1000.0;
                                    tracing::debug!(
                                        "🐌 连接太慢，关闭: {:?} (优先级: {}, 耗时: {:.3}ms)",
                                        protocol,
                                        protocol_index,
                                        ms
                                    );
                                    Self::disconnect_client_async(
                                        protocol,
                                        protocol_index,
                                        elapsed,
                                    );
                                }
                            }
                        }
                        Err(e) => {
                            let error_msg = e.to_string();
                            errors.push((protocol_index, protocol, e, elapsed));
                            let ms = elapsed.as_secs_f64() * 1000.0;
                            tracing::warn!(
                                "连接失败: {:?} (优先级: {}, 耗时: {:.3}ms): {}",
                                protocol,
                                protocol_index,
                                ms,
                                error_msg
                            );
                        }
                    }
                }
                Err(join_err) => {
                    tracing::error!("Task join error: {:?}", join_err);
                }
            }
        }

        Self::cleanup_inflight(&inflight).await;

        (first_success, successful_clients, errors)
    }

    /// 竞速提前结束或全部完成后，关闭仍在进行/已建立但未选中的连接，避免泄漏。
    async fn cleanup_remaining_race_handles(
        handles: Vec<tokio::task::JoinHandle<ConnectionTaskResult>>,
    ) {
        for handle in handles {
            match handle.await {
                Ok((_protocol, _index, Ok(mut client), _elapsed)) => {
                    client.set_disconnect_requested(true);
                    if let Err(e) = client.disconnect().await {
                        tracing::debug!("[HybridClient] race cleanup disconnect: {}", e);
                    }
                }
                Ok((_protocol, _index, Err(_), _elapsed)) => {}
                Err(join_err) => {
                    tracing::debug!("[HybridClient] race task join error: {:?}", join_err);
                }
            }
        }
    }

    /// 异步关闭客户端（不阻塞）
    fn disconnect_client_async(protocol: TransportProtocol, priority: usize, elapsed: Duration) {
        tracing::debug!(
            "🐌 连接太慢，关闭: {:?} (优先级: {}, 耗时: {:?})",
            protocol,
            priority,
            elapsed
        );
        // 注意：这里无法获取 client，因为它在 Result 中
        // 实际关闭会在 process_race_results 中处理
    }

    /// 处理竞速结果，选择最佳连接，然后发送 CONNECT 消息
    async fn process_race_results(
        first_success: Option<SuccessfulConnection>,
        mut successful_clients: Vec<SuccessfulConnection>,
        errors: Vec<FailedConnection>,
        shared_core: ClientCore,
        race_start: MonotonicInstant,
    ) -> Result<(Box<dyn Client>, TransportProtocol)> {
        // 打印所有协议的耗时信息
        Self::log_protocol_timings(&first_success, &successful_clients, &errors);

        if let Some((first_index, first_protocol, first_client, first_elapsed)) = first_success {
            // 合并所有成功的连接，选择优先级最高的
            successful_clients.push((first_index, first_protocol, first_client, first_elapsed));
            successful_clients.sort_by_key(|(index, _protocol, _client, _elapsed)| *index);

            // 选择优先级最高的（index 最小的）
            let (selected_index, selected_protocol, mut selected_client, selected_elapsed) =
                successful_clients.remove(0);

            let selected_ms = selected_elapsed.as_secs_f64() * 1000.0;
            let total_ms = race_start.elapsed().as_secs_f64() * 1000.0;
            tracing::info!(
                "✅ 协议竞速成功: {:?} (优先级: {}, 网络连接耗时: {:.3}ms, 总竞速时间: {:.3}ms)",
                selected_protocol,
                selected_index,
                selected_ms,
                total_ms
            );

            // 先由客户端主动断开所有未选中连接（await 完成），再在选中连接上发 CONNECT，
            // 避免未选中连接仍在线时服务端对其发 KICK 导致误报「被踢」和前端断连
            Self::disconnect_unselected_clients(successful_clients).await;

            // 现在发送 CONNECT 消息（网络连接已建立）
            tracing::debug!(
                "📤 发送 CONNECT 消息: {:?} (优先级: {})",
                selected_protocol,
                selected_index
            );
            selected_client.connect().await?;

            // 同步连接状态到共享的 core
            shared_core.state_manager.set_connected();

            Ok((selected_client, selected_protocol))
        } else {
            // 所有协议都失败了
            Self::build_all_failed_error(errors)
        }
    }

    /// 打印所有协议的耗时信息
    fn log_protocol_timings(
        first_success: &Option<SuccessfulConnection>,
        successful_clients: &[SuccessfulConnection],
        errors: &[FailedConnection],
    ) {
        tracing::info!("📊 协议竞速耗时统计：");

        // 收集所有协议的结果（成功 + 失败）
        let mut all_results: Vec<(usize, TransportProtocol, Option<Duration>, Option<String>)> =
            Vec::new();

        // 添加成功的连接
        if let Some((index, protocol, _, elapsed)) = first_success {
            all_results.push((*index, *protocol, Some(*elapsed), None));
        }

        for (index, protocol, _, elapsed) in successful_clients {
            all_results.push((*index, *protocol, Some(*elapsed), None));
        }

        // 添加失败的连接
        for (index, protocol, error, elapsed) in errors {
            all_results.push((*index, *protocol, Some(*elapsed), Some(error.to_string())));
        }

        // 按优先级排序
        all_results.sort_by_key(|(index, _, _, _)| *index);

        // 打印每个协议的耗时
        for (index, protocol, elapsed_opt, error_opt) in all_results {
            let protocol_name = match protocol {
                TransportProtocol::WebSocket => "WebSocket",
                TransportProtocol::QUIC => "QUIC",
                TransportProtocol::TCP => "TCP",
            };

            match (elapsed_opt, error_opt) {
                (Some(elapsed), None) => {
                    let ms = elapsed.as_secs_f64() * 1000.0;
                    // 检查是否是第一个成功的（会被选中）
                    let is_first = first_success
                        .as_ref()
                        .map(|(idx, _, _, _)| *idx == index)
                        .unwrap_or(false);
                    if is_first {
                        tracing::info!(
                            "  [{:2}] {:12} ✅ 成功 - {:.3}ms ⭐ (已选中)",
                            index,
                            protocol_name,
                            ms
                        );
                    } else {
                        tracing::info!(
                            "  [{:2}] {:12} ✅ 成功 - {:.3}ms",
                            index,
                            protocol_name,
                            ms
                        );
                    }
                }
                (Some(elapsed), Some(err)) => {
                    let ms = elapsed.as_secs_f64() * 1000.0;
                    // 截断错误信息，避免日志过长
                    let err_short = if err.len() > 50 {
                        format!("{}...", &err[..50])
                    } else {
                        err
                    };
                    tracing::info!(
                        "  [{:2}] {:12} ❌ 失败 - {:.3}ms - {}",
                        index,
                        protocol_name,
                        ms,
                        err_short
                    );
                }
                _ => {}
            }
        }
    }

    /// 关闭未选中的连接（客户端主动断开，并等待全部关闭完成）
    ///
    /// 先对每个未选中 client 置位 set_disconnect_requested(true)，再并行 await 各 disconnect，
    /// 确保在选中连接发送 CONNECT 之前未选中连接已关闭，避免服务端对未选中连接发 KICK 导致误报「被踢」。
    async fn disconnect_unselected_clients(clients: Vec<SuccessfulConnection>) {
        if clients.is_empty() {
            return;
        }

        tracing::info!(
            "正在主动关闭 {} 个未选中的连接（客户端先断）...",
            clients.len()
        );

        let mut handles = Vec::with_capacity(clients.len());
        for (index, protocol, mut client, elapsed) in clients {
            tracing::debug!(
                "关闭未选中的连接: {:?} (优先级: {}, 耗时: {:?})",
                protocol,
                index,
                elapsed
            );
            client.set_disconnect_requested(true);
            let handle = tokio::spawn(async move {
                let res = client.disconnect().await;
                (protocol, res)
            });
            handles.push(handle);
        }

        for handle in handles {
            if let Ok((protocol, res)) = handle.await {
                match res {
                    Ok(()) => tracing::debug!("✅ {:?} 未选中连接已关闭", protocol),
                    Err(e) => tracing::warn!("关闭 {:?} 连接时出错: {}", protocol, e),
                }
            }
        }

        tracing::info!("所有未选中的连接已主动关闭");
    }

    /// 构建所有协议都失败的错误信息
    fn build_all_failed_error(
        errors: Vec<FailedConnection>,
    ) -> Result<(Box<dyn Client>, TransportProtocol)> {
        let mut sorted_errors = errors;
        sorted_errors.sort_by_key(|(index, _protocol, _error, _elapsed)| *index);

        let error_details: Vec<String> = sorted_errors
            .iter()
            .map(|(index, protocol, e, elapsed)| {
                format!("[{}] {:?} (耗时: {:?}): {}", index, protocol, elapsed, e)
            })
            .collect();

        let error_msg = format!(
            "所有协议连接都失败（按优先级顺序）: {}",
            error_details.join(", ")
        );

        tracing::error!("❌ {}", error_msg);
        Err(FlareError::connection_failed(error_msg))
    }

    /// 获取当前使用的协议
    pub fn active_protocol(&self) -> TransportProtocol {
        self.active_protocol
    }

    /// 获取 ClientCore（用于外部访问）
    pub fn core(&self) -> &ClientCore {
        &self.core
    }

    /// 获取 ClientCore 的可变引用（用于外部修改）
    pub fn core_mut(&mut self) -> &mut ClientCore {
        &mut self.core
    }
}

#[async_trait]
impl Client for HybridClient {
    async fn connect(&mut self) -> Result<()> {
        let mut client = self.inner.lock().await;
        client.connect().await
    }

    async fn disconnect(&mut self) -> Result<()> {
        let mut client = self.inner.lock().await;
        client.disconnect().await
    }

    async fn send_frame(&mut self, frame: &Frame) -> Result<()> {
        let mut client = self.inner.lock().await;
        client.send_frame(frame).await
    }

    fn is_connected(&self) -> bool {
        // 优先使用 try_lock 避免阻塞，如果无法立即获取锁则使用 block_in_place
        // 这样可以避免在异步运行时中直接使用 blocking_lock 导致的 panic
        match self.inner.try_lock() {
            Ok(client) => client.is_connected(),
            Err(_) => {
                // 如果无法立即获取锁，使用 block_in_place 在专用线程中执行
                // 这会将阻塞操作移到专用线程，避免阻塞 Tokio 运行时
                tokio::task::block_in_place(|| {
                    let client = self.inner.blocking_lock();
                    client.is_connected()
                })
            }
        }
    }

    fn add_observer(&mut self, observer: ArcObserver) {
        // 通过 ClientCore 添加观察者
        self.core.add_observer(observer);
    }

    fn remove_observer(&mut self, observer: ArcObserver) {
        self.core.remove_observer(observer);
    }

    fn connection_id(&self) -> Option<String> {
        if let Ok(client) = self.inner.try_lock() {
            client.connection_id()
        } else {
            None
        }
    }
}

impl std::fmt::Debug for HybridClient {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("HybridClient")
            .field("active_protocol", &self.active_protocol)
            .finish_non_exhaustive()
    }
}

/// 创建混合客户端的便捷函数
impl HybridClient {
    /// 发送消息并等待服务端响应（按 Frame.message_id 匹配）
    pub async fn send_frame_and_wait(&mut self, frame: &Frame, timeout: Duration) -> Result<Frame> {
        if frame.message_id.is_empty() {
            return Err(FlareError::protocol_error(
                "message_id is empty".to_string(),
            ));
        }

        tracing::debug!(
            "[HybridClient] send_frame_and_wait: 注册等待响应, message_id={}, frame.message_id={}",
            frame.message_id,
            frame.message_id
        );

        // 在发送前注册等待器
        let rx = self.core.register_pending_response(&frame.message_id).await;
        tracing::debug!(
            "[HybridClient] send_frame_and_wait: 已注册等待响应, message_id={}",
            frame.message_id
        );

        // 发送消息
        {
            let mut client = self.inner.lock().await;
            client.send_frame(frame).await?;
        }

        tracing::debug!(
            "[HybridClient] send_frame_and_wait: 消息已发送, 等待响应, message_id={}, timeout={:?}",
            frame.message_id,
            timeout
        );

        // 等待响应或超时
        match platform_timeout(timeout, rx).await {
            Ok(Ok(resp)) => {
                tracing::debug!(
                    "[HybridClient] send_frame_and_wait: 收到响应, message_id={}, resp.message_id={}",
                    frame.message_id,
                    resp.message_id
                );
                Ok(resp)
            }
            Ok(Err(_)) => {
                // 发送失败（通道已关闭），清理等待项
                tracing::debug!(
                    "[HybridClient] send_frame_and_wait: 响应通道已关闭, message_id={}",
                    frame.message_id
                );
                self.core.cancel_pending_response(&frame.message_id).await;
                Err(FlareError::protocol_error(
                    "Response channel closed".to_string(),
                ))
            }
            Err(_) => {
                // 超时，清理等待项
                tracing::warn!(
                    "[HybridClient] send_frame_and_wait: 响应超时, message_id={}",
                    frame.message_id
                );
                self.core.cancel_pending_response(&frame.message_id).await;
                Err(FlareError::protocol_error(format!(
                    "Response timeout for message_id {}",
                    frame.message_id
                )))
            }
        }
    }

    /// 使用配置创建并连接（单协议）
    pub async fn connect_with_config(config: ClientConfig) -> Result<Self> {
        let mut client = Self::new(config)?;
        client.connect().await?;
        Ok(client)
    }

    /// 使用配置创建并连接（协议竞速）
    pub async fn connect_with_race(config: ClientConfig) -> Result<Self> {
        if !config.is_race_mode() {
            return Self::connect_with_config(config).await;
        }

        let core = ClientCore::new(&config);
        let race_timeout = config.race_timeout.unwrap_or(Duration::from_secs(5));
        let inflight: RaceInflight = Arc::new(StdMutex::new(Vec::new()));
        let inflight_cleanup = Arc::clone(&inflight);

        let race_result = platform_timeout(
            race_timeout,
            Self::race_connect(config, core.clone(), inflight),
        )
        .await;

        let (client, protocol) = match race_result {
            Ok(result) => result?,
            Err(_) => {
                Self::cleanup_inflight(&inflight_cleanup).await;
                return Err(FlareError::connection_failed(format!(
                    "Protocol race timed out after {:?}",
                    race_timeout
                )));
            }
        };

        Ok(Self {
            inner: Arc::new(Mutex::new(client)),
            active_protocol: protocol,
            core,
        })
    }
}

#[cfg(all(test, not(target_arch = "wasm32")))]
mod hybrid_race_tests {
    use super::*;
    use std::sync::atomic::{AtomicBool, Ordering};

    struct TrackedRaceClient {
        disconnected: Arc<AtomicBool>,
        disconnect_requested: Arc<AtomicBool>,
    }

    impl TrackedRaceClient {
        fn new(disconnected: Arc<AtomicBool>, disconnect_requested: Arc<AtomicBool>) -> Self {
            Self {
                disconnected,
                disconnect_requested,
            }
        }
    }

    #[async_trait]
    impl Client for TrackedRaceClient {
        async fn connect(&mut self) -> Result<()> {
            Ok(())
        }

        async fn disconnect(&mut self) -> Result<()> {
            self.disconnected.store(true, Ordering::SeqCst);
            Ok(())
        }

        async fn send_frame(&mut self, _frame: &Frame) -> Result<()> {
            Ok(())
        }

        fn is_connected(&self) -> bool {
            !self.disconnected.load(Ordering::SeqCst)
        }

        fn add_observer(&mut self, _observer: ArcObserver) {}

        fn remove_observer(&mut self, _observer: ArcObserver) {}

        fn set_disconnect_requested(&mut self, value: bool) {
            self.disconnect_requested.store(value, Ordering::SeqCst);
        }
    }

    fn ok_client(
        disconnected: Arc<AtomicBool>,
        disconnect_requested: Arc<AtomicBool>,
    ) -> Box<dyn Client> {
        Box::new(TrackedRaceClient::new(disconnected, disconnect_requested))
    }

    #[tokio::test]
    async fn cleanup_remaining_race_handles_disconnects_successful_clients() {
        let disconnected = Arc::new(AtomicBool::new(false));
        let disconnect_requested = Arc::new(AtomicBool::new(false));
        let client = ok_client(Arc::clone(&disconnected), Arc::clone(&disconnect_requested));

        let handle = tokio::spawn(async move {
            (
                TransportProtocol::QUIC,
                1,
                Ok(client),
                Duration::from_millis(1),
            )
        });

        HybridClient::cleanup_remaining_race_handles(vec![handle]).await;

        assert!(disconnected.load(Ordering::SeqCst));
        assert!(disconnect_requested.load(Ordering::SeqCst));
    }

    #[tokio::test]
    async fn cleanup_remaining_race_handles_ignores_failed_clients() {
        let handle = tokio::spawn(async move {
            (
                TransportProtocol::QUIC,
                0,
                Err(FlareError::connection_failed("mock failure".to_string())),
                Duration::ZERO,
            )
        });

        HybridClient::cleanup_remaining_race_handles(vec![handle]).await;
    }

    #[tokio::test]
    async fn wait_for_connections_cleans_up_inflight_losers_after_early_win() {
        let winner_disconnected = Arc::new(AtomicBool::new(false));
        let winner_requested = Arc::new(AtomicBool::new(false));
        let loser_disconnected = Arc::new(AtomicBool::new(false));
        let loser_requested = Arc::new(AtomicBool::new(false));

        let fast = tokio::spawn({
            let winner_disconnected = Arc::clone(&winner_disconnected);
            let winner_requested = Arc::clone(&winner_requested);
            async move {
                crate::common::platform::sleep(Duration::from_millis(5)).await;
                (
                    TransportProtocol::WebSocket,
                    0,
                    Ok(ok_client(winner_disconnected, winner_requested)),
                    Duration::from_millis(5),
                )
            }
        });

        let slow = tokio::spawn({
            let loser_disconnected = Arc::clone(&loser_disconnected);
            let loser_requested = Arc::clone(&loser_requested);
            async move {
                crate::common::platform::sleep(Duration::from_millis(200)).await;
                (
                    TransportProtocol::QUIC,
                    1,
                    Ok(ok_client(loser_disconnected, loser_requested)),
                    Duration::from_millis(200),
                )
            }
        });

        let (first, _also_fast, errors) = {
            let inflight: RaceInflight = Arc::new(StdMutex::new(vec![fast, slow]));
            HybridClient::wait_for_connections(inflight).await
        };

        assert!(first.is_some());
        assert!(errors.is_empty());
        assert!(
            !winner_disconnected.load(Ordering::SeqCst),
            "winner must not be disconnected during race cleanup"
        );
        assert!(
            loser_disconnected.load(Ordering::SeqCst),
            "in-flight loser must be disconnected after early win"
        );
        assert!(loser_requested.load(Ordering::SeqCst));
    }
}