aiops/backend/controller/pipeline_executor.go

package controller

import (
	"context"
	"encoding/json"
	"fmt"
	"os"
	"os/exec"
	"strings"
	"sync"
	"time"

	"ezops/model"

	"go.uber.org/zap"
)

// ==================== 流水线执行引擎 V2（Redis 实时日志推送）====================

// LogEntry 日志条目
type LogEntry struct {
	Time    string `json:"time"`
	Level   string `json:"level"`
	Stage   string `json:"stage"`
	Message string `json:"message"`
}

// StageStep 单个阶段定义
type StageStep struct {
	Name   string `json:"name"`
	Type   string `json:"type"` // checkout/build/test/image/deploy/verify
	Status string `json:"status"`
}

var logger *zap.Logger

func init() {
	logger, _ = zap.NewProduction()
}

// executePipeline 后台执行流水线（goroutine）— V2 Redis 版
func (h *PipelineHandler) executePipeline(pipeline model.Pipeline, run model.PipelineRun) {
	runID := run.ID
	ctx := context.Background()
	logKey := fmt.Sprintf("pipeline:run:%d:logs", runID)
	pubCh := fmt.Sprintf("pipeline:run:%d", runID)

	// 日志函数：同时写 Redis buffer + 发布 Pub/Sub
	log := func(level, stage, msg string) {
		entry := LogEntry{Time: time.Now().Format("15:04:05"), Level: level, Stage: stage, Message: msg}
		data, _ := json.Marshal(entry)

		// 追加到 Redis 列表
		if h.cache != nil {
			existing, _ := h.cache.Get(ctx, logKey)
			var entries []LogEntry
			if existing != "" {
				json.Unmarshal([]byte(existing), &entries)
			}
			entries = append(entries, entry)
			buf, _ := json.Marshal(entries)
			h.cache.Set(ctx, logKey, string(buf), 30*time.Minute)
			// Pub/Sub 实时推送
			h.cache.Publish(ctx, pubCh, string(data))
		}
	}

	updateStatus := func(status string, duration int) {
		now := time.Now()
		updates := map[string]interface{}{"status": status}
		if duration > 0 {
			updates["duration"] = duration
		}
		if status == "success" || status == "failed" || status == "cancelled" {
			updates["finished_at"] = &now
		}
		h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Updates(updates)
		// 推送状态变更
		if h.cache != nil {
			statusData, _ := json.Marshal(map[string]interface{}{"type": "status", "status": status, "duration": duration})
			h.cache.Publish(ctx, pubCh, string(statusData))
		}
	}

	updatePipelineStats := func(status string) {
		ups := map[string]interface{}{"last_status": status, "last_run_at": time.Now(), "run_count": pipeline.RunCount + 1}
		if status == "success" {
			ups["success_count"] = pipeline.SuccessCount + 1
		}
		if status == "failed" {
			ups["fail_count"] = pipeline.FailCount + 1
		}
		h.db.Model(&model.Pipeline{}).Where("id = ?", pipeline.ID).Updates(ups)
	}

	isCancelled := func() bool {
		var current model.PipelineRun
		h.db.First(&current, runID)
		return current.Status == "cancelled"
	}

	// 解析阶段
	var stages []StageStep
	if pipeline.Stages != "" {
		json.Unmarshal([]byte(pipeline.Stages), &stages)
	}
	if len(stages) == 0 {
		log("warn", "init", "没有配置阶段，流水线结束")
		log("success", "完成", "流水线执行成功 (0s)")
		updateStatus("success", 0)
		updatePipelineStats("success")
		h.dequeueNext(pipeline.ID)
		h.flushLogsToDB(ctx, runID, logKey)
		return
	}

	// 工作目录 — 按仓库名+分支+时间戳命名，方便区分不同构建
	branch := run.Branch
	if branch == "" {
		branch = pipeline.Branch
	}
	gitURL := pipeline.Source
	// 从 Git URL 提取仓库名 (e.g. https://gitee.com/attacker/demo-jar.git → demo-jar)
	repoName := extractRepoName(gitURL)
	if repoName == "" {
		repoName = pipeline.Code
	}
	ts := time.Now()
	tsDir := fmt.Sprintf("%d%02d%02d%02d%02d", ts.Year(), ts.Month(), ts.Day(), ts.Hour(), ts.Minute())
	workspace := fmt.Sprintf("/app/workspace/%s-%s", repoName, pipeline.Code)

	os.RemoveAll(workspace)
	os.MkdirAll(workspace, 0755)
	// 保存 workspace 路径到运行记录，方便终端查看
	h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Update("workspace_path", workspace)

	imageTag := fmt.Sprintf("%s-%s", repoName, tsDir)

	log("info", "init", fmt.Sprintf("触发方式: 手动 (admin)"))
	log("info", "init", fmt.Sprintf("分支: %s | 镜像标签: %s", branch, imageTag))

	totalStart := time.Now()

	for _, stage := range stages {
		if isCancelled() {
			log("warn", stage.Name, "流水线已被停止")
			updateStatus("cancelled", int(time.Since(totalStart).Seconds()))
			updatePipelineStats("cancelled")
			h.dequeueNext(pipeline.ID)
			h.flushLogsToDB(ctx, runID, logKey)
			return
		}

		log("info", stage.Name, fmt.Sprintf("===== 阶段: %s =====", stage.Name))
		stageStart := time.Now()

		var err error
		switch stage.Type {
		case "checkout":
			err = h.runCheckout(gitURL, branch, workspace, log)
		case "build":
			err = h.runBuild(workspace, pipeline.Dockerfile, log)
		case "test":
			err = h.runTest(workspace, log)
		case "image":
			err = h.runDockerBuild(workspace, pipeline, imageTag, log)
		case "deploy":
			err = h.runDeploy(workspace, pipeline, imageTag, log)
		case "verify":
			err = h.runVerify(workspace, log)
		default:
			log("info", stage.Name, "跳过未知阶段类型: "+stage.Type)
		}

		stageDur := time.Since(stageStart)
		if err != nil {
			log("error", stage.Name, fmt.Sprintf("✗ %s 失败 (%v): %v", stage.Name, stageDur.Round(time.Second), err))
			updateStatus("failed", int(time.Since(totalStart).Seconds()))
			updatePipelineStats("failed")
			h.dequeueNext(pipeline.ID)
			h.flushLogsToDB(ctx, runID, logKey)
			return
		}
		log("success", stage.Name, fmt.Sprintf("✓ %s 完成 (%v)", stage.Name, stageDur.Round(time.Second)))
	}

	totalDur := int(time.Since(totalStart).Seconds())
	updateStatus("success", totalDur)
	log("success", "完成", fmt.Sprintf("流水线执行成功 (%ds)", totalDur))

	updatePipelineStats("success")
	h.dequeueNext(pipeline.ID)
	h.flushLogsToDB(ctx, runID, logKey)
}

// flushLogsToDB Redis 日志批量写入 MySQL
func (h *PipelineHandler) flushLogsToDB(ctx context.Context, runID uint, logKey string) {
	if h.cache == nil {
		return
	}
	raw, err := h.cache.Get(ctx, logKey)
	if err != nil || raw == "" {
		return
	}
	var entries []LogEntry
	if err := json.Unmarshal([]byte(raw), &entries); err != nil {
		return
	}
	logsJSON, _ := json.Marshal(entries)
	h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Update("logs", string(logsJSON))
	// 清理 Redis 缓存
	h.cache.Del(ctx, logKey)
}

// runCheckout git clone
func (h *PipelineHandler) runCheckout(url, branch, workspace string, logFn func(string, string, string)) error {
	if url == "" {
		logFn("warn", "代码检出", "没有配置 Git URL，跳过代码检出")
		return nil
	}
	logFn("info", "代码检出", "> Git 拉取 "+url+" (分支: "+branch+")")

	cmd := exec.Command("git", "clone", "--depth", "1", "-b", branch, url, workspace)
	output, err := cmd.CombinedOutput()
	// 流式输出到日志
	for _, line := range strings.Split(strings.TrimSpace(string(output)), "\n") {
		if line != "" {
			logFn("info", "代码检出", line)
		}
	}
	if err != nil {
		return fmt.Errorf("git clone 失败: %v", err)
	}
	hashCmd := exec.Command("git", "-C", workspace, "rev-parse", "--short", "HEAD")
	hashOut, _ := hashCmd.Output()
	logFn("success", "代码检出", fmt.Sprintf("已拉取 commit: %s", strings.TrimSpace(string(hashOut))))
	return nil
}

// runBuild 构建（自动检测 Maven / npm / Go / Makefile）
func (h *PipelineHandler) runBuild(workspace, dockerfile string, logFn func(string, string, string)) error {
	buildDir := workspace
	if dockerfile != "" {
		if dir := dotdot(dockerfile); dir != "." && dir != "" {
			testDir := workspace + "/" + dir
			if _, err := os.Stat(testDir); err == nil {
				buildDir = testDir
				logFn("info", "构建", ">>> 切换到构建目录: "+dir)
			}
		}
	}

	// 如果根目录没找到构建文件，扫描一级子目录
	if !hasBuildFile(buildDir) {
		type subMatch struct {
			dir  string
			kind string // "maven", "npm", "go", "make"
		}
		var matches []subMatch
		entries, err := os.ReadDir(buildDir)
		if err == nil {
			for _, e := range entries {
				if !e.IsDir() || e.Name() == ".git" {
					continue
				}
				sub := buildDir + "/" + e.Name()
				if _, err := os.Stat(sub + "/pom.xml"); err == nil {
					matches = append(matches, subMatch{sub, "maven"})
				}
				if _, err := os.Stat(sub + "/package.json"); err == nil {
					matches = append(matches, subMatch{sub, "npm"})
				}
				if _, err := os.Stat(sub + "/go.mod"); err == nil {
					matches = append(matches, subMatch{sub, "go"})
				}
				if _, err := os.Stat(sub + "/Makefile"); err == nil {
					matches = append(matches, subMatch{sub, "make"})
				}
			}
		}
		if len(matches) > 0 {
			// 有多个匹配时列出，优先选择第一个
			if len(matches) > 1 {
				names := make([]string, len(matches))
				for i, m := range matches {
					names[i] = fmt.Sprintf("%s(%s)", m.dir[len(buildDir)+1:], m.kind)
				}
				logFn("info", "构建", ">>> 发现多个构建目录: "+strings.Join(names, ", "))
			}
			buildDir = matches[0].dir
			logFn("info", "构建", ">>> 自动检测到构建目录: "+buildDir[len(workspace)+1:])
		} else {
			logFn("info", "构建", "未检测到构建文件 (pom.xml/go.mod/package.json)，跳过构建")
			return nil
		}
	}

	if _, err := os.Stat(buildDir + "/pom.xml"); err == nil {
		logFn("info", "构建", ">>> Maven 构建 (mvn clean package -DskipTests)")
		logFn("info", "构建", "🔨 正在下载依赖并编译...")
		cmd := exec.Command("mvn", "clean", "package", "-DskipTests")
		cmd.Dir = buildDir
		out, err := cmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "构建", line)
			}
		}
		if err != nil {
			return fmt.Errorf("Maven 构建失败: %v", err)
		}
		logFn("success", "构建", "Maven 构建成功 ✅")
		return nil
	}

	if _, err := os.Stat(buildDir + "/package.json"); err == nil {
		logFn("info", "构建", ">>> npm 构建 (npm install && npm run build)")
		logFn("info", "构建", "📦 正在安装依赖...")
		installCmd := exec.Command("sh", "-c", "npm install --registry=https://registry.npmmirror.com --legacy-peer-deps 2>&1")
		installCmd.Dir = buildDir
		out, err := installCmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "构建", line)
			}
		}
		if err != nil {
			return fmt.Errorf("npm install 失败: %v", err)
		}
		logFn("info", "构建", "🏗️ 正在打包构建...")
		buildCmd := exec.Command("sh", "-c", "npm run build 2>&1")
		buildCmd.Dir = buildDir
		out, err = buildCmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "构建", line)
			}
		}
		if err != nil {
			return fmt.Errorf("npm build 失败: %v", err)
		}
		logFn("success", "构建", "npm 构建成功 ✅")
		return nil
	}

	if _, err := os.Stat(buildDir + "/go.mod"); err == nil {
		logFn("info", "构建", ">>> Go 构建 (go build ./...)")
		cmd := exec.Command("go", "build", "./...")
		cmd.Dir = buildDir
		out, err := cmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "构建", line)
			}
		}
		if err != nil {
			return fmt.Errorf("Go 构建失败: %v", err)
		}
		logFn("success", "构建", "Go 构建成功")
		return nil
	}

	logFn("info", "构建", "未检测到构建文件 (pom.xml/go.mod/package.json)，跳过构建")
	return nil
}

// hasBuildFile 检查目录是否包含可识别的构建文件
func hasBuildFile(dir string) bool {
	for _, f := range []string{"pom.xml", "package.json", "go.mod", "Makefile"} {
		if _, err := os.Stat(dir + "/" + f); err == nil {
			return true
		}
	}
	return false
}

// runTest 执行测试
func (h *PipelineHandler) runTest(workspace string, logFn func(string, string, string)) error {
	if _, err := os.Stat(workspace + "/pom.xml"); err == nil {
		logFn("info", "测试", ">>> Maven 测试 (mvn test)")
		cmd := exec.Command("mvn", "test")
		cmd.Dir = workspace
		out, _ := cmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "测试", line)
			}
		}
	}
	return nil
}

// runDockerBuild Docker 构建 & 推送
func (h *PipelineHandler) runDockerBuild(workspace string, pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
	dockerfile := pipeline.Dockerfile
	if dockerfile == "" {
		dockerfile = "Dockerfile"
	}

	var registryURL, imageName string
	if pipeline.RegistryID != nil {
		var reg model.PipelineRegistry
		if err := h.db.First(&reg, *pipeline.RegistryID).Error; err == nil {
			registryURL = reg.RegistryURL
			if reg.Namespace != "" {
				imageName = reg.Namespace + "/" + pipeline.Code
			}
		}
	}
	if imageName == "" {
		imageName = pipeline.Code
	}
	if registryURL == "" {
		registryURL = "registry.example.com"
	}

	fullImage := fmt.Sprintf("%s/%s:%s", registryURL, imageName, tag)
	logFn("info", "镜像打包", fmt.Sprintf("镜像地址: %s", fullImage))

	buildDir := workspace
	if dir := dotdot(dockerfile); dir != "." && dir != "" {
		buildDir = workspace + "/" + dir
	}
	dfPath := dockerfile
	if strings.Contains(dockerfile, "/") {
		dfPath = dockerfile[strings.LastIndex(dockerfile, "/")+1:]
	}

	logFn("info", "镜像打包", fmt.Sprintf(">>> Docker 构建 (dockerfile: %s)", dfPath))
	cmd := exec.Command("docker", "build", "-t", fullImage, "-f", dfPath, ".")
	cmd.Dir = buildDir
	out, err := cmd.CombinedOutput()
	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
		if line != "" {
			logFn("info", "镜像打包", line)
		}
	}
	if err != nil {
		return fmt.Errorf("Docker 构建失败: %v", err)
	}
	logFn("success", "镜像打包", "Docker 构建成功")

	logFn("info", "镜像打包", ">>> 推送镜像: "+fullImage)
	pushCmd := exec.Command("docker", "push", fullImage)
	pushOut, err := pushCmd.CombinedOutput()
	for _, line := range strings.Split(strings.TrimSpace(string(pushOut)), "\n") {
		if line != "" {
			logFn("info", "镜像打包", line)
		}
	}
	if err != nil {
		return fmt.Errorf("Docker 推送失败: %v", err)
	}
	logFn("success", "镜像打包", "镜像推送完成")
	return nil
}

// runDeploy 部署到 K8s / 服务器
func (h *PipelineHandler) runDeploy(workspace string, pipeline model.Pipeline, tag string, logFn func(string, string, string)) error {
	deployYamls := findFiles(workspace, ".yaml", ".yml")
	var k8sFiles []string
	for _, f := range deployYamls {
		content, _ := os.ReadFile(f)
		if strings.Contains(string(content), "kind:") &&
			(strings.Contains(string(content), "Deployment") ||
				strings.Contains(string(content), "Service") ||
				strings.Contains(string(content), "StatefulSet")) {
			k8sFiles = append(k8sFiles, f)
		}
	}

	if len(k8sFiles) == 0 {
		logFn("info", "部署", "未找到 K8s 部署文件，跳过部署")
		return nil
	}

	for _, yamlFile := range k8sFiles {
		logFn("info", "部署", ">>> kubectl apply -f "+yamlFile)
		cmd := exec.Command("kubectl", "apply", "-f", yamlFile)
		out, err := cmd.CombinedOutput()
		for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
			if line != "" {
				logFn("info", "部署", line)
			}
		}
		if err != nil {
			logFn("error", "部署", fmt.Sprintf("kubectl apply 失败: %v", err))
			return fmt.Errorf("kubectl apply 失败")
		}
	}
	logFn("success", "部署", "部署完成")
	return nil
}

// runVerify 部署后验证
func (h *PipelineHandler) runVerify(workspace string, logFn func(string, string, string)) error {
	logFn("info", "验证", ">>> 部署后健康检查")
	cmd := exec.Command("kubectl", "get", "pods", "-o", "wide")
	cmd.Dir = workspace
	out, err := cmd.CombinedOutput()
	if err != nil {
		logFn("warn", "验证", fmt.Sprintf("kubectl get pods 失败: %v", err))
	}
	for _, line := range strings.Split(strings.TrimSpace(string(out)), "\n") {
		if line != "" {
			logFn("info", "验证", line)
		}
	}
	return nil
}

// ======== helpers ========

func dotdot(path string) string {
	i := strings.LastIndex(path, "/")
	if i < 0 {
		return "."
	}
	return path[:i]
}

// extractRepoName 从 Git URL 提取仓库名
// e.g. https://gitee.com/attacker/demo-jar.git → demo-jar
func extractRepoName(url string) string {
	// Remove trailing .git
	clean := strings.TrimSuffix(url, ".git")
	// Get last path segment
	idx := strings.LastIndex(clean, "/")
	if idx < 0 {
		return ""
	}
	return clean[idx+1:]
}

// cleanOldWorkspaces 清理超过 maxAge 的 workspace 目录
func cleanOldWorkspaces(baseDir string, maxAge time.Duration) {
	entries, err := os.ReadDir(baseDir)
	if err != nil {
		return
	}
	now := time.Now()
	for _, e := range entries {
		if !e.IsDir() {
			continue
		}
		info, err := e.Info()
		if err != nil {
			continue
		}
		if now.Sub(info.ModTime()) > maxAge {
			os.RemoveAll(baseDir + "/" + e.Name())
		}
	}
}

// ==================== 流水线队列引擎（Jenkins 风格并发控制）====================

var (
	pipelineRunningCount = make(map[uint]int)    // pipelineID → 当前运行数
	pipelineQueues       = make(map[uint][]uint)  // pipelineID → 排队 runID 列表
	queueMu              sync.Mutex
)

// tryExecuteOrQueue 触发流水线时检查并发限制
// 如果当前运行数 < maxExecutors → 立即执行
// 否则 → 加入队列 (status=queued)
func (h *PipelineHandler) tryExecuteOrQueue(pipeline model.Pipeline, run model.PipelineRun) {
	max := pipeline.MaxExecutors
	if max <= 0 { max = 2 }

	queueMu.Lock()
	running := pipelineRunningCount[pipeline.ID]
	if running < max {
		pipelineRunningCount[pipeline.ID] = running + 1
		queueMu.Unlock()
		// 立即执行
		h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "running")
		go h.executePipeline(pipeline, run)
		return
	}
	// 加入队列
	pipelineQueues[pipeline.ID] = append(pipelineQueues[pipeline.ID], run.ID)
	queueMu.Unlock()
	h.db.Model(&model.PipelineRun{}).Where("id = ?", run.ID).Update("status", "queued")
}

// dequeueNext 流水线运行结束后，从队列中取下一个执行
func (h *PipelineHandler) dequeueNext(pipelineID uint) {
	queueMu.Lock()
	running := pipelineRunningCount[pipelineID]
	if running > 0 {
		pipelineRunningCount[pipelineID] = running - 1
	}

	queue := pipelineQueues[pipelineID]
	if len(queue) == 0 {
		queueMu.Unlock()
		return
	}
	nextRunID := queue[0]
	pipelineQueues[pipelineID] = queue[1:]
	pipelineRunningCount[pipelineID] = pipelineRunningCount[pipelineID] + 1
	queueMu.Unlock()

	// 加载 pipeline + run
	var pipeline model.Pipeline
	if err := h.db.First(&pipeline, pipelineID).Error; err != nil { return }
	var run model.PipelineRun
	if err := h.db.First(&run, nextRunID).Error; err != nil { return }

	h.db.Model(&run).Update("status", "running")
	go h.executePipeline(pipeline, run)
}

// GetQueueStatus 查询流水线队列状态（供 API 使用）
func (h *PipelineHandler) GetQueueStatus(pipelineID uint) map[string]interface{} {
	queueMu.Lock()
	defer queueMu.Unlock()
	running := pipelineRunningCount[pipelineID]
	queued := pipelineQueues[pipelineID]
	return map[string]interface{}{
		"running": running,
		"queued":  len(queued),
		"queue_ids": queued,
	}
}

// CancelQueued 取消排队中的构建
func (h *PipelineHandler) CancelQueued(pipelineID uint, runID uint) error {
	queueMu.Lock()
	defer queueMu.Unlock()
	queue := pipelineQueues[pipelineID]
	for i, id := range queue {
		if id == runID {
			pipelineQueues[pipelineID] = append(queue[:i], queue[i+1:]...)
			h.db.Model(&model.PipelineRun{}).Where("id = ?", runID).Update("status", "cancelled")
			return nil
		}
	}
	return fmt.Errorf("run %d not in queue", runID)
}

func findFiles(dir string, exts ...string) []string {
	var result []string
	entries, err := os.ReadDir(dir)
	if err != nil {
		return result
	}
	for _, e := range entries {
		full := dir + "/" + e.Name()
		if e.IsDir() {
			if e.Name() != ".git" {
				result = append(result, findFiles(full, exts...)...)
			}
		} else {
			for _, ext := range exts {
				if strings.HasSuffix(e.Name(), ext) {
					result = append(result, full)
					break
				}
			}
		}
	}
	return result
}