package controller import ( "net/http" "regexp" "strconv" "time" "ezops/model" "ezops/middleware" ezredis "ezops/pkg/redis" "ezops/pkg/toolinstaller" "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(&model.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 toolInstaller *toolinstaller.ToolInstaller } // NewPipelineHandler 创建流水线处理器 func NewPipelineHandler(db *gorm.DB, cache ezredis.Cache, ti *toolinstaller.ToolInstaller) *PipelineHandler { return &PipelineHandler{db: db, cache: cache, toolInstaller: ti} } // PipelineListItem 列表项(含最近一次运行摘要)// Logan.Li type PipelineListItem struct { model.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} model.Response{data=model.ListResponse} // ListPipelines 获取流水线列表 func (h *PipelineHandler) ListPipelines(c *gin.Context) { var query model.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(&model.Pipeline{}) if isAdmin != true && usernameStr != "" { ids, hasAccess := middleware.GetAccessiblePipelineIDs(h.db, usernameStr, uid) if !hasAccess { // 没有任何流水线权限 → 返回空列表 c.JSON(http.StatusOK, model.Response{Code: 200, Data: model.ListResponse{List: []model.Pipeline{}, Total: 0}}) return } db = db.Where("id IN ?", ids) } var total int64 db.Count(&total) var pipelines []model.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 []model.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]*model.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, model.Response{ Code: 200, Data: model.ListResponse{List: items, Total: total}, }) } // GetPipeline 获取流水线详情 // @Summary 获取流水线详情 // @Description 获取指定流水线的详细信息 // @Tags 流水线 // @Produce json // @Security Bearer // @Param id path int true "流水线ID" // @Success 200 {object} model.Response{data=model.Pipeline} // @Router /pipelines/{id} [get] func (h *PipelineHandler) GetPipeline(c *gin.Context) { id := getPipelineID(c) if id == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的流水线编码"}) return } var pipeline model.Pipeline if err := h.db.Preload("Environment").First(&pipeline, id).Error; err != nil { c.JSON(http.StatusNotFound, model.Response{ Code: 404, Message: "流水线不存在", }) return } c.JSON(http.StatusOK, model.Response{ Code: 200, Message: "success", Data: pipeline, }) } // CreatePipeline 创建流水线 // @Summary 创建流水线 // @Description 创建新流水线 // @Tags 流水线 // @Accept json // @Produce json // @Security Bearer // @Param request body model.CreatePipelineRequest true "创建请求" // @Success 200 {object} model.Response{data=model.Pipeline} // @Router /pipelines [post] func (h *PipelineHandler) CreatePipeline(c *gin.Context) { var req model.CreatePipelineRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, model.Response{ Code: 400, Message: "请求参数错误: " + err.Error(), }) return } // 校验编码格式 if msg := validateCode(h.db, req.Code, 0); msg != "" { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: msg}) return } pipeline := model.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, // 并发控制 MaxExecutors: req.MaxExecutors, Enabled: true, } if err := h.db.Create(&pipeline).Error; err != nil { c.JSON(http.StatusInternalServerError, model.Response{ Code: 500, Message: "创建失败", }) return } c.JSON(http.StatusOK, model.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 model.CreatePipelineRequest true "更新请求" // @Success 200 {object} model.Response{data=model.Pipeline} // @Router /pipelines/{id} [put] func (h *PipelineHandler) UpdatePipeline(c *gin.Context) { id := getPipelineID(c) if id == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的流水线编码"}) return } var req model.UpdatePipelineRequest if err := c.ShouldBindJSON(&req); err != nil { c.JSON(http.StatusBadRequest, model.Response{ Code: 400, Message: "请求参数错误: " + err.Error(), }) return } var pipeline model.Pipeline if err := h.db.First(&pipeline, id).Error; err != nil { c.JSON(http.StatusNotFound, model.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 req.MaxExecutors > 0 { updates["max_executors"] = req.MaxExecutors } if err := h.db.Model(&pipeline).Updates(updates).Error; err != nil { c.JSON(http.StatusInternalServerError, model.Response{ Code: 500, Message: "更新失败", }) return } h.db.First(&pipeline, id) c.JSON(http.StatusOK, model.Response{ Code: 200, Message: "更新成功", Data: pipeline, }) } // DeletePipeline 删除流水线 // @Summary 删除流水线 // @Description 删除指定流水线 // @Tags 流水线 // @Produce json // @Security Bearer // @Param id path int true "流水线ID" // @Success 200 {object} model.Response // @Router /pipelines/{id} [delete] func (h *PipelineHandler) DeletePipeline(c *gin.Context) { id := getPipelineID(c) if err := h.db.Delete(&model.Pipeline{}, id).Error; err != nil { c.JSON(http.StatusInternalServerError, model.Response{ Code: 500, Message: "删除失败", }) return } c.JSON(http.StatusOK, model.Response{ Code: 200, Message: "删除成功", }) } // TriggerPipeline 触发流水线 // @Summary 触发流水线 // @Description 手动触发流水线执行 // @Tags 流水线 // @Accept json // @Produce json // @Security Bearer // @Param id path int true "流水线ID" // @Param request body model.TriggerPipelineRequest true "触发请求" // @Success 200 {object} model.Response{data=model.PipelineRun} // TriggerPipeline 触发流水线 func (h *PipelineHandler) TriggerPipeline(c *gin.Context) { id := getPipelineID(c) if id == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的流水线编码"}) return } var pipeline model.Pipeline if err := h.db.First(&pipeline, id).Error; err != nil { c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: "流水线不存在"}) return } var req model.TriggerPipelineRequest c.ShouldBindJSON(&req) var maxBuildNumber int h.db.Model(&model.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 := model.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, model.Response{Code: 500, Message: "触发失败"}) return } // 通过队列引擎执行(尊重 max_executors 并发限制) go h.tryExecuteOrQueue(pipeline, run) c.JSON(http.StatusOK, model.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} model.Response{data=model.ListResponse} // @Router /pipelines/{id}/runs [get] func (h *PipelineHandler) ListRuns(c *gin.Context) { id := getPipelineID(c) var query model.PageQuery c.ShouldBindQuery(&query) db := h.db.Model(&model.PipelineRun{}).Where("pipeline_id = ?", id) var total int64 db.Count(&total) var runs []model.PipelineRun db.Order("started_at DESC"). Offset(query.GetOffset()). Limit(query.GetLimit()). Find(&runs) c.JSON(http.StatusOK, model.Response{ Code: 200, Message: "success", Data: model.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 model.Pipeline if jobName != "" { if err := h.db.Where("job_name = ? AND enabled = ?", jobName, true).First(&pipeline).Error; err != nil { c.JSON(http.StatusNotFound, model.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, model.Response{Code: 404, Message: "流水线不存在或已禁用"}) return } } else { c.JSON(http.StatusBadRequest, model.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, model.Response{Code: 403, Message: "Webhook Token 无效"}) return } } // 解析请求体获取 branch/tag var req model.TriggerPipelineRequest c.ShouldBindJSON(&req) var maxBuildNumber int h.db.Model(&model.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 := model.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, model.Response{Code: 500, Message: "触发失败"}) return } go h.executePipeline(pipeline, run) c.JSON(http.StatusOK, model.Response{ Code: 200, Message: "已触发", Data: map[string]interface{}{ "build_number": run.BuildNumber, "run_id": run.ID, }, }) } // GetPipelineQueue 查询流水线队列状态 func (h *PipelineHandler) GetPipelineQueue(c *gin.Context) { id := getPipelineID(c) if id == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "无效的流水线编码"}) return } status := h.GetQueueStatus(uint(id)) c.JSON(http.StatusOK, model.Response{Code: 200, Data: status}) } // CancelQueueItem 取消排队中的构建 func (h *PipelineHandler) CancelQueueItem(c *gin.Context) { id := getPipelineID(c) runID, _ := strconv.ParseUint(c.Param("runId"), 10, 64) if id == 0 || runID == 0 { c.JSON(http.StatusBadRequest, model.Response{Code: 400, Message: "参数错误"}) return } if err := h.CancelQueued(uint(id), uint(runID)); err != nil { c.JSON(http.StatusNotFound, model.Response{Code: 404, Message: err.Error()}) return } c.JSON(http.StatusOK, model.Response{Code: 200, Message: "已取消"}) }