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

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

/// 音频命令 channel 容量（18F）：200 帧 × 100ms = 20s 音频，满时丢最旧帧避免无限堆积。
const CMD_CAP: usize = 200;

/// 浮层窗口 label（仅 overlay 监听 asr 事件）。
const OVERLAY: &str = "overlay";

/// 定向 emit 到 overlay 窗口（18G）。
fn emit_overlay(app: &AppHandle, event: &str, payload: Value) {
    let _ = app.emit_to(EventTarget::webview_window(OVERLAY), event, payload);
}

#[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::Sender<WsCmd>,
}

impl WsHandle {
    /// 投递命令；channel 满时（音频积压）丢弃当前帧并记录，避免无限堆积（18F）。
    /// 对非音频控制帧仍尽量投递（try_send 满时记录但不阻塞）。
    pub fn send(&self, cmd: WsCmd) -> Result<(), ()> {
        match self.tx.try_send(cmd) {
            Ok(()) => Ok(()),
            Err(mpsc::error::TrySendError::Full(c)) => {
                log::warn!("ws cmd channel full, dropping frame: {c:?}");
                Err(())
            }
            Err(mpsc::error::TrySendError::Closed(_)) => Err(()),
        }
    }
}

/// 启动常驻连接任务（应用启动即预连接，按键时零握手）。
pub fn spawn(app: AppHandle) -> WsHandle {
    // bounded channel（18F）：容量 CMD_CAP，满时由 WsHandle::send 丢最旧/当前帧并记录。
    let (tx, mut rx) = mpsc::channel::<WsCmd>(CMD_CAP);
    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;
                }
            };
            // header 用 from_str 处理 Result（18D）：token/device_id 含非法字符（换行/中文）
            // 时不 unwrap panic（否则杀死唯一 WS 泵任务致听写永久静默），而是 emit 错误并跳过本次连接。
            let auth = match HeaderValue::from_str(&format!("Bearer {token}")) {
                Ok(v) => v,
                Err(e) => {
                    log::error!("invalid Authorization header (token): {e}");
                    emit_overlay(
                        &app,
                        "asr",
                        json!({"type":"error","code":"AUTH_HEADER","message":"登录凭证异常，请重新登录"}),
                    );
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    continue;
                }
            };
            let dev = match HeaderValue::from_str(&device_id) {
                Ok(v) => v,
                Err(e) => {
                    log::error!("invalid X-Device-ID header: {e}");
                    emit_overlay(
                        &app,
                        "asr",
                        json!({"type":"error","code":"DEVICE_HEADER","message":"设备标识异常"}),
                    );
                    tokio::time::sleep(Duration::from_secs(5)).await;
                    continue;
                }
            };
            req.headers_mut().insert("Authorization", auth);
            req.headers_mut().insert("X-Device-ID", dev);

            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) => {
                    // 401 握手被拒（18F）：清 token + emit UNAUTHORIZED + 停止无脑重连。
                    if is_unauthorized(&e) {
                        log::warn!("ws handshake unauthorized (401); clearing token");
                        clear_token(&app);
                        emit_overlay(
                            &app,
                            "asr",
                            json!({"type":"error","code":"UNAUTHORIZED","message":"登录已失效，请重新登录"}),
                        );
                        // token 已清空，loop 顶部会进入"未登录"分支等待重新登录后的 Reconnect。
                        continue;
                    }
                    log::warn!("ws connect failed: {e}; retry in {backoff:?}");
                    tokio::time::sleep(backoff).await;
                    backoff = (backoff * 2).min(Duration::from_secs(8));
                    continue;
                }
            };

            // 重连成功（18E）：旧会话在断连期间的 start/stop 已丢失，服务端不认旧 session。
            // 不续旧会话——清空积压的旧会话音频帧 + 停止当前录音 + emit 会话级错误，避免静默丢音频。
            abort_active_session(&app, &mut rx);

            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() {
                            emit_overlay(&app, "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);
                                    emit_overlay(&app, "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}");
                                emit_overlay(&app, "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 }
}

/// 握手错误是否为 401 未授权（18F）。
fn is_unauthorized(e: &tokio_tungstenite::tungstenite::Error) -> bool {
    matches!(
        e,
        tokio_tungstenite::tungstenite::Error::Http(resp)
            if resp.status() == tokio_tungstenite::tungstenite::http::StatusCode::UNAUTHORIZED
    )
}

/// 清空本地 token（18F）：401 后需重新登录，避免用失效 token 无脑重连。
fn clear_token(app: &AppHandle) {
    let store = app.state::<crate::settings::SettingsStore>();
    let mut s = store.get();
    if !s.token.is_empty() {
        s.token.clear();
        store.set(s);
    }
}

/// 重连成功后中止仍在进行的录音会话（18E）：
/// 通知 dictation 停止采集 + emit 会话级错误（让用户知道这次听写失败需重说），
/// 并清空积压的旧会话音频/控制帧，避免发往新连接被服务端静默丢弃。
fn abort_active_session(app: &AppHandle, rx: &mut mpsc::Receiver<WsCmd>) {
    if !crate::dictation::is_recording(app) {
        // 没有进行中的会话：仍清掉可能积压的陈旧帧。
        while rx.try_recv().is_ok() {}
        return;
    }
    log::warn!("ws reconnected mid-session; aborting active dictation session");
    // 停止采集（取消模式，丢弃已采集文本，不上屏）。
    crate::dictation::stop(app, true);
    // 清空积压的旧会话音频/控制帧（新连接无对应 session，服务端会静默丢弃）。
    while rx.try_recv().is_ok() {}
    // 会话级错误：告知用户本次听写因网络重连失败，需重说。
    emit_overlay(
        app,
        "asr",
        json!({"type":"error","code":"SESSION_ABORTED","message":"网络已重连，本次听写中断，请重试"}),
    );
}
