opsxsql/backend/internal/handler/instance_detail.go

package handler

import (
	"fmt"
	"net/http"
	"reflect"

	"sql-platform/internal/handler/resp"
	"sql-platform/internal/model"

	"bytes"
	"context"
	"encoding/json"
	"github.com/gin-gonic/gin"
	"go.mongodb.org/mongo-driver/v2/bson"
	"go.mongodb.org/mongo-driver/v2/mongo"
	"go.mongodb.org/mongo-driver/v2/mongo/options"
	"go.mongodb.org/mongo-driver/v2/mongo/readpref"
	"regexp"
	"sort"
	"sql-platform/internal/middleware"
	"strconv"
	"strings"
	"time"
)

// GET /api/v1/instances/:id/detail/databases
func DetailListDatabases(c *gin.Context) {
	id := c.Param("id")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, []string{})
		return
	}

	db, err := getDBConnection(&instance, "")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	if instance.DBType == "postgresql" || instance.DBType == "postgres" || instance.DBType == "pg" {
		pgRows, err := db.Query("SELECT datname FROM pg_database WHERE NOT datistemplate AND datallowconn ORDER BY 1")
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		defer pgRows.Close()
		var databases []string
		for pgRows.Next() {
			var name string
			pgRows.Scan(&name)
			databases = append(databases, name)
		}
		if databases == nil {
			databases = []string{}
		}
		c.JSON(http.StatusOK, databases)
		return
	}

	rows, err := db.Query("SHOW DATABASES")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer rows.Close()

	var databases []string
	for rows.Next() {
		var name string
		rows.Scan(&name)
		// Filter out system databases
		if name != "information_schema" && name != "performance_schema" && name != "mysql" && name != "sys" {
			databases = append(databases, name)
		}
	}
	if databases == nil {
		databases = []string{}
	}
	c.JSON(http.StatusOK, databases)
}

// GET /api/v1/instances/:id/detail/tables
func DetailListTables(c *gin.Context) {
	id := c.Param("id")
	dbName := c.Query("database")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, []interface{}{})
		return
	}

	if dbName == "" {
		if instance.Database != nil {
			dbName = *instance.Database
		}
	}
	if dbName == "" {
		resp.BadRequest(c, "database parameter required")
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	if instance.DBType == "postgresql" || instance.DBType == "postgres" || instance.DBType == "pg" {
		pgRows, err := db.Query(`SELECT c.relname AS name, c.reltuples::bigint AS rows,
			pg_total_relation_size(c.oid) AS data_length
			FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
			WHERE n.nspname NOT IN ('pg_catalog', 'information_schema') AND c.relkind = 'r'
			ORDER BY c.relname`)
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		defer pgRows.Close()
		var tables []map[string]interface{}
		for pgRows.Next() {
			var name string
			var rowCount, dataLen int64
			pgRows.Scan(&name, &rowCount, &dataLen)
			tables = append(tables, map[string]interface{}{
				"name": name, "rows": rowCount, "data_length": dataLen, "engine": "postgres",
			})
		}
		if tables == nil {
			tables = []map[string]interface{}{}
		}
		c.JSON(http.StatusOK, tables)
		return
	}

	rows, err := db.Query("SHOW TABLE STATUS")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer rows.Close()

	cols, _ := rows.Columns()
	var tables []map[string]interface{}
	for rows.Next() {
		values := make([]interface{}, len(cols))
		ptrValues := make([]interface{}, len(cols))
		for i := range values {
			ptrValues[i] = &values[i]
		}
		rows.Scan(ptrValues...)
		// Normalize to lowercase keys for frontend compatibility
		rowMap := make(map[string]interface{})
		for i, col := range cols {
			val := values[i]
			b, ok := val.([]byte)
			if ok {
				rowMap[strings.ToLower(col)] = string(b)
			} else {
				rowMap[strings.ToLower(col)] = val
			}
		}
		tables = append(tables, rowMap)
	}
	if tables == nil {
		tables = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, tables)
}

// GET /api/v1/instances/:id/detail/db-schema
func DetailDbSchema(c *gin.Context) {
	id := c.Param("id")
	dbName := c.Query("database")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, map[string]interface{}{})
		return
	}

	if dbName == "" {
		if instance.Database != nil {
			dbName = *instance.Database
		}
	}
	if dbName == "" {
		resp.BadRequest(c, "database parameter required")
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	schema := make(map[string][]string)

	if instance.DBType == "postgresql" || instance.DBType == "postgres" || instance.DBType == "pg" {
		// PG:一次查询拿全部 库内表.列(information_schema)
		colRows, err := db.Query(`SELECT table_name, column_name FROM information_schema.columns
			WHERE table_schema NOT IN ('pg_catalog', 'information_schema')
			ORDER BY table_name, ordinal_position`)
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		defer colRows.Close()
		for colRows.Next() {
			var tableName, columnName string
			colRows.Scan(&tableName, &columnName)
			schema[tableName] = append(schema[tableName], columnName)
		}
		c.JSON(http.StatusOK, schema)
		return
	}

	// Get all tables and their columns
	tableRows, err := db.Query("SHOW TABLES")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer tableRows.Close()

	for tableRows.Next() {
		var tableName string
		tableRows.Scan(&tableName)
		// Get columns for this table
		colRows, err := db.Query("SHOW COLUMNS FROM " + "`" + tableName + "`")
		if err != nil {
			schema[tableName] = []string{}
			continue
		}
		var cols []string
		for colRows.Next() {
			var field, typ, null, key string
			var extra interface{}
			var def interface{}
			colRows.Scan(&field, &typ, &null, &key, &def, &extra)
			cols = append(cols, field+" ("+typ+")")
		}
		colRows.Close()
		if cols == nil {
			cols = []string{}
		}
		schema[tableName] = cols
	}
	c.JSON(http.StatusOK, schema)
}

// GET /api/v1/instances/:id/detail/table-detail
func DetailTableDetail(c *gin.Context) {
	id := c.Param("id")
	dbName := c.Query("database")
	tableName := c.Query("table")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if dbName == "" {
		if instance.Database != nil {
			dbName = *instance.Database
		}
	}
	if dbName == "" || tableName == "" {
		resp.BadRequest(c, "database and table parameters required")
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	if instance.DBType == "postgresql" || instance.DBType == "postgres" || instance.DBType == "pg" {
		// PG:information_schema 拼列 + pg_indexes 拿索引(无 SHOW CREATE,DDL 以列清单形式给出)
		colRows, err := db.Query(`SELECT column_name, data_type, is_nullable,
			COALESCE(column_default, ''), character_maximum_length
			FROM information_schema.columns
			WHERE table_schema NOT IN ('pg_catalog', 'information_schema') AND table_name = $1
			ORDER BY ordinal_position`, tableName)
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		defer colRows.Close()
		var columns []map[string]interface{}
		var ddlLines []string
		for colRows.Next() {
			var name, dataType, nullable, def string
			var maxLen interface{}
			colRows.Scan(&name, &dataType, &nullable, &def, &maxLen)
			nullStr := "YES"
			if nullable != "YES" {
				nullStr = "NO"
			}
			key := ""
			pkRows, _ := db.Query(`SELECT a.attname FROM pg_index i
				JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
				WHERE i.indrelid = to_regclass($1) AND i.indisprimary`, tableName)
			if pkRows != nil {
				for pkRows.Next() {
					var pk string
					pkRows.Scan(&pk)
					if pk == name {
						key = "PRI"
					}
				}
				pkRows.Close()
			}
			columns = append(columns, map[string]interface{}{
				"Field": name, "Type": dataType, "Null": nullStr, "Key": key, "Default": def, "Extra": "",
			})
			ddlLines = append(ddlLines, "  "+name+" "+dataType+(func() string {
				if nullStr == "NO" {
					return " NOT NULL"
				}
				return ""
			})())
		}
		ddl := "CREATE TABLE " + tableName + " (\n" + strings.Join(ddlLines, ",\n") + "\n);"

		idxRows, err := db.Query(`SELECT indexname, indexdef FROM pg_indexes
			WHERE schemaname NOT IN ('pg_catalog', 'information_schema') AND tablename = $1
			ORDER BY indexname`, tableName)
		var indexes []map[string]interface{}
		if err == nil {
			defer idxRows.Close()
			for idxRows.Next() {
				var idxName, idxDef string
				idxRows.Scan(&idxName, &idxDef)
				nonUnique := !strings.Contains(idxDef, " UNIQUE INDEX ")
				indexes = append(indexes, map[string]interface{}{
					"Key_name": idxName, "Column_name": strings.Trim(idxDef[strings.LastIndex(idxDef, "(")+1:], ")"), "Non_unique": nonUnique, "Index_type": "btree",
				})
			}
		}
		if indexes == nil {
			indexes = []map[string]interface{}{}
		}
		if columns == nil {
			columns = []map[string]interface{}{}
		}
		c.JSON(http.StatusOK, gin.H{"ddl": ddl, "columns": columns, "indexes": indexes})
		return
	}

	// Get DDL
	var ddl string
	db.QueryRow("SHOW CREATE TABLE `"+tableName+"`").Scan(&tableName, &ddl)

	// Get columns
	colRows, err := db.Query("SHOW FULL COLUMNS FROM `" + tableName + "`")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer colRows.Close()

	colCols, _ := colRows.Columns()
	var columns []map[string]interface{}
	for colRows.Next() {
		values := make([]interface{}, len(colCols))
		ptrValues := make([]interface{}, len(colCols))
		for i := range values {
			ptrValues[i] = &values[i]
		}
		colRows.Scan(ptrValues...)
		rowMap := make(map[string]interface{})
		for i, col := range colCols {
			val := values[i]
			b, ok := val.([]byte)
			if ok {
				rowMap[col] = string(b)
			} else {
				rowMap[col] = val
			}
		}
		columns = append(columns, rowMap)
	}

	// Get indexes
	idxRows, err := db.Query("SHOW INDEX FROM `" + tableName + "`")
	if err == nil {
		defer idxRows.Close()
		idxCols, _ := idxRows.Columns()
		var indexes []map[string]interface{}
		for idxRows.Next() {
			values := make([]interface{}, len(idxCols))
			ptrValues := make([]interface{}, len(idxCols))
			for i := range values {
				ptrValues[i] = &values[i]
			}
			idxRows.Scan(ptrValues...)
			rowMap := make(map[string]interface{})
			for i, col := range idxCols {
				val := values[i]
				b, ok := val.([]byte)
				if ok {
					rowMap[col] = string(b)
				} else {
					rowMap[col] = val
				}
			}
			indexes = append(indexes, rowMap)
		}
		c.JSON(http.StatusOK, gin.H{
			"ddl":     ddl,
			"columns": columns,
			"indexes": indexes,
		})
	} else {
		c.JSON(http.StatusOK, gin.H{
			"ddl":     ddl,
			"columns": columns,
			"indexes": []interface{}{},
		})
	}
}

// GET /api/v1/instances/:id/detail/processlist
func DetailProcesslist(c *gin.Context) {
	id := c.Param("id")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, []interface{}{})
		return
	}

	dbName := ""
	if instance.Database != nil {
		dbName = *instance.Database
	}
	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW PROCESSLIST")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer rows.Close()

	cols, _ := rows.Columns()
	var processes []map[string]interface{}
	for rows.Next() {
		values := make([]interface{}, len(cols))
		ptrValues := make([]interface{}, len(cols))
		for i := range values {
			ptrValues[i] = &values[i]
		}
		rows.Scan(ptrValues...)
		rowMap := make(map[string]interface{})
		for i, col := range cols {
			val := values[i]
			b, ok := val.([]byte)
			if ok {
				rowMap[col] = string(b)
			} else {
				rowMap[col] = val
			}
		}
		processes = append(processes, rowMap)
	}
	if processes == nil {
		processes = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, processes)
}

// GET /api/v1/instances/:id/detail/status
func DetailStatus(c *gin.Context) {
	id := c.Param("id")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, gin.H{
			"key_metrics": map[string]string{},
			"all_status":  []interface{}{},
		})
		return
	}

	dbName := ""
	if instance.Database != nil {
		dbName = *instance.Database
	}
	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW GLOBAL STATUS")
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer rows.Close()

	keyMetrics := make(map[string]string)
	var allStatus []map[string]interface{}
	importantKeys := map[string]bool{
		"Threads_connected": true, "Threads_running": true, "Threads_cached": true,
		"Queries": true, "Questions": true, "Slow_queries": true,
		"Connections": true, "Uptime": true,
		"Bytes_received": true, "Bytes_sent": true,
		"Table_locks_waited": true, "Table_locks_immediate": true,
		"Innodb_row_lock_waits": true, "Innodb_buffer_pool_reads": true,
	}

	for rows.Next() {
		var name, val string
		rows.Scan(&name, &val)
		allStatus = append(allStatus, map[string]interface{}{"name": name, "value": val})
		if importantKeys[name] {
			keyMetrics[name] = val
		}
	}
	if allStatus == nil {
		allStatus = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, gin.H{
		"key_metrics": keyMetrics,
		"all_status":  allStatus,
	})
}

// POST /api/v1/instances/:id/detail/processlist/:pid/kill
func DetailKillProcess(c *gin.Context) {
	id := c.Param("id")
	pid := c.Param("pid")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	dbName := ""
	if instance.Database != nil {
		dbName = *instance.Database
	}
	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	_, err = db.Exec("KILL " + pid)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	c.JSON(http.StatusOK, gin.H{"message": "Process " + pid + " killed"})
}

// GET /api/v1/instances/:id/detail/users
func DetailListUsers(c *gin.Context) {
	id := c.Param("id")
	var instance model.Instance
	if err := model.DB.First(&instance, id).Error; err != nil {
		resp.NotFound(c, "Instance not found")
		return
	}

	if instance.DBType == "redis" {
		c.JSON(http.StatusOK, []interface{}{})
		return
	}

	dbName := ""
	if instance.Database != nil {
		dbName = *instance.Database
	}
	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer db.Close()

	rows, err := db.Query("SELECT User, Host FROM mysql.user")
	if err != nil {
		// Might not have permission - return empty
		c.JSON(http.StatusOK, []interface{}{})
		return
	}
	defer rows.Close()

	var users []map[string]interface{}
	for rows.Next() {
		var user, host string
		rows.Scan(&user, &host)
		users = append(users, map[string]interface{}{
			"user": user,
			"host": host,
		})
	}
	if users == nil {
		users = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, users)
}

// getDBConnection is defined in connection_tester.go or instance_detail.go
// Using the one from connection_tester.go

// POST /api/v1/query/execute
func ExecuteQuery(c *gin.Context) {
	userID, username, _ := middleware.GetCurrentUser(c)
	var input struct {
		InstanceID int    `json:"instance_id"`
		Database   string `json:"database"`
		SQL        string `json:"sql"`
		Limit      int    `json:"limit"`
		Offset     int    `json:"offset"`
	}
	if err := c.ShouldBindJSON(&input); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}
	if input.InstanceID == 0 || input.SQL == "" {
		resp.BadRequest(c, "instance_id and sql are required")
		return
	}
	// 查询行数上限统一由系统设置 query_limit_max 决定（默认 5000）。
	// 注意：历史上这里还有一段 `if input.Limit > 1000 { input.Limit = 1000 }` 的硬编码，
	// 导致管理员把上限调到 5000 也不生效（前端不传 limit 时永远被压回 1000）。
	// 现已移除，唯一口径就是设置值。
	maxLimit := getSettingInt("query_limit_max", 5000)
	if maxLimit <= 0 {
		maxLimit = 5000
	}
	if input.Limit <= 0 {
		input.Limit = maxLimit
	}
	if input.Limit > maxLimit {
		input.Limit = maxLimit
	}
	if input.Offset < 0 {
		input.Offset = 0
	}

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

	// Dangerous pattern check
	if getSetting("query_dangerous_pattern", "true") == "true" {
		upper := strings.ToUpper(input.SQL)
		for _, pat := range []string{"DELETE FROM", "UPDATE ", "DROP ", "TRUNCATE", "ALTER ", "RENAME "} {
			if strings.Contains(upper, pat) && !strings.Contains(upper, " WHERE ") && !strings.Contains(upper, " LIMIT ") {
				saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, 0, 0, "failed", "危险操作检测: "+pat+" 缺少 WHERE/LIMIT 条件，请在工单中提交", true)
				c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": 0,
					"error": "危险操作检测: " + pat + " 缺少 WHERE/LIMIT 条件，请在工单中提交"})
				return
			}
		}
	}

	start := time.Now()

	switch instance.DBType {
	case "mysql", "postgresql", "postgres", "pg":
		result, err := executeSQLQuery(&instance, input.Database, input.SQL, input.Limit, input.Offset)
		costMs := time.Since(start).Milliseconds()
		if err != nil {
			saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, 0, "failed", err.Error(), false)
			c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()})
			return
		}
		result["cost_ms"] = costMs
		rowsInterface, _ := result["rows"].([]interface{})
		applyDataMasking(result, &instance)
		saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, len(rowsInterface))
		saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, len(rowsInterface), "success", "", false)
		c.JSON(http.StatusOK, result)

	case "redis":
		result, err := executeRedisQuery(&instance, input.Database, input.SQL, input.Limit, input.Offset)
		costMs := time.Since(start).Milliseconds()
		if err != nil {
			saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, 0, "failed", err.Error(), false)
			c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()})
			return
		}
		result["cost_ms"] = costMs
		rowCount := 1
		if rowsArr, ok := result["rows"].([]interface{}); ok {
			rowCount = len(rowsArr)
		}
		applyDataMasking(result, &instance)
		saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, rowCount)
		saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, rowCount, "success", "", false)
		c.JSON(http.StatusOK, result)

	case "mongodb", "mongo":
		result, err := executeMongoQuery(&instance, input.Database, input.SQL, input.Limit, input.Offset)
		costMs := time.Since(start).Milliseconds()
		if err != nil {
			saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, 0, "failed", err.Error(), false)
			c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()})
			return
		}
		result["cost_ms"] = costMs
		rowsInterface, _ := result["rows"].([]interface{})
		applyDataMasking(result, &instance)
		saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, len(rowsInterface))
		saveQueryAuditLog(c, userID, username, &instance, input.Database, input.SQL, costMs, len(rowsInterface), "success", "", false)
		c.JSON(http.StatusOK, result)

	default:
		resp.BadRequest(c, "Unsupported database type: "+instance.DBType)
	}
}

func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, limit int, offset int) (gin.H, error) {
	db, err := getDBConnection(instance, dbName)
	if err != nil {
		return nil, fmt.Errorf("connection failed: %v", err)
	}
	defer db.Close()

	// Apply query timeout from platform settings
	timeoutSec := getSettingInt("query_timeout", 30)
	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(timeoutSec)*time.Second)
	defer cancel()
	_ = ctx

	upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))
	if ok, reason := isReadOnlySQL(sqlStr); !ok {
		return nil, fmt.Errorf("查询入口仅允许只读语句：%s。数据变更请提交工单", reason)
	}

	// Count total (only when no user LIMIT and offset=0)
	total := 0
	hasLimit := strings.Contains(upperSQL, " LIMIT ")

	if !hasLimit && offset == 0 {
		if strings.HasPrefix(upperSQL, "SELECT") {
			fromIdx := strings.Index(upperSQL, " FROM ")
			if fromIdx > 0 {
				countSQL := "SELECT COUNT(*) " + sqlStr[fromIdx:]
				upperCount := strings.ToUpper(countSQL)
				if idx := strings.Index(upperCount, " ORDER BY "); idx > 0 {
					countSQL = countSQL[:idx]
				}
				if idx := strings.Index(upperCount, " GROUP BY "); idx > 0 {
					countSQL = countSQL[:idx]
				}
				db.QueryRowContext(ctx, countSQL).Scan(&total)
			}
		}
	}

	// Inject LIMIT/OFFSET if not already present — only for SELECT statements
	// SHOW/DESC/EXPLAIN/CREATE/ALTER/DROP/etc do not support LIMIT syntax
	autoLimit := getSetting("query_auto_limit", "true") == "true"
	needsLimit := autoLimit && !hasLimit && isLimitableReadSQL(upperSQL)
	if needsLimit {
		sqlStr = fmt.Sprintf("%s LIMIT %d OFFSET %d", sqlStr, limit, offset)
	}

	rows, err := db.QueryContext(ctx, sqlStr)
	if err != nil {
		return nil, fmt.Errorf("query failed: %v", err)
	}
	defer rows.Close()

	cols, _ := rows.Columns()
	colNames := make([]string, len(cols))
	for i, col := range cols {
		colNames[i] = col
	}

	var resultRows []interface{}
	for rows.Next() {
		values := make([]interface{}, len(cols))
		ptrValues := make([]interface{}, len(cols))
		for i := range values {
			ptrValues[i] = &values[i]
		}
		rows.Scan(ptrValues...)
		rowMap := make(map[string]interface{})
		for i, col := range colNames {
			val := values[i]
			b, ok := val.([]byte)
			if ok {
				rowMap[col] = string(b)
			} else {
				rowMap[col] = val
			}
		}
		resultRows = append(resultRows, rowMap)
	}

	return gin.H{
		"columns": colNames,
		"rows":    resultRows,
		"total":   total,
		"count":   len(resultRows),
	}, nil
}

func isReadOnlySQL(sqlStr string) (bool, string) {
	cleaned := strings.TrimSpace(stripLeadingSQLComments(sqlStr))
	if cleaned == "" {
		return false, "空语句"
	}
	if hasMultipleSQLStatements(cleaned) {
		return false, "不允许一次执行多条语句"
	}

	upper := strings.ToUpper(cleaned)
	readPrefixes := []string{"SELECT", "SHOW", "DESCRIBE", "DESC", "EXPLAIN", "WITH"}
	allowed := false
	for _, prefix := range readPrefixes {
		if strings.HasPrefix(upper, prefix+" ") || upper == prefix {
			allowed = true
			break
		}
	}
	if !allowed {
		return false, "只允许 SELECT / SHOW / DESCRIBE / EXPLAIN"
	}

	blockedPatterns := []string{
		" FOR UPDATE", " LOCK IN SHARE MODE", " INTO OUTFILE", " INTO DUMPFILE",
		"EXPLAIN ANALYZE", "DELETE ", "UPDATE ", "INSERT ", "REPLACE ",
		"CREATE ", "ALTER ", "DROP ", "TRUNCATE ", "RENAME ", "GRANT ", "REVOKE ",
		"CALL ", "DO ", "LOAD DATA", "COPY ",
	}
	for _, pattern := range blockedPatterns {
		if strings.Contains(upper, pattern) {
			return false, strings.TrimSpace(pattern)
		}
	}
	return true, ""
}

func isLimitableReadSQL(upperSQL string) bool {
	return strings.HasPrefix(upperSQL, "SELECT") || strings.HasPrefix(upperSQL, "WITH")
}

func hasMultipleSQLStatements(sqlStr string) bool {
	trimmed := strings.TrimSpace(sqlStr)
	if strings.HasSuffix(trimmed, ";") {
		trimmed = strings.TrimSpace(strings.TrimSuffix(trimmed, ";"))
	}
	return strings.Contains(trimmed, ";")
}

func stripLeadingSQLComments(sqlStr string) string {
	s := strings.TrimSpace(sqlStr)
	for {
		switch {
		case strings.HasPrefix(s, "--"):
			if idx := strings.Index(s, "\n"); idx >= 0 {
				s = strings.TrimSpace(s[idx+1:])
				continue
			}
			return ""
		case strings.HasPrefix(s, "#"):
			if idx := strings.Index(s, "\n"); idx >= 0 {
				s = strings.TrimSpace(s[idx+1:])
				continue
			}
			return ""
		case strings.HasPrefix(s, "/*"):
			if idx := strings.Index(s, "*/"); idx >= 0 {
				s = strings.TrimSpace(s[idx+2:])
				continue
			}
			return s
		default:
			return s
		}
	}
}

// parseCommandParts splits a command string into parts, respecting quoted strings.
// e.g. HGET "my key" "my field" => ["HGET", "my key", "my field"]
func parseCommandParts(input string) []string {
	var parts []string
	var current strings.Builder
	inQuote := false
	quoteChar := byte(0)

	for i := 0; i < len(input); i++ {
		ch := input[i]
		if inQuote {
			if ch == quoteChar {
				inQuote = false
				quoteChar = 0
			} else {
				current.WriteByte(ch)
			}
		} else {
			switch ch {
			case '"', '\'':
				if current.Len() > 0 {
					parts = append(parts, current.String())
					current.Reset()
				}
				inQuote = true
				quoteChar = ch
			case ' ', '\t', '\n', '\r':
				if current.Len() > 0 {
					parts = append(parts, current.String())
					current.Reset()
				}
			default:
				current.WriteByte(ch)
			}
		}
	}
	if current.Len() > 0 {
		parts = append(parts, current.String())
	}
	return parts
}

// rowsOrDefault ensures rows is never nil (JSON null), always an empty slice
func rowsOrDefault(rows []interface{}) []interface{} {
	if rows == nil {
		return []interface{}{}
	}
	return rows
}

func executeRedisQuery(instance *model.Instance, dbSelect string, command string, limit int, offset int) (gin.H, error) {
	// Parse Redis DB number from dbSelect (e.g. "0", "1")
	redisDB := 0
	if dbSelect != "" {
		if n, err := strconv.Atoi(dbSelect); err == nil && n >= 0 {
			redisDB = n
		}
	}

	client, err := getRedisClient(instance, redisDB)
	if err != nil {
		return nil, err
	}
	defer client.Close()

	ctx := context.Background()

	parts := parseCommandParts(command)
	if len(parts) == 0 {
		return nil, fmt.Errorf("empty command")
	}

	cmd := strings.ToUpper(parts[0])
	keyName := ""
	if len(parts) > 1 {
		keyName = parts[1]
	}
	readOnly := map[string]bool{
		"GET": true, "MGET": true, "HGET": true, "HGETALL": true,
		"LRANGE": true, "SMEMBERS": true, "ZRANGE": true,
		"TYPE": true, "TTL": true, "PTTL": true, "EXISTS": true, "KEYS": true,
		"INFO": true, "DBSIZE": true, "SCAN": true, "STRLEN": true,
		"LLEN": true, "SCARD": true, "ZCARD": true, "HLEN": true,
		"HKEYS": true, "HVALS": true, "HSCAN": true, "SSCAN": true,
		"ZSCAN": true, "ZRANGEBYSCORE": true,
		"GETSET": true, "GETRANGE": true, "SUBSTR": true,
		"PING": true, "CONFIG": true, "CLIENT": true, "SLOWLOG": true,
		"CLUSTER": true, "MEMORY": true, "OBJECT": true, "DEBUG": true,
		"SRANDMEMBER": true, "RANDOMKEY": true, "LASTSAVE": true,
		"ROLE": true, "REPLCONF": true, "SENTINEL": true,
		"LATENCY": true, "MODULE": true, "COMMAND": true,
	}
	if !readOnly[cmd] {
		return nil, fmt.Errorf("only read-only commands allowed, got: %s", cmd)
	}

	// Build args: command name + all arguments
	args := make([]interface{}, 0, len(parts))
	args = append(args, cmd)
	for _, p := range parts[1:] {
		args = append(args, p)
	}

	// For KEYS command, use ScanAllKeys (cluster-aware) with pagination
	if cmd == "KEYS" {
		pattern := "*"
		if len(parts) > 1 {
			pattern = parts[1]
		}
		// Auto-append * if pattern has no wildcard
		if !strings.Contains(pattern, "*") && !strings.Contains(pattern, "?") && !strings.Contains(pattern, "[") {
			pattern = pattern + "*"
		}
		allKeys, err := client.ScanAllKeys(ctx, pattern, limit+offset)
		if err != nil {
			return nil, fmt.Errorf("SCAN failed: %v", err)
		}
		total := len(allKeys)
		end := offset + limit
		if end > total {
			end = total
		}
		if offset > total {
			offset = total
		}
		pagedKeys := allKeys[offset:end]
		rows := make([]interface{}, len(pagedKeys))
		for i, k := range pagedKeys {
			rows[i] = map[string]interface{}{"index": offset + i, "key": k}
		}
		return gin.H{
			"columns": []string{"index", "key"},
			"rows":    rows,
			"total":   total,
			"count":   len(rows),
		}, nil
	}

	// For LRANGE, inject pagination if user didn't specify start/stop
	if cmd == "LRANGE" && len(parts) < 4 {
		key := ""
		if len(parts) > 1 {
			key = parts[1]
		}
		result, err := client.LRange(ctx, key, int64(offset), int64(offset+limit-1)).Result()
		if err != nil {
			return nil, fmt.Errorf("command failed: %v", err)
		}
		// Get total via LLEN
		total, _ := client.LLen(ctx, key).Result()
		rows := make([]interface{}, len(result))
		for i, val := range result {
			rows[i] = map[string]interface{}{"index": offset + i, "value": val}
		}
		return gin.H{
			"columns": []string{"index", "value"},
			"rows":    rows,
			"total":   int(total),
			"count":   len(rows),
		}, nil
	}

	// For SMEMBERS with limit, use SSCAN
	if cmd == "SMEMBERS" && limit > 0 {
		key := ""
		if len(parts) > 1 {
			key = parts[1]
		}
		total, _ := client.SCard(ctx, key).Result()
		var members []string
		cursor := uint64(0)
		for len(members) < offset+limit {
			batch, nextCursor, err := client.SScan(ctx, key, cursor, "", int64(offset+limit-len(members))).Result()
			if err != nil {
				break
			}
			members = append(members, batch...)
			cursor = nextCursor
			if cursor == 0 {
				break
			}
		}
		end := offset + limit
		if end > len(members) {
			end = len(members)
		}
		paged := members[offset:end]
		rows := make([]interface{}, len(paged))
		for i, m := range paged {
			rows[i] = map[string]interface{}{"index": offset + i, "member": m}
		}
		return gin.H{
			"columns": []string{"index", "member"},
			"rows":    rows,
			"total":   int(total),
			"count":   len(rows),
		}, nil
	}

	result, err := client.Do(ctx, args...).Result()
	if err != nil {
		return nil, fmt.Errorf("command failed: %v", err)
	}

	keyMeta := gin.H{}
	if keyName != "" && cmd != "TYPE" && cmd != "TTL" && cmd != "PTTL" && cmd != "EXISTS" && cmd != "DBSIZE" && cmd != "INFO" {
		keyMeta["raw_key"] = keyName
		if t, err := client.Type(ctx, keyName).Result(); err == nil {
			keyMeta["raw_type"] = t
		}
		if ttl, err := client.TTL(ctx, keyName).Result(); err == nil {
			if ttl < 0 {
				keyMeta["raw_ttl"] = "-1"
			} else {
				keyMeta["raw_ttl"] = fmt.Sprintf("%d", int(ttl.Seconds()))
			}
		}
	}

	// Format result into rows/columns for frontend table display
	columns := []string{"key", "value"}
	var rows []interface{}

	switch v := result.(type) {
	case []interface{}:
		// ZRANGE WITHSCORES: Do() returns [[member1, score1], [member2, score2], ...]
		// Each element is a []interface{} pair
		if (cmd == "ZRANGE" || cmd == "ZRANGEBYSCORE") && len(v) >= 1 {
			withScores := false
			if pair, ok := v[0].([]interface{}); ok && len(pair) == 2 {
				// Check if second element is numeric (score)
				switch pair[1].(type) {
				case float64, int64, string:
					withScores = true
				}
			}
			if withScores {
				type jsonEntry struct {
					Member string
					Score  float64
					Fields map[string]interface{}
				}
				entries := make([]jsonEntry, 0, len(v))
				allKeys := map[string]bool{}

				for _, item := range v {
					pair, ok := item.([]interface{})
					if !ok || len(pair) < 2 {
						continue
					}
					member := fmt.Sprintf("%v", pair[0])
					var score float64
					switch s := pair[1].(type) {
					case float64:
						score = s
					case int64:
						score = float64(s)
					case string:
						score, _ = strconv.ParseFloat(s, 64)
					}
					entry := jsonEntry{Member: member, Score: score}

					var parsed interface{}
					if json.Unmarshal([]byte(member), &parsed) == nil {
						if m, ok := parsed.(map[string]interface{}); ok {
							entry.Fields = m
							for k := range m {
								allKeys[k] = true
							}
						}
					}
					entries = append(entries, entry)
				}

				if len(allKeys) > 0 {
					// JSON members → expand into table columns
					sortedKeys := make([]string, 0, len(allKeys))
					for k := range allKeys {
						sortedKeys = append(sortedKeys, k)
					}
					sort.Strings(sortedKeys)

					cols := []string{"#", "timestamp"}
					cols = append(cols, sortedKeys...)

					for idx, entry := range entries {
						row := map[string]interface{}{
							"#":         idx + 1,
							"timestamp": int64(entry.Score),
						}
						for _, k := range sortedKeys {
							if entry.Fields != nil {
								if fv, ok := entry.Fields[k]; ok {
									row[k] = fv
								} else {
									row[k] = nil
								}
							} else {
								row[k] = nil
							}
						}
						rows = append(rows, row)
					}
					columns = cols
				} else {
					// Plain members with scores
					for idx, item := range v {
						pair, ok := item.([]interface{})
						if !ok || len(pair) < 2 {
							continue
						}
						member := fmt.Sprintf("%v", pair[0])
						var score float64
						switch s := pair[1].(type) {
						case float64:
							score = s
						case int64:
							score = float64(s)
						case string:
							score, _ = strconv.ParseFloat(s, 64)
						}
						rows = append(rows, map[string]interface{}{
							"#":         idx + 1,
							"member":    member,
							"timestamp": int64(score),
						})
					}
					columns = []string{"#", "member", "timestamp"}
				}

				total := len(rows)
				return gin.H{
					"columns": columns,
					"rows":    rows,
					"total":   total,
					"count":   total,
				}, nil
			}
		}

		// Array result (MGET, LRANGE, ZRANGE without WITHSCORES, etc.)
		if cmd == "MGET" {
			for i, val := range v {
				rows = append(rows, map[string]interface{}{
					"index": i,
					"value": fmt.Sprintf("%v", val),
				})
			}
			columns = []string{"index", "value"}
		} else {
			for i, val := range v {
				rows = append(rows, map[string]interface{}{
					"index": i,
					"value": fmt.Sprintf("%v", val),
				})
			}
		}
	case map[string]string:
		// HGETALL returns map[string]string via some redis clients
		for k, val := range v {
			rows = append(rows, map[string]interface{}{
				"field": k,
				"value": val,
			})
		}
		columns = []string{"field", "value"}
	default:
		// Use reflect to handle any map type (map[interface{}]interface{}, etc.)
		rv := reflect.ValueOf(result)
		if rv.Kind() == reflect.Map {
			iter := rv.MapRange()
			for iter.Next() {
				rows = append(rows, map[string]interface{}{
					"field": fmt.Sprintf("%v", iter.Key().Interface()),
					"value": fmt.Sprintf("%v", iter.Value().Interface()),
				})
			}
			columns = []string{"field", "value"}
		} else {
			// Single value result (GET, HGET, TYPE, TTL, etc.)
			rawValue := fmt.Sprintf("%v", v)
			// Try to pretty-format JSON with Chinese characters preserved
			var prettyJSON string
			var parsed interface{}
			if json.Unmarshal([]byte(rawValue), &parsed) == nil {
				switch parsed.(type) {
				case map[string]interface{}, []interface{}:
					var buf bytes.Buffer
					enc := json.NewEncoder(&buf)
					enc.SetEscapeHTML(false)
					enc.SetIndent("", "  ")
					enc.Encode(parsed)
					prettyJSON = strings.TrimRight(buf.String(), "\n")
				}
			}
			result := gin.H{
				"columns":   []string{},
				"rows":      []interface{}{},
				"total":     0,
				"raw_value": prettyJSON != "",
				"raw_key":   keyName,
				"raw_text":  rawValue,
			}
			for k, v := range keyMeta {
				result[k] = v
			}
			return result, nil
		}
	}

	response := gin.H{
		"columns": columns,
		"rows":    rowsOrDefault(rows),
		"total":   len(rows),
		"count":   len(rows),
	}
	for k, v := range keyMeta {
		response[k] = v
	}
	return response, nil
}

// executeMongoQuery — MongoDB query with pagination
// Supports: db.collection.find({...}), db.collection.aggregate([...])
// Input format: db.collection.find({query}) or JSON array of pipeline stages
func executeMongoQuery(instance *model.Instance, dbName string, queryStr string, limit int, offset int) (gin.H, error) {
	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
	defer cancel()

	// Build connection URI
	uri := fmt.Sprintf("mongodb://%s:%d", instance.Host, instance.Port)
	if instance.Password != "" {
		pwd := instance.Password
		if strings.HasPrefix(pwd, "enc:") {
			if dec, err := decryptPassword(strings.TrimPrefix(pwd, "enc:")); err == nil {
				pwd = dec
			}
		}
		user := instance.Username
		if user == "" {
			user = "root"
		}
		uri = fmt.Sprintf("mongodb://%s:%s@%s:%d", user, pwd, instance.Host, instance.Port)
	}

	client, err := mongo.Connect(options.Client().ApplyURI(uri))
	if err != nil {
		return nil, fmt.Errorf("MongoDB connection failed: %v", err)
	}
	defer client.Disconnect(ctx)

	if err := client.Ping(ctx, readpref.Primary()); err != nil {
		return nil, fmt.Errorf("MongoDB ping failed: %v", err)
	}

	// Parse query: "db.collection.find({...})" or raw JSON
	collectionName, filter, pipeline, err := parseMongoQuery(queryStr)
	if err != nil {
		return nil, err
	}

	if dbName == "" {
		dbName = "test"
	}
	coll := client.Database(dbName).Collection(collectionName)

	var total int64
	var docs []bson.M

	if pipeline != nil {
		// Aggregation — add $facet for count + data
		facetStage := bson.D{
			{Key: "$facet", Value: bson.D{
				{Key: "data", Value: bson.A{
					bson.D{{Key: "$skip", Value: offset}},
					bson.D{{Key: "$limit", Value: limit}},
				}},
				{Key: "count", Value: bson.A{
					bson.D{{Key: "$count", Value: "total"}},
				}},
			}},
		}
		fullPipeline := append(pipeline, facetStage)
		var results []bson.M
		cursor, err := coll.Aggregate(ctx, fullPipeline)
		if err != nil {
			return nil, fmt.Errorf("aggregation failed: %v", err)
		}
		defer cursor.Close(ctx)
		cursor.All(ctx, &results)

		if len(results) > 0 {
			if dataArr, ok := results[0]["data"].(bson.A); ok {
				for _, d := range dataArr {
					if m, ok := d.(bson.M); ok {
						docs = append(docs, m)
					}
				}
			}
			if countArr, ok := results[0]["count"].(bson.A); ok && len(countArr) > 0 {
				if m, ok := countArr[0].(bson.M); ok {
					if t, ok := m["total"].(int32); ok {
						total = int64(t)
					} else if t, ok := m["total"].(int64); ok {
						total = t
					}
				}
			}
		}
	} else {
		// Find
		total, _ = coll.CountDocuments(ctx, filter)
		opts := options.Find().SetSkip(int64(offset)).SetLimit(int64(limit))
		cursor, err := coll.Find(ctx, filter, opts)
		if err != nil {
			return nil, fmt.Errorf("find failed: %v", err)
		}
		defer cursor.Close(ctx)
		cursor.All(ctx, &docs)
	}

	// Convert bson.M to flat maps for table display
	columnSet := make(map[string]bool)
	rowMaps := make([]interface{}, 0, len(docs))
	for _, doc := range docs {
		flat := flattenBSON(doc, "")
		for k := range flat {
			columnSet[k] = true
		}
		rowMaps = append(rowMaps, flat)
	}

	// Stable column order
	columns := make([]string, 0, len(columnSet))
	if _, hasID := columnSet["_id"]; hasID {
		columns = append(columns, "_id")
	}
	for k := range columnSet {
		if k != "_id" {
			columns = append(columns, k)
		}
	}

	return gin.H{
		"columns": columns,
		"rows":    rowMaps,
		"total":   int(total),
		"count":   len(rowMaps),
	}, nil
}

// parseMongoQuery parses "db.collection.find({...})" or "db.collection.aggregate([...])" or raw JSON
func parseMongoQuery(queryStr string) (collection string, filter bson.M, pipeline bson.A, err error) {
	queryStr = strings.TrimSpace(queryStr)

	// Pattern: db.collection.find({...})
	if re := regexpFromPattern(`db\.(\w+)\.find\((.+)?\)`); re != nil {
		matches := re.FindStringSubmatch(queryStr)
		if len(matches) >= 2 {
			collection = matches[1]
			if len(matches) >= 3 && strings.TrimSpace(matches[2]) != "" {
				if e := json.Unmarshal([]byte(matches[2]), &filter); e != nil {
					filter = bson.M{}
				}
			} else {
				filter = bson.M{}
			}
			return
		}
	}

	// Pattern: db.collection.aggregate([...])
	if re := regexpFromPattern(`db\.(\w+)\.aggregate\((.+)\)`); re != nil {
		matches := re.FindStringSubmatch(queryStr)
		if len(matches) >= 3 {
			collection = matches[1]
			if e := json.Unmarshal([]byte(matches[2]), &pipeline); e != nil {
				err = fmt.Errorf("invalid aggregation pipeline: %v", e)
				return
			}
			return
		}
	}

	// Fallback: raw JSON object treated as find on all collections? No — require db.xxx format
	err = fmt.Errorf("query format not supported. Use: db.collection.find({}) or db.collection.aggregate([...])")
	return
}

func regexpFromPattern(pattern string) *regexp.Regexp {
	// Import regexp lazily — actually let's just import at top
	// This is a simple helper
	return regexp.MustCompile(pattern)
}

func flattenBSON(doc bson.M, prefix string) map[string]interface{} {
	result := make(map[string]interface{})
	for k, v := range doc {
		fullKey := k
		if prefix != "" {
			fullKey = prefix + "." + k
		}
		switch child := v.(type) {
		case bson.M:
			for ck, cv := range flattenBSON(child, fullKey) {
				result[ck] = cv
			}
		case bson.A:
			// Serialize arrays as JSON string
			b, _ := json.Marshal(child)
			result[fullKey] = string(b)
		default:
			// Convert ObjectID to string
			s := fmt.Sprintf("%v", v)
			result[fullKey] = s
		}
	}
	return result
}

// POST /api/v1/query/export — Streaming CSV export for large datasets
func ExportQueryCSV(c *gin.Context) {
	var input struct {
		InstanceID int    `json:"instance_id"`
		Database   string `json:"database"`
		SQL        string `json:"sql"`
		Limit      int    `json:"limit"`
	}
	if err := c.ShouldBindJSON(&input); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}
	if input.InstanceID == 0 || input.SQL == "" {
		resp.BadRequest(c, "instance_id and sql are required")
		return
	}
	if input.Limit <= 0 {
		input.Limit = 10000
	}
	if input.Limit > 50000 {
		input.Limit = 50000
	}

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

	// Stream CSV response headers
	filename := fmt.Sprintf("query_export_%s.csv", time.Now().Format("20060102_150405"))
	c.Header("Content-Type", "text/csv; charset=utf-8")
	c.Header("Content-Disposition", fmt.Sprintf("attachment; filename=%s", filename))
	c.Writer.Write([]byte("\xEF\xBB\xBF")) // BOM for Excel

	switch instance.DBType {
	case "mysql", "postgresql", "postgres", "pg":
		exportSQLCSV(c, &instance, input.Database, input.SQL, input.Limit)
	case "redis":
		exportRedisCSV(c, &instance, input.Database, input.SQL, input.Limit)
	case "mongodb", "mongo":
		exportMongoCSV(c, &instance, input.Database, input.SQL, input.Limit)
	default:
		resp.BadRequest(c, "CSV export not supported for "+instance.DBType)
	}
}

func exportSQLCSV(c *gin.Context, instance *model.Instance, dbName string, sqlStr string, limit int) {
	db, err := getDBConnection(instance, dbName)
	if err != nil {
		c.Writer.Write([]byte("Error: connection failed\n"))
		return
	}
	defer db.Close()

	upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))
	if !strings.Contains(upperSQL, " LIMIT ") {
		sqlStr = fmt.Sprintf("%s LIMIT %d", sqlStr, limit)
	}

	rows, err := db.Query(sqlStr)
	if err != nil {
		c.Writer.Write([]byte("Error: query failed\n"))
		return
	}
	defer rows.Close()

	cols, _ := rows.Columns()
	c.Writer.Write([]byte(strings.Join(cols, ",") + "\n"))

	batchSize := 500
	batch := make([]string, 0, batchSize)
	rowCount := 0

	for rows.Next() {
		values := make([]interface{}, len(cols))
		ptrValues := make([]interface{}, len(cols))
		for i := range values {
			ptrValues[i] = &values[i]
		}
		rows.Scan(ptrValues...)

		cells := make([]string, len(cols))
		for i, val := range values {
			s := ""
			if b, ok := val.([]byte); ok {
				s = string(b)
			} else if val != nil {
				s = fmt.Sprintf("%v", val)
			}
			if strings.ContainsAny(s, ",\"\n\r") {
				s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
			}
			cells[i] = s
		}
		batch = append(batch, strings.Join(cells, ","))
		rowCount++

		if len(batch) >= batchSize {
			c.Writer.Write([]byte(strings.Join(batch, "\n") + "\n"))
			c.Writer.(http.Flusher).Flush()
			batch = batch[:0]
		}
	}
	if len(batch) > 0 {
		c.Writer.Write([]byte(strings.Join(batch, "\n") + "\n"))
	}
	c.Writer.(http.Flusher).Flush()
	saveQueryHistory(0, int(instance.ID), dbName, sqlStr, 0, rowCount)
}

func exportRedisCSV(c *gin.Context, instance *model.Instance, dbSelect string, command string, limit int) {
	result, err := executeRedisQuery(instance, dbSelect, command, limit, 0)
	if err != nil {
		c.Writer.Write([]byte("Error: " + err.Error() + "\n"))
		return
	}

	columns, _ := result["columns"].([]string)
	rows, _ := result["rows"].([]interface{})

	c.Writer.Write([]byte(strings.Join(columns, ",") + "\n"))

	for _, row := range rows {
		if m, ok := row.(map[string]interface{}); ok {
			cells := make([]string, len(columns))
			for i, col := range columns {
				s := fmt.Sprintf("%v", m[col])
				if strings.ContainsAny(s, ",\"\n\r") {
					s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
				}
				cells[i] = s
			}
			c.Writer.Write([]byte(strings.Join(cells, ",") + "\n"))
		}
	}
	c.Writer.(http.Flusher).Flush()
	rowCount := len(rows)
	saveQueryHistory(0, int(instance.ID), dbSelect, command, 0, rowCount)
}

func exportMongoCSV(c *gin.Context, instance *model.Instance, dbName string, queryStr string, limit int) {
	result, err := executeMongoQuery(instance, dbName, queryStr, limit, 0)
	if err != nil {
		c.Writer.Write([]byte("Error: " + err.Error() + "\n"))
		return
	}

	columns, _ := result["columns"].([]string)
	rows, _ := result["rows"].([]interface{})

	c.Writer.Write([]byte(strings.Join(columns, ",") + "\n"))

	for _, row := range rows {
		if m, ok := row.(map[string]interface{}); ok {
			cells := make([]string, len(columns))
			for i, col := range columns {
				s := fmt.Sprintf("%v", m[col])
				if strings.ContainsAny(s, ",\"\n\r") {
					s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
				}
				cells[i] = s
			}
			c.Writer.Write([]byte(strings.Join(cells, ",") + "\n"))
		}
	}
	c.Writer.(http.Flusher).Flush()
	rowCount := len(rows)
	saveQueryHistory(0, int(instance.ID), dbName, queryStr, 0, rowCount)
}

func saveQueryHistory(userID int, instanceID int, database string, sqlStr string, costMs int64, rowCount int) {
	var user model.User
	var inst model.Instance
	username := ""
	instName := ""
	dbType := ""
	if model.DB.First(&user, userID).Error == nil {
		username = user.Username
	}
	if model.DB.First(&inst, instanceID).Error == nil {
		instName = inst.Name
		dbType = inst.DBType
	}

	queryType := "SELECT"
	upper := strings.ToUpper(strings.TrimSpace(sqlStr))
	if strings.HasPrefix(upper, "SHOW") || strings.HasPrefix(upper, "DESCRIBE") || strings.HasPrefix(upper, "EXPLAIN") {
		queryType = strings.SplitN(upper, " ", 2)[0]
	}

	history := model.QueryHistory{
		UserID:          userID,
		Username:        username,
		InstanceID:      instanceID,
		InstanceName:    instName,
		DBType:          dbType,
		DatabaseName:    database,
		QueryContent:    sqlStr,
		QueryType:       queryType,
		ExecutionTimeMs: int(costMs),
		ResultRows:      rowCount,
		IsSuccess:       true,
	}
	model.DB.Create(&history)
}

func saveQueryAuditLog(c *gin.Context, userID int, username string, instance *model.Instance, database string, sqlStr string, costMs int64, rowCount int, status string, errMsg string, isRisky bool) {
	targetID := instance.ID
	cost := int(costMs)
	rows := rowCount
	action := "query"
	if status != "success" {
		action = "query_failed"
	}

	detail := fmt.Sprintf("数据查询: %s/%s, 返回 %d 行, 耗时 %dms", instance.Name, database, rowCount, costMs)
	if errMsg != "" {
		detail = fmt.Sprintf("数据查询失败: %s/%s, %s", instance.Name, database, errMsg)
	}

	auditLog := model.AuditLog{
		UserID:          userID,
		Username:        &username,
		Action:          action,
		TargetType:      "query",
		TargetID:        &targetID,
		Detail:          &detail,
		IPAddress:       auditStringPtr(c.ClientIP()),
		DBType:          &instance.DBType,
		DatabaseName:    &database,
		InstanceName:    &instance.Name,
		ExecutionTimeMs: &cost,
		RowsAffected:    &rows,
		ResultStatus:    &status,
		QueryContent:    &sqlStr,
		UserAgent:       auditStringPtr(c.GetHeader("User-Agent")),
		IsRisky:         &isRisky,
	}
	model.DB.Create(&auditLog)
}

func auditStringPtr(value string) *string {
	return &value
}

// applyDataMasking masks sensitive fields in query results with "*".
// It reads global settings (data_masking_enabled + sensitive fields) and
// merges with instance-level mask_fields.
func applyDataMasking(result gin.H, instance *model.Instance) {
	// Check global enable flag
	var enabledSetting model.SystemSetting
	if model.DB.Where("`key` = ?", "data_masking_enabled").First(&enabledSetting).Error != nil {
		return // setting not found, skip masking
	}
	if enabledSetting.Value != "true" {
		return // masking disabled
	}

	// Collect global sensitive fields
	var fieldsSetting model.SystemSetting
	globalFields := map[string]bool{}
	if model.DB.Where("`key` = ?", "data_masking_sensitive_fields").First(&fieldsSetting).Error == nil {
		for _, f := range strings.Split(fieldsSetting.Value, ",") {
			f = strings.TrimSpace(f)
			if f != "" {
				globalFields[strings.ToLower(f)] = true
			}
		}
	}

	// Merge instance-level mask_fields
	instanceFields := map[string]bool{}
	if instance.MaskFields != nil && *instance.MaskFields != "" {
		for _, f := range strings.Split(*instance.MaskFields, ",") {
			f = strings.TrimSpace(f)
			if f != "" {
				instanceFields[strings.ToLower(f)] = true
			}
		}
	}

	// Combine all mask fields
	allMaskFields := make(map[string]bool)
	for k := range globalFields {
		allMaskFields[k] = true
	}
	for k := range instanceFields {
		allMaskFields[k] = true
	}

	if len(allMaskFields) == 0 {
		return
	}

	// Apply masking to rows and collect masked column names
	rows, ok := result["rows"].([]interface{})
	if !ok {
		return
	}
	maskedCols := []string{}
	for _, rowIface := range rows {
		row, ok := rowIface.(map[string]interface{})
		if !ok {
			continue
		}
		for colName, val := range row {
			if allMaskFields[strings.ToLower(colName)] {
				row[colName] = maskValue(val)
				// Track masked column (deduplicate)
				found := false
				for _, mc := range maskedCols {
					if mc == colName {
						found = true
						break
					}
				}
				if !found {
					maskedCols = append(maskedCols, colName)
				}
			}
		}
	}
	if len(maskedCols) > 0 {
		result["masked_columns"] = maskedCols
	}
}

// maskValue replaces a value with asterisks, preserving length info for strings
func maskValue(val interface{}) string {
	if val == nil {
		return "***"
	}
	s := fmt.Sprintf("%v", val)
	if len(s) == 0 {
		return "*"
	}
	// Replace all chars with *, keep reasonable length
	if len(s) > 20 {
		return "********************"
	}
	return strings.Repeat("*", len(s))
}