FlowStack/backend/internal/handler/runs.go

package handler

import (
	"encoding/json"
	"fmt"
	"net/http"
	"sort"
	"time"

	"github.com/gin-gonic/gin"

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

// CreateRun POST /api/pipelines/:id/runs —— 创建运行并启动执行引擎
func CreateRun(hub *Hub) gin.HandlerFunc {
	return func(c *gin.Context) {
		var p model.Pipeline
		if err := database.DB.First(&p, c.Param("id")).Error; err != nil {
			c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"})
			return
		}
		var body struct {
			Branch      string `json:"branch"`
			Env         string `json:"env"`
			TriggerBy   string `json:"triggerBy"`
			TriggerType string `json:"triggerType"`
		}
		_ = c.ShouldBindJSON(&body)

		run, err := triggerPipelineRun(hub, p, body.Branch, body.Env, body.TriggerBy, body.TriggerType)
		if err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}
		c.JSON(http.StatusOK, run)
	}
}

// triggerPipelineRun 统一的运行创建入口：校验节点 → 建 Run/Stages → 启动执行引擎
func triggerPipelineRun(hub *Hub, p model.Pipeline, branch, env, triggerBy, triggerType string) (*model.PipelineRun, error) {
	if branch == "" {
		branch = p.SourceBranch
	}
	if env == "" {
		env = p.Env
	}
	if triggerBy == "" {
		triggerBy = "admin"
	}
	if triggerType == "" {
		triggerType = "manual"
	}

	var nodes []model.PipelineNode
	database.DB.Where("pipeline_id = ?", p.ID).Find(&nodes)
	if len(nodes) == 0 {
		return nil, fmt.Errorf("流水线没有节点，请先在画布中编排并保存")
	}
	var edges []model.PipelineEdge
	database.DB.Where("pipeline_id = ?", p.ID).Find(&edges)
	nodesJSON, _ := json.Marshal(nodes)
	edgesJSON, _ := json.Marshal(edges)

	run := model.PipelineRun{
		RunID: fmt.Sprintf("pr-%d", time.Now().UnixMilli()),
		PipelineID: p.ID, PipelineName: p.Name,
		Branch: branch, Env: env, DeployType: p.DeployType, Type: p.Type,
		TriggerBy: triggerBy, TriggerType: triggerType,
		Status: "running", StartedAt: time.Now(),
		NodesSnapshot: string(nodesJSON), EdgesSnapshot: string(edgesJSON),
	}
	if err := database.DB.Create(&run).Error; err != nil {
		return nil, err
	}

	// jenkins 类型流水线走真实 Jenkins 执行引擎
	if p.Type == "jenkins" && p.JenkinsJobName != "" {
		var server model.JenkinsServer
		if err := database.DB.First(&server, p.JenkinsServerID).Error; err == nil && server.Enabled {
			client := clientFromServer(server)
			go jenkinsRunEngine(hub, run, client, p.JenkinsJobName)
			return &run, nil
		}
	}

	// 按画布 X 坐标从左到右生成执行阶段
	sort.Slice(nodes, func(i, j int) bool { return nodes[i].PositionX < nodes[j].PositionX })
	stages := make([]model.RunStage, len(nodes))
	for i, n := range nodes {
		stages[i] = model.RunStage{RunID: run.ID, NodeID: n.NodeID, Name: n.Name, Status: "pending"}
		database.DB.Create(&stages[i])
	}

	go simulateRun(hub, run, stages)
	return &run, nil
}

// simulateRun 模拟执行：逐阶段 pending→running→success，逐条写日志并 SSE 推送
func simulateRun(hub *Hub, run model.PipelineRun, stages []model.RunStage) {
	start := time.Now()
	emit := func(level, msg string, stageID uint) {
		entry := model.LogEntry{RunID: run.ID, StageID: stageID, Level: level, Message: msg, Source: "runner", Timestamp: time.Now()}
		database.DB.Create(&entry)
		hub.Publish(run.RunID, "log", entry)
	}
	setStage := func(stage *model.RunStage, status string) {
		now := time.Now()
		updates := map[string]any{"status": status}
		if status == "running" {
			stage.StartedAt = &now
			updates["started_at"] = now
		} else {
			stage.CompletedAt = &now
			updates["completed_at"] = now
			exit := 0
			stage.ExitCode = &exit
			updates["exit_code"] = exit
		}
		database.DB.Model(&model.RunStage{}).Where("id = ?", stage.ID).Updates(updates)
		stage.Status = status
		hub.Publish(run.RunID, "stage", gin.H{"stageId": stage.ID, "nodeId": stage.NodeID, "name": stage.Name, "status": status})
	}
	setRun := func(status string) {
		now := time.Now()
		updates := map[string]any{"status": status, "duration_ms": time.Since(start).Milliseconds(), "completed_at": now}
		database.DB.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Updates(updates)
		hub.Publish(run.RunID, "run", gin.H{"runId": run.RunID, "status": status, "durationMs": time.Since(start).Milliseconds()})
		// 通知订阅了 pipeline.success / pipeline.failed 的出站 Webhook + 飞书发布卡片
		if status == "success" || status == "failed" {
			NotifyWebhookEvent("pipeline."+status, map[string]any{
				"runId": run.RunID, "pipeline": run.PipelineName,
				"branch": run.Branch, "durationMs": time.Since(start).Milliseconds(),
			})
		}
		go notifyLarkRelease(run, status)
	}

	emit("info", "Pipeline 触发成功，开始调度执行引擎…", 0)
	for i := range stages {
		st := &stages[i]
		setStage(st, "running")
		for _, line := range stageScript(st) {
			emit("info", line, st.ID)
			time.Sleep(500 * time.Millisecond)
		}
		setStage(st, "success")
	}
	setRun("success")
}

// stageScript 按节点类别生成贴近真实构建过程的日志
func stageScript(st *model.RunStage) []string {
	switch st.Name {
	case "git-clone":
		return []string{
			fmt.Sprintf("$ git clone --depth=50 --branch=master %s", "https://gitee.com/attacker/aiops.git"),
			"Cloning into 'source'...",
			"HEAD is now at 8f3c21a feat: release v2.4.0",
		}
	case "unit-test":
		return []string{
			"$ go test -v -race ./...",
			"ok  \tflowstack/internal/handler\t2.318s",
			"ok  \tflowstack/internal/model\t0.402s",
			"coverage: 87.4% of statements",
		}
	case "kaniko-build":
		return []string{
			"$ kaniko --dockerfile Dockerfile --destination registry.example.com/aiops/aiops-backend:" + time.Now().Format("200601021504"),
			"INFO[0002] Retrieving image manifest golang:1.24-alpine",
			"INFO[0018] Executing 12 build steps",
			"INFO[0041] Pushing image to registry.example.com",
		}
	case "trivy-scan":
		return []string{
			"$ trivy image --severity HIGH,CRITICAL registry.example.com/aiops/aiops-backend:latest",
			"alpine (apk): 0 HIGH, 0 CRITICAL",
			"Scan completed: no critical vulnerabilities found",
		}
	case "kubectl-deploy":
		return []string{
			"$ kubectl set image deployment/aiops-backend app=registry.example.com/aiops/aiops-backend:" + time.Now().Format("200601021504") + " -n aiops",
			"deployment.apps/aiops-backend image updated",
			"Waiting for rollout: 3/3 updated",
			"deployment \"aiops-backend\" successfully rolled out",
		}
	case "npm build":
		return []string{
			"$ npm ci --registry=https://registry.npmmirror.com",
			"$ npm run build",
			"vite v7.x building for production...",
			"✓ built in 18.4s",
		}
	case "npm audit":
		return []string{
			"$ npm audit --audit-level=high",
			"found 0 vulnerabilities",
		}
	case "主机部署":
		return []string{
			"$ tar zcf dist.tar.gz dist/ && scp dist.tar.gz production-group:/home/admin/app/",
			"$ rsync -avz --delete /tmp/dist/ /usr/share/nginx/html/",
			"$ nginx -t && nginx -s reload",
			"2/2 主机部署完成",
		}
	case "刷新 CDN":
		return []string{
			"$ curl -X POST https://cdn.aliyuncs.com/api/refresh",
			`{"urls":["https://aiops.example.com/*"]}`,
			"RefreshTaskId: 202610271200****",
		}
	default:
		return []string{
			fmt.Sprintf("[%s] 开始执行…", st.Name),
			fmt.Sprintf("[%s] 执行完成 (exit 0)", st.Name),
		}
	}
}

// ListRuns GET /api/runs?pipelineId=&status=
func ListRuns(c *gin.Context) {
	q := database.DB.Model(&model.PipelineRun{}).Order("started_at DESC").Limit(100)
	if v := c.Query("pipelineId"); v != "" {
		q = q.Where("pipeline_id = ?", v)
	}
	if v := c.Query("status"); v != "" && v != "all" {
		q = q.Where("status = ?", v)
	}
	var runs []model.PipelineRun
	q.Find(&runs)
	c.JSON(http.StatusOK, runs)
}

// GetRun GET /api/runs/:runId
func GetRun(c *gin.Context) {
	var run model.PipelineRun
	if err := database.DB.Where("run_id = ?", c.Param("runId")).First(&run).Error; err != nil {
		c.JSON(http.StatusNotFound, gin.H{"error": "run not found"})
		return
	}
	var stages []model.RunStage
	database.DB.Where("run_id = ?", run.ID).Order("id").Find(&stages)
	var logs []model.LogEntry
	database.DB.Where("run_id = ?", run.ID).Order("timestamp").Find(&logs)
	c.JSON(http.StatusOK, gin.H{"run": run, "stages": stages, "logs": logs})
}

// StreamLogs GET /api/runs/:runId/logs/stream —— SSE：先回放历史日志再实时推送
func StreamLogs(hub *Hub) gin.HandlerFunc {
	return func(c *gin.Context) {
		runId := c.Param("runId")
		var run model.PipelineRun
		if err := database.DB.Where("run_id = ?", runId).First(&run).Error; err != nil {
			c.JSON(http.StatusNotFound, gin.H{"error": "run not found"})
			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() } }

		fmt.Fprintf(c.Writer, "event: connected\ndata: {\"runId\":\"%s\"}\n\n", runId)
		flush()

		// 回放历史日志
		var history []model.LogEntry
		database.DB.Where("run_id = ?", run.ID).Order("timestamp").Find(&history)
		for _, entry := range history {
			data, _ := json.Marshal(entry)
			fmt.Fprintf(c.Writer, "event: log\ndata: %s\n\n", data)
		}
		if len(history) > 0 {
			flush()
		}

		ch := hub.Subscribe(runId)
		defer hub.Unsubscribe(runId, ch)

		heartbeat := time.NewTicker(15 * time.Second)
		defer heartbeat.Stop()

		ctx := c.Request.Context()
		for {
			select {
			case <-ctx.Done():
				return
			case <-heartbeat.C:
				fmt.Fprint(c.Writer, ": ping\n\n")
				flush()
			case frame := <-ch:
				c.Writer.Write(frame)
				flush()
			}
		}
	}
}

// RetryRun POST /api/runs/:runId/retry —— 以同一流水线与分支重新触发
func RetryRun(hub *Hub) gin.HandlerFunc {
	return func(c *gin.Context) {
		var prev model.PipelineRun
		if err := database.DB.Where("run_id = ?", c.Param("runId")).First(&prev).Error; err != nil {
			c.JSON(http.StatusNotFound, gin.H{"error": "run not found"})
			return
		}
		var p model.Pipeline
		if err := database.DB.First(&p, prev.PipelineID).Error; err != nil {
			c.JSON(http.StatusNotFound, gin.H{"error": "pipeline not found"})
			return
		}
		run, err := triggerPipelineRun(hub, p, prev.Branch, prev.Env, "admin", "manual")
		if err != nil {
			c.JSON(http.StatusBadRequest, gin.H{"error": err.Error()})
			return
		}
		c.JSON(http.StatusOK, run)
	}
}