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

package opencode

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io"
	"net/http"
	"os"
	"path/filepath"
	"strings"
	"time"
)

// DefaultOpenCodePort opencode 服务默认端口（用于测试）
const DefaultOpenCodePort = 8787

// DirectClient opencode API 客户端（直接 HTTP 调用，不使用 SDK）
type DirectClient struct {
	baseURL    string
	port       int
	httpClient *http.Client
}

// 确保 DirectClient 实现 OpenCodeClient 接口
var _ OpenCodeClient = (*DirectClient)(nil)

// NewDirectClient 创建新的直接 HTTP opencode 客户端
func NewDirectClient(port int) (*DirectClient, error) {
	baseURL := fmt.Sprintf("http://127.0.0.1:%d", port)

	// 测试连接
	if err := testDirectConnection(baseURL); err != nil {
		return nil, fmt.Errorf("无法连接到 opencode 服务: %w", err)
	}

	return &DirectClient{
		baseURL: baseURL,
		port:    port,
		httpClient: &http.Client{
			Timeout: 30 * time.Second,
		},
	}, nil
}

// testDirectConnection 测试连接是否可用
func testDirectConnection(baseURL string) error {
	client := &http.Client{Timeout: 5 * time.Second}
	resp, err := client.Get(baseURL + "/global/health")
	if err != nil {
		return err
	}
	defer resp.Body.Close()

	if resp.StatusCode != 200 {
		return fmt.Errorf("服务不可用，状态码: %d", resp.StatusCode)
	}

	return nil
}

// CreateSession 创建新会话（直接 HTTP 调用）
func (c *DirectClient) CreateSession(ctx context.Context, title string) (*Session, error) {
	// 构造请求体
	reqBody := map[string]interface{}{
		"title": title,
	}
	jsonBody, err := json.Marshal(reqBody)
	if err != nil {
		return nil, fmt.Errorf("编码请求失败: %w", err)
	}

	// 发送 HTTP 请求
	req, err := http.NewRequestWithContext(ctx, "POST", c.baseURL+"/session", bytes.NewBuffer(jsonBody))
	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("HTTP请求失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("创建会话失败，状态码: %d, 响应体: %s", resp.StatusCode, string(body))
	}

	// 解析响应
	var session Session
	body, _ := io.ReadAll(resp.Body)
	if err := json.Unmarshal(body, &session); err != nil {
		return nil, fmt.Errorf("解析会话响应失败: %w", err)
	}

	fmt.Fprintf(os.Stderr, "[opencode-direct-client] 创建会话成功: %s\n", session.ID)
	return &session, nil
}

// SendPrompt 发送消息（同步，直接 HTTP 调用）
func (c *DirectClient) SendPrompt(ctx context.Context, sessionID string, prompt *PromptRequest) (*PromptResponse, error) {
	// 序列化请求体
	reqBody, err := json.Marshal(prompt)
	if err != nil {
		return nil, fmt.Errorf("编码请求失败: %w", err)
	}

	// 发送 HTTP 请求到 /session/{id}/message 端点（基于 svc-worker 测试）
	url := fmt.Sprintf("%s/session/%s/message", c.baseURL, sessionID)
	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(reqBody))
	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("发送消息失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("请求失败，状态码: %d, 响应体: %s", resp.StatusCode, string(body))
	}

	// 解析响应
	body, _ := io.ReadAll(resp.Body)

	// 首先尝试解析为完整的 PromptResponse
	var response PromptResponse
	if err := json.Unmarshal(body, &response); err == nil && response.Info.ID != "" {
		fmt.Fprintf(os.Stderr, "[opencode-direct-client] 发送消息成功，消息ID: %s\n", response.Info.ID)
		return &response, nil
	}

	// 如果失败，尝试解析为直接的消息响应
	var directResponse struct {
		ID      string `json:"id"`
		Role    string `json:"role"`
		Content string `json:"content"`
		Parts   []struct {
			Type string `json:"type"`
			Text string `json:"text"`
		} `json:"parts"`
		Model struct {
			ID         string `json:"id"`
			ProviderID string `json:"providerID"`
		} `json:"model"`
	}

	if err := json.Unmarshal(body, &directResponse); err != nil {
		return nil, fmt.Errorf("解析响应失败: %w", err)
	}

	// 构造标准的 PromptResponse
	response = PromptResponse{
		Info: AssistantMessage{
			ID:         directResponse.ID,
			Role:       directResponse.Role,
			SessionID:  sessionID,
			Content:    directResponse.Content,
			Agent:      "opencode",
			ModelID:    directResponse.Model.ID,
			ProviderID: directResponse.Model.ProviderID,
			Tokens: TokenInfo{
				Input:  0,
				Output: 0,
			},
			Time: map[string]interface{}{
				"created": time.Now().Unix(),
			},
		},
	}

	// 转换 parts
	if len(directResponse.Parts) > 0 {
		for _, part := range directResponse.Parts {
			response.Parts = append(response.Parts, map[string]string{
				"type": part.Type,
				"text": part.Text,
			})
		}
	}

	fmt.Fprintf(os.Stderr, "[opencode-direct-client] 发送消息成功，消息ID: %s\n", response.Info.ID)
	return &response, nil
}

// SendPromptStream 发送消息（流式，直接 HTTP 调用）
func (c *DirectClient) SendPromptStream(ctx context.Context, sessionID string, prompt *PromptRequest) (<-chan string, error) {
	reqBody, err := json.Marshal(prompt)
	if err != nil {
		return nil, fmt.Errorf("编码请求失败: %w", err)
	}

	fmt.Printf("🔍 [opencode.DirectClient] 发送流式请求到 session: %s\n", sessionID)
	fmt.Printf("🔍 [opencode.DirectClient] 请求体: %s\n", string(reqBody))
	fmt.Printf("🔍 [opencode.DirectClient] 端口: %d\n", c.port)

	// 测试异步端点
	url := fmt.Sprintf("%s/session/%s/prompt_async", c.baseURL, sessionID)
	req, err := http.NewRequestWithContext(ctx, "POST", url, bytes.NewBuffer(reqBody))
	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}
	req.Header.Set("Content-Type", "application/json")
	req.Header.Set("Accept", "text/event-stream")

	fmt.Printf("🔍 [opencode.DirectClient] 请求 URL: %s\n", url)
	fmt.Printf("🔍 [opencode.DirectClient] 请求头: %v\n", req.Header)

	// 为SSE流创建独立的httpClient，不设置超时限制
	sseClient := &http.Client{
		// 不设置Timeout，允许长连接
		// Timeout: 0 表示无超时限制
	}
	resp, err := sseClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("发送请求失败: %w", err)
	}

	fmt.Printf("🔍 [opencode.DirectClient] 响应状态: %d\n", resp.StatusCode)
	fmt.Printf("🔍 [opencode.DirectClient] 响应头: %v\n", resp.Header)

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		resp.Body.Close()
		fmt.Printf("🔍 [opencode.DirectClient] 错误响应体: %s\n", string(body))

		// 如果返回 204，尝试其他流式端点
		if resp.StatusCode == http.StatusNoContent {
			fmt.Printf("🔍 [opencode.DirectClient] 异步端点返回 204，尝试 /global/event SSE 流\n")
			return c.subscribeGlobalEvents(ctx)
		}

		return nil, fmt.Errorf("请求失败，状态码: %d", resp.StatusCode)
	}

	ch := make(chan string, 100)
	go func() {
		defer resp.Body.Close()
		defer close(ch)

		reader := bufio.NewReader(resp.Body)
		eventCount := 0

		for {
			line, err := reader.ReadString('\n')
			if err != nil {
				if err == io.EOF {
					fmt.Printf("🔍 [opencode.DirectClient] SSE流结束，共收到 %d 个事件\n", eventCount)
				} else {
					// 区分正常取消和错误
					if ctx.Err() != nil {
						fmt.Printf("🔍 [opencode.DirectClient] SSE流正常结束（上下文取消）\n")
					} else {
						fmt.Printf("🔍 [opencode.DirectClient] 读取错误: %v\n", err)
					}
				}
				return
			}

			line = strings.TrimSpace(line)
			if line == "" {
				continue
			}

			if strings.HasPrefix(line, "data: ") {
				data := strings.TrimPrefix(line, "data: ")
				eventCount++
				fmt.Printf("🔍 [opencode.DirectClient] 收到SSE数据[%d]: %s\n", eventCount, data)

				// 写入日志文件用于分析
				writeStreamLog(sessionID, data)

				select {
				case ch <- data:
				case <-ctx.Done():
					fmt.Printf("🔍 [opencode.DirectClient] 上下文取消\n")
					return
				}
			} else {
				fmt.Printf("🔍 [opencode.DirectClient] 忽略非数据行: %s\n", line)
			}
		}
	}()

	return ch, nil
}

// subscribeGlobalEvents 订阅全局事件流
func (c *DirectClient) subscribeGlobalEvents(ctx context.Context) (<-chan string, error) {
	url := c.baseURL + "/global/event"
	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return nil, fmt.Errorf("创建事件订阅请求失败: %w", err)
	}
	req.Header.Set("Accept", "text/event-stream")

	fmt.Printf("🔍 [opencode.DirectClient] 订阅全局事件流: %s\n", url)

	// 为SSE连接创建独立的httpClient，不设置超时限制
	sseClient := &http.Client{
		// 不设置Timeout，允许长连接
		// Timeout: 0 表示无超时限制
	}

	resp, err := sseClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("订阅事件流失败: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		resp.Body.Close()
		return nil, fmt.Errorf("事件流订阅失败，状态码: %d", resp.StatusCode)
	}

	ch := make(chan string, 100)
	go func() {
		defer resp.Body.Close()
		defer close(ch)

		scanner := bufio.NewScanner(resp.Body)
		eventCount := 0

		for scanner.Scan() {
			line := scanner.Text()
			if strings.HasPrefix(line, "data: ") {
				data := strings.TrimPrefix(line, "data: ")
				eventCount++
				fmt.Printf("🔍 [opencode.DirectClient] 收到全局事件[%d]: %s\n", eventCount, data)

				// 写入日志文件用于分析
				writeStreamLog("", data)

				select {
				case ch <- data:
				case <-ctx.Done():
					fmt.Printf("🔍 [opencode.DirectClient] 全局事件上下文取消\n")
					return
				}
			}
		}

		if err := scanner.Err(); err != nil {
			// 区分正常取消和错误
			if ctx.Err() != nil {
				fmt.Printf("🔍 [opencode.DirectClient] 全局事件流正常结束（上下文取消）\n")
			} else {
				fmt.Printf("🔍 [opencode.DirectClient] 扫描事件流错误: %v\n", err)
			}
		}
	}()

	return ch, nil
}

// GetSession 获取会话信息
func (c *DirectClient) GetSession(ctx context.Context, sessionID string) (*Session, error) {
	url := fmt.Sprintf("%s/session/%s", c.baseURL, sessionID)

	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("获取会话失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("获取会话失败，状态码: %d, 响应体: %s", resp.StatusCode, string(body))
	}

	var session Session
	if err := json.NewDecoder(resp.Body).Decode(&session); err != nil {
		return nil, fmt.Errorf("解析会话响应失败: %w", err)
	}

	return &session, nil
}

// ListSessions 获取会话列表
func (c *DirectClient) ListSessions(ctx context.Context) ([]Session, error) {
	url := c.baseURL + "/session"

	req, err := http.NewRequestWithContext(ctx, "GET", url, nil)
	if err != nil {
		return nil, fmt.Errorf("创建请求失败: %w", err)
	}

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return nil, fmt.Errorf("获取会话列表失败: %w", err)
	}
	defer resp.Body.Close()

	if resp.StatusCode != http.StatusOK {
		body, _ := io.ReadAll(resp.Body)
		return nil, fmt.Errorf("获取会话列表失败，状态码: %d, 响应体: %s", resp.StatusCode, string(body))
	}

	var sessions []Session
	if err := json.NewDecoder(resp.Body).Decode(&sessions); err != nil {
		return nil, fmt.Errorf("解析会话列表失败: %w", err)
	}

	return sessions, nil
}

// GetBaseURL 获取基础URL
func (c *DirectClient) GetBaseURL() string {
	return c.baseURL
}

// GetPort 获取端口
func (c *DirectClient) GetPort() int {
	return c.port
}

// writeStreamLog 将流式数据写入日志文件用于分析
func writeStreamLog(sessionID string, data string) {
	// 创建日志目录
	logDir := "/Users/kenqdy/Documents/v-bdx-workspace/svc-code/logs"
	if err := os.MkdirAll(logDir, 0755); err != nil {
		fmt.Printf("🔍 [opencode-direct-client] 创建日志目录失败: %v\n", err)
		return
	}

	// 生成日志文件名，按日期和会话ID组织
	dateStr := time.Now().Format("20060102")
	hourStr := time.Now().Format("15") // 小时
	var filename string

	if sessionID == "" {
		// 全局事件按小时组织
		filename = fmt.Sprintf("stream-global-%s-%s.log", dateStr, hourStr)
	} else {
		// 会话事件按会话ID和日期组织
		filename = fmt.Sprintf("stream-session-%s-%s.log", sessionID, dateStr)
	}

	filepath := filepath.Join(logDir, filename)

	// 追加写入数据
	file, err := os.OpenFile(filepath, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		fmt.Printf("🔍 [opencode-direct-client] 打开日志文件失败: %v\n", err)
		return
	}
	defer file.Close()

	// 写入时间戳和数据
	logLine := fmt.Sprintf("[%s] %s\n", time.Now().Format("15:04:05.000"), data)
	if _, err := file.WriteString(logLine); err != nil {
		fmt.Printf("🔍 [opencode-direct-client] 写入日志失败: %v\n", err)
	}
}