//! 公网隧道：把本机本地 HTTP 端口以 <前缀>.<中继域名> 形式映射到公网（反向代理）。
//!
//! 传输范式与 mobile 模块一致：无 tokio，纯 std 线程 + 每流一线程 + generation 代际取消。
//! 拨出到中继的 `wss://<中继域名>/__ccgui_tunnel__`，注册为隧道连接（携带注册密钥），
//! 中继把浏览器请求解复用成 REQ_*/U_* 帧下发，本端每条流开一条到 127.0.0.1:<本地端口> 的
//! TCP 双向对拷。安全边界：本端只按「本地已启用映射表」用子域名前缀反查端口，绝不使用中继下发的端口。
//!
//! 分工：
//! - 普通 HTTP：本端重建请求（改写 Host 为本地 + 注入 Connection: close），解析响应头后按
//!   content-length / chunked / close 定界把响应体回发；
//! - WebSocket/Upgrade（含 dev server 的 HMR）：握手后即不透明字节管道，本端不解析。

use std::collections::HashMap;
use std::io::{ErrorKind, Read, Write};
use std::net::TcpStream;
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::mpsc::{channel, Receiver, Sender, TryRecvError};
use std::sync::{Arc, Mutex};
use std::time::Duration;

use serde::Serialize;
use serde_json::{json, Map, Value};
use tauri::{AppHandle, Manager};
use tungstenite::stream::MaybeTlsStream;
use tungstenite::{Message, WebSocket};

use crate::store::{self, TunnelMapping};
use crate::AppState;

/// 桌面隧道控制连接的固定路径（须与中继 server.js 的 TUNNEL_PATH 一致）
const TUNNEL_PATH: &str = "/__ccgui_tunnel__";

// 数据面帧类型（二进制帧：[type:1][streamId:u32 大端][payload]，须与中继 F 对齐）
const REQ_HEAD: u8 = 0x01;
const REQ_BODY: u8 = 0x02;
const REQ_END: u8 = 0x03;
const REQ_ABORT: u8 = 0x04;
const U_OPEN: u8 = 0x20;
const PAUSE: u8 = 0x30;
const RESUME: u8 = 0x31;
const RES_HEAD: u8 = 0x10;
const RES_BODY: u8 = 0x11;
const RES_END: u8 = 0x12;
const RES_ABORT: u8 = 0x13;
const U_DATA: u8 = 0x22;
const U_CLOSE: u8 = 0x23;

/// 保留前缀（与中继侧一致，避免与基础设施冲突）
const RESERVED: [&str; 10] = ["www", "api", "healthz", "internal", "relay", "mail", "ftp", "ns", "ns1", "ns2"];

/// 隧道托管态（由 lib.rs 以 .manage() 注入）
#[derive(Default)]
pub struct TunnelState {
    /// 隧道拨出代际：翻转开关 / 改映射 / 改中继地址时 +1，令旧链路线程优雅退出（独立于 mobile 的各代际）
    generation: Arc<AtomicU64>,
    /// 是否已拨通中继（tunnel-registered 后置 true）
    online: Arc<AtomicBool>,
    /// 中继确认「已激活」的前缀
    active: Arc<Mutex<Vec<String>>>,
    /// 与他人在线连接冲突、未激活的前缀
    conflicts: Arc<Mutex<Vec<String>>>,
}

impl TunnelState {
    pub fn new() -> Self {
        Self::default()
    }
    /// 递增代际：让当前运行的隧道链路线程与所有流线程优雅退出
    pub fn bump_generation(&self) {
        self.generation.fetch_add(1, Ordering::Relaxed);
    }
}

// ---------- 帧构造 ----------

fn frame(t: u8, id: u32, payload: &[u8]) -> Message {
    let mut v = Vec::with_capacity(5 + payload.len());
    v.push(t);
    v.extend_from_slice(&id.to_be_bytes());
    v.extend_from_slice(payload);
    Message::Binary(v)
}

/// 逐跳头（不透传）
fn is_hop(lk: &str) -> bool {
    matches!(
        lk,
        "connection" | "keep-alive" | "proxy-authenticate" | "proxy-authorization" | "te" | "trailer" | "transfer-encoding" | "upgrade"
    )
}

// ---------- 子域名 -> 本地端口（安全闸：只认本地已启用映射）----------

fn port_of(app: &AppHandle, sub: &str) -> Option<u16> {
    let state = app.state::<AppState>();
    let cfg = state.config.lock().unwrap();
    if !cfg.settings.tunnel_enabled {
        return None;
    }
    cfg.settings
        .tunnel_mappings
        .iter()
        .find(|m| m.enabled && m.subdomain.eq_ignore_ascii_case(sub))
        .map(|m| m.local_port)
}

// ---------- 流线程与主会话 ----------

/// 主循环 → 流线程 的入站信号
enum StreamIn {
    /// 请求体字节（http）/ 升级连接字节（raw）
    Body(Vec<u8>),
    /// 请求体结束（http），关闭本地写半
    End,
    /// 关闭该流
    Close,
}

struct StreamHandle {
    tx: Sender<StreamIn>,
    /// 响应方向背压：中继 PAUSE 置真、RESUME 置假；流线程据此暂停读取本地
    paused: Arc<AtomicBool>,
}

/// 通用隧道会话循环：读中继帧、派发到各流线程、回发出站帧。
/// 对 LAN 不适用（隧道仅走中继拨出），但保持与 mobile 一致的泛型签名。
fn run_tunnel_session<S: Read + Write>(app: &AppHandle, ws: &mut WebSocket<S>, gen: &Arc<AtomicU64>, my_gen: u64) {
    let (out_tx, out_rx): (Sender<Message>, Receiver<Message>) = channel();
    // 流线程结束后通知主循环清理（修复「本地 EOF 结束的流不清理」的泄漏）
    let (done_tx, done_rx): (Sender<u32>, Receiver<u32>) = channel();
    let mut streams: HashMap<u32, StreamHandle> = HashMap::new();

    loop {
        if gen.load(Ordering::Relaxed) != my_gen {
            break;
        }
        match ws.read() {
            Ok(Message::Text(txt)) => {
                if let Ok(v) = serde_json::from_str::<Value>(&txt) {
                    match v.get("type").and_then(|t| t.as_str()) {
                        Some("tunnel-registered") => {
                            let ts = app.state::<TunnelState>();
                            ts.online.store(true, Ordering::Relaxed);
                            *ts.active.lock().unwrap() = str_vec(v.get("active"));
                            *ts.conflicts.lock().unwrap() = str_vec(v.get("conflicts"));
                        }
                        Some("error") => break, // 密钥不符 / 未启用等：断链，交由外层重连
                        _ => {}
                    }
                }
            }
            Ok(Message::Binary(buf)) => {
                if buf.len() < 5 {
                    continue;
                }
                let t = buf[0];
                let id = u32::from_be_bytes([buf[1], buf[2], buf[3], buf[4]]);
                let payload = &buf[5..];
                match t {
                    REQ_HEAD | U_OPEN => {
                        let head: Value = serde_json::from_slice(payload).unwrap_or(Value::Null);
                        let sub = head.get("sub").and_then(|x| x.as_str()).unwrap_or("").to_string();
                        match port_of(app, &sub) {
                            Some(port) => {
                                let (tx, rx) = channel::<StreamIn>();
                                let paused = Arc::new(AtomicBool::new(false));
                                streams.insert(id, StreamHandle { tx, paused: paused.clone() });
                                if t == REQ_HEAD {
                                    spawn_http_stream(id, port, head, rx, out_tx.clone(), done_tx.clone(), paused, gen.clone(), my_gen);
                                } else {
                                    spawn_raw_stream(id, port, head, rx, out_tx.clone(), done_tx.clone(), paused, gen.clone(), my_gen);
                                }
                            }
                            // 本地无此启用映射（刚被关掉等）：关掉这条流
                            None => {
                                let _ = out_tx.send(frame(if t == REQ_HEAD { RES_ABORT } else { U_CLOSE }, id, &[]));
                            }
                        }
                    }
                    REQ_BODY | U_DATA => {
                        if let Some(h) = streams.get(&id) {
                            let _ = h.tx.send(StreamIn::Body(payload.to_vec()));
                        }
                    }
                    REQ_END => {
                        if let Some(h) = streams.get(&id) {
                            let _ = h.tx.send(StreamIn::End);
                        }
                    }
                    REQ_ABORT | U_CLOSE => {
                        if let Some(h) = streams.remove(&id) {
                            let _ = h.tx.send(StreamIn::Close);
                        }
                    }
                    PAUSE => {
                        if let Some(h) = streams.get(&id) {
                            h.paused.store(true, Ordering::Relaxed);
                        }
                    }
                    RESUME => {
                        if let Some(h) = streams.get(&id) {
                            h.paused.store(false, Ordering::Relaxed);
                        }
                    }
                    _ => {}
                }
            }
            Ok(Message::Close(_)) => break,
            Ok(_) => {}
            Err(tungstenite::Error::Io(ref e)) if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut => {}
            Err(_) => break,
        }

        // 出站信箱：把各流线程产生的帧发给中继
        while let Ok(m) = out_rx.try_recv() {
            if ws.send(m).is_err() {
                app.state::<TunnelState>().online.store(false, Ordering::Relaxed);
                return;
            }
        }
        // 清理已结束的流
        while let Ok(id) = done_rx.try_recv() {
            streams.remove(&id);
        }
        let _ = ws.flush();
    }

    app.state::<TunnelState>().online.store(false, Ordering::Relaxed);
}

fn str_vec(v: Option<&Value>) -> Vec<String> {
    v.and_then(|x| x.as_array())
        .map(|a| a.iter().filter_map(|s| s.as_str().map(|s| s.to_string())).collect())
        .unwrap_or_default()
}

// ---------- HTTP 流：重建请求 + 解析响应回发 ----------

fn spawn_http_stream(
    id: u32,
    port: u16,
    head: Value,
    rx: Receiver<StreamIn>,
    out_tx: Sender<Message>,
    done_tx: Sender<u32>,
    paused: Arc<AtomicBool>,
    gen: Arc<AtomicU64>,
    my_gen: u64,
) {
    std::thread::spawn(move || {
        let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port));
        let mut sock = match TcpStream::connect_timeout(&addr, Duration::from_secs(5)) {
            Ok(s) => s,
            Err(_) => {
                let _ = out_tx.send(frame(RES_ABORT, id, &[]));
                let _ = done_tx.send(id);
                return;
            }
        };
        let _ = sock.set_read_timeout(Some(Duration::from_millis(20)));
        // 重建请求：改写 Host 为本地 + 强制 Connection: close（响应靠 EOF/长度/分块定界）
        if sock.write_all(&build_request_head(&head, port)).is_err() {
            let _ = out_tx.send(frame(RES_ABORT, id, &[]));
            let _ = done_tx.send(id);
            return;
        }

        let mut parser = RespParser::new();
        let mut buf = [0u8; 16384];
        loop {
            if gen.load(Ordering::Relaxed) != my_gen {
                break;
            }
            // 1) 请求体（浏览器 -> 本地）
            let mut closed = false;
            loop {
                match rx.try_recv() {
                    Ok(StreamIn::Body(b)) => {
                        if sock.write_all(&b).is_err() {
                            closed = true;
                            break;
                        }
                    }
                    Ok(StreamIn::End) => {
                        // 请求体结束：不在此半关闭写端。GET 无 body、POST 靠 Content-Length/分块自带定界，
                        // 本地服务据此即可判定请求已完。若此处 shutdown(Write)，nginx 会读到客户端 FIN，
                        // 对需反代到后端的接口（proxy_ignore_client_abort 默认 off）误判为「客户端中止」，
                        // 直接掐断上游、不回响应 → 隧道侧表现为 502（静态文件瞬时返回不受影响）。
                    }
                    Ok(StreamIn::Close) => {
                        let _ = sock.shutdown(std::net::Shutdown::Both);
                        closed = true;
                        break;
                    }
                    Err(TryRecvError::Empty) => break,
                    Err(TryRecvError::Disconnected) => break,
                }
            }
            if closed {
                break;
            }
            // 2) 响应（本地 -> 浏览器）；被中继 PAUSE 时不读，把 TCP 背压反压给本地
            if paused.load(Ordering::Relaxed) {
                std::thread::sleep(Duration::from_millis(10));
                continue;
            }
            match sock.read(&mut buf) {
                Ok(0) => {
                    parser.on_eof(id, &out_tx);
                    break;
                }
                Ok(n) => {
                    if !parser.feed(&buf[..n], id, &out_tx) {
                        break;
                    }
                    if parser.done {
                        break;
                    }
                }
                Err(ref e) if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut => {}
                Err(_) => {
                    parser.on_eof(id, &out_tx);
                    break;
                }
            }
        }
        let _ = done_tx.send(id);
    });
}

// ---------- 升级连接（WebSocket/HMR）：不透明字节双向对拷 ----------

fn spawn_raw_stream(
    id: u32,
    port: u16,
    head: Value,
    rx: Receiver<StreamIn>,
    out_tx: Sender<Message>,
    done_tx: Sender<u32>,
    paused: Arc<AtomicBool>,
    gen: Arc<AtomicU64>,
    my_gen: u64,
) {
    std::thread::spawn(move || {
        let addr = std::net::SocketAddr::from(([127, 0, 0, 1], port));
        let mut sock = match TcpStream::connect_timeout(&addr, Duration::from_secs(5)) {
            Ok(s) => s,
            Err(_) => {
                let _ = out_tx.send(frame(U_CLOSE, id, &[]));
                let _ = done_tx.send(id);
                return;
            }
        };
        let _ = sock.set_read_timeout(Some(Duration::from_millis(20)));
        // 重建升级请求：改写 Host / Origin 为本地，其余头（Upgrade/Connection/Sec-WebSocket-*）原样保留
        if sock.write_all(&build_upgrade_head(&head, port)).is_err() {
            let _ = out_tx.send(frame(U_CLOSE, id, &[]));
            let _ = done_tx.send(id);
            return;
        }

        let mut buf = [0u8; 16384];
        let mut closed = false;
        loop {
            if gen.load(Ordering::Relaxed) != my_gen {
                break;
            }
            // 浏览器 -> 本地
            loop {
                match rx.try_recv() {
                    Ok(StreamIn::Body(b)) => {
                        if sock.write_all(&b).is_err() {
                            closed = true;
                            break;
                        }
                    }
                    Ok(StreamIn::End) => {}
                    Ok(StreamIn::Close) => {
                        let _ = sock.shutdown(std::net::Shutdown::Both);
                        closed = true;
                        break;
                    }
                    Err(TryRecvError::Empty) => break,
                    Err(TryRecvError::Disconnected) => break,
                }
            }
            if closed {
                break;
            }
            if paused.load(Ordering::Relaxed) {
                std::thread::sleep(Duration::from_millis(10));
                continue;
            }
            // 本地 -> 浏览器
            match sock.read(&mut buf) {
                Ok(0) => {
                    let _ = out_tx.send(frame(U_CLOSE, id, &[]));
                    break;
                }
                Ok(n) => {
                    let _ = out_tx.send(frame(U_DATA, id, &buf[..n]));
                }
                Err(ref e) if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut => {}
                Err(_) => {
                    let _ = out_tx.send(frame(U_CLOSE, id, &[]));
                    break;
                }
            }
        }
        let _ = done_tx.send(id);
    });
}

// ---------- 请求头/升级头重建 ----------

fn append_header(s: &mut String, k: &str, v: &Value) {
    match v {
        Value::String(val) => {
            s.push_str(k);
            s.push_str(": ");
            s.push_str(val);
            s.push_str("\r\n");
        }
        Value::Array(arr) => {
            for it in arr {
                if let Some(val) = it.as_str() {
                    s.push_str(k);
                    s.push_str(": ");
                    s.push_str(val);
                    s.push_str("\r\n");
                }
            }
        }
        _ => {}
    }
}

fn build_request_head(head: &Value, port: u16) -> Vec<u8> {
    let method = head.get("method").and_then(|x| x.as_str()).unwrap_or("GET");
    let url = head.get("url").and_then(|x| x.as_str()).unwrap_or("/");
    let mut s = format!("{} {} HTTP/1.1\r\n", method, url);
    if let Some(hs) = head.get("headers").and_then(|x| x.as_object()) {
        for (k, v) in hs {
            let lk = k.to_ascii_lowercase();
            if lk == "host" || lk == "connection" || lk == "proxy-connection" {
                continue;
            }
            append_header(&mut s, k, v);
        }
    }
    s.push_str(&format!("Host: 127.0.0.1:{}\r\n", port));
    s.push_str("Connection: close\r\n\r\n");
    s.into_bytes()
}

fn build_upgrade_head(head: &Value, port: u16) -> Vec<u8> {
    let method = head.get("method").and_then(|x| x.as_str()).unwrap_or("GET");
    let url = head.get("url").and_then(|x| x.as_str()).unwrap_or("/");
    let mut s = format!("{} {} HTTP/1.1\r\n", method, url);
    if let Some(hs) = head.get("headers").and_then(|x| x.as_object()) {
        for (k, v) in hs {
            let lk = k.to_ascii_lowercase();
            if lk == "host" {
                continue;
            }
            if lk == "origin" {
                // 改写 Origin 为本地，避免 dev server 的同源校验拒绝 WS/HMR
                s.push_str(&format!("Origin: http://127.0.0.1:{}\r\n", port));
                continue;
            }
            append_header(&mut s, k, v);
        }
    }
    s.push_str(&format!("Host: 127.0.0.1:{}\r\n\r\n", port));
    s.into_bytes()
}

// ---------- 响应解析器：解析响应头 + 按 content-length / chunked / close 定界回发 ----------

enum BodyMode {
    Pending,
    Length(usize),
    Chunked,
    Close,
}

struct RespParser {
    head_buf: Vec<u8>,
    head_sent: bool,
    mode: BodyMode,
    chunk_buf: Vec<u8>,
    done: bool,
}

impl RespParser {
    fn new() -> Self {
        Self {
            head_buf: Vec::new(),
            head_sent: false,
            mode: BodyMode::Pending,
            chunk_buf: Vec::new(),
            done: false,
        }
    }

    fn feed(&mut self, data: &[u8], id: u32, out: &Sender<Message>) -> bool {
        if self.done {
            return true;
        }
        if !self.head_sent {
            self.head_buf.extend_from_slice(data);
            if let Some(pos) = find_sub(&self.head_buf, b"\r\n\r\n") {
                let head = self.head_buf[..pos].to_vec();
                let rest = self.head_buf[pos + 4..].to_vec();
                if !self.send_head(&head, id, out) {
                    return false;
                }
                self.head_sent = true;
                self.feed_body(&rest, id, out);
            }
            // 头未收全：等待更多字节
            true
        } else {
            self.feed_body(data, id, out);
            true
        }
    }

    fn send_head(&mut self, head: &[u8], id: u32, out: &Sender<Message>) -> bool {
        let text = String::from_utf8_lossy(head);
        let mut lines = text.split("\r\n");
        let status_line = lines.next().unwrap_or("");
        let mut parts = status_line.splitn(3, ' ');
        let _http = parts.next();
        let status: u16 = parts.next().and_then(|s| s.trim().parse().ok()).unwrap_or(502);
        let status_text = parts.next().unwrap_or("").to_string();

        let mut headers = Map::new();
        let mut te_chunked = false;
        let mut content_length: Option<usize> = None;
        for line in lines {
            if line.is_empty() {
                continue;
            }
            if let Some(ci) = line.find(':') {
                let k = line[..ci].trim().to_string();
                let v = line[ci + 1..].trim().to_string();
                let lk = k.to_ascii_lowercase();
                if lk == "transfer-encoding" && v.to_ascii_lowercase().contains("chunked") {
                    te_chunked = true;
                }
                if lk == "content-length" {
                    content_length = v.parse().ok();
                }
                if is_hop(&lk) {
                    continue; // 去 hop-by-hop（含 transfer-encoding），由中继 res 自行成帧
                }
                insert_header(&mut headers, k, v);
            }
        }

        if te_chunked {
            self.mode = BodyMode::Chunked;
        } else if let Some(n) = content_length {
            headers.insert("Content-Length".into(), Value::String(n.to_string()));
            self.mode = BodyMode::Length(n);
            if n == 0 {
                self.done = true;
            }
        } else {
            self.mode = BodyMode::Close;
        }

        let payload = json!({ "status": status, "statusText": status_text, "headers": Value::Object(headers) });
        let _ = out.send(frame(RES_HEAD, id, payload.to_string().as_bytes()));
        if self.done {
            let _ = out.send(frame(RES_END, id, &[]));
        }
        true
    }

    fn feed_body(&mut self, data: &[u8], id: u32, out: &Sender<Message>) {
        if self.done || data.is_empty() {
            return;
        }
        match &mut self.mode {
            BodyMode::Length(rem) => {
                let take = data.len().min(*rem);
                if take > 0 {
                    let _ = out.send(frame(RES_BODY, id, &data[..take]));
                    *rem -= take;
                }
                if *rem == 0 {
                    self.done = true;
                    let _ = out.send(frame(RES_END, id, &[]));
                }
            }
            BodyMode::Chunked => {
                self.chunk_buf.extend_from_slice(data);
                loop {
                    let nl = match find_sub(&self.chunk_buf, b"\r\n") {
                        Some(n) => n,
                        None => break,
                    };
                    let size_line = String::from_utf8_lossy(&self.chunk_buf[..nl]);
                    let size = match usize::from_str_radix(size_line.trim().split(';').next().unwrap_or("").trim(), 16) {
                        Ok(s) => s,
                        Err(_) => {
                            // 分块头损坏：丢弃剩余，避免死循环
                            self.chunk_buf.clear();
                            break;
                        }
                    };
                    if size == 0 {
                        self.done = true;
                        let _ = out.send(frame(RES_END, id, &[]));
                        self.chunk_buf.clear();
                        break;
                    }
                    if self.chunk_buf.len() < nl + 2 + size + 2 {
                        break; // 该分块尚未收全
                    }
                    let body = self.chunk_buf[nl + 2..nl + 2 + size].to_vec();
                    let _ = out.send(frame(RES_BODY, id, &body));
                    self.chunk_buf.drain(..nl + 2 + size + 2);
                }
            }
            BodyMode::Close => {
                let _ = out.send(frame(RES_BODY, id, data));
            }
            BodyMode::Pending => {}
        }
    }

    fn on_eof(&mut self, id: u32, out: &Sender<Message>) {
        if !self.head_sent {
            let _ = out.send(frame(RES_ABORT, id, &[]));
            return;
        }
        if !self.done {
            self.done = true;
            let _ = out.send(frame(RES_END, id, &[]));
        }
    }
}

fn find_sub(hay: &[u8], needle: &[u8]) -> Option<usize> {
    if needle.is_empty() || hay.len() < needle.len() {
        return None;
    }
    hay.windows(needle.len()).position(|w| w == needle)
}

/// 累积同名头（如 set-cookie）为数组，供中继 res.writeHead 正确输出
fn insert_header(map: &mut Map<String, Value>, k: String, v: String) {
    match map.get_mut(&k) {
        Some(Value::Array(arr)) => arr.push(Value::String(v)),
        Some(existing) => {
            let old = std::mem::replace(existing, Value::Null);
            *existing = Value::Array(vec![old, Value::String(v)]);
        }
        None => {
            map.insert(k, Value::String(v));
        }
    }
}

// ---------- 拨出与重连 ----------

/// 从 MaybeTlsStream 取底层 TcpStream（设读超时）
fn tcp_of(s: &MaybeTlsStream<TcpStream>) -> Option<&TcpStream> {
    match s {
        MaybeTlsStream::Plain(t) => Some(t),
        MaybeTlsStream::Rustls(t) => Some(&t.sock),
        _ => None,
    }
}

/// 隧道控制连接地址：由中继地址派生（wss://域名/ → wss://域名/__ccgui_tunnel__）
fn tunnel_ctrl_url(app: &AppHandle) -> String {
    let url = {
        let state = app.state::<AppState>();
        let cfg = state.config.lock().unwrap();
        cfg.settings.mobile_relay_url.trim().to_string()
    };
    if url.is_empty() {
        return String::new();
    }
    format!("{}{}", url.trim_end_matches('/'), TUNNEL_PATH)
}

/// 启动隧道拨出线程：连中继 → 注册（携密钥+映射快照）→ 跑会话循环；断线自动重连，受 generation 守护。
pub fn start_tunnel_link(app: AppHandle) {
    let ts = app.state::<TunnelState>();
    let my_gen = ts.generation.load(Ordering::Relaxed);
    let generation = ts.generation.clone();
    let online = ts.online.clone();
    let url = tunnel_ctrl_url(&app);
    if url.is_empty() {
        return;
    }
    let _ = std::thread::Builder::new().name("tunnel-link".into()).spawn(move || {
        while generation.load(Ordering::Relaxed) == my_gen {
            if let Err(e) = tunnel_once(&app, &url, &generation, my_gen) {
                eprintln!("隧道链路结束：{e}");
            }
            online.store(false, Ordering::Relaxed);
            // 约 3s 退避重连，其间响应代际停机
            for _ in 0..30 {
                if generation.load(Ordering::Relaxed) != my_gen {
                    break;
                }
                std::thread::sleep(Duration::from_millis(100));
            }
        }
    });
}

fn tunnel_once(app: &AppHandle, url: &str, gen: &Arc<AtomicU64>, my_gen: u64) -> Result<(), Box<dyn std::error::Error>> {
    let (secret, mappings) = {
        let state = app.state::<AppState>();
        let cfg = state.config.lock().unwrap();
        (cfg.settings.tunnel_secret.clone(), cfg.settings.tunnel_mappings.clone())
    };
    let (mut ws, _resp) = tungstenite::connect(url)?;
    if let Some(tcp) = tcp_of(ws.get_ref()) {
        let _ = tcp.set_read_timeout(Some(Duration::from_millis(20)));
    }
    // 注册：全量「子域名 -> 启用」快照 + 注册密钥 + 复用镜像的房间号/令牌
    let room = crate::mobile::store::load_or_init_room();
    let token = crate::mobile::store::load_or_init_token();
    let tunnels: Vec<Value> = mappings
        .iter()
        .map(|m| json!({ "sub": m.subdomain, "enabled": m.enabled }))
        .collect();
    ws.send(Message::Text(
        json!({ "type": "tunnel-register", "secret": secret, "room": room, "token": token, "tunnels": tunnels }).to_string(),
    ))?;
    ws.flush()?;
    run_tunnel_session(app, &mut ws, gen, my_gen);
    Ok(())
}

// ---------- 前缀校验 / URL 派生 ----------

fn valid_sub(s: &str) -> bool {
    let b = s.as_bytes();
    if b.is_empty() || b.len() > 63 {
        return false;
    }
    if RESERVED.iter().any(|r| *r == s) {
        return false;
    }
    if !b.iter().all(|&c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == b'-') {
        return false;
    }
    let first = b[0];
    let last = b[b.len() - 1];
    (first.is_ascii_lowercase() || first.is_ascii_digit()) && (last.is_ascii_lowercase() || last.is_ascii_digit())
}

fn normalize_sub(s: &str) -> Result<String, String> {
    let sub = s.trim().to_ascii_lowercase();
    if !valid_sub(&sub) {
        return Err("子域名前缀只能含小写字母/数字/连字符，首尾为字母或数字，且不能是保留字".into());
    }
    Ok(sub)
}

/// 由中继地址取基域 host（去协议、端口、路径）
fn base_host(relay_url: &str) -> String {
    let h = relay_url.trim().trim_end_matches('/');
    let h = h
        .strip_prefix("wss://")
        .or_else(|| h.strip_prefix("ws://"))
        .or_else(|| h.strip_prefix("https://"))
        .or_else(|| h.strip_prefix("http://"))
        .unwrap_or(h);
    let h = h.split('/').next().unwrap_or(h);
    h.split(':').next().unwrap_or(h).to_string()
}

fn public_url(base: &str, sub: &str) -> String {
    if base.is_empty() {
        String::new()
    } else {
        format!("https://{}.{}/", sub, base)
    }
}

// ---------- Tauri 命令 ----------

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TunnelEntry {
    pub id: String,
    pub subdomain: String,
    pub local_port: u16,
    pub enabled: bool,
    /// 中继确认已激活（启用 + 中继在线 + 未冲突）
    pub active: bool,
    /// 前缀被别的在线桌面占用（中继返回 conflicts）
    pub conflict: bool,
    /// 公网地址 https://<前缀>.<基域>/
    pub url: String,
}

#[derive(Serialize)]
#[serde(rename_all = "camelCase")]
pub struct TunnelStatus {
    /// 隧道总开关
    pub enabled: bool,
    /// 是否已开启公网中继（隧道页可用的前提）
    pub relay_enabled: bool,
    /// 是否已拨通中继
    pub online: bool,
    /// 是否已设置注册密钥
    pub secret_set: bool,
    /// 基域（由中继地址派生）
    pub base_domain: String,
    pub entries: Vec<TunnelEntry>,
}

fn build_status(app: &AppHandle) -> TunnelStatus {
    let (enabled, relay_enabled, secret, relay_url, mappings) = {
        let state = app.state::<AppState>();
        let cfg = state.config.lock().unwrap();
        (
            cfg.settings.tunnel_enabled,
            cfg.settings.mobile_relay_enabled,
            cfg.settings.tunnel_secret.clone(),
            cfg.settings.mobile_relay_url.clone(),
            cfg.settings.tunnel_mappings.clone(),
        )
    };
    let ts = app.state::<TunnelState>();
    let online = ts.online.load(Ordering::Relaxed);
    let active = ts.active.lock().unwrap().clone();
    let conflicts = ts.conflicts.lock().unwrap().clone();
    let base = base_host(&relay_url);
    let entries = mappings
        .iter()
        .map(|m| {
            let is_conflict = online && m.enabled && conflicts.iter().any(|a| a.eq_ignore_ascii_case(&m.subdomain));
            let is_active = online && m.enabled && !is_conflict && active.iter().any(|a| a.eq_ignore_ascii_case(&m.subdomain));
            TunnelEntry {
                id: m.id.clone(),
                subdomain: m.subdomain.clone(),
                local_port: m.local_port,
                enabled: m.enabled,
                active: is_active,
                conflict: is_conflict,
                url: public_url(&base, &m.subdomain),
            }
        })
        .collect();
    TunnelStatus {
        enabled,
        relay_enabled,
        online,
        secret_set: !secret.is_empty(),
        base_domain: base,
        entries,
    }
}

/// 改配置 → 存盘 → 重启隧道链路（重连即把最新映射快照推给中继）
fn mutate_and_restart<F: FnOnce(&mut store::Settings) -> Result<(), String>>(app: &AppHandle, f: F) -> Result<(), String> {
    let (relay_on, tunnel_on) = {
        let state = app.state::<AppState>();
        let mut cfg = state.config.lock().unwrap();
        f(&mut cfg.settings)?;
        store::save(&cfg)?;
        (cfg.settings.mobile_relay_enabled, cfg.settings.tunnel_enabled)
    };
    let ts = app.state::<TunnelState>();
    ts.bump_generation();
    ts.online.store(false, Ordering::Relaxed);
    ts.active.lock().unwrap().clear();
    ts.conflicts.lock().unwrap().clear();
    if relay_on && tunnel_on {
        start_tunnel_link(app.clone());
    }
    Ok(())
}

// ---------- 供 Tauri 命令与移动端网关共用的核心逻辑 ----------

pub fn status(app: &AppHandle) -> TunnelStatus {
    build_status(app)
}

pub fn set_master(app: &AppHandle, enabled: bool) -> Result<TunnelStatus, String> {
    mutate_and_restart(app, |s| {
        s.tunnel_enabled = enabled;
        Ok(())
    })?;
    Ok(build_status(app))
}

pub fn add(app: &AppHandle, subdomain: &str, local_port: u16) -> Result<TunnelStatus, String> {
    let sub = normalize_sub(subdomain)?;
    if local_port == 0 {
        return Err("本地端口需为 1-65535".into());
    }
    mutate_and_restart(app, |s| {
        if s.tunnel_mappings.iter().any(|m| m.subdomain.eq_ignore_ascii_case(&sub)) {
            return Err("该子域名前缀已存在".into());
        }
        s.tunnel_mappings.push(TunnelMapping {
            id: uuid::Uuid::new_v4().simple().to_string(),
            subdomain: sub,
            local_port,
            enabled: true,
        });
        Ok(())
    })?;
    Ok(build_status(app))
}

pub fn update(app: &AppHandle, id: &str, subdomain: &str, local_port: u16) -> Result<TunnelStatus, String> {
    let sub = normalize_sub(subdomain)?;
    if local_port == 0 {
        return Err("本地端口需为 1-65535".into());
    }
    mutate_and_restart(app, |s| {
        if s.tunnel_mappings.iter().any(|m| m.id != id && m.subdomain.eq_ignore_ascii_case(&sub)) {
            return Err("该子域名前缀已存在".into());
        }
        match s.tunnel_mappings.iter_mut().find(|m| m.id == id) {
            Some(m) => {
                m.subdomain = sub;
                m.local_port = local_port;
                Ok(())
            }
            None => Err("映射不存在".into()),
        }
    })?;
    Ok(build_status(app))
}

pub fn remove(app: &AppHandle, id: &str) -> Result<TunnelStatus, String> {
    mutate_and_restart(app, |s| {
        s.tunnel_mappings.retain(|m| m.id != id);
        Ok(())
    })?;
    Ok(build_status(app))
}

pub fn toggle(app: &AppHandle, id: &str, enabled: bool) -> Result<TunnelStatus, String> {
    mutate_and_restart(app, |s| {
        match s.tunnel_mappings.iter_mut().find(|m| m.id == id) {
            Some(m) => {
                m.enabled = enabled;
                Ok(())
            }
            None => Err("映射不存在".into()),
        }
    })?;
    Ok(build_status(app))
}

// ---------- Tauri 命令（薄封装）----------

#[tauri::command(async)]
pub fn tunnel_status(app: AppHandle) -> TunnelStatus {
    status(&app)
}

#[tauri::command(async)]
pub fn tunnel_set_master(app: AppHandle, enabled: bool) -> Result<TunnelStatus, String> {
    set_master(&app, enabled)
}

#[tauri::command(async)]
pub fn tunnel_set_secret(app: AppHandle, secret: String) -> Result<TunnelStatus, String> {
    mutate_and_restart(&app, |s| {
        s.tunnel_secret = secret.trim().to_string();
        Ok(())
    })?;
    Ok(build_status(&app))
}

#[tauri::command(async)]
pub fn tunnel_add(app: AppHandle, subdomain: String, local_port: u16) -> Result<TunnelStatus, String> {
    add(&app, &subdomain, local_port)
}

#[tauri::command(async)]
pub fn tunnel_update(app: AppHandle, id: String, subdomain: String, local_port: u16) -> Result<TunnelStatus, String> {
    update(&app, &id, &subdomain, local_port)
}

#[tauri::command(async)]
pub fn tunnel_remove(app: AppHandle, id: String) -> Result<TunnelStatus, String> {
    remove(&app, &id)
}

#[tauri::command(async)]
pub fn tunnel_toggle(app: AppHandle, id: String, enabled: bool) -> Result<TunnelStatus, String> {
    toggle(&app, &id, enabled)
}

#[cfg(test)]
mod tests {
    use super::*;
    use std::sync::mpsc::channel;

    // 把收集到的出站帧拆成 (type, id, payload)
    fn collect(rx: &Receiver<Message>) -> Vec<(u8, u32, Vec<u8>)> {
        let mut out = Vec::new();
        while let Ok(m) = rx.try_recv() {
            if let Message::Binary(b) = m {
                out.push((b[0], u32::from_be_bytes([b[1], b[2], b[3], b[4]]), b[5..].to_vec()));
            }
        }
        out
    }
    fn body_of(f: &[(u8, u32, Vec<u8>)]) -> Vec<u8> {
        f.iter().filter(|x| x.0 == RES_BODY).flat_map(|x| x.2.clone()).collect()
    }

    #[test]
    fn resp_length_mode() {
        let (tx, rx) = channel();
        let mut p = RespParser::new();
        assert!(p.feed(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\n\r\nhello", 7, &tx));
        assert!(p.done);
        let f = collect(&rx);
        assert_eq!(f[0].0, RES_HEAD);
        let head: Value = serde_json::from_slice(&f[0].2).unwrap();
        assert_eq!(head["status"].as_u64(), Some(200));
        assert_eq!(head["headers"]["Content-Length"], "5");
        assert_eq!(body_of(&f), b"hello");
        assert_eq!(f.last().unwrap().0, RES_END);
    }

    #[test]
    fn resp_chunked_mode() {
        let (tx, rx) = channel();
        let mut p = RespParser::new();
        p.feed(b"HTTP/1.1 200 OK\r\nTransfer-Encoding: chunked\r\n\r\n5\r\nhello\r\n6\r\n world\r\n0\r\n\r\n", 1, &tx);
        assert!(p.done);
        let f = collect(&rx);
        let head: Value = serde_json::from_slice(&f[0].2).unwrap();
        // chunked 应被去除、且不带 content-length（交由中继 res 自行成帧）
        assert!(head["headers"].get("Transfer-Encoding").is_none());
        assert!(head["headers"].get("Content-Length").is_none());
        assert_eq!(body_of(&f), b"hello world");
        assert!(f.iter().any(|x| x.0 == RES_END));
    }

    #[test]
    fn resp_close_mode_eof() {
        let (tx, rx) = channel();
        let mut p = RespParser::new();
        p.feed(b"HTTP/1.1 200 OK\r\n\r\nhel", 2, &tx);
        p.feed(b"lo", 2, &tx);
        assert!(!p.done);
        p.on_eof(2, &tx);
        assert!(p.done);
        let f = collect(&rx);
        assert_eq!(body_of(&f), b"hello");
        assert_eq!(f.last().unwrap().0, RES_END);
    }

    #[test]
    fn resp_split_head() {
        let (tx, rx) = channel();
        let mut p = RespParser::new();
        p.feed(b"HTTP/1.1 404 Not Found\r\nContent-Len", 3, &tx);
        assert!(!p.head_sent);
        p.feed(b"gth: 3\r\n\r\nabc", 3, &tx);
        assert!(p.done);
        let f = collect(&rx);
        let head: Value = serde_json::from_slice(&f[0].2).unwrap();
        assert_eq!(head["status"].as_u64(), Some(404));
        assert_eq!(body_of(&f), b"abc");
    }

    #[test]
    fn sub_validation() {
        assert!(valid_sub("demo"));
        assert!(valid_sub("a"));
        assert!(valid_sub("my-app-1"));
        assert!(!valid_sub("-bad"));
        assert!(!valid_sub("bad-"));
        assert!(!valid_sub("UP")); // 大写不合法
        assert!(!valid_sub("www")); // 保留字
        assert!(!valid_sub("a_b")); // 下划线
        assert!(!valid_sub(""));
    }

    #[test]
    fn url_derive() {
        assert_eq!(base_host("wss://relay.example.com/"), "relay.example.com");
        assert_eq!(base_host("https://example.com:8443/x"), "example.com");
        assert_eq!(base_host("example.com"), "example.com");
        assert_eq!(public_url("example.com", "demo"), "https://demo.example.com/");
        assert_eq!(public_url("", "demo"), "");
    }
}
