terminalX/packages/TXTransport/Sources/TXTransport/MoshSession.swift

import Foundation
import CMosh

/// MoshSession — 驱动 vendored mosh 客户端（MoshCore.xcframework 的 `mosh_main`）。
///
/// 桥接模型：`mosh_main` 是阻塞式主循环，用 `FILE*` 读用户输入、写终端输出。
/// - 输入：Swift `send(_:)` 写入 inPipe[1] → mosh 从 `fileno(f_in)=inPipe[0]` 读。
/// - 输出：mosh 写 `f_out=outPipe[1]` → 后台读线程从 outPipe[0] 读 → `onBytes`。
/// - resize：更新共享 `winsize` 结构后 `pthread_kill(SIGWINCH)`，mosh 的 select 循环据此重绘。
///
/// 连接目标由调用方给定 ip:port —— mosh-over-tsnet 时传 `127.0.0.1:<relay 本地端口>`
/// （见 TsnetBridge 的 MoshRelay），mosh 感知不到 tailnet 的存在。
public final class MoshSession: @unchecked Sendable {
    /// mosh 输出的终端字节（送入 libghostty）。
    public var onBytes: (@Sendable ([UInt8]) -> Void)?
    /// mosh_main 返回（会话结束）时回调，携带返回码（0=成功退出）。
    public var onClosed: (@Sendable (Int32) -> Void)?

    private let ip: String
    private let port: String
    private let key: String
    private let predictMode: String

    private let winPtr = UnsafeMutablePointer<winsize>.allocate(capacity: 1)
    private let lock = NSLock()
    private var inWriteFD: Int32 = -1     // Swift 写用户输入
    private var outReadFD: Int32 = -1     // Swift 读终端输出
    private var moshThread: pthread_t?
    private var started = false
    private var closed = false

    /// - Parameters:
    ///   - ip/port: mosh-server 的可达地址（tsnet 场景为 loopback relay 地址）。
    ///   - key: `mosh-server new` 打印的 MOSH_KEY（base64 会话密钥）。
    ///   - predictMode: always | never | adaptive | experimental（默认 adaptive）。
    public init(ip: String, port: Int, key: String, cols: Int, rows: Int, predictMode: String = "adaptive") {
        self.ip = ip
        self.port = String(port)
        self.key = key
        self.predictMode = predictMode
        winPtr.pointee = winsize(ws_row: UInt16(rows), ws_col: UInt16(cols), ws_xpixel: 0, ws_ypixel: 0)
    }

    deinit {
        winPtr.deallocate()
    }

    /// 建立 pipe、启动读线程与 mosh 主循环线程。可多次调用无副作用（仅首次生效）。
    public func start() {
        lock.lock()
        if started { lock.unlock(); return }
        started = true
        lock.unlock()

        // 写关闭的 pipe 会触发 SIGPIPE（默认杀进程）；改为返回 EPIPE。
        signal(SIGPIPE, SIG_IGN)
        // mosh 要求 UTF-8 native locale，否则 is_utf8_locale() 失败 → exit(1)。
        setenv("LANG", "en_US.UTF-8", 1)
        setenv("LC_ALL", "en_US.UTF-8", 1)

        var inPipe: [Int32] = [-1, -1]
        var outPipe: [Int32] = [-1, -1]
        _ = inPipe.withUnsafeMutableBufferPointer { pipe($0.baseAddress) }
        _ = outPipe.withUnsafeMutableBufferPointer { pipe($0.baseAddress) }

        lock.lock()
        inWriteFD = inPipe[1]
        outReadFD = outPipe[0]
        lock.unlock()

        startReader(fd: outPipe[0])
        startMosh(inReadFD: inPipe[0], outWriteFD: outPipe[1])
    }

    /// 写入用户输入字节（键盘）。
    public func send(_ bytes: [UInt8]) {
        lock.lock(); let fd = inWriteFD; lock.unlock()
        guard fd >= 0, !bytes.isEmpty else { return }
        var off = 0
        bytes.withUnsafeBytes { buf in
            while off < buf.count {
                let n = write(fd, buf.baseAddress!.advanced(by: off), buf.count - off)
                if n <= 0 { break }
                off += n
            }
        }
    }

    /// 更新终端尺寸并通知 mosh（SIGWINCH → process_resize）。
    public func resize(cols: Int, rows: Int) {
        winPtr.pointee.ws_col = UInt16(cols)
        winPtr.pointee.ws_row = UInt16(rows)
        lock.lock(); let tid = moshThread; lock.unlock()
        if let tid { pthread_kill(tid, SIGWINCH) }
    }

    /// 关闭会话：关掉 pipe（mosh 侧 select 遇 EOF/EPIPE），读线程随即结束。
    public func close() {
        lock.lock()
        if closed { lock.unlock(); return }
        closed = true
        let w = inWriteFD, r = outReadFD
        inWriteFD = -1; outReadFD = -1
        lock.unlock()
        if w >= 0 { Foundation.close(w) }
        if r >= 0 { Foundation.close(r) }
    }

    // MARK: - 私有

    private func setMoshThread(_ tid: pthread_t) { lock.lock(); moshThread = tid; lock.unlock() }

    private func handleExit(_ rc: Int32) {
        lock.lock(); moshThread = nil; lock.unlock()
        onClosed?(rc)
    }

    /// mosh 主循环线程：fdopen 两端 FILE*，调用阻塞的 mosh_main。用 Foundation.Thread
    /// （而非 pthread_create + Unmanaged 上下文）以规避 Swift6 SendNonSendable pass 的编译器崩溃；
    /// 线程内 pthread_self() 取得 tid 供 SIGWINCH。
    private func startMosh(inReadFD: Int32, outWriteFD: Int32) {
        let ip = self.ip, port = self.port, key = self.key, predict = self.predictMode
        let winPtr = self.winPtr
        let t = Thread { [weak self] in
            self?.setMoshThread(pthread_self())
            guard let fin = fdopen(inReadFD, "r"), let fout = fdopen(outWriteFD, "w") else {
                self?.handleExit(-1); return
            }
            setvbuf(fout, nil, _IONBF, 0)   // mosh 用 fwrite 写 out_fd，须无缓冲才能实时流出
            let rc = ip.withCString { ipC in
                port.withCString { portC in
                    key.withCString { keyC in
                        predict.withCString { prC in
                            mosh_main(fin, fout, winPtr, ipC, portC, keyC, prC)
                        }
                    }
                }
            }
            fflush(fout)
            self?.handleExit(rc)
        }
        t.name = "mosh-main"
        t.stackSize = 2 * 1024 * 1024
        t.start()
    }

    /// 后台读线程：把 mosh 输出字节泵给 onBytes。
    private func startReader(fd: Int32) {
        let t = Thread { [weak self] in
            var buf = [UInt8](repeating: 0, count: 16384)
            while true {
                let n = buf.withUnsafeMutableBytes { read(fd, $0.baseAddress, $0.count) }
                if n <= 0 { break }
                let chunk = Array(buf[0..<n])
                self?.onBytes?(chunk)
            }
        }
        t.name = "mosh-reader"
        t.stackSize = 512 * 1024
        t.start()
    }
}