opsxsql/backend/internal/handler/ticket.go

package handler

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"regexp"
	"strconv"
	"strings"
	"time"

	"sql-platform/internal/ghost"
	"sql-platform/internal/handler/resp"
	"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)
		// 无执行步配置的工单(默认流程):approved 状态下 admin/dba 为隐式执行人
		if isExec == 0 && execCount == 0 && (role == "admin" || role == "dba") && t.Status != nil && *t.Status == "approved" {
			isExec = 1
		}
		// Only count PENDING approvals (action IS NULL) — already-approved records should not show approve button
		model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type != 'execute' AND step_type != 'submit' AND approver_id = ? AND action IS NULL", 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 {
		resp.NotFound(c, "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" {
		resp.Forbidden(c, "权限不足：审计员不能提交工单")
		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 {
		resp.BadRequest(c, 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) {
				resp.BadRequest(c, "数据变更工单不支持 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 {
				resp.BadRequest(c, 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 {
		resp.InternalErr(c, err)
		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" {
		resp.Forbidden(c, "权限不足：审核操作需要管理员、DBA 或开发者角色")
		return
	}

	var ticket model.Ticket
	if err := model.DB.First(&ticket, id).Error; err != nil {
		resp.NotFound(c, "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 {
		resp.BadRequest(c, err.Error())
		return
	}

	now := time.Now()

	// Find PENDING approval for this user (action IS NULL)
	var approval model.TicketApproval
	if err := model.DB.Where("ticket_id = ? AND approver_id = ? AND action IS NULL AND step_type != 'execute'", id, userID).First(&approval).Error; err != nil {
		// Already approved or no pending approval for this user
		resp.BadRequest(c, "没有待处理的审批记录（你可能已经审批过了）")
		return
	}

	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" {
		resp.Forbidden(c, "权限不足：执行操作需要管理员或 DBA 角色")
		return
	}

	var ticket model.Ticket
	if err := model.DB.First(&ticket, id).Error; err != nil {
		resp.NotFound(c, "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" {
		resp.BadRequest(c, "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 {
		// 无执行步配置的工单(默认流程):admin/dba 为隐式执行人(与列表 is_executor 判定一致)
		var execStepCount int64
		model.DB.Model(&model.TicketApproval{}).Where("ticket_id = ? AND step_type = 'execute'", id).Count(&execStepCount)
		_, _, role := middleware.GetCurrentUser(c)
		if execStepCount > 0 || (role != "admin" && role != "dba") {
			resp.Forbidden(c, "你不是此工单的执行人")
			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 {
		resp.BadRequest(c, "还有未完成的审核步骤")
		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 {
		resp.BadRequest(c, "审核已被拒绝，无法执行")
		return
	}

	// Get instance info
	var instance model.Instance
	if err := model.DB.First(&instance, ticket.InstanceID).Error; err != nil {
		resp.BadRequest(c, "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 {
		resp.NotFound(c, "Ticket not found")
		return
	}

	// Only applicant or admin can cancel
	if ticket.ApplicantID != userID {
		_, _, role := middleware.GetCurrentUser(c)
		if role != "admin" {
			resp.Forbidden(c, "Only applicant or admin can cancel")
			return
		}
	}

	if ticket.Status != nil && (*ticket.Status != "pending" && *ticket.Status != "reviewing" && *ticket.Status != "approved") {
		resp.BadRequest(c, "只能撤销待审核、审核中或已通过（待执行）的工单")
		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 {
		resp.NotFound(c, "Ticket not found")
		return
	}

	if ticket.RollbackSQL == nil || *ticket.RollbackSQL == "" {
		resp.BadRequest(c, "No rollback SQL available")
		return
	}

	// Get instance
	var instance model.Instance
	if err := model.DB.First(&instance, ticket.InstanceID).Error; err != nil {
		resp.BadRequest(c, "Instance not found")
		return
	}

	if ticket.DBType == "redis" {
		resp.BadRequest(c, "Redis rollback not supported")
		return
	}

	db, err := getDBConnection(&instance, ticket.TargetDatabase)
	if err != nil {
		resp.InternalErr(c, 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 {
		resp.NotFound(c, "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 {
		resp.InternalErr(c, err)
		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 <slot> <host>:<port>
	if !strings.HasPrefix(errMsg, "MOVED ") {
		return ""
	}
	parts := strings.SplitN(errMsg, " ", 3)
	if len(parts) == 3 {
		return parts[2]
	}
	return ""
}