go-svc-code/internal/api/events_stream.go

package api

import (
	"bufio"
	"bytes"
	"encoding/json"
	"fmt"
	"log"
	"net/http"
	"strings"
	"time"

	"git.x2erp.com/qdy/go-svc-code/internal/opencode/container"
)

// EventsStreamHandler 创建OpenCode事件流代理处理器，包含会话过滤
func EventsStreamHandler(manager *container.InstanceManager) http.HandlerFunc {
	return func(w http.ResponseWriter, r *http.Request) {
		// 从URL路径中提取项目ID
		// 路径格式：/api/opencode/projects/{id}/events
		pathParts := strings.Split(strings.Trim(r.URL.Path, "/"), "/")
		if len(pathParts) < 5 {
			http.Error(w, "无效的路径格式", http.StatusBadRequest)
			return
		}

		projectID := pathParts[3]
		// 从查询参数获取会话ID
		sessionID := r.URL.Query().Get("session_id")
		if sessionID == "" {
			sessionID = "default"
		}

		log.Printf("OpenCode事件流请求: 项目ID=%s, 会话ID=%s", projectID, sessionID)

		// 获取实例
		instance := manager.GetInstance(projectID)
		if instance == nil {
			http.Error(w, fmt.Sprintf("项目 %s 的OpenCode实例未启动", projectID), http.StatusNotFound)
			return
		}

		if instance.Status != container.StatusRunning {
			http.Error(w, fmt.Sprintf("项目 %s 的OpenCode实例不在运行状态", projectID), http.StatusServiceUnavailable)
			return
		}

		// 构建OpenCode事件流URL
		eventsURL := fmt.Sprintf("http://localhost:%d/api/sessions/%s/events", instance.Port, sessionID)

		// 创建HTTP请求到OpenCode实例
		req, err := http.NewRequest("GET", eventsURL, nil)
		if err != nil {
			http.Error(w, fmt.Sprintf("创建HTTP请求失败: %v", err), http.StatusInternalServerError)
			return
		}

		// 复制原始请求的头
		for name, values := range r.Header {
			for _, value := range values {
				req.Header.Add(name, value)
			}
		}

		// 添加OpenCode认证头（如果需要）
		if instance.Token != "" {
			req.Header.Set("Authorization", "Bearer "+instance.Token)
		}

		// 发送请求
		client := &http.Client{}
		resp, err := client.Do(req)
		if err != nil {
			http.Error(w, fmt.Sprintf("连接到OpenCode事件流失败: %v", err), http.StatusBadGateway)
			return
		}
		defer resp.Body.Close()

		// 检查响应状态
		if resp.StatusCode != http.StatusOK {
			http.Error(w, fmt.Sprintf("OpenCode事件流返回错误状态码: %d", resp.StatusCode), http.StatusBadGateway)
			return
		}

		// 设置响应头
		w.Header().Set("Content-Type", "text/event-stream")
		w.Header().Set("Cache-Control", "no-cache")
		w.Header().Set("Connection", "keep-alive")
		w.Header().Set("Access-Control-Allow-Origin", "*")

		flusher, ok := w.(http.Flusher)
		if !ok {
			http.Error(w, "流式响应不支持", http.StatusInternalServerError)
			return
		}

		// 建立连接事件
		connectEvent := map[string]interface{}{
			"type":       "connected",
			"project_id": projectID,
			"session_id": sessionID,
			"timestamp":  timestamp(),
		}
		sendSSEEvent(w, flusher, connectEvent)

		// 使用bufio.Scanner读取SSE事件流
		scanner := bufio.NewScanner(resp.Body)
		scanner.Buffer(make([]byte, 64*1024), 1024*1024) // 增加缓冲区大小

		var eventBuffer bytes.Buffer
		var inEvent bool

		for scanner.Scan() {
			line := scanner.Text()

			if line == "" {
				// 空行表示事件结束
				if inEvent && eventBuffer.Len() > 0 {
					data := eventBuffer.String()
					eventBuffer.Reset()
					inEvent = false

					// 处理SSE事件
					processedEvent := processSSEEvent(projectID, sessionID, data)
					if processedEvent != "" {
						// 发送处理后的事件
						fmt.Fprintf(w, "%s\n", processedEvent)
						flusher.Flush()
					}
				}
				continue
			}

			if !inEvent {
				inEvent = true
			}

			// 累积事件行
			eventBuffer.WriteString(line)
			eventBuffer.WriteString("\n")
		}

		if err := scanner.Err(); err != nil {
			log.Printf("读取OpenCode事件流失败: 项目=%s, 会话=%s, 错误=%v", projectID, sessionID, err)
		}
	}
}

// processSSEEvent 处理SSE事件，添加项目ID和会话ID信息
func processSSEEvent(projectID, sessionID, data string) string {
	// 解析SSE事件格式：data: {...}
	lines := strings.Split(data, "\n")

	for _, line := range lines {
		if strings.HasPrefix(line, "data: ") {
			eventData := strings.TrimPrefix(line, "data: ")

			// 尝试解析JSON
			var eventObj map[string]interface{}
			if err := json.Unmarshal([]byte(eventData), &eventObj); err != nil {
				// 如果不是JSON，直接返回原始数据
				return fmt.Sprintf("data: %s\n", eventData)
			}

			// 添加项目ID和会话ID信息
			eventObj["project_id"] = projectID
			eventObj["session_id"] = sessionID
			eventObj["timestamp"] = timestamp()

			// 重新序列化
			processedData, err := json.Marshal(eventObj)
			if err != nil {
				log.Printf("序列化处理后的事件失败: %v", err)
				return fmt.Sprintf("data: %s\n", eventData)
			}

			return fmt.Sprintf("data: %s\n", string(processedData))
		}
	}

	return data
}

// sendSSEEvent 发送SSE事件
func sendSSEEvent(w http.ResponseWriter, flusher http.Flusher, event map[string]interface{}) {
	eventJSON, err := json.Marshal(event)
	if err != nil {
		log.Printf("序列化SSE事件失败: %v", err)
		return
	}

	fmt.Fprintf(w, "data: %s\n\n", string(eventJSON))
	flusher.Flush()
}

// timestamp 获取当前时间戳
func timestamp() string {
	return fmt.Sprintf("%d", time.Now().UnixNano()/int64(time.Millisecond))
}