opsxsql/backend/internal/handler/redis.go

package handler

import (
	"context"
	"fmt"
	"net"
	"net/http"
	"strconv"
	"strings"
	"time"

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

	"github.com/gin-gonic/gin"
	"github.com/redis/go-redis/v9"
)

// Helper: get Redis client from instance (auto-detect standalone vs cluster)
type redisClientWrapper struct {
	Client        *redis.Client
	ClusterClient *redis.ClusterClient
	isCluster     bool
	// MasterAddrs 集群各 master 地址(排序);单机为单地址。扫描分片与游标校验依赖
	MasterAddrs []string
	Username    string
	Password    string
}

func (w *redisClientWrapper) IsCluster() bool { return w.isCluster }

func (w *redisClientWrapper) Close() {
	if w.Client != nil {
		w.Client.Close()
	}
	if w.ClusterClient != nil {
		w.ClusterClient.Close()
	}
}

func (w *redisClientWrapper) Ping(ctx context.Context) *redis.StatusCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Ping(ctx)
	}
	return w.Client.Ping(ctx)
}

func (w *redisClientWrapper) Info(ctx context.Context, sections ...string) *redis.StringCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Info(ctx, sections...)
	}
	return w.Client.Info(ctx, sections...)
}

func (w *redisClientWrapper) DBSize(ctx context.Context) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.DBSize(ctx)
	}
	return w.Client.DBSize(ctx)
}

func (w *redisClientWrapper) Scan(ctx context.Context, cursor uint64, match string, count int64) *redis.ScanCmd {
	return w.Client.Scan(ctx, cursor, match, count)
}

func (w *redisClientWrapper) ScanAllKeys(ctx context.Context, match string, count int) ([]string, error) {
	if w.isCluster && w.ClusterClient != nil {
		var allKeys []string
		err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, shard *redis.Client) error {
			cursor := uint64(0)
			for {
				keys, nextCursor, err := shard.Scan(ctx, cursor, match, int64(count)).Result()
				if err != nil {
					return err
				}
				allKeys = append(allKeys, keys...)
				cursor = nextCursor
				if cursor == 0 {
					break
				}
			}
			return nil
		})
		return allKeys, err
	}

	var allKeys []string
	cursor := uint64(0)
	for {
		keys, nextCursor, err := w.Client.Scan(ctx, cursor, match, int64(count)).Result()
		if err != nil {
			return allKeys, err
		}
		allKeys = append(allKeys, keys...)
		cursor = nextCursor
		if cursor == 0 {
			break
		}
	}
	return allKeys, nil
}

func (w *redisClientWrapper) Type(ctx context.Context, key string) *redis.StatusCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Type(ctx, key)
	}
	return w.Client.Type(ctx, key)
}

func (w *redisClientWrapper) TTL(ctx context.Context, key string) *redis.DurationCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.TTL(ctx, key)
	}
	return w.Client.TTL(ctx, key)
}

func (w *redisClientWrapper) Get(ctx context.Context, key string) *redis.StringCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Get(ctx, key)
	}
	return w.Client.Get(ctx, key)
}

func (w *redisClientWrapper) LRange(ctx context.Context, key string, start, stop int64) *redis.StringSliceCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.LRange(ctx, key, start, stop)
	}
	return w.Client.LRange(ctx, key, start, stop)
}

func (w *redisClientWrapper) SMembers(ctx context.Context, key string) *redis.StringSliceCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.SMembers(ctx, key)
	}
	return w.Client.SMembers(ctx, key)
}

func (w *redisClientWrapper) ZRangeWithScores(ctx context.Context, key string, start, stop int64) *redis.ZSliceCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.ZRangeWithScores(ctx, key, start, stop)
	}
	return w.Client.ZRangeWithScores(ctx, key, start, stop)
}

func (w *redisClientWrapper) HGetAll(ctx context.Context, key string) *redis.MapStringStringCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.HGetAll(ctx, key)
	}
	return w.Client.HGetAll(ctx, key)
}

// SlowLogGet 集群模式逐 master 聚合(单机连接打不到全部分片)
func (w *redisClientWrapper) SlowLogGet(ctx context.Context, count int64) ([]redis.SlowLog, error) {
	if w.isCluster && w.ClusterClient != nil {
		var all []redis.SlowLog
		err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, shard *redis.Client) error {
			logs, err := shard.SlowLogGet(ctx, count).Result()
			if err != nil {
				return err
			}
			all = append(all, logs...)
			return nil
		})
		return all, err
	}
	return w.Client.SlowLogGet(ctx, count).Result()
}

// ClientList 集群模式逐 master 聚合
func (w *redisClientWrapper) ClientList(ctx context.Context) (string, error) {
	if w.isCluster && w.ClusterClient != nil {
		var sb strings.Builder
		err := w.ClusterClient.ForEachMaster(ctx, func(ctx context.Context, shard *redis.Client) error {
			list, err := shard.ClientList(ctx).Result()
			if err != nil {
				return err
			}
			sb.WriteString(list)
			if !strings.HasSuffix(list, "\n") {
				sb.WriteString("\n")
			}
			return nil
		})
		return sb.String(), err
	}
	return w.Client.ClientList(ctx).Result()
}

func (w *redisClientWrapper) Do(ctx context.Context, args ...interface{}) *redis.Cmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Do(ctx, args...)
	}
	return w.Client.Do(ctx, args...)
}

func (w *redisClientWrapper) Set(ctx context.Context, key string, value interface{}, expiration time.Duration) *redis.StatusCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Set(ctx, key, value, expiration)
	}
	return w.Client.Set(ctx, key, value, expiration)
}

func (w *redisClientWrapper) RPush(ctx context.Context, key string, values ...interface{}) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.RPush(ctx, key, values...)
	}
	return w.Client.RPush(ctx, key, values...)
}

func (w *redisClientWrapper) SAdd(ctx context.Context, key string, members ...interface{}) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.SAdd(ctx, key, members...)
	}
	return w.Client.SAdd(ctx, key, members...)
}

func (w *redisClientWrapper) HSet(ctx context.Context, key string, values ...interface{}) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.HSet(ctx, key, values...)
	}
	return w.Client.HSet(ctx, key, values...)
}

func (w *redisClientWrapper) ZAdd(ctx context.Context, key string, members ...redis.Z) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.ZAdd(ctx, key, members...)
	}
	return w.Client.ZAdd(ctx, key, members...)
}

func (w *redisClientWrapper) Persist(ctx context.Context, key string) *redis.BoolCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Persist(ctx, key)
	}
	return w.Client.Persist(ctx, key)
}

func (w *redisClientWrapper) HDel(ctx context.Context, key string, fields ...string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.HDel(ctx, key, fields...)
	}
	return w.Client.HDel(ctx, key, fields...)
}

func (w *redisClientWrapper) Expire(ctx context.Context, key string, expiration time.Duration) *redis.BoolCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Expire(ctx, key, expiration)
	}
	return w.Client.Expire(ctx, key, expiration)
}

func (w *redisClientWrapper) Del(ctx context.Context, keys ...string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.Del(ctx, keys...)
	}
	return w.Client.Del(ctx, keys...)
}

func (w *redisClientWrapper) LLen(ctx context.Context, key string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.LLen(ctx, key)
	}
	return w.Client.LLen(ctx, key)
}

func (w *redisClientWrapper) SCard(ctx context.Context, key string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.SCard(ctx, key)
	}
	return w.Client.SCard(ctx, key)
}

func (w *redisClientWrapper) SScan(ctx context.Context, key string, cursor uint64, match string, count int64) *redis.ScanCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.SScan(ctx, key, cursor, match, count)
	}
	return w.Client.SScan(ctx, key, cursor, match, count)
}

// HScan 分批遍历 hash —— 大 key 不能用 HGETALL 整体读，会打爆内存。
func (w *redisClientWrapper) HScan(ctx context.Context, key string, cursor uint64, match string, count int64) *redis.ScanCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.HScan(ctx, key, cursor, match, count)
	}
	return w.Client.HScan(ctx, key, cursor, match, count)
}

// ZScan 分批遍历 zset 成员（不含 score，仅在需要成员集合时使用）。
func (w *redisClientWrapper) ZScan(ctx context.Context, key string, cursor uint64, match string, count int64) *redis.ScanCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.ZScan(ctx, key, cursor, match, count)
	}
	return w.Client.ZScan(ctx, key, cursor, match, count)
}

// 以下几个 count/长度类命令用于分页时给出 total，都是 O(1)，不会因为 key 大而变慢。
func (w *redisClientWrapper) HLen(ctx context.Context, key string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.HLen(ctx, key)
	}
	return w.Client.HLen(ctx, key)
}

func (w *redisClientWrapper) ZCard(ctx context.Context, key string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.ZCard(ctx, key)
	}
	return w.Client.ZCard(ctx, key)
}

func (w *redisClientWrapper) StrLen(ctx context.Context, key string) *redis.IntCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.StrLen(ctx, key)
	}
	return w.Client.StrLen(ctx, key)
}

// GetRange 按字节区间读取字符串，用于超大 string 的分页预览与流式导出。
func (w *redisClientWrapper) GetRange(ctx context.Context, key string, start, end int64) *redis.StringCmd {
	if w.isCluster && w.ClusterClient != nil {
		return w.ClusterClient.GetRange(ctx, key, start, end)
	}
	return w.Client.GetRange(ctx, key, start, end)
}

func writeRedisKeyValue(ctx context.Context, client *redisClientWrapper, key string, keyType string, value interface{}) error {
	switch keyType {
	case "string":
		return client.Set(ctx, key, fmt.Sprintf("%v", value), 0).Err()
	case "list":
		values, ok := value.([]interface{})
		if !ok {
			return fmt.Errorf("list value must be an array")
		}
		if len(values) == 0 {
			return nil
		}
		items := make([]interface{}, 0, len(values))
		for _, v := range values {
			items = append(items, fmt.Sprintf("%v", v))
		}
		return client.RPush(ctx, key, items...).Err()
	case "set":
		values, ok := value.([]interface{})
		if !ok {
			return fmt.Errorf("set value must be an array")
		}
		if len(values) == 0 {
			return nil
		}
		items := make([]interface{}, 0, len(values))
		for _, v := range values {
			items = append(items, fmt.Sprintf("%v", v))
		}
		return client.SAdd(ctx, key, items...).Err()
	case "hash":
		values, ok := value.(map[string]interface{})
		if !ok {
			return fmt.Errorf("hash value must be an object")
		}
		if len(values) == 0 {
			return nil
		}
		return client.HSet(ctx, key, values).Err()
	case "zset":
		values, ok := value.([]interface{})
		if !ok {
			return fmt.Errorf("zset value must be an array")
		}
		members := make([]redis.Z, 0, len(values))
		for _, row := range values {
			m, ok := row.(map[string]interface{})
			if !ok {
				return fmt.Errorf("zset rows must be objects with member and score")
			}
			member, ok := m["member"]
			if !ok {
				member = m["Member"]
			}
			scoreRaw, ok := m["score"]
			if !ok {
				scoreRaw = m["Score"]
			}
			score, err := strconv.ParseFloat(fmt.Sprintf("%v", scoreRaw), 64)
			if err != nil {
				return fmt.Errorf("invalid zset score: %v", scoreRaw)
			}
			members = append(members, redis.Z{Member: fmt.Sprintf("%v", member), Score: score})
		}
		if len(members) == 0 {
			return nil
		}
		return client.ZAdd(ctx, key, members...).Err()
	default:
		return fmt.Errorf("unsupported redis type: %s", keyType)
	}
}

func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, error) {
	// 防御：非 Redis 实例不能走 Redis 协议。
	// 若前端误调 /api/v1/redis/:id/*（如共用组件的 MySQL 实例），
	// 用 Redis 协议连 3306 会收到 MySQL 握手包，报
	// `redis: invalid reply: "J\x00\x00\x00\n"` 并抛 500。
	// 这里提前拦截，给出可读错误。
	if instance.DBType != "redis" {
		return nil, resp.UserFacing(fmt.Sprintf(
			"实例「%s」的类型是 %s，不是 redis。Redis 接口只适用于 redis 实例；"+
				"MySQL/PostgreSQL 请使用 /api/v1/instances/%d/databases 等对应接口",
			instance.Name, instance.DBType, instance.ID))
	}

	password := instance.Password
	// Decrypt if encrypted
	if strings.HasPrefix(password, "enc:") {
		decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
		if err == nil {
			password = decrypted
		}
	}
	addr := fmt.Sprintf("%s:%d", instance.Host, instance.Port)

	// Redis 6.0+ ACL: username is optional (empty = use "default" traditional auth)
	redisUsername := instance.Username
	if redisUsername == "default" || redisUsername == "" {
		redisUsername = "" // go-redis treats empty Username as no ACL auth
	}

	client := redis.NewClient(&redis.Options{
		Addr:         addr,
		Username:     redisUsername,
		Password:     password,
		DB:           db,
		DialTimeout:  5 * time.Second,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 5 * time.Second,
	})
	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()
	if err := client.Ping(ctx).Err(); err != nil {
		client.Close()
		return nil, fmt.Errorf("Redis connection failed: %v", err)
	}

	// Check cluster mode flag for wrapper behavior (ScanAllKeys needs to know)
	isCluster := instance.ClusterMode != nil && *instance.ClusterMode
	addrs := []string{addr}
	if slots, err := client.ClusterSlots(ctx).Result(); err == nil && len(slots) > 0 {
		isCluster = true
		seen := map[string]bool{}
		addrs = addrs[:0]
		for _, slot := range slots {
			if len(slot.Nodes) == 0 {
				continue
			}
				nodeAddr := slot.Nodes[0].Addr
				// 只在 Redis 上报空/回环地址时才用配置 host 兜底（NAT / 端口映射场景）。
				// 多机集群必须保留 Redis 上报的真实主机，否则会把跨机节点全部指向种子机，
				// 导致 "dial tcp <种子机>:<其他机器的端口>: connection refused"。
				if h, p, err := net.SplitHostPort(nodeAddr); err == nil {
					if h == "" || h == "127.0.0.1" || h == "0.0.0.0" {
						nodeAddr = net.JoinHostPort(instance.Host, p)
					}
				}
			if !seen[nodeAddr] {
				seen[nodeAddr] = true
				addrs = append(addrs, nodeAddr)
			}
		}
		if len(addrs) == 0 {
			addrs = []string{addr}
		}
	}

	wrapper := &redisClientWrapper{
		Client:      client,
		isCluster:   isCluster,
		MasterAddrs: []string{addr},
		Username:    redisUsername,
		Password:    password,
	}
	if !isCluster {
		return wrapper, nil
	}

	clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
		Addrs:        addrs,
		Username:     redisUsername,
		Password:     password,
		DialTimeout:  5 * time.Second,
		ReadTimeout:  5 * time.Second,
		WriteTimeout: 5 * time.Second,
	})
	if err := clusterClient.Ping(ctx).Err(); err != nil {
		clusterClient.Close()
		client.Close()
		return nil, fmt.Errorf("Redis cluster connection failed: %v", err)
	}
	wrapper.ClusterClient = clusterClient
	wrapper.MasterAddrs = addrs
	return wrapper, nil
}

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

	client, err := getRedisClient(&instance, 0)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer client.Close()

	ctx := context.Background()
	info, err := client.Info(ctx).Result()
	if err != nil {
		resp.InternalErr(c, err)
		return
	}

	// Parse Redis INFO into both a flat map and section groups.
	result := make(map[string]interface{})
	grouped := make(map[string]map[string]interface{})
	currentSection := "General"
	sections := strings.Split(info, "\n# ")
	for _, section := range sections {
		lines := strings.Split(section, "\n")
		for lineIndex, line := range lines {
			line = strings.TrimSpace(line)
			if line == "" {
				continue
			}
			if strings.HasPrefix(line, "#") {
				currentSection = strings.TrimSpace(strings.TrimPrefix(line, "#"))
				continue
			}
			if lineIndex == 0 && !strings.Contains(line, ":") {
				currentSection = strings.TrimSpace(line)
				continue
			}
			parts := strings.SplitN(line, ":", 2)
			if len(parts) == 2 {
				key := strings.TrimSpace(parts[0])
				value := strings.TrimSpace(parts[1])
				result[key] = value
				if _, ok := grouped[currentSection]; !ok {
					grouped[currentSection] = make(map[string]interface{})
				}
				grouped[currentSection][key] = value
			}
		}
	}

	// Add db_size
	dbSize, _ := client.DBSize(ctx).Result()
	result["db_size"] = dbSize
	if _, ok := grouped["Keyspace"]; !ok {
		grouped["Keyspace"] = make(map[string]interface{})
	}
	grouped["Keyspace"]["db_size"] = dbSize

	c.JSON(http.StatusOK, gin.H{
		"info":     result,
		"sections": grouped,
	})
}

// GET /api/v1/redis/:id/dbsize
func RedisDbsize(c *gin.Context) {
	id := c.Param("id")
	dbParam := c.Query("db")
	db := 0
	if dbParam != "" {
		db, _ = strconv.Atoi(dbParam)
	}

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

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

	ctx := context.Background()
	dbSize, err := client.DBSize(ctx).Result()
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	c.JSON(http.StatusOK, gin.H{"db_size": dbSize})
}

// GET /api/v1/redis/:id/keys/:key
// Redis 键详情分页参数：默认每页 200 条，硬上限 2000 条。
// 大 key（如 PERP:market:depth:* 盘口快照）必须分页，否则 HGETALL / LRANGE 0 -1
// 会把整个值读进内存再整包序列化，极易把 Pod 内存打爆、前端也会卡死。
const (
	redisKeyPageDefault = 200
	redisKeyPageMax     = 2000
	// string 类型无法「分页」，改为给一个预览上限，超出时标记 truncated 并给出总字节数。
	redisStringPreviewBytes = 256 << 10
)

// GET /api/v1/redis/:id/keys/:key?db=0&count=200&token=<分页令牌>
//
// token 语义由类型决定：
//   - list / zset → 十进制 offset（偏移量）
//   - hash / set  → Redis SCAN 游标
//   - string      → 忽略
//
// 返回 token（下一页令牌，空串=没有更多）、total、truncated。
func RedisGetKeyDetail(c *gin.Context) {
	id := c.Param("id")
	key := c.Param("key")
	dbParam := c.Query("db")
	db := 0
	if dbParam != "" {
		db, _ = strconv.Atoi(dbParam)
	}

	count, _ := strconv.Atoi(c.DefaultQuery("count", strconv.Itoa(redisKeyPageDefault)))
	if count <= 0 {
		count = redisKeyPageDefault
	}
	if count > redisKeyPageMax {
		count = redisKeyPageMax
	}
	pageToken := c.Query("token")

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

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

	ctx := context.Background()
	keyType, _ := client.Type(ctx, key).Result()
	ttl, _ := client.TTL(ctx, key).Result()
	ttlMs := int64(-1)
	if ttl > 0 {
		ttlMs = ttl.Milliseconds()
	}

	var value interface{}
	var total int64 = -1
	nextToken := ""
	truncated := false

	offset := int64(0)
	if pageToken != "" {
		if n, err := strconv.ParseInt(pageToken, 10, 64); err == nil && n > 0 {
			offset = n
		}
	}
	scanCursor := uint64(0)
	if pageToken != "" {
		if n, err := strconv.ParseUint(pageToken, 10, 64); err == nil {
			scanCursor = n
		}
	}

	switch keyType {
	case "string":
		size, _ := client.StrLen(ctx, key).Result()
		total = size
		// 用 GETRANGE 只取预览区间，避免超大字符串整串读入
		if size > redisStringPreviewBytes {
			value, _ = client.GetRange(ctx, key, 0, redisStringPreviewBytes-1).Result()
			truncated = true
		} else {
			value, _ = client.Get(ctx, key).Result()
		}

	case "list":
		total, _ = client.LLen(ctx, key).Result()
		items, err := client.LRange(ctx, key, offset, offset+int64(count)-1).Result()
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		if items == nil {
			items = []string{}
		}
		value = items
		if offset+int64(len(items)) < total {
			nextToken = strconv.FormatInt(offset+int64(len(items)), 10)
			truncated = true
		}

	case "set":
		total, _ = client.SCard(ctx, key).Result()
		members, next, err := client.SScan(ctx, key, scanCursor, "", int64(count)).Result()
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		if members == nil {
			members = []string{}
		}
		value = members
		if next != 0 {
			nextToken = strconv.FormatUint(next, 10)
			truncated = true
		}

	case "zset":
		total, _ = client.ZCard(ctx, key).Result()
		items, err := client.ZRangeWithScores(ctx, key, offset, offset+int64(count)-1).Result()
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		if items == nil {
			items = []redis.Z{}
		}
		value = items
		if offset+int64(len(items)) < total {
			nextToken = strconv.FormatInt(offset+int64(len(items)), 10)
			truncated = true
		}

	case "hash":
		total, _ = client.HLen(ctx, key).Result()
		flat, next, err := client.HScan(ctx, key, scanCursor, "", int64(count)).Result()
		if err != nil {
			resp.InternalErr(c, err)
			return
		}
		// HScan 返回扁平 [field, value, field, value, ...]，这里还原成 map，
		// 与旧的 HGetAll 返回结构保持一致，前端无需改动取值方式。
		m := make(map[string]string, len(flat)/2)
		for i := 0; i+1 < len(flat); i += 2 {
			m[flat[i]] = flat[i+1]
		}
		value = m
		if next != 0 {
			nextToken = strconv.FormatUint(next, 10)
			truncated = true
		}

	default:
		value = nil
	}

	c.JSON(http.StatusOK, gin.H{
		"key":        key,
		"type":       keyType,
		"value":      value,
		"ttl":        ttlMs,
		"total":      total,
		"count":      count,
		"nextToken":  nextToken,
		"truncated":  truncated,
		"pageToken":  pageToken,
		"previewCap": redisStringPreviewBytes,
	})
}

// GET /api/v1/redis/:id/slowlog
func RedisSlowlog(c *gin.Context) {
	id := c.Param("id")
	count, _ := strconv.Atoi(c.DefaultQuery("count", "10"))

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

	client, err := getRedisClient(&instance, 0)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer client.Close()

	ctx := context.Background()
	logs, err := client.SlowLogGet(ctx, int64(count))
	if err != nil {
		resp.InternalErr(c, err)
		return
	}

	var result []map[string]interface{}
	for _, log := range logs {
		result = append(result, map[string]interface{}{
			"id":       log.ID,
			"time":     log.Time.Unix(),
			"duration": log.Duration.Microseconds(),
			"command":  fmt.Sprintf("%v", log.Args),
			"client":   log.ClientAddr,
		})
	}
	if result == nil {
		result = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, result)
}

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

	client, err := getRedisClient(&instance, 0)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer client.Close()

	ctx := context.Background()
	clientList, err := client.ClientList(ctx)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}

	// Parse client list
	var clients []map[string]interface{}
	for _, line := range strings.Split(clientList, "\n") {
		line = strings.TrimSpace(line)
		if line == "" {
			continue
		}
		clientInfo := make(map[string]interface{})
		for _, field := range strings.Split(line, " ") {
			parts := strings.SplitN(field, "=", 2)
			if len(parts) == 2 {
				clientInfo[parts[0]] = parts[1]
			}
		}
		clients = append(clients, clientInfo)
	}
	if clients == nil {
		clients = []map[string]interface{}{}
	}
	c.JSON(http.StatusOK, clients)
}

// POST /api/v1/redis/:id/command — 只读命令行(引号感知拆参,白名单按首词判定)
func RedisExecuteCommand(c *gin.Context) {
	id := c.Param("id")

	var input struct {
		DB      int      `json:"db"`
		Command string   `json:"command" binding:"required"`
		Args    []string `json:"args"` // 兼容旧调用:显式参数
	}
	if err := c.ShouldBindJSON(&input); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}
	if v := c.Query("db"); v != "" && input.DB == 0 {
		input.DB, _ = strconv.Atoi(v)
	}

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

	client, err := getRedisClient(&instance, input.DB)
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	defer client.Close()

	ctx := context.Background()

	// 命令解析:优先按整串引号感知拆分;整串为空时兼容旧的显式 args 调用
	var verb string
	var args []string
	if _, parts := splitRedisCommand(input.Command); len(parts) > 0 {
		verb = strings.ToUpper(parts[0])
		args = parts[1:]
	} else if len(input.Args) > 0 {
		verb = strings.ToUpper(strings.TrimSpace(input.Args[0]))
		args = input.Args[1:]
	}
	if verb == "" {
		resp.BadRequest(c, "命令不能为空")
		return
	}

	readOnly := map[string]bool{
		"GET": true, "MGET": true, "HGET": true, "HGETALL": true, "HKEYS": true, "HVALS": true, "HSTRLEN": true,
		"LRANGE": true, "LINDEX": true, "LLEN": true, "SMEMBERS": true, "SISMEMBER": true, "SCARD": true,
		"ZRANGE": true, "ZREVRANGE": true, "ZSCORE": true, "ZCARD": true,
		"TYPE": true, "TTL": true, "PTTL": true, "EXISTS": true, "STRLEN": true,
		"INFO": true, "DBSIZE": true, "SCAN": true, "HSCAN": true, "SSCAN": true, "ZSCAN": true,
		"GETRANGE": true, "OBJECT": true, "RANDOMKEY": true, "PING": true, "ECHO": true,
		"XRANGE": true, "XLEN": true, "MEMORY": true, "BITCOUNT": true, "GETBIT": true,
	}
	if !readOnly[verb] {
		resp.Forbidden(c, "仅允许只读命令,数据变更请走工单流程")
		return
	}

	cmdArgs := make([]interface{}, 0, len(args)+1)
	cmdArgs = append(cmdArgs, verb)
	for _, a := range args {
		cmdArgs = append(cmdArgs, a)
	}
	result, err := client.Do(ctx, cmdArgs...).Result()
	if err != nil {
		c.JSON(http.StatusOK, gin.H{"result": nil, "error": redisErrHint(err).Error()})
		return
	}
	c.JSON(http.StatusOK, gin.H{"result": result})
}

// splitRedisCommand 按空白拆分命令串(双引号/单引号感知,支持引号内空格)
func splitRedisCommand(cmdline string) (verb string, args []string) {
	var parts []string
	var cur strings.Builder
	inQuote := byte(0)
	flush := func() {
		if cur.Len() > 0 {
			parts = append(parts, cur.String())
			cur.Reset()
		}
	}
	for i := 0; i < len(cmdline); i++ {
		ch := cmdline[i]
		switch {
		case inQuote != 0:
			if ch == inQuote {
				inQuote = 0
			} else {
				cur.WriteByte(ch)
			}
		case ch == '"' || ch == '\'':
			inQuote = ch
		case ch == ' ' || ch == '\t':
			flush()
		default:
			cur.WriteByte(ch)
		}
	}
	flush()
	if len(parts) == 0 {
		return "", nil
	}
	return strings.ToUpper(parts[0]), parts
}

func stringsToInterfaces(ss []string) []interface{} {
	result := make([]interface{}, len(ss))
	for i, s := range ss {
		result[i] = s
	}
	return result
}

// POST /api/v1/redis/:id/keys
func RedisCreateKey(c *gin.Context) {
	id := c.Param("id")
	dbParam := c.Query("db")
	db := 0
	if dbParam != "" {
		db, _ = strconv.Atoi(dbParam)
	}

	var input struct {
		Key   string      `json:"key" binding:"required"`
		Type  string      `json:"type" binding:"required"`
		Value interface{} `json:"value"`
		TTL   int         `json:"ttl"`
	}
	if err := c.ShouldBindJSON(&input); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}

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

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

	ctx := context.Background()

	if err := writeRedisKeyValue(ctx, client, input.Key, input.Type, input.Value); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}

	if input.TTL > 0 {
		client.Expire(ctx, input.Key, time.Duration(input.TTL)*time.Second)
	}

	c.JSON(http.StatusOK, gin.H{"message": "Key created"})
}

// PUT /api/v1/redis/:id/keys/:key
func RedisUpdateKey(c *gin.Context) {
	id := c.Param("id")
	key := c.Param("key")
	dbParam := c.Query("db")
	db := 0
	if dbParam != "" {
		db, _ = strconv.Atoi(dbParam)
	}

	var input struct {
		Type  string      `json:"type" binding:"required"`
		Value interface{} `json:"value"`
		TTL   int         `json:"ttl"`
	}
	if err := c.ShouldBindJSON(&input); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}

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

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

	ctx := context.Background()

	// 记录原 TTL:未显式指定新 TTL 时保留原值(旧行为会静默丢 TTL)
	prevTTL, _ := client.TTL(ctx, key).Result()

	// string 覆盖写无需 DEL;容器类型可能收缩,需先清空再重写
	if input.Type != "string" {
		if err := client.Del(ctx, key).Err(); err != nil {
			resp.InternalErr(c, err)
			return
		}
	}
	if err := writeRedisKeyValue(ctx, client, key, input.Type, input.Value); err != nil {
		resp.BadRequest(c, err.Error())
		return
	}
	switch {
	case input.TTL > 0:
		client.Expire(ctx, key, time.Duration(input.TTL)*time.Second)
	case input.TTL == -1:
		client.Persist(ctx, key)
	case prevTTL > 0:
		client.Expire(ctx, key, prevTTL)
	}

	c.JSON(http.StatusOK, gin.H{"message": "Key updated"})
}

// DELETE /api/v1/redis/:id/keys/:key
func RedisDeleteKey(c *gin.Context) {
	id := c.Param("id")
	key := c.Param("key")
	dbParam := c.Query("db")
	db := 0
	if dbParam != "" {
		db, _ = strconv.Atoi(dbParam)
	}

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

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

	ctx := context.Background()
	deleted, err := client.Del(ctx, key).Result()
	if err != nil {
		resp.InternalErr(c, err)
		return
	}
	c.JSON(http.StatusOK, gin.H{"message": "Key deleted", "deleted": deleted})
}

// POST /api/v1/workflows/:id/clone
func CloneWorkflow(c *gin.Context) {
	id, _ := strconv.Atoi(c.Param("id"))
	var wf model.WorkflowDefinition
	if err := model.DB.First(&wf, id).Error; err != nil {
		resp.NotFound(c, "Workflow not found")
		return
	}
	// clone
	newWf := model.WorkflowDefinition{
		Name:        wf.Name + " (副本)",
		Description: wf.Description,
		Category:    wf.Category,
		IsActive:    wf.IsActive,
		FlowConfig:  wf.FlowConfig,
	}
	model.DB.Create(&newWf)
	c.JSON(http.StatusCreated, newWf)
}