diff --git a/crates/df-tunnel/src/events.rs b/crates/df-tunnel/src/events.rs index 0cf2289..af5cba3 100644 --- a/crates/df-tunnel/src/events.rs +++ b/crates/df-tunnel/src/events.rs @@ -11,6 +11,44 @@ use serde::{Deserialize, Serialize}; +/// 握手身份宣告帧(对齐 df-relay::relay::Hello) +/// +/// 桌面端 WS 连 `/ws/device` 后,首条消息发此 JSON,relay 校验 kind 与 token。 +/// 字段名/serde rename 与 df-relay 严格一致(`#[serde(rename_all="snake_case")]`), +/// 否则 relay 反序列化失败 → 握手被拒。 +/// +/// 注意:本 crate 不依赖 df-relay(避免 lib→server 反向依赖),自行镜像协议结构。 +/// 协议字段如有变更须同步 df-relay::relay::Hello 与本类型。 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Hello { + /// 客户端类型("device" / "miniapp"),桌面端恒为 Device + pub kind: HelloKind, + /// 配对绑定的设备 ID + pub device_id: String, + /// 配对 token(env DF_RELAY_TOKEN 同源) + pub token: String, +} + +/// Hello.kind 传输表示(serde 字符串,与 df-relay::ClientKindWire 一致) +#[derive(Debug, Clone, Copy, Serialize, Deserialize, PartialEq, Eq)] +#[serde(rename_all = "snake_case")] +pub enum HelloKind { + Device, + Miniapp, +} + +/// relay 控制面错误帧(握手失败时 relay 发送) +/// +/// relay 在校验失败/连接异常时发 `{"kind":"control","error":""}` 文本帧。 +/// 桌面端据此识别握手失败原因(不依赖 error 帧亦可:连接被 Close 即视为失败)。 +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct RelayControlError { + /// 固定 "control"(用于与正常 BroadcastMessage 区分,relay 错误帧简化为两字段) + pub kind: String, + /// 错误原因(handshake_failed / kind_mismatch / auth_failed) + pub error: String, +} + /// 隧道透传的事件类型(AiChatEvent 简化子集) /// /// 变体语义对齐桌面端 ai-chat-event: diff --git a/crates/df-tunnel/src/lib.rs b/crates/df-tunnel/src/lib.rs index 53d9086..b24aa04 100644 --- a/crates/df-tunnel/src/lib.rs +++ b/crates/df-tunnel/src/lib.rs @@ -22,5 +22,5 @@ pub mod tunnel; // 顶层再导出常用项,简化调用方 use pub use error::{Result, TunnelError}; -pub use events::{TunnelCommand, TunnelEvent, TunnelMessage}; +pub use events::{Hello, HelloKind, RelayControlError, TunnelCommand, TunnelEvent, TunnelMessage}; pub use tunnel::{TunnelClient, WsTunnelClient}; diff --git a/crates/df-tunnel/src/tunnel.rs b/crates/df-tunnel/src/tunnel.rs index ff20d23..5c193d9 100644 --- a/crates/df-tunnel/src/tunnel.rs +++ b/crates/df-tunnel/src/tunnel.rs @@ -1,35 +1,73 @@ -//! df-tunnel 隧道客户端 +//! df-tunnel 隧道客户端(Phase2 完整实现) //! -//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://your-server/ws/device`, -//! 穿 NAT 无需端口映射。本模块定义 `TunnelClient` trait + tokio-tungstenite WS 连接骨架。 +//! 设计依据:设计文档「Layer1」—— 桌面端主动出站连云后端 `wss://host/ws/device`, +//! 穿 NAT 无需端口映射。tokio-tungstenite WS + Hello 握手 + 收发循环 + 指数退避重连。 //! -//! 脚手架范围:trait 定义 + 连接骨架(占位实现),Phase2 填: -//! - 实际 WS 握手 + 鉴权(token / device_id) -//! - 断线指数退避重连 -//! - 事件桥接(app.emit 同步 clone 推隧道) -//! - 指令路由(收到 TunnelCommand 调 Tauri command) +//! ## 协议对齐(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 才是真正的 TunnelCommand +//! - 出站消息:桌面端发 TunnelEvent 的 JSON(relay 把它当 payload 包成 BroadcastMessage) +//! +//! ## AiChatEvent 透传原则 +//! payload 用 serde_json::Value 透传,本 crate 不依赖 src-tauri/df-types(避跨 crate 强耦合)。 +//! TunnelEvent 仅是高频子集,扩展变体不破坏向后兼容。 +//! +//! ## 并发模型 +//! 内部 spawn 收发循环 task,通过 mpsc 解耦 send_event(调用方)与 socket 写入。 +//! 收到 TunnelCommand 走 on_command 回调(调用方在 connect 时注册),不阻塞收发循环。 +//! `&self` + `Arc>` 持连接状态,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; -use crate::events::{TunnelCommand, TunnelEvent}; +use crate::error::{Result, TunnelError}; +use crate::events::{Hello, HelloKind, RelayControlError, TunnelCommand, TunnelEvent}; + +/// 收到 TunnelCommand 时的回调类型(Boxed Future,在收发循环 task 内 await) +/// +/// 设计为 async 便于调用方执行 Tauri command 异步路由。回调内 panic 会中断收发循环, +/// 调用方应自行处理错误(回调返回值忽略,失败由调用方记录)。 +pub type CommandHandler = Arc futures_util::future::BoxFuture<'static, ()> + Send + Sync>; /// 隧道客户端抽象 /// /// 设计为 trait 而非具体结构,便于: /// 1. 测试用 mock 实现(不依赖真实 WS) -/// 2. 未来替换底层实现(如换 sse / quic)不改调用方 +/// 2. 未来替换底层实现(如换 quic / sse)不改调用方 /// -/// 所有方法 async + 返回 Result,失败显式上抛不吞错。 +/// 状态语义: +/// - `connect` 一次建立连接 + spawn 收发循环,重复 connect 视为重连(先 disconnect) +/// - `is_connected` 轻量查询,不发心跳 +/// - `send_event` 非阻塞入队,实际 socket 写入由收发循环异步完成 +/// - `disconnect` 优雅关闭(Close 帧 + 停 task + 清资源) #[async_trait] pub trait TunnelClient: Send + Sync { - /// 主动连接云后端(出站长连接,穿 NAT) + /// 主动连接云后端 + 注册指令回调 /// - /// `url` 形如 `wss://your-server/ws/device?token=xxx`。 - /// 成功后进入「已连接」状态,失败返回 TunnelError::Connect。 - async fn connect(&self, url: &str) -> Result<()>; + /// `url` 形如 `wss://host/ws/device`(query 不带 token,token 走 Hello 帧)。 + /// 成功后进入「已连接」状态并 spawn 收发循环。失败返回 TunnelError::Connect/Auth。 + /// `on_command` 收到云后端转发的 TunnelCommand 时被调用(在收发 task 内异步执行)。 + async fn connect( + &self, + url: &str, + device_id: &str, + token: &str, + on_command: CommandHandler, + ) -> Result<()>; - /// 主动断开(优雅关闭 WS,释放底层 TCP) + /// 主动断开(优雅 Close 帧 + 释放 task / 通道) async fn disconnect(&self) -> Result<()>; /// 是否处于已连接状态(轻量查询,不阻塞,不发心跳) @@ -37,39 +75,72 @@ pub trait TunnelClient: Send + Sync { /// 发送事件(桌面端 → 云后端 → 小程序) /// - /// 设计为按事件粒度发送,不聚合;调用方在 app.emit 的同时 clone 推隧道。 + /// 非阻塞入队,实际 socket 写入异步完成。连接断开时返回 NotConnected。 async fn send_event(&self, event: TunnelEvent) -> Result<()>; - - /// 接收指令(云后端 → 桌面端) - /// - /// 阻塞等待下一条指令。返回 None 表示连接已关闭(EOF)。 - async fn recv_command(&self) -> Result>; } -/// 默认 WS 隧道客户端骨架(tokio-tungstenite) -/// -/// 脚手架阶段:字段占位,方法返回 `TunnelError::Other("未实现")`。 -/// Phase2 替换为完整实现: -/// - `WebSocketStream>` 持有连接 -/// - `mpsc` 通道解耦 send/recv 任务 -/// - 鉴权握手 + 心跳 + 重连 +/// 握手探测窗口(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, + /// 收发循环 task handle(disconnect 时 abort) + task: JoinHandle<()>, +} + +/// 收发循环出站帧(SendLoop 内部消费) +enum OutFrame { + /// 业务事件(序列化后写 socket) + Event(TunnelEvent), + /// 优雅关闭 + Close, +} + +/// 默认 WS 隧道客户端(完整实现) pub struct WsTunnelClient { - /// 目标服务器 URL(连接后保存,重连用) - #[allow(dead_code)] - server_url: Option, - /// 连接状态标志(轻量 is_connected 查询用) - #[allow(dead_code)] - connected: bool, + /// 目标服务器 URL(connect 后保存,重连用;字段供未来 auto-reconnect 读) + /// 注:当前 connect 一次性建立,未实现自动后台重连(手动 connect 即重连), + /// 该字段保留以备 Phase3「断线后台自动重连」消费。 + server_url: Mutex>, + device_id: Mutex>, + token: Mutex>, + /// 当前活跃连接状态(None = 未连接) + conn: Mutex>, + /// 连接状态标志(与收发循环 task 共享同一 Arc,is_connected 无锁查询; + /// task 退出时置 false,connect 时置 true) + connected: Arc, } impl WsTunnelClient { /// 创建未连接的客户端实例 pub fn new() -> Self { Self { - server_url: None, - connected: false, + 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 { @@ -80,35 +151,364 @@ impl Default for WsTunnelClient { #[async_trait] impl TunnelClient for WsTunnelClient { - async fn connect(&self, url: &str) -> Result<()> { - // 脚手架占位:Phase2 实现 tokio_tungstenite::connect_async + 鉴权握手。 - // 此处仅记录意图,返回未实现错误(不静默假成功,避免掩盖调用方逻辑)。 - tracing::warn!(url = %url, "WsTunnelClient.connect 尚未实现(Phase2 填充)"); - Err(crate::error::TunnelError::Other(anyhow::anyhow!( - "WsTunnelClient.connect 尚未实现(Phase2)" - ))) + 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 = 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::(); + // 复用 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<()> { - tracing::warn!("WsTunnelClient.disconnect 尚未实现(Phase2 填充)"); - Err(crate::error::TunnelError::Other(anyhow::anyhow!( - "WsTunnelClient.disconnect 尚未实现(Phase2)" - ))) + self.cleanup_conn().await; + tracing::info!("WsTunnelClient 已断开"); + Ok(()) } fn is_connected(&self) -> bool { - self.connected + self.connected.load(Ordering::Relaxed) } - async fn send_event(&self, _event: TunnelEvent) -> Result<()> { - Err(crate::error::TunnelError::Other(anyhow::anyhow!( - "WsTunnelClient.send_event 尚未实现(Phase2)" - ))) - } - - async fn recv_command(&self) -> Result> { - Err(crate::error::TunnelError::Other(anyhow::anyhow!( - "WsTunnelClient.recv_command 尚未实现(Phase2)" - ))) + 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 反序列化 TunnelCommand → 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>, + WsMessage, + >, + mut stream: futures_util::stream::SplitStream< + tokio_tungstenite::WebSocketStream>, + >, + first_frame: Option, + mut rx: mpsc::UnboundedReceiver, + on_command: CommandHandler, + connected: Arc, +) { + // 心跳 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) => { + // relay 转发的是完整 BroadcastMessage JSON + // (含 device_id/kind/source/from/payload/ts,payload 才是 TunnelCommand) + // 先尝试按 BroadcastMessage 镜像解析取 payload;失败则尝试直接解析为 TunnelCommand + // (兼容 relay 未来直发 payload 的变更) + let cmd_opt = parse_command_from_broadcast(text) + .or_else(|| serde_json::from_str::(text).ok()); + + if let Some(cmd) = cmd_opt { + // 回调内执行指令路由(Tauri command);回调失败不影响收发循环 + let fut = on_command(cmd); + fut.await; + } else { + // 非 Command 消息(Control/Event 回环/未知):忽略,不中断 + tracing::debug!(text = %text.chars().take(200).collect::(), "收到非 Command 入站消息,忽略"); + } + true + } + WsMessage::Binary(_) => { + tracing::debug!("收到二进制帧,忽略"); + true + } + WsMessage::Ping(_) => { + // tungstenite 协议层自动应答 Pong,这里不处理 + true + } + WsMessage::Pong(_) => true, + WsMessage::Close(_) => false, + WsMessage::Frame(_) => true, // 原始帧,上层已聚合,忽略 + } +} + +/// 从 BroadcastMessage JSON 提取 TunnelCommand +/// +/// df-relay 入站包成 `BroadcastMessage { ..., payload: <原文本解析为 Value> }`, +/// 故 payload 字段即 TunnelCommand 序列化后的 JSON 对象(tag="kind")。 +/// 此处只镜像解析 payload 字段(避免依赖 df-relay crate),payload 用 Value 透传不耦合。 +fn parse_command_from_broadcast(raw: &str) -> Option { + // 仅取 payload 字段,避整结构强类型耦合(device_id/source 等字段本客户端不关心) + #[derive(serde::Deserialize)] + struct BroadcastLike { + #[serde(default)] + payload: Option, + #[serde(default)] + kind: Option, + } + let parsed: BroadcastLike = serde_json::from_str(raw).ok()?; + // 仅 kind == "command" 时才尝试提 TunnelCommand(Event/Control 不提) + if parsed.kind.as_deref() != Some("command") { + return None; + } + let payload = parsed.payload?; + serde_json::from_value::(payload).ok() +} + +/// 判断是否为 relay 控制面错误帧 +fn parse_relay_error(msg: &WsMessage) -> Option { + if let WsMessage::Text(t) = msg { + serde_json::from_str::(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_command_from_broadcast_extracts_payload() { + // 模拟 relay 转发的 BroadcastMessage + let raw = r#"{"device_id":"dev-1","kind":"command","source":{"0":42},"from":"miniapp","payload":{"kind":"send","conv_id":"c1","content":"hi"},"ts":1700000000000}"#; + let cmd = parse_command_from_broadcast(raw).expect("应提取出 TunnelCommand"); + match cmd { + TunnelCommand::Send { conv_id, content } => { + assert_eq!(conv_id, "c1"); + assert_eq!(content, "hi"); + } + other => panic!("预期 Send,实际 {other:?}"), + } + } + + #[test] + fn parse_command_ignores_event_kind() { + let raw = r#"{"device_id":"dev-1","kind":"event","source":{"0":1},"from":"device","payload":{"kind":"text_delta","conv_id":"c1","delta":"x"},"ts":1}"#; + assert!(parse_command_from_broadcast(raw).is_none()); + } + + #[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"); } }