dudu/desktop/src-tauri/src/ws.rs

//! WS 客户端（10C）：预连接 + 心跳 + 指数退避重连；
//! 上行音频二进制帧 / 控制 JSON，下行 ServerMsg → emit("asr") + 维护听写文本状态。

use futures_util::{SinkExt, StreamExt};
use serde_json::json;
use std::time::Duration;
use tauri::{AppHandle, Emitter, Manager};
use tokio::sync::mpsc;
use tokio_tungstenite::tungstenite::client::IntoClientRequest;
use tokio_tungstenite::tungstenite::Message;

#[derive(Debug)]
pub enum WsCmd {
    Start { session_id: String },
    Audio(Vec<u8>),
    Stop { session_id: String },
    Cancel { session_id: String },
    Reconnect, // 设置变更（token/服务器）后强制重连
}

#[derive(Clone)]
pub struct WsHandle {
    pub tx: mpsc::UnboundedSender<WsCmd>,
}

/// 启动常驻连接任务（应用启动即预连接，按键时零握手）。
pub fn spawn(app: AppHandle) -> WsHandle {
    let (tx, mut rx) = mpsc::unbounded_channel::<WsCmd>();
    tauri::async_runtime::spawn(async move {
        let mut backoff = Duration::from_millis(500);
        loop {
            let (url, token, device_id) = {
                let store = app.state::<crate::settings::SettingsStore>();
                let s = store.get();
                let ws_url = s.server_url.replace("http://", "ws://").replace("https://", "wss://");
                (format!("{ws_url}/v1/asr/stream"), s.token, s.device_id)
            };
            if token.is_empty() {
                // 未登录：等待命令但丢弃音频，定期重查
                tokio::time::sleep(Duration::from_secs(2)).await;
                while let Ok(cmd) = rx.try_recv() {
                    drop(cmd);
                }
                continue;
            }

            let mut req = match url.clone().into_client_request() {
                Ok(r) => r,
                Err(e) => {
                    log::error!("bad ws url: {e}");
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    continue;
                }
            };
            req.headers_mut().insert(
                "Authorization",
                format!("Bearer {token}").parse().unwrap(),
            );
            req.headers_mut()
                .insert("X-Device-ID", device_id.parse().unwrap());

            let conn = tokio_tungstenite::connect_async(req).await;
            let (mut sink, mut stream) = match conn {
                Ok((ws, _)) => {
                    log::info!("ws connected");
                    backoff = Duration::from_millis(500);
                    ws.split()
                }
                Err(e) => {
                    log::warn!("ws connect failed: {e}; retry in {backoff:?}");
                    tokio::time::sleep(backoff).await;
                    backoff = (backoff * 2).min(Duration::from_secs(8));
                    continue;
                }
            };

            let mut ping = tokio::time::interval(Duration::from_secs(30));
            'conn: loop {
                tokio::select! {
                    cmd = rx.recv() => {
                        let Some(cmd) = cmd else { return };
                        let msg = match cmd {
                            WsCmd::Start { session_id } => Message::Text(
                                json!({"type":"start","session_id":session_id,"sample_rate":16000,"format":"pcm16"}).to_string()),
                            WsCmd::Audio(frame) => Message::Binary(frame),
                            WsCmd::Stop { session_id } => Message::Text(
                                json!({"type":"stop","session_id":session_id}).to_string()),
                            WsCmd::Cancel { session_id } => Message::Text(
                                json!({"type":"cancel","session_id":session_id}).to_string()),
                            WsCmd::Reconnect => break 'conn,
                        };
                        if sink.send(msg).await.is_err() {
                            let _ = app.emit("asr", json!({"type":"error","code":"NETWORK"}));
                            break 'conn;
                        }
                    }
                    incoming = stream.next() => {
                        match incoming {
                            Some(Ok(Message::Text(text))) => {
                                if let Ok(v) = serde_json::from_str::<serde_json::Value>(&text) {
                                    crate::dictation::on_server_msg(&app, &v);
                                    let _ = app.emit("asr", v);
                                }
                            }
                            Some(Ok(Message::Ping(p))) => { let _ = sink.send(Message::Pong(p)).await; }
                            Some(Ok(_)) => {}
                            Some(Err(e)) => {
                                log::warn!("ws read error: {e}");
                                let _ = app.emit("asr", json!({"type":"error","code":"NETWORK"}));
                                break 'conn;
                            }
                            None => break 'conn,
                        }
                    }
                    _ = ping.tick() => {
                        if sink.send(Message::Ping(Vec::new())).await.is_err() { break 'conn; }
                    }
                }
            }
            tokio::time::sleep(backoff).await;
            backoff = (backoff * 2).min(Duration::from_secs(8));
        }
    });
    WsHandle { tx }
}