feat: enhance redis and production IA

2026-07-07 10:39:33 +08:00
1 parent 644981eef4
commit 58bbc394e9
@@ -184,6 +184,13 @@ func (w *redisClientWrapper) HSet(ctx context.Context, key string, values ...int
 	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) Expire(ctx context.Context, key string, expiration time.Duration) *redis.BoolCmd {
 	if w.isCluster && w.ClusterClient != nil {
 		return w.ClusterClient.Expire(ctx, key, expiration)
@@ -219,6 +226,79 @@ func (w *redisClientWrapper) SScan(ctx context.Context, key string, cursor uint6
 	return w.Client.SScan(ctx, key, cursor, match, count)
 }

+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) {
 	password := instance.Password
 	// Decrypt if encrypted
@@ -237,12 +317,16 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
 	}

 	client := redis.NewClient(&redis.Options{
-		Addr:     addr,
-		Username: redisUsername,
-		Password: password,
-		DB:       db,
+		Addr:         addr,
+		Username:     redisUsername,
+		Password:     password,
+		DB:           db,
+		DialTimeout:  5 * time.Second,
+		ReadTimeout:  5 * time.Second,
+		WriteTimeout: 5 * time.Second,
 	})
-	ctx := context.Background()
+	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)
@@ -250,16 +334,9 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro

 	// Check cluster mode flag for wrapper behavior (ScanAllKeys needs to know)
 	isCluster := instance.ClusterMode != nil && *instance.ClusterMode
-	wrapper := &redisClientWrapper{Client: client, isCluster: isCluster}
-	if !isCluster {
-		return wrapper, nil
-	}
-
-	// Redis Cluster routes key commands by slot. Build the address list from
-	// CLUSTER SLOTS, but replace advertised internal node IPs with the configured
-	// host so Docker/NAT deployments can still reach the mapped node ports.
 	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 {
@@ -280,10 +357,18 @@ func getRedisClient(instance *model.Instance, db int) (*redisClientWrapper, erro
 		}
 	}

+	wrapper := &redisClientWrapper{Client: client, isCluster: isCluster}
+	if !isCluster {
+		return wrapper, nil
+	}
+
 	clusterClient := redis.NewClusterClient(&redis.ClusterOptions{
-		Addrs:    addrs,
-		Username: redisUsername,
-		Password: password,
+		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()
@@ -710,25 +795,9 @@ func RedisCreateKey(c *gin.Context) {

 	ctx := context.Background()

-	switch input.Type {
-	case "string":
-		client.Set(ctx, input.Key, input.Value, 0)
-	case "list":
-		if vals, ok := input.Value.([]interface{}); ok {
-			for _, v := range vals {
-				client.RPush(ctx, input.Key, fmt.Sprintf("%v", v))
-			}
-		}
-	case "set":
-		if vals, ok := input.Value.([]interface{}); ok {
-			for _, v := range vals {
-				client.SAdd(ctx, input.Key, fmt.Sprintf("%v", v))
-			}
-		}
-	case "hash":
-		if vals, ok := input.Value.(map[string]interface{}); ok {
-			client.HSet(ctx, input.Key, vals)
-		}
+	if err := writeRedisKeyValue(ctx, client, input.Key, input.Type, input.Value); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
+		return
 	}

 	if input.TTL > 0 {
@@ -738,6 +807,55 @@ func RedisCreateKey(c *gin.Context) {
 	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 {
+		c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
+		return
+	}
+
+	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
+	}
+
+	client, err := getRedisClient(&instance, db)
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
+		return
+	}
+	defer client.Close()
+
+	ctx := context.Background()
+	if err := client.Del(ctx, key).Err(); err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"detail": err.Error()})
+		return
+	}
+	if err := writeRedisKeyValue(ctx, client, key, input.Type, input.Value); err != nil {
+		c.JSON(http.StatusBadRequest, gin.H{"detail": err.Error()})
+		return
+	}
+	if input.TTL > 0 {
+		client.Expire(ctx, key, time.Duration(input.TTL)*time.Second)
+	}
+
+	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")