style(backend/redis): gofumpt formatting and whitespace fixes in redis_common.go

Author: hyp3rd

pkg/backend/backend.go

@@ -23,7 +23,7 @@ import (
 // It restricts the generic type parameter to supported backend types, ensuring
 // type safety and proper implementation at compile time.
 type IBackendConstrain interface {
-	InMemory | Redis
+	InMemory | Redis | RedisCluster
 }
 
 // IBackend defines the contract that all cache backends must implement.

pkg/backend/options.go

@@ -22,6 +22,11 @@ func (rb *Redis) setCapacity(capacity int) {
 	rb.capacity = capacity
 }
 
+// setCapacity sets the `Capacity` field of the `RedisCluster` backend.
+func (rc *RedisCluster) setCapacity(capacity int) {
+	rc.capacity = capacity
+}
+
 // Option is a function type that can be used to configure the `HyperCache` struct.
 type Option[T IBackendConstrain] func(*T)
 
@@ -64,3 +69,24 @@ func WithSerializer[T Redis](serializer serializer.ISerializer) Option[Redis] {
 		backend.Serializer = serializer
 	}
 }
+
+// WithRedisClusterClient sets the redis cluster client to use.
+func WithRedisClusterClient[T RedisCluster](client *redis.ClusterClient) Option[RedisCluster] {
+	return func(backend *RedisCluster) {
+		backend.rdb = client
+	}
+}
+
+// WithClusterKeysSetName sets the name of the set for cluster backend keys.
+func WithClusterKeysSetName[T RedisCluster](keysSetName string) Option[RedisCluster] {
+	return func(backend *RedisCluster) {
+		backend.keysSetName = keysSetName
+	}
+}
+
+// WithClusterSerializer sets the serializer for the cluster backend.
+func WithClusterSerializer[T RedisCluster](ser serializer.ISerializer) Option[RedisCluster] {
+	return func(backend *RedisCluster) {
+		backend.Serializer = ser
+	}
+}

pkg/backend/redis.go

@@ -125,120 +125,17 @@ func (cacheBackend *Redis) Get(ctx context.Context, key string) (*cache.Item, bo
 
 // Set stores the Item in the cacheBackend.
 func (cacheBackend *Redis) Set(ctx context.Context, item *cache.Item) error {
-	pipe := cacheBackend.rdb.TxPipeline()
-
-	// Check if the item is valid
-	err := item.Valid()
-	if err != nil {
-		// Return the item to the pool
-		return err
-	}
-
-	// Serialize the item
-	data, err := cacheBackend.Serializer.Marshal(item)
-	if err != nil {
-		// Return the item to the pool
-		return err
-	}
-
-	expiration := item.Expiration.String()
-
-	// Store the item in the cacheBackend
-	err = pipe.HSet(ctx, item.Key, map[string]any{
-		"data":       data,
-		"expiration": expiration,
-	}).Err()
-	if err != nil {
-		return ewrap.Wrap(err, "failed to set item in redis")
-	}
-	// Add the key to the set of keys associated with the cache prefix
-	pipe.SAdd(ctx, cacheBackend.keysSetName, item.Key)
-	// Set the expiration if it is not zero
-	if item.Expiration > 0 {
-		pipe.Expire(ctx, item.Key, item.Expiration)
-	}
-
-	_, err = pipe.Exec(ctx)
-	if err != nil {
-		return ewrap.Wrap(err, "failed to execute redis pipeline")
-	}
-
-	return nil
+	return redisSet(ctx, cacheBackend.rdb, cacheBackend.keysSetName, item, cacheBackend.Serializer)
 }
 
 // List returns a list of all the items in the cacheBackend that match the given filter options.
 func (cacheBackend *Redis) List(ctx context.Context, filters ...IFilter) ([]*cache.Item, error) {
-	// Get the set of keys held in the cacheBackend with the given `keysSetName`
-	keys, err := cacheBackend.rdb.SMembers(ctx, cacheBackend.keysSetName).Result()
-	if err != nil {
-		return nil, ewrap.Wrap(err, "failed to get keys from redis")
-	}
-
-	// Execute the Redis commands in a pipeline transaction to improve performance and reduce the number of round trips
-	cmds, err := cacheBackend.rdb.Pipelined(ctx, func(pipe redis.Pipeliner) error {
-		// Get the items from the cacheBackend
-		for _, key := range keys {
-			pipe.HGetAll(ctx, key)
-		}
-
-		return nil
-	})
-	if err != nil {
-		return nil, ewrap.Wrap(err, "failed to execute redis pipeline while listing")
-	}
-
-	// Create a slice to hold the items
-	items := make([]*cache.Item, 0, len(keys))
-
-	// Deserialize the items and add them to the slice of items to return
-	for _, cmd := range cmds {
-		command, ok := cmd.(*redis.MapStringStringCmd)
-		if !ok {
-			continue
-		}
-
-		data, err := command.Result() // Change the type assertion to match HGetAll
-		if err != nil {
-			return nil, ewrap.Wrap(err, "failed to get item data from redis")
-		}
-
-		item := cacheBackend.itemPoolManager.Get()
-		// Return the item to the pool
-		defer cacheBackend.itemPoolManager.Put(item)
-
-		err = cacheBackend.Serializer.Unmarshal([]byte(data["data"]), item)
-		if err == nil {
-			items = append(items, item)
-		}
-	}
-
-	// Apply the filters
-	if len(filters) > 0 {
-		for _, filter := range filters {
-			items, err = filter.ApplyFilter(constants.RedisBackend, items)
-		}
-	}
-
-	return items, err
+	return redisList(ctx, cacheBackend.rdb, cacheBackend.keysSetName, cacheBackend.Serializer, cacheBackend.itemPoolManager, filters...)
 }
 
 // Remove removes an item from the cache with the given key.
 func (cacheBackend *Redis) Remove(ctx context.Context, keys ...string) error {
-	pipe := cacheBackend.rdb.TxPipeline()
-
-	_, err := pipe.SRem(ctx, cacheBackend.keysSetName, keys).Result()
-	if err != nil {
-		return ewrap.Wrap(err, "removing keys from set")
-	}
-
-	_, err = pipe.Del(ctx, keys...).Result()
-	if err != nil {
-		return ewrap.Wrap(err, "removing keys")
-	}
-
-	_, err = pipe.Exec(ctx)
-
-	return ewrap.Wrap(err, "executing pipeline")
+	return redisRemove(ctx, cacheBackend.rdb, cacheBackend.keysSetName, keys...)
 }
 
 // Clear removes all items from the cache.

pkg/backend/redis_cluster.go

@@ -0,0 +1,134 @@
+package backend
+
+import (
+	"context"
+	"errors"
+	"time"
+
+	"github.com/redis/go-redis/v9"
+
+	"github.com/hyp3rd/ewrap"
+
+	"github.com/hyp3rd/hypercache/internal/constants"
+	"github.com/hyp3rd/hypercache/internal/libs/serializer"
+	"github.com/hyp3rd/hypercache/internal/sentinel"
+	"github.com/hyp3rd/hypercache/pkg/cache"
+)
+
+const (
+	clusterMaxRetries   = 3
+	clusterRetriesDelay = 100 * time.Millisecond
+)
+
+// RedisCluster is a cache backend that stores items in a Redis Cluster.
+// It mirrors the single-node Redis backend semantics but uses go-redis ClusterClient.
+type RedisCluster struct {
+	rdb             *redis.ClusterClient // redis cluster client
+	capacity        int                  // capacity of the cache
+	itemPoolManager *cache.ItemPoolManager
+	keysSetName     string
+	Serializer      serializer.ISerializer
+}
+
+// NewRedisCluster creates a new Redis Cluster backend with the given options.
+func NewRedisCluster(redisOptions ...Option[RedisCluster]) (IBackend[RedisCluster], error) {
+	rc := &RedisCluster{itemPoolManager: cache.NewItemPoolManager()}
+
+	ApplyOptions(rc, redisOptions...)
+
+	if rc.rdb == nil {
+		return nil, sentinel.ErrNilClient
+	}
+
+	if rc.capacity < 0 {
+		return nil, sentinel.ErrInvalidCapacity
+	}
+
+	if rc.keysSetName == "" {
+		rc.keysSetName = constants.RedisBackend
+	}
+
+	if rc.Serializer == nil {
+		var err error
+
+		rc.Serializer, err = serializer.New("msgpack")
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	return rc, nil
+}
+
+// SetCapacity sets the capacity of the cluster backend.
+func (cacheBackend *RedisCluster) SetCapacity(capacity int) {
+	if capacity < 0 {
+		return
+	}
+
+	cacheBackend.capacity = capacity
+}
+
+// Capacity returns the capacity of the cluster backend.
+func (cacheBackend *RedisCluster) Capacity() int {
+	return cacheBackend.capacity
+}
+
+// Count returns the number of keys stored.
+func (cacheBackend *RedisCluster) Count(ctx context.Context) int {
+	count, err := cacheBackend.rdb.DBSize(ctx).Result()
+	if err != nil {
+		return 0
+	}
+
+	return int(count)
+}
+
+// Get retrieves an item by key.
+func (cacheBackend *RedisCluster) Get(ctx context.Context, key string) (*cache.Item, bool) {
+	isMember, err := cacheBackend.rdb.SIsMember(ctx, cacheBackend.keysSetName, key).Result()
+	if err != nil || !isMember {
+		return nil, false
+	}
+
+	item := cacheBackend.itemPoolManager.Get()
+	defer cacheBackend.itemPoolManager.Put(item)
+
+	data, err := cacheBackend.rdb.HGet(ctx, key, "data").Bytes()
+	if err != nil {
+		if errors.Is(err, redis.Nil) {
+			return nil, false
+		}
+
+		return nil, false
+	}
+
+	err = cacheBackend.Serializer.Unmarshal(data, item)
+	if err != nil {
+		return nil, false
+	}
+
+	return item, true
+}
+
+// Set stores an item in the cluster.
+func (cacheBackend *RedisCluster) Set(ctx context.Context, item *cache.Item) error {
+	return redisSet(ctx, cacheBackend.rdb, cacheBackend.keysSetName, item, cacheBackend.Serializer)
+}
+
+// List returns items matching optional filters.
+func (cacheBackend *RedisCluster) List(ctx context.Context, filters ...IFilter) ([]*cache.Item, error) {
+	return redisList(ctx, cacheBackend.rdb, cacheBackend.keysSetName, cacheBackend.Serializer, cacheBackend.itemPoolManager, filters...)
+}
+
+// Remove deletes the specified keys.
+func (cacheBackend *RedisCluster) Remove(ctx context.Context, keys ...string) error {
+	return redisRemove(ctx, cacheBackend.rdb, cacheBackend.keysSetName, keys...)
+}
+
+// Clear flushes the database.
+func (cacheBackend *RedisCluster) Clear(ctx context.Context) error {
+	_, err := cacheBackend.rdb.FlushDB(ctx).Result()
+
+	return ewrap.Wrap(err, "flushing database", ewrap.WithRetry(clusterMaxRetries, clusterRetriesDelay))
+}