opsxsql/backend/internal/handler/instance_detail.go

package handler

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

	"sql-platform/internal/model"

	"github.com/gin-gonic/gin"
	"time"
	"context"
	"sql-platform/internal/middleware"
	"strings"
	"strconv"
	"encoding/json"
	"bytes"
	"regexp"
	"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"
)

// 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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
		return
	}

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

	db, err := getDBConnection(&instance, "")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW DATABASES")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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 == "" {
		c.JSON(http.StatusBadRequest, gin.H{"detail": "database parameter required"})
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW TABLE STATUS")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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 == "" {
		c.JSON(http.StatusBadRequest, gin.H{"detail": "database parameter required"})
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	// Get all tables and their columns
	tableRows, err := db.Query("SHOW TABLES")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		return
	}
	defer tableRows.Close()

	schema := make(map[string][]string)
	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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
		return
	}

	if dbName == "" {
		if instance.Database != nil {
			dbName = *instance.Database
		}
	}
	if dbName == "" || tableName == "" {
		c.JSON(http.StatusBadRequest, gin.H{"detail": "database and table parameters required"})
		return
	}

	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	// 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 {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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 {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW PROCESSLIST")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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 {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	rows, err := db.Query("SHOW GLOBAL STATUS")
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
		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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
		return
	}

	dbName := ""
	if instance.Database != nil {
		dbName = *instance.Database
	}
	db, err := getDBConnection(&instance, dbName)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", err)})
		return
	}
	defer db.Close()

	_, err = db.Exec("KILL " + pid)
	if err != nil {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Kill failed: %v", 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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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 {
		c.JSON(http.StatusInternalServerError, gin.H{"detail": fmt.Sprintf("Connection failed: %v", 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, _, _ := 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 {
		c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
		return
	}
	if input.InstanceID == 0 || input.SQL == "" {
		c.JSON(http.StatusBadRequest, gin.H{"detail": "instance_id and sql are required"})
		return
	}
	// Default pagination
	if input.Limit <= 0 {
		input.Limit = 100
	}
	if input.Limit > 1000 {
		input.Limit = 1000
	}
	if input.Offset < 0 {
		input.Offset = 0
	}

	var instance model.Instance
	if err := model.DB.First(&instance, input.InstanceID).Error; err != nil {
		c.JSON(http.StatusNotFound, gin.H{"detail": "Instance not found"})
		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 {
			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))
		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 {
			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)
		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 {
			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))
		c.JSON(http.StatusOK, result)

	default:
		c.JSON(http.StatusBadRequest, gin.H{"detail": "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()

	upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))

	// 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.QueryRow(countSQL).Scan(&total)
			}
		}
	}

	// Inject LIMIT/OFFSET if not already present
	if !hasLimit {
		sqlStr = fmt.Sprintf("%s LIMIT %d OFFSET %d", sqlStr, limit, offset)
	}

	rows, err := db.Query(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
}

// 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
}

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])
	readOnly := map[string]bool{
		"GET": true, "MGET": true, "HGET": true, "HGETALL": true,
		"LRANGE": true, "SMEMBERS": true, "ZRANGE": true,
		"TYPE": true, "TTL": 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,
	}
	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)
	}

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

	switch v := result.(type) {
	case []interface{}:
		// Array result (MGET, LRANGE, 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.)
			keyName := ""
			if len(parts) > 1 {
				keyName = parts[1]
			}
			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")
				}
			}
			return gin.H{
				"columns":   []string{},
				"rows":      []interface{}{},
				"total":     0,
				"raw_value": prettyJSON != "",
				"raw_key":   keyName,
				"raw_text":  rawValue,
			}, nil
		}
	}

	return gin.H{
		"columns": columns,
		"rows":    rows,
		"total":   len(rows),
		"count":   len(rows),
	}, 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{
			{"$facet", bson.D{
				{"data", bson.A{
					bson.D{{"$skip", offset}},
					bson.D{{"$limit", limit}},
				}},
				{"count", bson.A{
					bson.D{{"$count", "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 {
		c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
		return
	}
	if input.InstanceID == 0 || input.SQL == "" {
		c.JSON(http.StatusBadRequest, gin.H{"detail": "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 {
		c.JSON(http.StatusNotFound, gin.H{"detail": "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:
		c.JSON(http.StatusBadRequest, gin.H{"detail": "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)
}

// 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))
}