go-svc-code/internal/opencode/process.go

package opencode

import (
	"bufio"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"strconv"
	"sync"
	"time"

	"git.x2erp.com/qdy/go-base/logger"
)

// Process 表示 opencode 进程
type Process struct {
	cmd     *exec.Cmd
	port    int
	logChan chan string
	mu      sync.Mutex
	running bool
}

// GetPort 返回 opencode 服务的端口
func (p *Process) GetPort() int {
	return p.port
}

// GetLogs 返回日志通道
func (p *Process) GetLogs() <-chan string {
	return p.logChan
}

// Close 停止 opencode 进程（实现 io.Closer 接口）
func (p *Process) Close() error {
	return p.Stop()
}

// Stop 停止 opencode 进程
func (p *Process) Stop() error {
	p.mu.Lock()
	defer p.mu.Unlock()

	if !p.running || p.cmd == nil {
		return nil
	}

	logger.Info("正在停止 opencode 进程...")

	// 尝试优雅停止
	if err := p.cmd.Process.Signal(os.Interrupt); err != nil {
		logger.Warn(fmt.Sprintf("发送中断信号失败: %v", err))
		// 强制终止
		if err := p.cmd.Process.Kill(); err != nil {
			logger.Error(fmt.Sprintf("强制终止进程失败: %v", err))
		}
	}

	// 等待进程退出
	done := make(chan error, 1)
	go func() {
		done <- p.cmd.Wait()
	}()

	select {
	case err := <-done:
		p.running = false
		if err != nil {
			logger.Debug(fmt.Sprintf("opencode 进程退出: %v", err))
		} else {
			logger.Debug("opencode 进程已停止")
		}
	case <-time.After(5 * time.Second):
		logger.Warn("opencode 进程在5秒内未退出，强制终止")
		if err := p.cmd.Process.Kill(); err != nil {
			logger.Error(fmt.Sprintf("最终强制终止失败: %v", err))
		}
		p.running = false
	}

	close(p.logChan)
	return nil
}

// Start 启动 opencode 进程
func Start(port int) (*Process, error) {
	cmd := exec.Command("opencode", "serve",
		"--hostname", "127.0.0.1",
		"--port", strconv.Itoa(port),
		"--log-level", "INFO",
	)

	logChan := make(chan string, 100)

	// 捕获标准输出
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return nil, fmt.Errorf("获取标准输出管道失败: %w", err)
	}

	// 捕获标准错误
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return nil, fmt.Errorf("获取标准错误管道失败: %w", err)
	}

	// 启动进程
	if err := cmd.Start(); err != nil {
		return nil, fmt.Errorf("启动 opencode 进程失败: %w", err)
	}

	process := &Process{
		cmd:     cmd,
		port:    port,
		logChan: logChan,
		running: true,
	}

	// 读取日志
	go readLogs(stdout, logChan, "stdout")
	go readLogs(stderr, logChan, "stderr")

	// 监控进程退出
	go func() {
		err := cmd.Wait()
		process.mu.Lock()
		process.running = false
		process.mu.Unlock()

		if err != nil {
			select {
			case logChan <- fmt.Sprintf("进程异常退出: %v", err):
			default:
			}
		} else {
			select {
			case logChan <- "进程正常退出":
			default:
			}
		}
		close(logChan)
	}()

	logger.Debug(fmt.Sprintf("opencode 进程已启动，端口: %d，PID: %d", port, cmd.Process.Pid))
	return process, nil
}

// readLogs 读取进程输出并发送到日志通道
func readLogs(reader io.Reader, logChan chan<- string, source string) {
	scanner := bufio.NewScanner(reader)
	for scanner.Scan() {
		line := scanner.Text()
		select {
		case logChan <- fmt.Sprintf("[%s] %s", source, line):
		default:
			// 如果通道满，丢弃旧日志（创建临时通道）
			tempChan := make(chan string, 1)
			select {
			case tempChan <- fmt.Sprintf("[%s] %s", source, line):
				// 无法发送，直接丢弃
			default:
			}
		}
	}
	if err := scanner.Err(); err != nil {
		select {
		case logChan <- fmt.Sprintf("[%s] 读取错误: %v", source, err):
		default:
		}
	}
}

// GetAvailablePort 获取可用端口
func GetAvailablePort() (int, error) {
	// 使用简化的端口获取，无需解析TCP地址
	listener, err := net.Listen("tcp", "127.0.0.1:0")
	if err != nil {
		return 0, err
	}
	defer listener.Close()

	addr := listener.Addr().(*net.TCPAddr)
	return addr.Port, nil
}

// WaitForReady 等待 opencode 服务就绪
func (p *Process) WaitForReady(timeout time.Duration) error {
	url := fmt.Sprintf("http://127.0.0.1:%d/global/health", p.port)
	client := &http.Client{Timeout: 5 * time.Second}

	start := time.Now()
	for time.Since(start) < timeout {
		resp, err := client.Get(url)
		if err == nil && resp.StatusCode == 200 {
			resp.Body.Close()
			return nil
		}
		if resp != nil {
			resp.Body.Close()
		}
		time.Sleep(100 * time.Millisecond)
	}

	return fmt.Errorf("opencode 服务在 %v 内未就绪", timeout)
}