package handler import ( "context" "encoding/json" "fmt" "net/http" "regexp" "strconv" "strings" "time" "sql-platform/internal/ghost" "sql-platform/internal/middleware" "sql-platform/internal/model" "github.com/gin-gonic/gin" "github.com/redis/go-redis/v9" ) // SqlExecItem records the execution result of a single SQL statement type SqlExecItem struct { SQL string `json:"sql"` Status string `json:"status"` // success / fail AffectedRows int64 `json:"affected_rows"` DurationMs int64 `json:"duration_ms"` Error string `json:"error,omitempty"` } // SqlExecResult is the structured execution result stored as JSON type SqlExecResult struct { Status string `json:"status"` // success / failed TotalMs int64 `json:"total_ms"` Items []SqlExecItem `json:"items"` StartTime string `json:"start_time,omitempty"` EndTime string `json:"end_time,omitempty"` } // GET /api/v1/tickets/statistics func TicketStatistics(c *gin.Context) { userID, _, role := middleware.GetCurrentUser(c) type statusCount struct { Status string `json:"status"` Count int64 `json:"count"` } var counts []statusCount query := model.DB.Model(&model.Ticket{}).Select("status, count(*) as count") // Filter by user role switch role { case "developer": query = query.Where("applicant_id = ?", userID) case "dba", "auditor": // DBA sees tickets they submitted + tickets they need to approve query = query.Where( "applicant_id = ? OR id IN (SELECT ticket_id FROM ticket_approvals WHERE approver_id = ?)", userID, userID, ) // admin sees everything } query.Group("status").Scan(&counts) stats := model.TicketStatistics{} for _, c := range counts { switch c.Status { case "pending": stats.Pending = int(c.Count) case "reviewing": stats.Reviewing = int(c.Count) case "executing": stats.Executing = int(c.Count) case "success": stats.Completed = int(c.Count) case "failed": stats.Failed = int(c.Count) case "rejected": stats.Rejected = int(c.Count) } } stats.Total = stats.Pending + stats.Reviewing + stats.Executing + stats.Completed + stats.Failed + stats.Rejected c.JSON(http.StatusOK, stats) } // GET /api/v1/tickets func ListTickets(c *gin.Context) { userID, _, role := middleware.GetCurrentUser(c) status := c.Query("status") page, _ := strconv.Atoi(c.DefaultQuery("page", "1")) limit, _ := strconv.Atoi(c.DefaultQuery("page_size", "20")) if page < 1 { page = 1 } if limit < 1 || limit > 100 { limit = 20 } query := model.DB.Model(&model.Ticket{}).Order("id DESC") // keyword search keyword := c.Query("keyword") if keyword != "" { query = query.Where("title LIKE ? OR ticket_no LIKE ?", "%"+keyword+"%", "%"+keyword+"%") } // Filter by user role switch role { case "admin": // admin sees all tickets default: query = query.Where( "applicant_id = ? OR id IN (SELECT ticket_id FROM ticket_approvals WHERE approver_id = ?)", userID, userID, ) } if status != "" && status != "all" { query = query.Where("status = ?", status) } var total int64 query.Count(&total) var tickets []model.Ticket query.Offset((page - 1) * limit).Limit(limit).Find(&tickets) // Enrich tickets with has_audit_step info type TicketWithMeta struct { model.Ticket HasAuditStep bool `json:"has_audit_step"` HasExecuteStep bool `json:"has_execute_step"` IsExecutor bool `json:"is_executor"` IsApprover bool `json:"is_approver"` } var result []TicketWithMeta for _, t := range tickets { var auditCount, execCount int64 model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type = 'audit'", t.ID).Count(&auditCount) model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type = 'execute'", t.ID).Count(&execCount) var isExec, isAppr int64 model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type = 'execute' AND approver_id = ?", t.ID, userID).Count(&isExec) model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type != 'execute' AND step_type != 'submit' AND approver_id = ?", t.ID, userID).Count(&isAppr) result = append(result, TicketWithMeta{ Ticket: t, HasAuditStep: auditCount > 0, HasExecuteStep: execCount > 0, IsExecutor: isExec > 0, IsApprover: isAppr > 0, }) } c.JSON(http.StatusOK, gin.H{ "items": result, "total": total, "page": page, "page_size": limit, }) } // GET /api/v1/tickets/:id func GetTicket(c *gin.Context) { id := c.Param("id") var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } // Get approvals — enrich with user display names var approvals []model.TicketApproval model.DB.Where("ticket_id = ?", id).Order("step_number, id").Find(&approvals) type ApprovalWithNames struct { model.TicketApproval ApproverName string `json:"approver_name"` ActualApproverName string `json:"actual_approver_name"` } var approvalsResult []ApprovalWithNames for _, a := range approvals { item := ApprovalWithNames{TicketApproval: a} if a.ApproverID != nil { var u model.User if model.DB.First(&u, *a.ApproverID).Error == nil { item.ApproverName = u.DisplayName } } if a.ActualApproverID != nil { var u model.User if model.DB.First(&u, *a.ActualApproverID).Error == nil { item.ActualApproverName = u.DisplayName } } approvalsResult = append(approvalsResult, item) } // Get SQL items var sqlItems []model.TicketSQLItem model.DB.Where("ticket_id = ?", id).Order("seq").Find(&sqlItems) // Flatten ticket fields into response + add nested data result := gin.H{} // Marshal ticket to JSON then unmarshal to map for flattening ticketBytes, _ := json.Marshal(ticket) json.Unmarshal(ticketBytes, &result) result["approvals"] = approvalsResult result["sql_items"] = sqlItems // Add applicant_name and instance_name for frontend var applicant model.User if err := model.DB.First(&applicant, ticket.ApplicantID).Error; err == nil { result["applicant_name"] = applicant.DisplayName } var inst model.Instance if err := model.DB.First(&inst, ticket.InstanceID).Error; err == nil { result["instance_name"] = inst.Name } c.JSON(http.StatusOK, result) } // POST /api/v1/tickets func CreateTicket(c *gin.Context) { userID, _, userRole := middleware.GetCurrentUser(c) // auditor cannot create tickets if userRole == "auditor" { c.JSON(http.StatusForbidden, gin.H{"detail": "权限不足:审计员不能提交工单"}) return } var input struct { Title string `json:"title" binding:"required"` DBType string `json:"db_type" binding:"required"` InstanceID int `json:"instance_id" binding:"required"` TargetDatabase string `json:"target_database" binding:"required"` SQLContent string `json:"sql_content" binding:"required"` RollbackSQL *string `json:"rollback_sql"` Priority *string `json:"priority"` Reason *string `json:"reason"` WorkflowID *int `json:"workflow_id"` RiskLevel *string `json:"risk_level"` ExecStrategy *string `json:"exec_strategy"` } if err := c.ShouldBindJSON(&input); err != nil { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) return } // Reject read-only SQL statements upperSQL := strings.ToUpper(strings.TrimSpace(input.SQLContent)) if input.DBType == "mysql" || input.DBType == "postgresql" { readOnlyPrefixes := []string{"SELECT", "SHOW", "DESC", "DESCRIBE", "EXPLAIN"} for _, prefix := range readOnlyPrefixes { if strings.HasPrefix(upperSQL, prefix) { c.JSON(http.StatusBadRequest, gin.H{"detail": "数据变更工单不支持 SELECT/SHOW/DESC 等只读语句,请使用查询数据功能"}) return } } } else if input.DBType == "redis" { // Redis: reject read-only commands (GET/HGET/KEYS/SCAN/TTL/TYPE/LLEN/LRANGE/INFO etc.) redisReadOnlyCmds := []string{"GET", "HGET", "MGET", "HGETALL", "KEYS", "SCAN", "SSCAN", "HSCAN", "ZSCAN", "TTL", "PTTL", "TYPE", "EXISTS", "LLEN", "LRANGE", "SRANDMEMBER", "SCARD", "SISMEMBER", "SMEMBERS", "STRLEN", "INFO", "DBSIZE", "PING", "ECHO", "OBJECT", "BITCOUNT", "GETRANGE", "SUBSTR"} cmdUpper := upperSQL if idx := strings.IndexByte(cmdUpper, ' '); idx > 0 { cmdUpper = cmdUpper[:idx] } for _, ro := range redisReadOnlyCmds { if cmdUpper == ro { c.JSON(http.StatusBadRequest, gin.H{"detail": fmt.Sprintf("数据变更工单不支持只读命令 %s,请使用查询功能", cmdUpper)}) return } } } // ── SQL审核规则检查 (from audit_rules table) ── statements := splitSQL(input.SQLContent) sqlToCheck := input.SQLContent // For Redis, check as a single command if input.DBType == "redis" { cmdParts := parseCommandParts(input.SQLContent) if len(cmdParts) > 0 { sqlToCheck = strings.ToUpper(cmdParts[0]) + " " + strings.Join(cmdParts[1:], " ") } } var rules []model.AuditRule model.DB.Where("db_type = ? AND is_enabled = ? AND pattern IS NOT NULL AND pattern != ''", input.DBType, true).Find(&rules) for _, stmt := range statements { checkStmt := strings.TrimSpace(stmt) if checkStmt == "" { continue } // For Redis, the whole sql_content is one command if input.DBType == "redis" { checkStmt = sqlToCheck } for _, rule := range rules { if rule.Pattern == nil || *rule.Pattern == "" { continue } matched, _ := regexp.MatchString(*rule.Pattern, checkStmt) if matched { // Special: WHERE-check rules — reject only if no WHERE clause if rule.Value != nil && *rule.Value == "where_check" { if strings.Contains(strings.ToUpper(checkStmt), "WHERE") { continue // has WHERE, safe } c.JSON(http.StatusBadRequest, gin.H{ "detail": fmt.Sprintf("审核规则拦截: %s — %s", rule.Name, rule.Message), }) return } // value="reject" → block submission, "review"/"warn" → allow but record action := "reject" if rule.Value != nil { action = *rule.Value } if action == "reject" { c.JSON(http.StatusBadRequest, gin.H{ "detail": fmt.Sprintf("审核规则拦截: %s — %s", rule.Name, rule.Message), }) return } // For warn/review rules, we could attach warnings later // For now, reject-only rules block, others pass through } } // Redis only has one command per ticket, break after first if input.DBType == "redis" { break } } // Auto-fill workflow_id from instance if not provided if input.WorkflowID == nil || *input.WorkflowID == 0 { var inst model.Instance if model.DB.First(&inst, input.InstanceID).Error == nil && inst.WorkflowID != nil { input.WorkflowID = inst.WorkflowID } } status := "pending" priority := "medium" riskLevel := "low" execStrategy := "stop_on_error" if input.Priority != nil { priority = *input.Priority } if input.RiskLevel != nil { riskLevel = *input.RiskLevel } if input.ExecStrategy != nil { execStrategy = *input.ExecStrategy } // Generate ticket number now := time.Now() ticketNo := now.Format("20060102") + "-" + fmt.Sprintf("%04d", time.Now().UnixNano()%10000) ticket := model.Ticket{ Title: input.Title, DBType: input.DBType, InstanceID: input.InstanceID, TargetDatabase: input.TargetDatabase, SQLContent: input.SQLContent, RollbackSQL: input.RollbackSQL, Status: &status, Priority: &priority, ApplicantID: userID, Reason: input.Reason, WorkflowID: input.WorkflowID, RiskLevel: &riskLevel, ExecStrategy: &execStrategy, TicketNo: &ticketNo, } if err := model.DB.Create(&ticket).Error; err != nil { c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) return } // Create approval records based on workflow or default to admin hasAuditStep := false if input.WorkflowID != nil && *input.WorkflowID > 0 { // Load workflow steps var wfSteps []model.WorkflowStep model.DB.Where("workflow_id = ?", *input.WorkflowID).Order("step_number").Find(&wfSteps) for _, ws := range wfSteps { if ws.StepType == "submit" { continue // submit step needs no approver } if ws.StepType != "execute" { hasAuditStep = true } // Parse approver_ids from JSON var approverIDs []int if ws.ApproverIDs != nil && *ws.ApproverIDs != "" { json.Unmarshal([]byte(*ws.ApproverIDs), &approverIDs) } // Create one approval record per approver (any one can approve) if len(approverIDs) > 0 { for _, uid := range approverIDs { uidCopy := uid stepName := ws.Name stepType := ws.StepType approval := model.TicketApproval{ TicketID: ticket.ID, StepNumber: ws.StepNumber, StepName: &stepName, StepType: &stepType, ApproverID: &uidCopy, } model.DB.Create(&approval) } } else { // No specific users — use role-based fallback stepName := ws.Name stepType := ws.StepType var roleUser model.User role := "admin" if ws.ApproverRole != nil && *ws.ApproverRole != "" { role = *ws.ApproverRole } if model.DB.Where("role = ? AND is_active = 1", role).First(&roleUser).Error == nil { rid := roleUser.ID approval := model.TicketApproval{ TicketID: ticket.ID, StepNumber: ws.StepNumber, StepName: &stepName, StepType: &stepType, ApproverID: &rid, ApproverRole: &role, } model.DB.Create(&approval) } } } } else { // No workflow — default: first admin as approver stepName := "审核" stepType := "audit" approverRole := "admin" var adminUser model.User if err := model.DB.Where("role = ? AND is_active = 1", "admin").First(&adminUser).Error; err == nil { adminID := adminUser.ID approval := model.TicketApproval{ TicketID: ticket.ID, StepNumber: 1, StepName: &stepName, StepType: &stepType, ApproverID: &adminID, ApproverRole: &approverRole, } model.DB.Create(&approval) } } // If workflow has no audit step (only submit + execute), skip to reviewing if !hasAuditStep { reviewing := "reviewing" model.DB.Model(&ticket).Update("status", reviewing) ticket.Status = &reviewing } c.JSON(http.StatusCreated, ticket) } // POST /api/v1/tickets/:id/approve func ApproveTicket(c *gin.Context) { id := c.Param("id") userID, _, userRole := middleware.GetCurrentUser(c) // Role check: admin/dba/developer can approve if userRole != "admin" && userRole != "dba" && userRole != "developer" { c.JSON(http.StatusForbidden, gin.H{"detail": "权限不足:审核操作需要管理员、DBA 或开发者角色"}) return } var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } var input struct { Action string `json:"action" binding:"required"` // approve, reject Comment *string `json:"comment"` } if err := c.ShouldBindJSON(&input); err != nil { c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()}) return } now := time.Now() // Update approval record — auto-create if missing (legacy tickets) var approval model.TicketApproval if err := model.DB.Where("ticket_id = ? AND approver_id = ? AND action IS NULL", id, userID).First(&approval).Error; err != nil { // No approval record — authorized role can create one stepName := "审核" stepType := "audit" approverRole := userRole approval = model.TicketApproval{ TicketID: ticket.ID, StepNumber: 1, StepName: &stepName, StepType: &stepType, ApproverID: &userID, ApproverRole: &approverRole, } model.DB.Create(&approval) } approval.Action = &input.Action approval.Comment = input.Comment approval.ActualApproverID = &userID approval.CompletedAt = &now model.DB.Save(&approval) // Update ticket status newStatus := "approved" if input.Action == "reject" { newStatus = "rejected" } else if input.Action == "approve" { // Check if there are more pending AUDIT approvals (exclude execute type) var pendingAuditCount int64 model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND action IS NULL AND step_type != 'execute'", id).Count(&pendingAuditCount) if pendingAuditCount > 0 { newStatus = "reviewing" } // If all audit steps done → "approved" (ready for execution) } ticket.Status = &newStatus ticket.ReviewTime = &now model.DB.Save(&ticket) c.JSON(http.StatusOK, gin.H{"ticket": ticket, "approval": approval}) } // POST /api/v1/tickets/:id/execute func ExecuteTicket(c *gin.Context) { id := c.Param("id") // Role check: only admin/dba can execute _, _, userRole := middleware.GetCurrentUser(c) if userRole != "admin" && userRole != "dba" { c.JSON(http.StatusForbidden, gin.H{"detail": "权限不足:执行操作需要管理员或 DBA 角色"}) return } var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } // Check ticket status — allow approved OR reviewing (will auto-approve execute step) if ticket.Status != nil && *ticket.Status != "approved" && *ticket.Status != "reviewing" { c.JSON(http.StatusBadRequest, gin.H{"detail": "Ticket must be approved before execution"}) return } // Auto-approve any pending execute-type approval for the current user userID, _, _ := middleware.GetCurrentUser(c) // Verify current user is an executor for this ticket var executorCheck int64 model.DB.Model(&model.TicketApproval{}). Where("ticket_id = ? AND step_type = 'execute' AND (approver_id = ? OR approver_id IS NULL)", id, userID). Count(&executorCheck) if executorCheck == 0 { c.JSON(http.StatusForbidden, gin.H{"detail": "你不是此工单的执行人"}) return } var pendingExec model.TicketApproval if err := model.DB.Where("ticket_id = ? AND step_type = 'execute' AND action IS NULL", id).First(&pendingExec).Error; err == nil { // Auto-approve the execute step now := time.Now() action := "approve" comment := "自动通过(执行触发)" pendingExec.Action = &action pendingExec.Comment = &comment pendingExec.ActualApproverID = &userID pendingExec.CompletedAt = &now model.DB.Save(&pendingExec) } // Check if all audit steps are approved var pendingAudit int64 model.DB.Model(&model.TicketApproval{}). Where("ticket_id = ? AND step_type = 'audit' AND action IS NULL", id). Count(&pendingAudit) if pendingAudit > 0 { c.JSON(http.StatusBadRequest, gin.H{"detail": "还有未完成的审核步骤"}) return } // Check for rejected audit steps var rejectedAudit int64 model.DB.Model(&model.TicketApproval{}). Where("ticket_id = ? AND step_type = 'audit' AND action = 'reject'", id). Count(&rejectedAudit) if rejectedAudit > 0 { c.JSON(http.StatusBadRequest, gin.H{"detail": "审核已被拒绝,无法执行"}) return } // Get instance info var instance model.Instance if err := model.DB.First(&instance, ticket.InstanceID).Error; err != nil { c.JSON(http.StatusBadRequest, gin.H{"detail": "Instance not found"}) return } status := "executing" ticket.Status = &status now := time.Now() ticket.ExecuteTime = &now model.DB.Save(&ticket) // Execute SQL based on db_type var execResult string var execErr string if ticket.DBType == "redis" { // Execute Redis command (write operations allowed in ticket workflow) redisDB := 0 if ticket.TargetDatabase != "" { if n, err := strconv.Atoi(strings.TrimPrefix(ticket.TargetDatabase, "db")); err == nil && n >= 0 { redisDB = n } } client, err := getRedisClient(&instance, redisDB) if err != nil { execErr = fmt.Sprintf("Redis connection failed: %v", err) } else { defer client.Close() ctx := context.Background() parts := parseCommandParts(ticket.SQLContent) if len(parts) == 0 { execErr = "empty command" } else { args := make([]interface{}, len(parts)) for i, p := range parts { args[i] = p } execStart := time.Now() result, err := client.Do(ctx, args...).Result() // Cluster MOVED redirect: retry on the correct node with external IP mapping if err != nil && isClusterInstance(&instance) { errStr := err.Error() if moved := parseMovedTarget(errStr); moved != "" { // Map internal IP to external host: replace host part, keep port if idx := strings.LastIndex(moved, ":"); idx >= 0 { port := moved[idx+1:] externalAddr := instance.Host + ":" + port password := instance.Password if strings.HasPrefix(password, "enc:") { if dec, e := decryptPassword(strings.TrimPrefix(password, "enc:")); e == nil { password = dec } } targetClient := redis.NewClient(&redis.Options{ Addr: externalAddr, Username: instance.Username, Password: password, }) defer targetClient.Close() result, err = targetClient.Do(ctx, args...).Result() if err != nil { err = fmt.Errorf("集群节点重定向失败(MOVED→%s): %w — 集群节点端口(%s)可能未暴露给外部,需检查Docker端口映射", moved, err, port) } } } } stmtMs := time.Since(execStart).Milliseconds() execRes := SqlExecResult{ Status: "success", TotalMs: stmtMs, StartTime: execStart.UTC().Format(time.RFC3339), EndTime: time.Now().UTC().Format(time.RFC3339), } if err != nil { execRes.Status = "failed" execRes.Items = []SqlExecItem{{ SQL: ticket.SQLContent, Status: "fail", DurationMs: stmtMs, Error: err.Error(), }} execErr = err.Error() } else { resultStr := fmt.Sprintf("%v", result) execRes.Items = []SqlExecItem{{ SQL: ticket.SQLContent, Status: "success", DurationMs: stmtMs, Error: resultStr, }} } if jsonBytes, jsonErr := json.Marshal(execRes); jsonErr == nil { execResult = string(jsonBytes) } } } } else { // MySQL/PG execution // ── Ghost/OSC channel: if ghost_enabled and SQL is ALTER TABLE → use gh-ost ── if instance.DBType == "mysql" && isGhostEnabled() { upperSQL := strings.TrimSpace(strings.ToUpper(ticket.SQLContent)) if strings.HasPrefix(upperSQL, "ALTER TABLE") { // Parse ALTER TABLE → extract table name and alter clause tableName, alterClause, parseErr := parseAlterTable(ticket.SQLContent) if parseErr != nil { execErr = fmt.Sprintf("ALTER TABLE 解析失败: %v", parseErr) } else { ghostTask, buildErr := buildGhOstTask(int(instance.ID), ticket.TargetDatabase, tableName, alterClause) if buildErr != nil { execErr = fmt.Sprintf("gh-ost 任务构建失败: %v", buildErr) } else { ghostTask.Execute = true applyGhostSettings(ghostTask) // Apply executor timeout execTimeout := getSettingInt("executor_timeout", 300) ctx, cancel := context.WithTimeout(context.Background(), time.Duration(execTimeout)*time.Second) defer cancel() execStart := time.Now() result, ghErr := ghost.RunGhOst(ctx, *ghostTask, getGhostPath()) stmtMs := time.Since(execStart).Milliseconds() ghRes := SqlExecResult{ Status: "success", TotalMs: stmtMs, StartTime: execStart.UTC().Format(time.RFC3339), EndTime: time.Now().UTC().Format(time.RFC3339), } if ghErr != nil { ghRes.Status = "failed" ghRes.Items = []SqlExecItem{{ SQL: ticket.SQLContent, Status: "fail", DurationMs: stmtMs, Error: fmt.Sprintf("%s\n%s", result.Error, ghErr.Error()), }} execErr = ghErr.Error() } else { ghRes.Items = []SqlExecItem{{ SQL: ticket.SQLContent, Status: "success", DurationMs: stmtMs, Error: result.Output, }} } if jsonBytes, jsonErr := json.Marshal(ghRes); jsonErr == nil { execResult = string(jsonBytes) } // Skip normal execution path goto executionDone } } } } db, err := getDBConnection(&instance, ticket.TargetDatabase) if err != nil { execErr = fmt.Sprintf("DB connection failed: %v", err) } else { defer db.Close() // Apply executor timeout from platform settings execTimeout := getSettingInt("executor_timeout", 300) ctx, cancel := context.WithTimeout(context.Background(), time.Duration(execTimeout)*time.Second) defer cancel() // Dry-run: EXPLAIN pre-check if getSetting("executor_dry_run", "true") == "true" { ddlPrefixes := []string{"CREATE", "ALTER", "DROP", "TRUNCATE", "RENAME"} statements := splitSQL(ticket.SQLContent) // If batch contains ANY DDL, skip EXPLAIN precheck entirely — // DDL changes the schema so subsequent DML can't be EXPLAINed beforehand hasDDL := false for _, stmt := range statements { upper := strings.ToUpper(strings.TrimSpace(stmt)) for _, prefix := range ddlPrefixes { if strings.HasPrefix(upper, prefix) { hasDDL = true break } } if hasDDL { break } } if !hasDDL { for _, stmt := range statements { stmt = strings.TrimSpace(stmt) if stmt == "" { continue } upperStmt := strings.ToUpper(stmt) if strings.HasPrefix(upperStmt, "EXPLAIN") { continue } rows, err := db.QueryContext(ctx, "EXPLAIN "+stmt) if err != nil { execErr = fmt.Sprintf("预检查失败(EXPLAIN): %v", err) break } rows.Close() } } if execErr != "" { failStatus := "failed" ticket.Status = &failStatus ticket.ExecuteResult = &execErr model.DB.Save(&ticket) c.JSON(http.StatusOK, gin.H{"message": "Pre-check failed", "ticket": ticket}) return } } // Split multi-statement by semicolons — wrap in transaction for atomic rollback execStart := time.Now() statements := splitSQL(ticket.SQLContent) execRes := SqlExecResult{Status: "success", StartTime: execStart.UTC().Format(time.RFC3339)} // Begin transaction for multi-statement execution tx, txErr := db.BeginTx(ctx, nil) if txErr != nil { execErr = fmt.Sprintf("开启事务失败: %v", txErr) } else { committed := false for _, stmt := range statements { stmt = strings.TrimSpace(stmt) if stmt == "" { continue } stmtStart := time.Now() sqlResult, err := tx.ExecContext(ctx, stmt) stmtMs := time.Since(stmtStart).Milliseconds() if err != nil { execRes.Items = append(execRes.Items, SqlExecItem{ SQL: stmt, Status: "fail", DurationMs: stmtMs, Error: err.Error(), }) execRes.Status = "failed" execErr = err.Error() // Rollback on any failure tx.Rollback() break } affected, _ := sqlResult.RowsAffected() execRes.Items = append(execRes.Items, SqlExecItem{ SQL: stmt, Status: "success", AffectedRows: affected, DurationMs: stmtMs, }) } if execErr == "" { if commitErr := tx.Commit(); commitErr != nil { execRes.Status = "failed" execErr = fmt.Sprintf("事务提交失败: %v", commitErr) } else { committed = true } } _ = committed } execRes.TotalMs = time.Since(execStart).Milliseconds() execRes.EndTime = time.Now().UTC().Format(time.RFC3339) if jsonBytes, err := json.Marshal(execRes); err == nil { execResult = string(jsonBytes) } } } executionDone: // Update final status if execErr != "" { failStatus := "failed" ticket.Status = &failStatus ticket.ExecuteResult = &execErr } else { doneStatus := "success" ticket.Status = &doneStatus ticket.ExecuteResult = &execResult } model.DB.Save(&ticket) c.JSON(http.StatusOK, gin.H{"message": "Execution completed", "ticket": ticket}) } func splitSQL(sql string) []string { return strings.Split(sql, ";") } func truncateStr(s string, maxLen int) string { if len(s) <= maxLen { return s } return s[:maxLen] + "..." } // parseAlterTable parses "ALTER TABLE [db.]table ..." → returns (tableName, alterClause, error) // Handles backtick-quoted names: ALTER TABLE `my_db`.`my_table` ADD COLUMN ... func parseAlterTable(sql string) (string, string, error) { // Normalize whitespace s := strings.TrimSpace(sql) upper := strings.ToUpper(s) // Strip leading ALTER TABLE if !strings.HasPrefix(upper, "ALTER TABLE") { return "", "", fmt.Errorf("not an ALTER TABLE statement") } rest := strings.TrimSpace(s[len("ALTER TABLE"):]) // Parse table name (may include db.table or `db`.`table`) tableName := "" alterStart := 0 if rest[0] == '`' { // Backtick-quoted end := strings.Index(rest[1:], "`") if end < 0 { return "", "", fmt.Errorf("unclosed backtick in table name") } tableName = rest[1 : end+1] alterStart = end + 2 } else { // Unquoted — find first space idx := strings.Index(rest, " ") if idx < 0 { return "", "", fmt.Errorf("no alter clause found") } tableName = rest[:idx] alterStart = idx } // Strip database prefix if present (db.table → table) if parts := strings.SplitN(tableName, ".", 2); len(parts) == 2 { tableName = strings.Trim(parts[1], "`") } else { tableName = strings.Trim(tableName, "`") } alterClause := strings.TrimSpace(rest[alterStart:]) if alterClause == "" { return "", "", fmt.Errorf("empty alter clause") } return tableName, alterClause, nil } // POST /api/v1/tickets/:id/cancel func CancelTicket(c *gin.Context) { id := c.Param("id") userID, _, _ := middleware.GetCurrentUser(c) var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } // Only applicant or admin can cancel if ticket.ApplicantID != userID { _, _, role := middleware.GetCurrentUser(c) if role != "admin" { c.JSON(http.StatusForbidden, gin.H{"detail": "Only applicant or admin can cancel"}) return } } if ticket.Status != nil && (*ticket.Status != "pending" && *ticket.Status != "reviewing" && *ticket.Status != "approved") { c.JSON(http.StatusBadRequest, gin.H{"detail": "只能撤销待审核、审核中或已通过(待执行)的工单"}) return } status := "cancelled" ticket.Status = &status now := time.Now() ticket.ReviewTime = &now model.DB.Save(&ticket) c.JSON(http.StatusOK, gin.H{"message": "Ticket cancelled", "ticket": ticket}) } // POST /api/v1/tickets/:id/rollback func RollbackTicket(c *gin.Context) { id := c.Param("id") var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } if ticket.RollbackSQL == nil || *ticket.RollbackSQL == "" { c.JSON(http.StatusBadRequest, gin.H{"detail": "No rollback SQL available"}) return } // Get instance var instance model.Instance if err := model.DB.First(&instance, ticket.InstanceID).Error; err != nil { c.JSON(http.StatusBadRequest, gin.H{"detail": "Instance not found"}) return } if ticket.DBType == "redis" { c.JSON(http.StatusBadRequest, gin.H{"detail": "Redis rollback not supported"}) return } db, err := getDBConnection(&instance, ticket.TargetDatabase) if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)}) return } defer db.Close() // Execute rollback SQL statements := splitSQL(*ticket.RollbackSQL) var results []string var execErr string for _, stmt := range statements { stmt = strings.TrimSpace(stmt) if stmt == "" { continue } _, err := db.Exec(stmt) if err != nil { results = append(results, fmt.Sprintf("[FAIL] %s: %v", truncateStr(stmt, 50), err)) execErr = err.Error() break } else { results = append(results, fmt.Sprintf("[OK] %s", truncateStr(stmt, 50))) } } if execErr != "" { c.JSON(http.StatusInternalServerError, gin.H{"detail": "Rollback failed", "error": execErr, "results": results}) return } status := "rollback" ticket.Status = &status execResult := strings.Join(results, "\n") ticket.ExecuteResult = &execResult model.DB.Save(&ticket) c.JSON(http.StatusOK, gin.H{"message": "Rollback completed", "results": results}) } // GET /api/v1/tickets/:id/timeline func GetTicketTimeline(c *gin.Context) { id := c.Param("id") var ticket model.Ticket if err := model.DB.First(&ticket, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"detail": "Ticket not found"}) return } type TimelineEntry struct { ID int `json:"id"` Action string `json:"action"` User string `json:"user"` Comment *string `json:"comment"` Timestamp string `json:"timestamp"` Status string `json:"status"` } var timeline []TimelineEntry // 1. Created event timeline = append(timeline, TimelineEntry{ ID: 1, Action: "created", User: getApplicantName(ticket.ApplicantID), Timestamp: ticket.CreateTime.Format(time.RFC3339), Status: "pending", }) // 2. Approval events from ticket_approvals var approvals []model.TicketApproval model.DB.Where("ticket_id = ?", id).Order("step_number").Find(&approvals) for _, a := range approvals { action := "pending_review" userName := getUserName(a.ActualApproverID) if a.Action != nil { action = *a.Action } ts := a.CreatedAt.Format(time.RFC3339) if a.CompletedAt != nil { ts = a.CompletedAt.Format(time.RFC3339) } timeline = append(timeline, TimelineEntry{ ID: a.ID, Action: action, User: userName, Comment: a.Comment, Timestamp: ts, }) } // 3. Executed event if ticket.ExecuteTime != nil { execAction := "executed" if ticket.Status != nil && *ticket.Status == "failed" { execAction = "execution_failed" } else if ticket.Status != nil && *ticket.Status == "completed" { execAction = "executed" } timeline = append(timeline, TimelineEntry{ ID: 99, Action: execAction, Timestamp: ticket.ExecuteTime.Format(time.RFC3339), Status: func() string { if ticket.Status != nil { return *ticket.Status }; return "" }(), }) } c.JSON(http.StatusOK, gin.H{"timeline": timeline}) } // GET /api/v1/tickets/:id/approvals func GetTicketApprovals(c *gin.Context) { id := c.Param("id") var approvals []model.TicketApproval if err := model.DB.Where("ticket_id = ?", id).Order("step_number").Find(&approvals).Error; err != nil { c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()}) return } // Enrich with approver names type ApprovalResponse struct { model.TicketApproval ApproverName string `json:"approver_name"` } var result []ApprovalResponse for _, a := range approvals { ar := ApprovalResponse{TicketApproval: a} ar.ApproverName = getUserName(a.ApproverID) result = append(result, ar) } c.JSON(http.StatusOK, result) } // Helper: get user name by ID (pointer-safe) func getUserName(userID *int) string { if userID == nil { return "" } var user model.User if model.DB.First(&user, *userID).Error == nil { return user.DisplayName } return "" } // Helper: get applicant name by ID func getApplicantName(userID int) string { var user model.User if model.DB.First(&user, userID).Error == nil { return user.DisplayName } return "" } // isClusterInstance checks if a Redis instance is configured as cluster mode func isClusterInstance(instance *model.Instance) bool { return instance.ClusterMode != nil && *instance.ClusterMode } // parseMovedTarget extracts the target address from a MOVED response // Input: "MOVED 8480 172.19.0.4:9002" → Output: "172.19.0.4:9002" func parseMovedTarget(errMsg string) string { // MOVED : if !strings.HasPrefix(errMsg, "MOVED ") { return "" } parts := strings.SplitN(errMsg, " ", 3) if len(parts) == 3 { return parts[2] } return "" }