deepseek-tray/supervisor.go

package main

import (
	"bufio"
	"errors"
	"fmt"
	"io"
	"net"
	"os"
	"os/exec"
	"path/filepath"
	"strconv"
	"strings"
	"sync"
	"time"
)

// 稳定运行的判定阈值：超过它则清零失败计数（退避复位）。
const healthyUptime = 30 * time.Second

// Status 服务状态快照。
type Status struct {
	Running   bool
	PID       int
	Restarts  int
	Uptime    time.Duration
	LastError string
	URL       string
	Updating  bool
	External  bool // 端口被外部实例占用（非本程序启动）
}

// Supervisor 守护 dsh web：隐藏终端启动，异常退出自动重启（指数退避）。
type Supervisor struct {
	mu        sync.Mutex
	probeMu   sync.Mutex // 串行化入口探测，避免并发重复探测
	cfg       *Config
	log       *Logger
	pid       int
	running   bool
	desired   bool
	restarts  int
	failures  int
	startedAt time.Time
	lastError string
	updating  bool
	external  bool
	binCache  string
	nodeCache string
	webURL    string // dsh web 输出的完整 URL（含访问 token，仅存内存）

	autoMu   sync.Mutex
	autoLoop bool // 定期更新循环是否在运行（保证单例）
}

func NewSupervisor(cfg *Config, log *Logger) *Supervisor {
	return &Supervisor{cfg: cfg, log: log}
}

// Start 设置期望运行并拉起守护循环。
func (s *Supervisor) Start() {
	s.mu.Lock()
	if s.desired {
		s.mu.Unlock()
		return
	}
	s.desired = true
	s.failures = 0
	s.mu.Unlock()
	go s.monitor()
}

// Stop 同步停止：结束进程树并等待其真正退出（最多 8 秒）。
func (s *Supervisor) Stop() {
	s.mu.Lock()
	s.desired = false
	pid := s.pid
	s.webURL = ""
	s.mu.Unlock()
	if pid > 0 {
		s.log.Printf("停止 dsh web pid=%d", pid)
		_ = killTree(pid)
		s.waitStopped(pid, 8*time.Second)
	}
}

// Restart 重启并等待端口释放，避免新旧实例争抢端口。
func (s *Supervisor) Restart() {
	s.log.Printf("重启 dsh web")
	s.Stop()
	s.waitPortFree(5 * time.Second)
	s.Start()
}

func (s *Supervisor) Status() Status {
	s.mu.Lock()
	defer s.mu.Unlock()
	url := s.baseURL()
	if s.webURL != "" {
		url = s.webURL // 优先使用带 token 的地址
	}
	var uptime time.Duration
	if s.running && !s.startedAt.IsZero() {
		uptime = time.Since(s.startedAt)
	}
	return Status{
		Running:   s.running,
		PID:       s.pid,
		Restarts:  s.restarts,
		Uptime:    uptime,
		LastError: s.lastError,
		URL:       url,
		Updating:  s.updating,
		External:  s.external,
	}
}

func (s *Supervisor) baseURL() string {
	host := s.cfg.WebHost
	if host == "" {
		host = "127.0.0.1"
	}
	return fmt.Sprintf("http://%s:%d", host, s.cfg.WebPort)
}

func (s *Supervisor) setError(err error) {
	s.mu.Lock()
	s.lastError = err.Error()
	s.mu.Unlock()
	s.log.Printf("守护错误: %v", err)
}

// backoff 指数退避：base * 2^failures（上限 2 分钟），避免故障时重启风暴。
func (s *Supervisor) backoff() time.Duration {
	s.mu.Lock()
	failures := s.failures
	base := s.cfg.RestartDelaySec
	s.mu.Unlock()
	if base <= 0 {
		base = 5
	}
	d := time.Duration(base) * time.Second
	for i := 0; i < failures && i < 5; i++ {
		d *= 2
	}
	if d > 2*time.Minute {
		d = 2 * time.Minute
	}
	return d
}

// portBusyBackoff 端口占用时的温和退避（上限 30 秒）：不是故障，只是等待外部实例释放。
func (s *Supervisor) portBusyBackoff() time.Duration {
	d := s.backoff()
	if d > 30*time.Second {
		d = 30 * time.Second
	}
	return d
}

// monitor 守护循环：端口预检 -> 启动 -> 等待退出 -> 期望运行时退避重启。
func (s *Supervisor) monitor() {
	for {
		if !s.isDesired() {
			return
		}

		// 端口预检：已有实例监听时不再启动，避免端口冲突导致的重启风暴
		if isPortOpen(s.cfg.WebHost, s.cfg.WebPort) {
			s.mu.Lock()
			s.running = false
			s.pid = 0
			s.external = true
			s.lastError = "端口已被占用（可能已有 dsh web 在运行）"
			s.mu.Unlock()
			s.log.Printf("端口 %s 已被占用，跳过启动（等待释放）", s.baseURL())
			s.bumpFailure()
			s.sleepInterruptible(s.portBusyBackoff())
			continue
		}
		s.mu.Lock()
		s.external = false
		s.mu.Unlock()

		exe, args, err := s.resolveCommand()
		if err != nil {
			s.setError(err)
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}

		cmd := exec.Command(exe, args...)
		hideWindow(cmd)
		stdout, errOut := cmd.StdoutPipe()
		stderr, errErr := cmd.StderrPipe()
		if errOut != nil || errErr != nil {
			s.setError(errors.New("无法创建输出管道"))
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}
		if err := cmd.Start(); err != nil {
			s.setError(err)
			s.bumpFailure()
			s.sleepInterruptible(s.backoff())
			continue
		}

		pid := cmd.Process.Pid
		startedAt := time.Now()
		s.mu.Lock()
		s.pid = pid
		s.running = true
		s.startedAt = startedAt
		s.lastError = ""
		s.webURL = ""
		s.mu.Unlock()
		s.log.Printf("dsh web 已启动 pid=%d: %s %s", pid, exe, strings.Join(args, " "))

		go s.pipeLog("out", stdout)
		go s.pipeLog("err", stderr)

		waitErr := cmd.Wait()
		ranFor := time.Since(startedAt)

		s.mu.Lock()
		s.running = false
		s.pid = 0
		stillDesired := s.desired
		if ranFor >= healthyUptime {
			s.failures = 0 // 稳定运行过，退避复位
		} else {
			s.failures++
		}
		if stillDesired {
			s.restarts++
		}
		s.mu.Unlock()
		s.log.Printf("dsh web 退出（err=%v，运行 %s），期望运行=%v", waitErr, ranFor.Round(time.Second), stillDesired)

		if !stillDesired {
			return
		}
		delay := s.backoff()
		s.log.Printf("%s 后自动重启", delay)
		s.sleepInterruptible(delay)
	}
}

func (s *Supervisor) isDesired() bool {
	s.mu.Lock()
	defer s.mu.Unlock()
	return s.desired
}

// bumpFailure 增加失败计数（探测/启动阶段失败）。
func (s *Supervisor) bumpFailure() {
	s.mu.Lock()
	s.failures++
	s.mu.Unlock()
}

func (s *Supervisor) sleepInterruptible(d time.Duration) {
	deadline := time.Now().Add(d)
	for time.Now().Before(deadline) {
		if !s.isDesired() {
			return
		}
		time.Sleep(200 * time.Millisecond)
	}
}

func (s *Supervisor) waitStopped(pid int, timeout time.Duration) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if !processAlive(pid) {
			return
		}
		time.Sleep(100 * time.Millisecond)
	}
	s.log.Printf("等待 pid=%d 退出超时（继续）", pid)
}

func (s *Supervisor) waitPortFree(timeout time.Duration) {
	deadline := time.Now().Add(timeout)
	for time.Now().Before(deadline) {
		if !isPortOpen(s.cfg.WebHost, s.cfg.WebPort) {
			return
		}
		time.Sleep(150 * time.Millisecond)
	}
}

// pipeLog 把子进程输出写入日志（token 由 Logger 统一脱敏），并捕获访问地址。
func (s *Supervisor) pipeLog(tag string, r io.ReadCloser) {
	if r == nil {
		return
	}
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 0, 64*1024), 1024*1024)
	for sc.Scan() {
		line := sc.Text()
		s.log.Printf("[dsh %s] %s", tag, line)
		if url := extractHTTPURL(line); url != "" {
			s.mu.Lock()
			if url != s.webURL {
				s.webURL = url
				s.log.Printf("捕获 Web 访问地址（含 token，仅存内存）")
			}
			s.mu.Unlock()
		}
	}
}

// extractHTTPURL 从一行输出里取出 http(s) URL（去除尾随分隔符）。
func extractHTTPURL(line string) string {
	idx := strings.Index(line, "http")
	if idx < 0 {
		return ""
	}
	url := strings.TrimSpace(line[idx:])
	if i := strings.IndexAny(url, " \t"); i > 0 {
		url = url[:i]
	}
	if strings.HasPrefix(url, "http://") || strings.HasPrefix(url, "https://") {
		return url
	}
	return ""
}

// resolveCommand 解析启动命令：优先 node + dsh 入口 JS（无 shell 包装，避免弹窗与注入面）。
func (s *Supervisor) resolveCommand() (string, []string, error) {
	node := s.cfg.NodePath
	if node == "" {
		node = s.nodePath()
	}
	binJS := s.cfg.DshBinJS
	if binJS == "" {
		binJS = s.dshBinJS()
	}
	port := strconv.Itoa(s.cfg.WebPort)
	if node != "" && binJS != "" {
		return node, []string{binJS, "web", "--no-open", "--host", s.cfg.WebHost, "--port", port}, nil
	}
	if p, err := exec.LookPath("dsh.cmd"); err == nil {
		return "cmd.exe", []string{"/c", p, "web", "--no-open", "--port", port}, nil
	}
	if p, err := exec.LookPath("dsh"); err == nil {
		return p, []string{"web", "--no-open", "--port", port}, nil
	}
	return "", nil, errors.New("未找到 dsh 入口：请确认已全局安装 @deepseek-ai/dsh，或在配置中指定 nodePath / dshBinJs")
}

func (s *Supervisor) nodePath() string {
	s.mu.Lock()
	if s.nodeCache != "" {
		v := s.nodeCache
		s.mu.Unlock()
		return v
	}
	s.mu.Unlock()
	p, err := exec.LookPath("node")
	if err != nil {
		return ""
	}
	s.mu.Lock()
	s.nodeCache = p
	s.mu.Unlock()
	return p
}

// dshBinJS 探测 dsh 的 bin.js（APPDATA\npm 与 npm root -g）；probeMu 保证并发只探测一次。
func (s *Supervisor) dshBinJS() string {
	s.probeMu.Lock()
	defer s.probeMu.Unlock()

	s.mu.Lock()
	if s.binCache != "" {
		v := s.binCache
		s.mu.Unlock()
		return v
	}
	s.mu.Unlock()

	var candidates []string
	if appdata := os.Getenv("APPDATA"); appdata != "" {
		candidates = append(candidates, filepath.Join(appdata, "npm", "node_modules", "@deepseek-ai", "dsh", "lib", "bin.js"))
	}
	if root := npmGlobalRoot(); root != "" {
		candidates = append(candidates, filepath.Join(root, "@deepseek-ai", "dsh", "lib", "bin.js"))
	}
	for _, c := range candidates {
		if fileExists(c) {
			s.mu.Lock()
			s.binCache = c
			s.mu.Unlock()
			s.log.Printf("dsh 入口: %s", c)
			return c
		}
	}
	return ""
}

// invalidateProbe 使入口缓存失效（更新后调用）。
func (s *Supervisor) invalidateProbe() {
	s.mu.Lock()
	s.binCache = ""
	s.mu.Unlock()
}

// installedVersion 读取已安装 dsh 版本。
func (s *Supervisor) installedVersion() string {
	binJS := s.cfg.DshBinJS
	if binJS == "" {
		binJS = s.dshBinJS()
	}
	if binJS == "" {
		return ""
	}
	return readPackageVersion(filepath.Join(filepath.Dir(filepath.Dir(binJS)), "package.json"))
}

func fileExists(p string) bool {
	st, err := os.Stat(p)
	return err == nil && !st.IsDir()
}

// isPortOpen 检测端口是否有监听者。
func isPortOpen(host string, port int) bool {
	if host == "" {
		host = "127.0.0.1"
	}
	conn, err := net.DialTimeout("tcp", fmt.Sprintf("%s:%d", host, port), 1200*time.Millisecond)
	if err != nil {
		return false
	}
	_ = conn.Close()
	return true
}