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} } // 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) c.JSON(http.StatusOK, models.Response{ Code: 200, Data: models.ListResponse{List: pipelines, 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.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, }, }) }