Files
opsxsql/backend/internal/handler/instance_detail.go
T
logan 5ea86fe334 feat: 查询分页(limit/offset)+流式CSV导出+去复制按钮
后端:
- ExecuteQuery 添加 limit/offset 参数,MySQL 自动注入 LIMIT/OFFSET
- 首页自动 COUNT 查总数(total),翻页页不再重复 count
- 新增 POST /query/export 流式 CSV 导出端点,游标分批写不占内存
- 默认 limit=100, max=1000,防止大数据量 OOM

前端:
- 查询结果加分页控件(首页/上一页/下一页/末页 + 每页条数选择)
- CSV 导出改走流式端点(全量数据),不限于当前页
- 去掉复制按钮(Copy/Check/coped state),Redis 单值也不显示复制
- 工具栏显示 总行数/行每页/列数
2026-05-28 15:51:22 +08:00

936 lines
24 KiB
Go

package handler
import (
"fmt"
"net/http"
"reflect"
"sql-platform/internal/model"
"github.com/gin-gonic/gin"
"time"
"context"
"github.com/redis/go-redis/v9"
"sql-platform/internal/middleware"
"strings"
"strconv"
"encoding/json"
"bytes"
)
// 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{})
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)
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)
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
}
func executeRedisQuery(instance *model.Instance, dbSelect string, command string) (gin.H, error) {
password := instance.Password
if strings.HasPrefix(password, "enc:") {
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
if err == nil {
password = decrypted
}
}
// 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 := redis.NewClient(&redis.Options{
Addr: fmt.Sprintf("%s:%d", instance.Host, instance.Port),
Password: password,
DB: redisDB,
})
defer client.Close()
ctx := context.Background()
if err := client.Ping(ctx).Err(); err != nil {
return nil, fmt.Errorf("Redis connection failed: %v", err)
}
parts := strings.Fields(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)
}
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),
}, nil
}
// 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 // Export max 10k rows
}
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
}
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
}
db, err := getDBConnection(&instance, input.Database)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Connection failed: " + err.Error()})
return
}
defer db.Close()
// Inject LIMIT if not present
upperSQL := strings.ToUpper(strings.TrimSpace(input.SQL))
if !strings.Contains(upperSQL, " LIMIT ") {
input.SQL = fmt.Sprintf("%s LIMIT %d", input.SQL, input.Limit)
}
rows, err := db.Query(input.SQL)
if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Query failed: " + err.Error()})
return
}
defer rows.Close()
cols, _ := rows.Columns()
colNames := make([]string, len(cols))
for i, col := range cols {
colNames[i] = col
}
// 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
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)
}
// CSV escape: wrap in quotes if contains comma, quote, or newline
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]
}
}
// Flush remaining
if len(batch) > 0 {
c.Writer.Write([]byte(strings.Join(batch, "\n") + "\n"))
}
c.Writer.(http.Flusher).Flush()
saveQueryHistory(0, input.InstanceID, input.Database, input.SQL, 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)
}