feat: Valkey cluster support with room key ensure RPC

broadcast-worker/deploy/docker-compose.yml

@@ -15,7 +15,7 @@ services:
       # Set USER_CACHE_SIZE=0 to disable caching.
       - USER_CACHE_SIZE=10000
       - USER_CACHE_TTL=5m
-      - VALKEY_ADDR=valkey:6379
+      - VALKEY_ADDRS=valkey:6379
       - VALKEY_KEY_GRACE_PERIOD=24h
       - BOOTSTRAP_STREAMS=true
       - ENCRYPTION_ENABLED=${ENCRYPTION_ENABLED:-false}

broadcast-worker/main.go

@@ -40,7 +40,7 @@ type config struct {
 	UserCacheTTL         time.Duration           `env:"USER_CACHE_TTL"            envDefault:"5m"`
 	RoomMetaCacheSize    int                     `env:"ROOM_META_CACHE_SIZE"      envDefault:"10000"`
 	RoomMetaCacheTTL     time.Duration           `env:"ROOM_META_CACHE_TTL"       envDefault:"2m"`
-	ValkeyAddr           string                  `env:"VALKEY_ADDR"`
+	ValkeyAddrs          []string                `env:"VALKEY_ADDRS"              envSeparator:","`
 	ValkeyPassword       string                  `env:"VALKEY_PASSWORD"           envDefault:""`
 	ValkeyKeyGracePeriod time.Duration           `env:"VALKEY_KEY_GRACE_PERIOD" envDefault:"24h"`
 	Consumer             stream.ConsumerSettings `envPrefix:"CONSUMER_"`
@@ -88,14 +88,17 @@ func main() {
 
 	var keyStore roomkeystore.RoomKeyStore
 	if cfg.Encryption.Enabled {
-		if cfg.ValkeyAddr == "" || cfg.ValkeyKeyGracePeriod <= 0 {
-			slog.Error("encryption enabled but VALKEY_ADDR is empty or VALKEY_KEY_GRACE_PERIOD is not a positive duration",
-				"valkey_addr_set", cfg.ValkeyAddr != "",
+		if len(cfg.ValkeyAddrs) == 0 {
+			slog.Error("encryption enabled but VALKEY_ADDRS is empty")
+			os.Exit(1)
+		}
+		if cfg.ValkeyKeyGracePeriod <= 0 {
+			slog.Error("VALKEY_KEY_GRACE_PERIOD must be a positive duration",
 				"valkey_key_grace_period", cfg.ValkeyKeyGracePeriod)
 			os.Exit(1)
 		}
-		keyStore, err = roomkeystore.NewValkeyStore(roomkeystore.Config{
-			Addr:        cfg.ValkeyAddr,
+		keyStore, err = roomkeystore.NewValkeyClusterStore(roomkeystore.ClusterConfig{
+			Addrs:       cfg.ValkeyAddrs,
 			Password:    cfg.ValkeyPassword,
 			GracePeriod: cfg.ValkeyKeyGracePeriod,
 		})

docker-local/compose.deps.yaml

@@ -143,20 +143,33 @@ services:
       - chat-local
 
   # Valkey backs the per-user restricted-rooms cache in search-service (5-min
-  # TTL, lazy-populated from Elasticsearch on miss). Persistence is disabled:
-  # the cache is derivative of the authoritative user-room ES doc and survives
-  # restart only through the lazy-populate path.
+  # TTL, lazy-populated from Elasticsearch on miss) and room key pairs for
+  # encryption. Persistence is disabled: the cache is derivative of the
+  # authoritative user-room ES doc and survives restart only through the
+  # lazy-populate path.
+  # Single-node cluster-mode: the entrypoint starts valkey-server with
+  # --cluster-enabled, waits for it to accept connections, then assigns all
+  # 16384 hash slots so it forms a valid one-node cluster.
   valkey:
     image: valkey/valkey:8-alpine
     container_name: chat-local-valkey
     ports:
       - "6379:6379"
-    command: ["valkey-server", "--save", "", "--appendonly", "no"]
+    entrypoint:
+      - sh
+      - -c
+      - |
+        valkey-server --cluster-enabled yes --cluster-config-file /tmp/nodes.conf --cluster-node-timeout 5000 --save '' --appendonly no &
+        until valkey-cli ping > /dev/null 2>&1; do sleep 0.1; done
+        if ! valkey-cli CLUSTER INFO | grep -q 'cluster_slots_assigned:16384'; then
+          valkey-cli CLUSTER ADDSLOTSRANGE 0 16383
+        fi
+        wait
     healthcheck:
-      test: ["CMD", "valkey-cli", "ping"]
+      test: ["CMD-SHELL", "valkey-cli CLUSTER INFO | grep 'cluster_state:ok'"]
       interval: 5s
       timeout: 3s
-      retries: 5
+      retries: 10
     networks:
       - chat-local