aiops/backend/internal/api/handlers/pipeline.go

package handlers

import (
	"net/http"
	"regexp"
	"time"

	"ezops/internal/models"
	"ezops/internal/api/middleware"
	ezredis "ezops/pkg/redis"

	"github.com/gin-gonic/gin"
	"gorm.io/gorm"
)

// getPipelineID 从 context 获取中间件注入的流水线 ID
func getPipelineID(c *gin.Context) uint {
	if v, exists := c.Get("pipelineID"); exists {
		if id, ok := v.(uint); ok {
			return id
		}
	}
	return 0
}

// codePattern 流水线编码规则：字母+数字+连字符，不能以连字符开头/结尾
var codePattern = regexp.MustCompile(`^[a-zA-Z0-9]([a-zA-Z0-9-]*[a-zA-Z0-9])?$`)

// validateCode 校验流水线编码格式并查重
func validateCode(db *gorm.DB, code string, excludeID uint) string {
	if code == "" {
		return "流水线编码不能为空"
	}
	if len(code) > 64 {
		return "编码长度不能超过 64 字符"
	}
	if !codePattern.MatchString(code) {
		return "编码格式不正确：只能使用字母、数字、连字符(-)，不能以连字符开头或结尾\n示例: My-Pipeline, demo-jar-backend"
	}
	// 查重
	var count int64
	q := db.Model(&models.Pipeline{}).Where("code = ?", code)
	if excludeID > 0 {
		q = q.Where("id != ?", excludeID)
	}
	q.Count(&count)
	if count > 0 {
		return "编码已存在，请使用其他编码"
	}
	return ""
}

// PipelineHandler 流水线管理器
type PipelineHandler struct {
	db    *gorm.DB
	cache ezredis.Cache
}

// NewPipelineHandler 创建流水线处理器
func NewPipelineHandler(db *gorm.DB, cache ezredis.Cache) *PipelineHandler {
	return &PipelineHandler{db: db, cache: cache}
}

// PipelineListItem 列表项（含最近一次运行摘要）// Logan.Li
type PipelineListItem struct {
	models.Pipeline
	LastRunStages   string `json:"last_run_stages,omitempty"`   // JSON: [{name,status}]
	LastImageTag    string `json:"last_image_tag,omitempty"`
	LastCommitID    string `json:"last_commit_id,omitempty"`
	LastRunDuration int    `json:"last_duration,omitempty"`
	LastRunBranch   string `json:"last_run_branch,omitempty"`
}

// ListPipelines 获取流水线列表
// @Summary 获取流水线列表
// @Description 获取所有流水线列表
// @Tags 流水线
// @Produce json
// @Security Bearer
// @Param page query int false "页码"
// @Param page_size query int false "每页数量"
// @Success 200 {object} models.Response{data=models.ListResponse}
// ListPipelines 获取流水线列表
func (h *PipelineHandler) ListPipelines(c *gin.Context) {
	var query models.PageQuery
	c.ShouldBindQuery(&query)

	// 管理员看全部，非管理员只看自己有权限的
	isAdmin, _ := c.Get("isAdmin")
	username, _ := c.Get("username")
	usernameStr, _ := username.(string)
	userID, _ := c.Get("userID")
	uid, _ := userID.(uint)

	db := h.db.Model(&models.Pipeline{})
	if isAdmin != true && usernameStr != "" {
		ids, hasAccess := middleware.GetAccessiblePipelineIDs(h.db, usernameStr, uid)
		if !hasAccess {
			// 没有任何流水线权限 → 返回空列表
			c.JSON(http.StatusOK, models.Response{Code: 200, Data: models.ListResponse{List: []models.Pipeline{}, Total: 0}})
			return
		}
		db = db.Where("id IN ?", ids)
	}

	var total int64
	db.Count(&total)

	var pipelines []models.Pipeline
	db.Preload("Environment").Offset(query.GetOffset()).Limit(query.GetLimit()).Order("updated_at DESC").Find(&pipelines)

	// 批量获取每个流水线最近一次运行的摘要（阶段状态 + 镜像 + 耗时）// Logan.Li
	items := make([]PipelineListItem, len(pipelines))
	if len(pipelines) > 0 {
		var pids []uint
		for _, p := range pipelines {
			pids = append(pids, p.ID)
		}
		// 子查询：每个 pipeline_id 的最新 run id
		var latestRuns []models.PipelineRun
		h.db.Where("id IN (SELECT MAX(id) FROM ops_pipeline_runs WHERE pipeline_id IN ? AND status IN ('success','failed','running','pending') GROUP BY pipeline_id)", pids).
			Find(&latestRuns)
		runMap := map[uint]*models.PipelineRun{}
		for i := range latestRuns {
			runMap[latestRuns[i].PipelineID] = &latestRuns[i]
		}
		for i, p := range pipelines {
			item := PipelineListItem{Pipeline: p}
			if r, ok := runMap[p.ID]; ok {
				item.LastRunStages = r.Stages
				item.LastImageTag = r.ImageTag
				item.LastCommitID = r.CommitID
				item.LastRunDuration = r.Duration
				item.LastRunBranch = r.Branch
			}
			items[i] = item
		}
	} else {
		for i, p := range pipelines {
			items[i] = PipelineListItem{Pipeline: p}
		}
	}

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Data:    models.ListResponse{List: items, Total: total},
	})
}

// GetPipeline 获取流水线详情
// @Summary 获取流水线详情
// @Description 获取指定流水线的详细信息
// @Tags 流水线
// @Produce json
// @Security Bearer
// @Param id path int true "流水线ID"
// @Success 200 {object} models.Response{data=models.Pipeline}
// @Router /pipelines/{id} [get]
func (h *PipelineHandler) GetPipeline(c *gin.Context) {
	id := getPipelineID(c)
	if id == 0 {
		c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的流水线编码"})
		return
	}

	var pipeline models.Pipeline
	if err := h.db.Preload("Environment").First(&pipeline, id).Error; err != nil {
		c.JSON(http.StatusNotFound, models.Response{
			Code:    404,
			Message: "流水线不存在",
		})
		return
	}

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "success",
		Data:    pipeline,
	})
}

// CreatePipeline 创建流水线
// @Summary 创建流水线
// @Description 创建新流水线
// @Tags 流水线
// @Accept json
// @Produce json
// @Security Bearer
// @Param request body models.CreatePipelineRequest true "创建请求"
// @Success 200 {object} models.Response{data=models.Pipeline}
// @Router /pipelines [post]
func (h *PipelineHandler) CreatePipeline(c *gin.Context) {
	var req models.CreatePipelineRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, models.Response{
			Code:    400,
			Message: "请求参数错误: " + err.Error(),
		})
		return
	}

	// 校验编码格式
	if msg := validateCode(h.db, req.Code, 0); msg != "" {
		c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: msg})
		return
	}

	pipeline := models.Pipeline{
		Name:         req.Name,
		Code:         req.Code,
		JobName:      req.JobName,
		Description:  req.Description,
		Type:         req.Type,
		Source:       req.Source,
		Branch:       req.Branch,
		Dockerfile:   req.Dockerfile,
		Stages:       req.Stages,
		Variables:    req.Variables,
		DeployMode:   req.DeployMode,
		DeployTarget: req.DeployTarget,
		ServerIDs:    req.ServerIDs,
		K8sClusterID: req.K8sClusterID,
		K8sNamespace: req.K8sNamespace,
		RepoID:       req.RepoID,
		RegistryID:   req.RegistryID,
		GitCredentialID:      req.GitCredentialID,
		RegistryCredentialID: req.RegistryCredentialID,
		EnvironmentID: req.EnvironmentID,
		// 触发配置
		TriggerMode:   req.TriggerMode,
		WatchBranches: req.WatchBranches,
		ExcludeBranch: req.ExcludeBranch,
		CronExpr:      req.CronExpr,
		WebhookToken:  req.WebhookToken,
		// 缓存
		CachePaths: req.CachePaths,
		Enabled:    true,
	}

	if err := h.db.Create(&pipeline).Error; err != nil {
		c.JSON(http.StatusInternalServerError, models.Response{
			Code:    500,
			Message: "创建失败",
		})
		return
	}

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "创建成功",
		Data:    pipeline,
	})
}

// UpdatePipeline 更新流水线
// @Summary 更新流水线
// @Description 更新指定流水线
// @Tags 流水线
// @Accept json
// @Produce json
// @Security Bearer
// @Param id path int true "流水线ID"
// @Param request body models.CreatePipelineRequest true "更新请求"
// @Success 200 {object} models.Response{data=models.Pipeline}
// @Router /pipelines/{id} [put]
func (h *PipelineHandler) UpdatePipeline(c *gin.Context) {
	id := getPipelineID(c)
	if id == 0 {
		c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的流水线编码"})
		return
	}

	var req models.UpdatePipelineRequest
	if err := c.ShouldBindJSON(&req); err != nil {
		c.JSON(http.StatusBadRequest, models.Response{
			Code:    400,
			Message: "请求参数错误: " + err.Error(),
		})
		return
	}

	var pipeline models.Pipeline
	if err := h.db.First(&pipeline, id).Error; err != nil {
		c.JSON(http.StatusNotFound, models.Response{
			Code:    404,
			Message: "流水线不存在",
		})
		return
	}

	updates := map[string]interface{}{}
	if req.Name != "" { updates["name"] = req.Name }
	if req.JobName != "" { updates["job_name"] = req.JobName }
	if req.Description != "" { updates["description"] = req.Description }
	if req.Type != "" { updates["type"] = req.Type }
	if req.Source != "" { updates["source"] = req.Source }
	if req.Branch != "" { updates["branch"] = req.Branch }
	if req.Dockerfile != "" { updates["dockerfile"] = req.Dockerfile }
	if req.Stages != "" { updates["stages"] = req.Stages }
	if req.Variables != "" { updates["variables"] = req.Variables }
	if req.DeployMode != "" { updates["deploy_mode"] = req.DeployMode }
	if req.DeployTarget != "" { updates["deploy_target"] = req.DeployTarget }
	if req.ServerIDs != "" { updates["server_ids"] = req.ServerIDs }
	if req.K8sNamespace != "" { updates["k8s_namespace"] = req.K8sNamespace }
	if req.K8sClusterID != nil { updates["k8s_cluster_id"] = req.K8sClusterID }
	if req.RepoID != nil { updates["repo_id"] = req.RepoID }
	if req.RegistryID != nil { updates["registry_id"] = req.RegistryID }
	if req.GitCredentialID != nil { updates["git_credential_id"] = req.GitCredentialID }
	if req.RegistryCredentialID != nil { updates["registry_credential_id"] = req.RegistryCredentialID }
	if req.EnvironmentID != nil { updates["environment_id"] = req.EnvironmentID }
	if req.TriggerMode != "" { updates["trigger_mode"] = req.TriggerMode }
	if req.WatchBranches != "" { updates["watch_branches"] = req.WatchBranches }
	if req.ExcludeBranch != "" { updates["exclude_branch"] = req.ExcludeBranch }
	if req.CronExpr != "" { updates["cron_expr"] = req.CronExpr }
	if req.WebhookToken != "" { updates["webhook_token"] = req.WebhookToken }
	if req.CachePaths != "" { updates["cache_paths"] = req.CachePaths }
	if req.Enabled != nil { updates["enabled"] = req.Enabled }

	if err := h.db.Model(&pipeline).Updates(updates).Error; err != nil {
		c.JSON(http.StatusInternalServerError, models.Response{
			Code:    500,
			Message: "更新失败",
		})
		return
	}

	h.db.First(&pipeline, id)

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "更新成功",
		Data:    pipeline,
	})
}

// DeletePipeline 删除流水线
// @Summary 删除流水线
// @Description 删除指定流水线
// @Tags 流水线
// @Produce json
// @Security Bearer
// @Param id path int true "流水线ID"
// @Success 200 {object} models.Response
// @Router /pipelines/{id} [delete]
func (h *PipelineHandler) DeletePipeline(c *gin.Context) {
	id := getPipelineID(c)

	if err := h.db.Delete(&models.Pipeline{}, id).Error; err != nil {
		c.JSON(http.StatusInternalServerError, models.Response{
			Code:    500,
			Message: "删除失败",
		})
		return
	}

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "删除成功",
	})
}

// TriggerPipeline 触发流水线
// @Summary 触发流水线
// @Description 手动触发流水线执行
// @Tags 流水线
// @Accept json
// @Produce json
// @Security Bearer
// @Param id path int true "流水线ID"
// @Param request body models.TriggerPipelineRequest true "触发请求"
// @Success 200 {object} models.Response{data=models.PipelineRun}
// TriggerPipeline 触发流水线
func (h *PipelineHandler) TriggerPipeline(c *gin.Context) {
	id := getPipelineID(c)
	if id == 0 {
		c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "无效的流水线编码"})
		return
	}

	var pipeline models.Pipeline
	if err := h.db.First(&pipeline, id).Error; err != nil {
		c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "流水线不存在"})
		return
	}

	var req models.TriggerPipelineRequest
	c.ShouldBindJSON(&req)

	var maxBuildNumber int
	h.db.Model(&models.PipelineRun{}).Where("pipeline_id = ?", id).
		Select("COALESCE(MAX(build_number), 0)").Scan(&maxBuildNumber)

	branch := req.Branch
	tag := req.Tag
	ref := branch
	if req.Tag != "" {
		ref = tag
	} else if branch == "" {
		ref = pipeline.Branch
	}

	username, _ := c.Get("username")
	userStr, _ := username.(string)

	now := time.Now()
	run := models.PipelineRun{
		PipelineID:  uint(id),
		BuildNumber: maxBuildNumber + 1,
		Status:      "running",
		Branch:      ref,
		ImageTag:    tag, // 发布 Tag
		TriggeredBy: userStr,
		StartedAt:   &now,
	}

	if err := h.db.Create(&run).Error; err != nil {
		c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "触发失败"})
		return
	}

	// 后台异步执行流水线
	go h.executePipeline(pipeline, run)

	c.JSON(http.StatusOK, models.Response{Code: 200, Message: "已触发", Data: run})
}

// ListRuns 获取运行记录列表
// @Summary 获取运行记录列表
// @Description 获取指定流水线的运行记录
// @Tags 流水线
// @Produce json
// @Security Bearer
// @Param id path int true "流水线ID"
// @Param page query int false "页码"
// @Param page_size query int false "每页数量"
// @Success 200 {object} models.Response{data=models.ListResponse}
// @Router /pipelines/{id}/runs [get]
func (h *PipelineHandler) ListRuns(c *gin.Context) {
	id := getPipelineID(c)
	var query models.PageQuery
	c.ShouldBindQuery(&query)

	db := h.db.Model(&models.PipelineRun{}).Where("pipeline_id = ?", id)

	var total int64
	db.Count(&total)

	var runs []models.PipelineRun
	db.Order("started_at DESC").
		Offset(query.GetOffset()).
		Limit(query.GetLimit()).
		Find(&runs)

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "success",
		Data: models.ListResponse{
			List:  runs,
			Total: total,
		},
	})
}

// WebhookTrigger 外部 Webhook 触发流水线 (支持 URL: /webhook/job/:jobName 和 /webhook/pipeline/:code)
func (h *PipelineHandler) WebhookTrigger(c *gin.Context) {
	jobName := c.Param("jobName")
	code := c.Param("code")

	var pipeline models.Pipeline
	if jobName != "" {
		if err := h.db.Where("job_name = ? AND enabled = ?", jobName, true).First(&pipeline).Error; err != nil {
			c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "Job 不存在或已禁用"})
			return
		}
	} else if code != "" {
		if err := h.db.Where("code = ? AND enabled = ?", code, true).First(&pipeline).Error; err != nil {
			c.JSON(http.StatusNotFound, models.Response{Code: 404, Message: "流水线不存在或已禁用"})
			return
		}
	} else {
		c.JSON(http.StatusBadRequest, models.Response{Code: 400, Message: "缺少 jobName 或 code"})
		return
	}

	// 验证 Webhook Token
	if pipeline.WebhookToken != "" {
		token := c.GetHeader("X-Webhook-Token")
		if token == "" {
			token = c.Query("token")
		}
		if token != pipeline.WebhookToken {
			c.JSON(http.StatusForbidden, models.Response{Code: 403, Message: "Webhook Token 无效"})
			return
		}
	}

	// 解析请求体获取 branch/tag
	var req models.TriggerPipelineRequest
	c.ShouldBindJSON(&req)

	var maxBuildNumber int
	h.db.Model(&models.PipelineRun{}).Where("pipeline_id = ?", pipeline.ID).
		Select("COALESCE(MAX(build_number), 0)").Scan(&maxBuildNumber)

	ref := req.Branch
	if req.Tag != "" {
		ref = req.Tag
	} else if ref == "" {
		ref = pipeline.Branch
	}

	triggeredBy := "webhook"
	if reqPayload, exists := c.Get("raw_body"); exists {
		// 尝试从 webhook payload 中提取提交者信息
		if body, ok := reqPayload.(map[string]interface{}); ok {
			if pusher, ok := body["pusher"].(map[string]interface{}); ok {
				if name, ok := pusher["name"].(string); ok {
					triggeredBy = name
				}
			}
		}
	}

	now := time.Now()
	run := models.PipelineRun{
		PipelineID:  pipeline.ID,
		BuildNumber: maxBuildNumber + 1,
		Status:      "running",
		Branch:      ref,
		ImageTag:    req.Tag,
		TriggeredBy: triggeredBy,
		StartedAt:   &now,
	}

	if err := h.db.Create(&run).Error; err != nil {
		c.JSON(http.StatusInternalServerError, models.Response{Code: 500, Message: "触发失败"})
		return
	}

	go h.executePipeline(pipeline, run)

	c.JSON(http.StatusOK, models.Response{
		Code:    200,
		Message: "已触发",
		Data: map[string]interface{}{
			"build_number": run.BuildNumber,
			"run_id":       run.ID,
		},
	})
}