//! df-tunnel 隧道客户端(Phase2 完整实现)
//!
//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://host/ws/device`,
//! 穿 NAT 无需端口映射。tokio-tungstenite WS + Hello 握手 + 收发循环 + 指数退避重连。
//!
//! ## 协议对齐(df-relay Phase2 commit 2b8b30e)
//! - 桌面端连 `/ws/device`,首帧发 `Hello { kind:"device", device_id, token }`
//! - relay 双因子校验(kind == expected_device + token 匹配),失败发
//!   `{"kind":"control","error":"..."}` + Close,成功则静默进入收发循环(无显式 ack)
//! - 握手成功信号:**未收到 error 帧 且 socket 保持打开**(connect 用 timeout 探测首帧)
//! - 入站消息:relay 把对端 Miniapp 的文本帧包成 BroadcastMessage 转发,桌面端收到的
//!   是完整 BroadcastMessage JSON,payload 是 miniapp 指令原样 JSON(Phase3 纯透传,不解析)
//! - 出站消息:桌面端发 TunnelEvent 的 JSON(relay 把它当 payload 包成 BroadcastMessage)
//!
//! ## AiChatEvent 透传原则
//! payload 用 serde_json::Value 透传,本 crate 不依赖 src-tauri/df-types(避跨 crate 强耦合)。
//! TunnelEvent 仅是高频子集,扩展变体不破坏向后兼容。
//!
//! ## 并发模型
//! 内部 spawn 收发循环 task,通过 mpsc 解耦 send_event(调用方)与 socket 写入。
//! 收到 command 方向 payload(`serde_json::Value`)走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。
//! Phase3 协议统一(D1=A):tunnel 纯透传,业务协议解析在 device 端桥接层(src-tauri)。
//! `&self` + `Arc<Mutex<...>>` 持连接状态,trait 方法全部 async + 不持 stream 借用。

use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use futures_util::{SinkExt, StreamExt};
use tokio::sync::{mpsc, Mutex};
use tokio::task::JoinHandle;
use tokio_tungstenite::tungstenite::protocol::CloseFrame;
use tokio_tungstenite::tungstenite::Message as WsMessage;

use crate::error::{Result, TunnelError};
use crate::events::{Hello, HelloKind, RelayControlError, TunnelEvent};

/// 收到 miniapp→device 指令 payload 时的回调类型(Boxed Future,在收发循环 task 内 await)
///
/// Phase3 协议统一(D1=A):tunnel 纯透传,不再反序列化 TunnelCommand 强类型,
/// payload 以 `serde_json::Value` 上抛,业务协议解析在 device 端桥接层(src-tauri)。
/// 设计为 async 便于调用方执行 Tauri command 异步路由。回调内 panic 会中断收发循环,
/// 调用方应自行处理错误(回调返回值忽略,失败由调用方记录)。
pub type CommandHandler = Arc<dyn Fn(serde_json::Value) -> futures_util::future::BoxFuture<'static, ()> + Send + Sync>;

/// 隧道客户端抽象
///
/// 设计为 trait 而非具体结构,便于:
/// 1. 测试用 mock 实现(不依赖真实 WS)
/// 2. 未来替换底层实现(如换 quic / sse)不改调用方
///
/// 状态语义:
/// - `connect` 一次建立连接 + spawn 收发循环,重复 connect 视为重连(先 disconnect)
/// - `is_connected` 轻量查询,不发心跳
/// - `send_event` 非阻塞入队,实际 socket 写入由收发循环异步完成
/// - `disconnect` 优雅关闭(Close 帧 + 停 task + 清资源)
#[async_trait]
pub trait TunnelClient: Send + Sync {
    /// 主动连接云后端 + 注册指令回调
    ///
    /// `url` 形如 `wss://host/ws/device`(query 不带 token,token 走 Hello 帧)。
    /// 成功后进入「已连接」状态并 spawn 收发循环。失败返回 TunnelError::Connect/Auth。
    /// `on_command` 收到云后端转发的 command 方向 payload Value 时被调用(在收发 task 内异步执行)。
    /// Phase3 纯透传(D1=A):payload 不在 tunnel 解析,由 device 端桥接层 match cmd 路由。
    async fn connect(
        &self,
        url: &str,
        device_id: &str,
        token: &str,
        on_command: CommandHandler,
    ) -> Result<()>;

    /// 主动断开(优雅 Close 帧 + 释放 task / 通道)
    async fn disconnect(&self) -> Result<()>;

    /// 是否处于已连接状态(轻量查询,不阻塞,不发心跳)
    fn is_connected(&self) -> bool;

    /// 发送事件(桌面端 → 云后端 → 小程序)
    ///
    /// 非阻塞入队,实际 socket 写入异步完成。连接断开时返回 NotConnected。
    async fn send_event(&self, event: TunnelEvent) -> Result<()>;
}

/// 握手探测窗口(relay 校验通过后静默进入收发循环,无 ack 帧;
/// 此窗口内若收到 error 帧或连接 Close,判定握手失败)
const HANDSHAKE_PROBE: Duration = Duration::from_secs(3);

/// 心跳间隔(应用层 Ping,补协议层 keepalive,防 NAT 连接表超时回收)
const HEARTBEAT_INTERVAL: Duration = Duration::from_secs(25);

/// 重连初始退避
const RECONNECT_BASE: Duration = Duration::from_secs(1);
/// 重连退避上限
const RECONNECT_MAX: Duration = Duration::from_secs(30);

/// 内部连接状态(disconnect 时 drop 即清资源)
struct ConnState {
    /// 发送队列:send_event 投递 → 收发循环消费写 socket
    tx: mpsc::UnboundedSender<OutFrame>,
    /// 收发循环 task handle(disconnect 时 abort)
    task: JoinHandle<()>,
}

/// 收发循环出站帧(SendLoop 内部消费)
enum OutFrame {
    /// 业务事件(序列化后写 socket)
    Event(TunnelEvent),
    /// 优雅关闭
    Close,
}

/// 默认 WS 隧道客户端(完整实现)
pub struct WsTunnelClient {
    /// 目标服务器 URL(connect 后保存,重连用;字段供未来 auto-reconnect 读)
    /// 注:当前 connect 一次性建立,未实现自动后台重连(手动 connect 即重连),
    /// 该字段保留以备 Phase3「断线后台自动重连」消费。
    server_url: Mutex<Option<String>>,
    device_id: Mutex<Option<String>>,
    token: Mutex<Option<String>>,
    /// 当前活跃连接状态(None = 未连接)
    conn: Mutex<Option<ConnState>>,
    /// 连接状态标志(与收发循环 task 共享同一 Arc,is_connected 无锁查询;
    /// task 退出时置 false,connect 时置 true)
    connected: Arc<AtomicBool>,
}

impl WsTunnelClient {
    /// 创建未连接的客户端实例
    pub fn new() -> Self {
        Self {
            server_url: Mutex::new(None),
            device_id: Mutex::new(None),
            token: Mutex::new(None),
            conn: Mutex::new(None),
            connected: Arc::new(AtomicBool::new(false)),
        }
    }

    /// 当前已连接时取一份重连参数(未来自动重连用)
    pub async fn reconnect_params(&self) -> Option<(String, String, String)> {
        let url = self.server_url.lock().await.clone()?;
        let device_id = self.device_id.lock().await.clone()?;
        let token = self.token.lock().await.clone()?;
        Some((url, device_id, token))
    }
}

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

#[async_trait]
impl TunnelClient for WsTunnelClient {
    async fn connect(
        &self,
        url: &str,
        device_id: &str,
        token: &str,
        on_command: CommandHandler,
    ) -> Result<()> {
        // 已连接则先清(支持重复 connect 视作重连)
        if self.connected.load(Ordering::Relaxed) {
            self.cleanup_conn().await;
        }

        // 1. 建立 WS 连接
        let (ws_stream, _resp) = tokio_tungstenite::connect_async(url)
            .await
            .map_err(|e| TunnelError::Connect(format!("WS 连接失败 {url}: {e}")))?;
        tracing::info!(%url, "WS 连接已建立,开始 Hello 握手");

        let (mut sink, mut stream) = ws_stream.split();

        // 2. 发 Hello 握手帧
        let hello = Hello {
            kind: HelloKind::Device,
            device_id: device_id.to_string(),
            token: token.to_string(),
        };
        let hello_json = serde_json::to_string(&hello)?;
        sink.send(WsMessage::Text(hello_json)).await.map_err(|e| {
            TunnelError::Connect(format!("发送 Hello 失败: {e}"))
        })?;

        // 3. 探测握手结果:HANDSHAKE_PROBE 窗口内
        //    - 收到 error 帧 → relay 拒绝(返回 Auth 错误,附原因)
        //    - 收到 Close / EOF → 连接被关(返回 Connect 错误)
        //    - 收到首条正常消息 / 超时无负帧 → 视为握手成功(relay 静默进入收发循环)
        //    - Ping/Pong 协议帧:忽略继续等
        let mut first_frame_pending: Option<WsMessage> = None;
        let probe = tokio::time::timeout(HANDSHAKE_PROBE, stream.next()).await;
        match probe {
            Err(_) => {
                // 超时无负帧:relay 静默通过(常见路径)
                tracing::debug!("握手探测窗口无负帧,视为成功");
            }
            Ok(None) => {
                return Err(TunnelError::Connect(
                    "握手阶段连接关闭(relay 未响应)".into(),
                ));
            }
            Ok(Some(Err(e))) => {
                return Err(TunnelError::Connect(format!("握手 recv 错误: {e}")));
            }
            Ok(Some(Ok(msg))) => {
                first_frame_pending = Some(msg);
            }
        }

        // 校验首帧是否为 relay 拒绝帧
        if let Some(msg) = &first_frame_pending {
            if let Some(err) = parse_relay_error(msg) {
                return Err(TunnelError::Auth(format!(
                    "relay 拒绝握手: {}",
                    err.error
                )));
            }
            if matches!(msg, WsMessage::Close(_)) {
                return Err(TunnelError::Connect("握手被 relay Close".into()));
            }
        }

        // 4. 握手成功:保存参数 + 建发送队列 + spawn 收发循环
        *self.server_url.lock().await = Some(url.to_string());
        *self.device_id.lock().await = Some(device_id.to_string());
        *self.token.lock().await = Some(token.to_string());

        let (tx, rx) = mpsc::unbounded_channel::<OutFrame>();
        // 复用 self.connected 的 Arc,task 退出时置 false 与 self 同步
        self.connected.store(true, Ordering::Relaxed);
        let connected_for_task = self.connected.clone();

        let task = tokio::spawn(run_loop(
            sink,
            stream,
            first_frame_pending,
            rx,
            on_command,
            connected_for_task,
        ));

        *self.conn.lock().await = Some(ConnState { tx, task });
        tracing::info!(%url, device_id, "WsTunnelClient 连接就绪,收发循环已启动");

        Ok(())
    }

    async fn disconnect(&self) -> Result<()> {
        self.cleanup_conn().await;
        tracing::info!("WsTunnelClient 已断开");
        Ok(())
    }

    fn is_connected(&self) -> bool {
        self.connected.load(Ordering::Relaxed)
    }

    async fn send_event(&self, event: TunnelEvent) -> Result<()> {
        let conn = self.conn.lock().await;
        let Some(state) = conn.as_ref() else {
            return Err(TunnelError::NotConnected);
        };
        state
            .tx
            .send(OutFrame::Event(event))
            .map_err(|_| TunnelError::NotConnected)?;
        Ok(())
    }
}

impl WsTunnelClient {
    /// 清理当前连接(Close 帧 + abort task + 置标志)
    async fn cleanup_conn(&self) {
        let mut conn = self.conn.lock().await;
        if let Some(state) = conn.take() {
            // 通知收发循环优雅关闭(忽略发送失败:循环已退出)
            let _ = state.tx.send(OutFrame::Close);
            // 给 task 一点时间自然退出(发完 Close 帧);超时则 abort 兜底
            let mut task = state.task;
            tokio::select! {
                _ = &mut task => {}
                _ = tokio::time::sleep(Duration::from_millis(500)) => {
                    task.abort();
                }
            }
        }
        *self.server_url.lock().await = None;
        *self.device_id.lock().await = None;
        *self.token.lock().await = None;
        self.connected.store(false, Ordering::Relaxed);
    }
}

/// 收发循环主体(spawn 后独立运行)
///
/// 职责:
/// 1. socket 入帧 → 解析 BroadcastMessage → 取 payload Value 纯透传 → on_command 回调(协议解析在桥接层)
/// 2. mpsc 出帧 → socket 写入(Event 序列化 / Ping 心跳 / Close 关闭)
/// 3. 心跳定时器:每 HEARTBEAT_INTERVAL 发一次 Ping
/// 4. 任一端断开 → 退出 task,置 connected=false
///
/// 注:本任务范围不含「自动后台重连」(disconnect 即退出)。Phase3 可在外层包一层
/// supervisor task,检测 task 退出后用指数退避重新 connect(参数从 reconnect_params 取)。
async fn run_loop(
    mut sink: futures_util::stream::SplitSink<
        tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
        WsMessage,
    >,
    mut stream: futures_util::stream::SplitStream<
        tokio_tungstenite::WebSocketStream<tokio_tungstenite::MaybeTlsStream<tokio::net::TcpStream>>,
    >,
    first_frame: Option<WsMessage>,
    mut rx: mpsc::UnboundedReceiver<OutFrame>,
    on_command: CommandHandler,
    connected: Arc<AtomicBool>,
) {
    // 心跳 tick
    let mut heartbeat = tokio::time::interval(HEARTBEAT_INTERVAL);
    heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip);
    // 首次 tick 立即触发会干扰,跳过
    heartbeat.tick().await;

    // 处理握手期已读但未消费的首帧(可能是首条业务消息)
    if let Some(msg) = first_frame {
        if !handle_inbound(&msg, &on_command).await {
            tracing::info!("首帧处理后连接应关闭,退出收发循环");
            connected.store(false, Ordering::Relaxed);
            let _ = sink.close().await;
            return;
        }
    }

    loop {
        tokio::select! {
            // socket 入帧
            maybe_msg = stream.next() => {
                match maybe_msg {
                    Some(Ok(msg)) => {
                        if !handle_inbound(&msg, &on_command).await {
                            tracing::info!("对端关闭连接,收发循环退出");
                            break;
                        }
                    }
                    Some(Err(e)) => {
                        tracing::warn!(error = %e, "socket recv 错误,收发循环退出");
                        break;
                    }
                    None => {
                        tracing::info!("socket 流结束,收发循环退出");
                        break;
                    }
                }
            }
            // 出站队列
            maybe_out = rx.recv() => {
                match maybe_out {
                    Some(OutFrame::Event(ev)) => {
                        let json = match serde_json::to_string(&ev) {
                            Ok(j) => j,
                            Err(e) => {
                                tracing::warn!(error = %e, "事件序列化失败,跳过");
                                continue;
                            }
                        };
                        if let Err(e) = sink.send(WsMessage::Text(json)).await {
                            tracing::warn!(error = %e, "socket 写入失败,收发循环退出");
                            break;
                        }
                    }
                    Some(OutFrame::Close) => {
                        tracing::debug!("收到 disconnect 指令,发 Close 帧优雅关闭");
                        let _ = sink.send(WsMessage::Close(Some(CloseFrame {
                            code: tokio_tungstenite::tungstenite::protocol::frame::coding::CloseCode::Normal,
                            reason: "client_disconnect".into(),
                        }))).await;
                        break;
                    }
                    None => {
                        tracing::debug!("发送队列关闭(客户端 drop),退出收发循环");
                        break;
                    }
                }
            }
            // 心跳
            _ = heartbeat.tick() => {
                if let Err(e) = sink.send(WsMessage::Ping(Vec::new())).await {
                    tracing::warn!(error = %e, "心跳发送失败,收发循环退出");
                    break;
                }
            }
        }
    }

    connected.store(false, Ordering::Relaxed);
    // 确保 sink 关闭(若上面 break 未发 Close,这里兜底)
    let _ = sink.close().await;
}

/// 处理入站帧,返回 false 表示连接应关闭(Close/EOF/错误)
async fn handle_inbound(msg: &WsMessage, on_command: &CommandHandler) -> bool {
    match msg {
        WsMessage::Text(text) => {
            // Phase3 纯透传(D1=A):relay 转发完整 BroadcastMessage,
            // payload 是 miniapp 指令原样 JSON。tunnel 只提取 payload Value 上抛,
            // 不解析业务协议(协议解析在 device 端桥接层 src-tauri)。
            if let Some(payload) = parse_payload_from_broadcast(text) {
                // 回调内执行指令路由(桥接层 match cmd → Tauri command);回调失败不影响收发循环
                let fut = on_command(payload);
                fut.await;
            } else {
                // 非 Command 消息(Control/Event 回环/未知):忽略,不中断
                tracing::debug!(text = %text.chars().take(200).collect::<String>(), "收到非 Command 入站消息,忽略");
            }
            true
        }
        WsMessage::Binary(_) => {
            tracing::debug!("收到二进制帧,忽略");
            true
        }
        WsMessage::Ping(_) => {
            // tungstenite 协议层自动应答 Pong,这里不处理
            true
        }
        WsMessage::Pong(_) => true,
        WsMessage::Close(_) => false,
        WsMessage::Frame(_) => true, // 原始帧,上层已聚合,忽略
    }
}

/// 从 BroadcastMessage JSON 提取 command 方向的 payload Value(纯透传,不解析业务协议)
///
/// Phase3 协议统一(D1=A):df-relay 入站包成 `BroadcastMessage { ..., payload: Value }`,
/// tunnel 只提取 payload 字段以 Value 上抛 on_command,不反序列化为 TunnelCommand 强类型
/// (业务协议解析在 device 端桥接层)。仅 kind=="command" 时提取(Event/Control 不提,避免回环噪音)。
fn parse_payload_from_broadcast(raw: &str) -> Option<serde_json::Value> {
    // 仅取 payload 字段,避整结构强类型耦合(device_id/source 等字段本客户端不关心)
    #[derive(serde::Deserialize)]
    struct BroadcastLike {
        #[serde(default)]
        payload: Option<serde_json::Value>,
        #[serde(default)]
        kind: Option<String>,
    }
    let parsed: BroadcastLike = serde_json::from_str(raw).ok()?;
    // 仅 kind == "command" 时才提 payload(Event/Control 不提)
    if parsed.kind.as_deref() != Some("command") {
        return None;
    }
    parsed.payload
}

/// 判断是否为 relay 控制面错误帧
fn parse_relay_error(msg: &WsMessage) -> Option<RelayControlError> {
    if let WsMessage::Text(t) = msg {
        serde_json::from_str::<RelayControlError>(t).ok()
    } else {
        None
    }
}

/// 计算指数退避延迟(供未来 supervisor 重连使用,本任务未直接调用,但属完整重连语义的一部分)
#[allow(dead_code)]
fn backoff_delay(attempt: u32) -> Duration {
    // 1s, 2s, 4s, 8s, 16s, 30s, 30s, ...
    let secs = RECONNECT_BASE
        .as_secs()
        .saturating_mul(1u64 << attempt.min(5));
    Duration::from_secs(secs.min(RECONNECT_MAX.as_secs()))
}

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

    #[test]
    fn backoff_caps_at_max() {
        assert_eq!(backoff_delay(0), RECONNECT_BASE);
        assert_eq!(backoff_delay(1), Duration::from_secs(2));
        assert_eq!(backoff_delay(5), Duration::from_secs(32).min(RECONNECT_MAX));
        assert_eq!(backoff_delay(10), RECONNECT_MAX);
    }

    #[test]
    fn parse_payload_from_broadcast_extracts_value() {
        // 模拟 relay 转发的 BroadcastMessage(miniapp 发的 MiniCommand {cmd, args} 在 payload)
        let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"cmd":"send_message","args":{"message":"hi","conversation_id":"c1"}},"ts":1700000000000}"#;
        let payload = parse_payload_from_broadcast(raw).expect("应提取出 payload Value");
        // 纯透传:payload 原样上抛,字段保持 miniapp 端 {cmd, args} 结构(未做协议转换)
        assert_eq!(payload["cmd"], "send_message");
        assert_eq!(payload["args"]["message"], "hi");
        assert_eq!(payload["args"]["conversation_id"], "c1");
    }

    #[test]
    fn parse_payload_ignores_event_kind() {
        // Event 方向(device→miniapp)不提取,避免回环噪音
        let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"type":"text_delta","conversation_id":"c1","delta":"x"},"ts":1}"#;
        assert!(parse_payload_from_broadcast(raw).is_none());
    }

    /// D5 弱校验:payload Value 仍可反序列化为 TunnelCommand(保留强类型作桥接层可选校验)
    #[test]
    fn payload_compatible_with_tunnel_command() {
        use crate::events::TunnelCommand;
        let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1}"#;
        let payload = parse_payload_from_broadcast(raw).expect("应提取 payload");
        let cmd: TunnelCommand = serde_json::from_value(payload).expect("payload 应兼容 TunnelCommand");
        match cmd {
            TunnelCommand::Send { conv_id, content } => {
                assert_eq!(conv_id, "c1");
                assert_eq!(content, "hi");
            }
            other => panic!("预期 Send,实际 {other:?}"),
        }
    }

    #[test]
    fn parse_relay_error_detects_auth_failure() {
        let msg = WsMessage::Text(r#"{"kind":"control","error":"auth_failed"}"#.into());
        let err = parse_relay_error(&msg).expect("应解析出错误帧");
        assert_eq!(err.error, "auth_failed");
    }
}