diff --git a/backend/go.mod b/backend/go.mod
index be9f4f4..70aa3e2 100644
--- a/backend/go.mod
+++ b/backend/go.mod
@@ -38,6 +38,7 @@ require (
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
github.com/json-iterator/go v1.1.12 // indirect
+ github.com/klauspost/compress v1.17.6 // indirect
github.com/klauspost/cpuid/v2 v2.2.10 // indirect
github.com/kr/text v0.2.0 // indirect
github.com/leodido/go-urn v1.4.0 // indirect
@@ -48,6 +49,11 @@ require (
github.com/pelletier/go-toml/v2 v2.2.3 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.12 // indirect
+ github.com/xdg-go/pbkdf2 v1.0.0 // indirect
+ github.com/xdg-go/scram v1.2.0 // indirect
+ github.com/xdg-go/stringprep v1.0.4 // indirect
+ github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
+ go.mongodb.org/mongo-driver/v2 v2.6.0 // indirect
go.uber.org/atomic v1.11.0 // indirect
golang.org/x/arch v0.15.0 // indirect
golang.org/x/net v0.50.0 // indirect
diff --git a/backend/internal/handler/instance_detail.go b/backend/internal/handler/instance_detail.go
index 24021de..0636f05 100644
--- a/backend/internal/handler/instance_detail.go
+++ b/backend/internal/handler/instance_detail.go
@@ -16,6 +16,11 @@ import (
"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
@@ -549,13 +554,30 @@ func ExecuteQuery(c *gin.Context) {
c.JSON(http.StatusOK, result)
case "redis":
- result, err := executeRedisQuery(&instance, input.Database, input.SQL)
+ 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
}
- saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, 1)
+ result["cost_ms"] = costMs
+ rowCount := 1
+ if rowsArr, ok := result["rows"].([]interface{}); ok {
+ rowCount = len(rowsArr)
+ }
+ 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{})
+ saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, len(rowsInterface))
c.JSON(http.StatusOK, result)
default:
@@ -639,7 +661,7 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
}, nil
}
-func executeRedisQuery(instance *model.Instance, dbSelect string, command string) (gin.H, error) {
+func executeRedisQuery(instance *model.Instance, dbSelect string, command string, limit int, offset int) (gin.H, error) {
password := instance.Password
if strings.HasPrefix(password, "enc:") {
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
@@ -695,6 +717,109 @@ func executeRedisQuery(instance *model.Instance, dbSelect string, command string
args = append(args, p)
}
+ // For KEYS command, use SCAN with COUNT instead for safety + pagination
+ if cmd == "KEYS" && strings.Contains(parts[len(parts)-1], "*") {
+ // Replace KEYS * with SCAN-based pagination
+ pattern := "*"
+ if len(parts) > 1 {
+ pattern = parts[1]
+ }
+ var allKeys []string
+ cursor := uint64(0)
+ maxScan := offset + limit
+ for {
+ keys, nextCursor, err := client.Scan(ctx, cursor, pattern, int64(maxScan-len(allKeys))).Result()
+ if err != nil {
+ return nil, fmt.Errorf("SCAN failed: %v", err)
+ }
+ allKeys = append(allKeys, keys...)
+ cursor = nextCursor
+ if cursor == 0 || len(allKeys) >= maxScan {
+ break
+ }
+ }
+ total := len(allKeys)
+ end := offset + limit
+ if end > len(allKeys) {
+ end = len(allKeys)
+ }
+ if offset > len(allKeys) {
+ offset = len(allKeys)
+ }
+ 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)
@@ -780,9 +905,208 @@ func executeRedisQuery(instance *model.Instance, dbSelect string, command string
"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 {
@@ -800,7 +1124,7 @@ func ExportQueryCSV(c *gin.Context) {
return
}
if input.Limit <= 0 {
- input.Limit = 10000 // Export max 10k rows
+ input.Limit = 10000
}
if input.Limit > 50000 {
input.Limit = 50000
@@ -812,48 +1136,47 @@ func ExportQueryCSV(c *gin.Context) {
return
}
- if instance.DBType != "mysql" && instance.DBType != "postgresql" && instance.DBType != "postgres" && instance.DBType != "pg" {
- c.JSON(http.StatusBadRequest, gin.H{"detail": "CSV export only supports MySQL/PostgreSQL"})
- 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
- db, err := getDBConnection(&instance, input.Database)
+ 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.JSON(http.StatusInternalServerError, gin.H{"detail": "Connection failed: " + err.Error()})
+ c.Writer.Write([]byte("Error: connection failed\n"))
return
}
defer db.Close()
- // Inject LIMIT if not present
- upperSQL := strings.ToUpper(strings.TrimSpace(input.SQL))
+ upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))
if !strings.Contains(upperSQL, " LIMIT ") {
- input.SQL = fmt.Sprintf("%s LIMIT %d", input.SQL, input.Limit)
+ sqlStr = fmt.Sprintf("%s LIMIT %d", sqlStr, limit)
}
- rows, err := db.Query(input.SQL)
+ rows, err := db.Query(sqlStr)
if err != nil {
- c.JSON(http.StatusInternalServerError, gin.H{"detail": "Query failed: " + err.Error()})
+ c.Writer.Write([]byte("Error: query failed\n"))
return
}
defer rows.Close()
cols, _ := rows.Columns()
- colNames := make([]string, len(cols))
- for i, col := range cols {
- colNames[i] = col
- }
+ c.Writer.Write([]byte(strings.Join(cols, ",") + "\n"))
- // Stream CSV response
- 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))
- // BOM for Excel UTF-8
- c.Writer.Write([]byte("\xEF\xBB\xBF"))
-
- // Header row
- c.Writer.Write([]byte(strings.Join(colNames, ",") + "\n"))
-
- // Data rows — write in chunks to avoid OOM
batchSize := 500
batch := make([]string, 0, batchSize)
rowCount := 0
@@ -874,7 +1197,6 @@ func ExportQueryCSV(c *gin.Context) {
} else if val != nil {
s = fmt.Sprintf("%v", val)
}
- // CSV escape: wrap in quotes if contains comma, quote, or newline
if strings.ContainsAny(s, ",\"\n\r") {
s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
}
@@ -889,13 +1211,71 @@ func ExportQueryCSV(c *gin.Context) {
batch = batch[:0]
}
}
- // Flush remaining
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)
+}
- saveQueryHistory(0, input.InstanceID, input.Database, input.SQL, 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) {
diff --git a/frontend/src/pages/QueriesPage.tsx b/frontend/src/pages/QueriesPage.tsx
index 656f7db..a68f288 100644
--- a/frontend/src/pages/QueriesPage.tsx
+++ b/frontend/src/pages/QueriesPage.tsx
@@ -494,7 +494,7 @@ export default function QueriesPage() {
{result.columns?.length || 0} 列
)}
- {result.rows?.length > 0 && dbType !== 'redis' && (
+ {result.rows?.length > 0 && (