FlowStack/backend/internal/handler/aistream.go

package handler

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strings"
	"time"

	"github.com/gin-gonic/gin"

	"flowstack/internal/database"
	"flowstack/internal/llm"
	"flowstack/internal/model"
)

// buildPlatformContext 平台实时数据注入（降低 AI 幻觉）：流水线概况 + 最近运行
func buildPlatformContext() string {
	var pipelines []model.Pipeline
	database.DB.Find(&pipelines)
	var runs []model.PipelineRun
	database.DB.Order("started_at DESC").Limit(5).Find(&runs)
	running := 0
	for _, r := range runs {
		if r.Status == "running" || r.Status == "pending" {
			running++
		}
	}
	var b strings.Builder
	fmt.Fprintf(&b, "【平台实时数据】\n流水线总数: %d（运行中 %d）\n", len(pipelines), running)
	if len(runs) > 0 {
		b.WriteString("最近运行:\n")
		for _, r := range runs {
			fmt.Fprintf(&b, "- %s %s %s %s %s\n", r.PipelineName, r.RunID, r.Branch, r.Status, fmtDurMs(r.DurationMs))
		}
	}
	return b.String()
}

func fmtDurMs(ms int64) string {
	if ms <= 0 {
		return "-"
	}
	return fmt.Sprintf("%ds", ms/1000)
}

// AiChatStream POST /api/ai/chat/stream —— 流式对话（SSE）
func AiChatStream(c *gin.Context) {
	var body struct {
		Messages []ChatMsg `json:"messages" binding:"required"`
	}
	if err := c.ShouldBindJSON(&body); err != nil || len(body.Messages) == 0 {
		c.JSON(http.StatusBadRequest, gin.H{"error": "messages 不能为空"})
		return
	}
	var lastUser string
	for i := len(body.Messages) - 1; i >= 0; i-- {
		if body.Messages[i].Role == "user" {
			lastUser = body.Messages[i].Content
			break
		}
	}
	if lastUser == "" {
		c.JSON(http.StatusBadRequest, gin.H{"error": "没有用户消息"})
		return
	}

	candidates := allCandidates()
	names := make([]string, 0, len(candidates))
	if len(candidates) > 200 {
		candidates = candidates[:200]
	}
	for _, cd := range candidates {
		names = append(names, cd.Name)
	}
	ruleRes := parseByRules(lastUser, candidates)

	client := llmClientFromSettings()
	if client == nil {
		// 未配置 LLM：直接返回规则结果（不流式）
		reply, action := ruleReply(ruleRes, candidates)
		c.JSON(http.StatusOK, gin.H{"reply": reply, "viaAI": false, "action": action})
		return
	}

	c.Header("Content-Type", "text/event-stream")
	c.Header("Cache-Control", "no-cache")
	c.Header("Connection", "keep-alive")
	c.Header("X-Accel-Buffering", "no")
	flush := func() {
		if f, ok := c.Writer.(http.Flusher); ok {
			f.Flush()
		}
	}
	emit := func(event string, data any) {
		b, _ := json.Marshal(data)
		fmt.Fprintf(c.Writer, "event: %s\ndata: %s\n\n", event, b)
		flush()
	}

	system := fmt.Sprintf(`你是 FlowStack 发布系统的智能发布助手。用户会用自然语言描述要发布的服务。
%s
可发布的服务清单（流水线与 Jenkins Job 名称）如下：
%s

规则：
- 若用户意图是发布/构建服务：只输出 JSON（不要 markdown 代码块、不要多余文字）：
  {"reply":"一句中文确认语","jobs":["服务名1"],"branch":"分支或留空","env":"环境或留空"}
  jobs 必须来自上面的清单按原文匹配；用户提到的服务不在清单时 jobs 留空并在 reply 中说明。
- 其他问题（如询问状态、概念解释）：用简洁中文自然语言回答，禁止编造清单中不存在的服务或字段。`,
		buildPlatformContext(), strings.Join(names, "\n"))

	emit("start", gin.H{"viaAI": true})

	ctx := c.Request.Context()
	full, err := client.ChatStream(ctx, system, lastUser, func(d llm.StreamDelta) {
		if d.Content != "" || d.Reasoning != "" {
			emit("delta", gin.H{"content": d.Content, "reasoning": d.Reasoning})
		}
	})

	if err != nil {
		// LLM 失败：降级规则结果
		reply, action := ruleReply(ruleRes, candidates)
		emit("done", gin.H{"reply": reply + "\n（AI 调用失败已降级本地规则: " + err.Error() + "）", "viaAI": false, "action": action})
		return
	}

	reply, action := parseLLMJson(full, candidates)
	if action == nil && len(ruleRes.Jobs) > 0 {
		// LLM 没识别出发布指令但规则识别到了，合并规则结果
		_, action = ruleReply(ruleRes, candidates)
	}
	emit("done", gin.H{"reply": reply, "viaAI": true, "action": action})
}

func ruleReply(ruleRes ruleParseResult, candidates []Candidate) (string, *aiReleaseAction) {
	if len(ruleRes.Jobs) == 0 {
		return "未识别到要发布的服务。请直接告诉我服务名，例如：发布 cex-agent-server、cex-agent-api 到 pre 分支。", nil
	}
	jobs := make([]aiActionJob, 0, len(ruleRes.Jobs))
	for _, name := range ruleRes.Jobs {
		m := matchCandidate(name, candidates)
		jobs = append(jobs, aiActionJob{Name: name, Matched: m != nil, Source: sourceOf(m)})
	}
	action := &aiReleaseAction{Type: "release", Jobs: jobs, Branch: ruleRes.Branch, Env: ruleRes.Env}
	return fmt.Sprintf("识别到 %d 个服务%s，确认后开始发布。", len(jobs), branchSuffix(ruleRes.Branch)), action
}

// analyzeFailure 发布失败时用 LLM 分析日志尾部（opsx-loki AnalyzeJenkins 同款 prompt）
func analyzeFailure(run model.PipelineRun) {
	client := llmClientFromSettings()
	if client == nil {
		return
	}
	var logs []model.LogEntry
	database.DB.Where("run_id = ?", run.ID).Order("timestamp DESC").Limit(100).Find(&logs)
	if len(logs) == 0 {
		return
	}
	// 倒序取出后反转为正序
	for i, j := 0, len(logs)-1; i < j; i, j = i+1, j-1 {
		logs[i], logs[j] = logs[j], logs[i]
	}
	var b strings.Builder
	for _, l := range logs {
		b.WriteString(l.Message)
		b.WriteString("\n")
	}
	system := `你是资深发布助手。根据流水线执行日志分析发布失败原因。用简洁中文输出，结构固定：
1. 结论：一句话说明失败原因
2. 失败阶段或关键步骤
3. 根因（如编译错误、依赖缺失、测试失败、镜像推送失败、部署超时）
4. 处理建议（1-3 条，可执行）
不要复述整段日志。`
	user := fmt.Sprintf("流水线: %s\n分支: %s\n状态: %s\n\n日志尾部:\n%s", run.PipelineName, run.Branch, run.Status, b.String())

	ctx, cancel := context.WithTimeout(context.Background(), 90*time.Second)
	defer cancel()
	reply, err := client.Chat(ctx, system, user)
	if err != nil {
		return
	}
	database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("analysis", reply)
}