package handler import ( "fmt" "net/http" "reflect" "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 { 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 } // Read query limits from platform settings maxLimit := getSettingInt("query_limit_max", 1000) if input.Limit <= 0 { input.Limit = maxLimit if input.Limit > 1000 { input.Limit = 1000 } } if input.Limit > maxLimit { input.Limit = maxLimit } if input.Offset < 0 { input.Offset = 0 } // 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 ") { c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": 0, "error": "危险操作检测: " + pat + " 缺少 WHERE/LIMIT 条件,请在工单中提交"}) return } } } 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() // 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]) 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) } // 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.) 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") } } // ARDM-style: auto-fetch key type and TTL for KeyHeader var keyType, keyTTL string if keyName != "" && cmd != "TYPE" && cmd != "TTL" && cmd != "PTTL" && cmd != "EXISTS" && cmd != "DBSIZE" && cmd != "INFO" { if t, err := client.Type(ctx, keyName).Result(); err == nil { keyType = t } if ttl, err := client.TTL(ctx, keyName).Result(); err == nil { if ttl < 0 { keyTTL = "永不过期" } else { keyTTL = fmt.Sprintf("%vs", int(ttl.Seconds())) } } } result := gin.H{ "columns": []string{}, "rows": []interface{}{}, "total": 0, "raw_value": prettyJSON != "", "raw_key": keyName, "raw_text": rawValue, } if keyType != "" { result["raw_type"] = keyType } if keyTTL != "" { result["raw_ttl"] = keyTTL } return result, nil } } return gin.H{ "columns": columns, "rows": rowsOrDefault(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)) }