Repository: centrifugal/centrifuge

Issue:
[feature] Support rediss:// which enables TLS in addition to tls_enabled param
**Is your feature request related to a problem? Please describe.**

Currently it's achievable over `?tls_enabled=true`, but `rediss://` is quite common.



Discussion:
@FZambia: Part of https://github.com/centrifugal/centrifuge/releases/tag/v0.35.0

Solution patch:
diff --git a/_examples/redis_broker_presence/main.go b/_examples/redis_broker_presence/main.go
index 36d00fa..4828933 100644
--- a/_examples/redis_broker_presence/main.go
+++ b/_examples/redis_broker_presence/main.go
@@ -97,15 +97,9 @@ func main() {
 	})
 
 	redisShardConfigs := []centrifuge.RedisShardConfig{
-		{Address: "localhost:6379"},
-		//{Address: "localhost:6380"},
-		//{
-		//	ClusterAddresses:  []string{"localhost:7000", "localhost:7001", "localhost:7002"},
-		//},
-		//{
-		//	SentinelAddresses:  []string{"localhost:26379"},
-		//	SentinelMasterName: "mymaster",
-		//},
+		{Address: "redis://localhost:6379"},
+		//{Address: "redis://localhost:6380"}, // Uncomment for client-side sharding.
+		//{Address: "redis+sentinel://localhost:26379?sentinel_master_name=mymaster"}, // Example for Redis Sentinel.
 	}
 	var redisShards []*centrifuge.RedisShard
 	for _, redisConf := range redisShardConfigs {
@@ -159,6 +153,7 @@ func main() {
 	http.Handle("/", http.FileServer(http.Dir("./")))
 
 	go func() {
+		log.Print("Starting server, visit http://localhost:" + strconv.Itoa(*port))
 		if err := http.ListenAndServe(":"+strconv.Itoa(*port), nil); err != nil {
 			log.Fatal(err)
 		}
diff --git a/broker_redis.go b/broker_redis.go
index 3dae1c1..33817f8 100644
--- a/broker_redis.go
+++ b/broker_redis.go
@@ -395,8 +395,19 @@ func (b *RedisBroker) Close(_ context.Context) error {
 	return nil
 }
 
+func getPubSubStartLogFields(s *RedisShard, logFields map[string]any) map[string]any {
+	startLogFields := make(map[string]any, len(logFields))
+	for k, v := range logFields {
+		startLogFields[k] = v
+	}
+	if s.isCluster {
+		startLogFields["cluster"] = true
+	}
+	return startLogFields
+}
+
 func (b *RedisBroker) runControlPubSub(s *RedisShard, logFields map[string]any, eventHandler BrokerEventHandler, startOnce func(error)) {
-	b.node.logger.log(newLogEntry(LogLevelDebug, "running Redis control PUB/SUB", logFields))
+	b.node.logger.log(newLogEntry(LogLevelDebug, "running Redis control PUB/SUB", getPubSubStartLogFields(s, logFields)))
 	defer func() {
 		b.node.logger.log(newLogEntry(LogLevelDebug, "stopping Redis control PUB/SUB", logFields))
 	}()
@@ -511,7 +522,7 @@ func (b *RedisBroker) runPubSub(s *shardWrapper, logFields map[string]any, event
 		if useShardedPubSub {
 			debugLogValues["cluster_shard_index"] = clusterShardIndex
 		}
-		b.node.logger.log(newLogEntry(LogLevelDebug, "running Redis PUB/SUB", logFields, debugLogValues))
+		b.node.logger.log(newLogEntry(LogLevelDebug, "running Redis PUB/SUB", getPubSubStartLogFields(s.shard, logFields), debugLogValues))
 		defer func() {
 			b.node.logger.log(newLogEntry(LogLevelDebug, "stopping Redis PUB/SUB", logFields, debugLogValues))
 		}()
diff --git a/docker-compose.yml b/docker-compose.yml
index 9029163..5025f16 100644
--- a/docker-compose.yml
+++ b/docker-compose.yml
@@ -5,26 +5,6 @@ services:
     image: redis:${REDIS_VERSION:-7}-alpine
     ports:
       - "6379:6379"
-  valkey:
-    image: valkey/valkey:${VALKEY_VERSION:-8}-alpine
-    ports:
-      - "8379:6379"
-  dragonflydb:
-    image: docker.dragonflydb.io/dragonflydb/dragonfly:v1.19.0
-    ports:
-      - "7379:6379"
-  sentinel:
-    image: redis:${REDIS_VERSION:-7}-alpine
-    entrypoint:
-      - /bin/sh
-      - -c
-      - |
-        redis-server --save "" --appendonly no --port 6380 &
-        echo "sentinel monitor mymaster 127.0.0.1 6380 2\n" > sentinel.conf
-        redis-server sentinel.conf --sentinel
-    ports:
-      - "6380:6380"
-      - "26379:26379"
   cluster:
     image: redis:${REDIS_VERSION:-7}-alpine
     entrypoint:
@@ -40,18 +20,43 @@ services:
       - "7000:7000"
       - "7001:7001"
       - "7002:7002"
+  sentinel:
+    image: redis:${REDIS_VERSION:-7}-alpine
+    entrypoint:
+      - /bin/sh
+      - -c
+      - |
+        redis-server --save "" --appendonly no --port 6380 &
+        echo "sentinel monitor mymaster 127.0.0.1 6380 2\n" > sentinel.conf
+        redis-server sentinel.conf --sentinel
+    ports:
+      - "6380:6380"
+      - "26379:26379"
+  valkey:
+    image: valkey/valkey:${VALKEY_VERSION:-8}-alpine
+    ports:
+      - "16379:6379"
   valkey-cluster:
     image: valkey/valkey:${VALKEY_VERSION:-8}-alpine
     entrypoint:
       - /bin/sh
       - -c
       - |
-        valkey-server --port 8000 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 8000.conf &
-        valkey-server --port 8001 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 8001.conf &
-        valkey-server --port 8002 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 8002.conf &
-        while ! valkey-cli --cluster create 127.0.0.1:8000 127.0.0.1:8001 127.0.0.1:8002 --cluster-yes; do sleep 1; done
+        valkey-server --port 17000 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 17000.conf &
+        valkey-server --port 17001 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 17001.conf &
+        valkey-server --port 17002 --save "" --appendonly no --cluster-enabled yes --cluster-config-file 17002.conf &
+        while ! valkey-cli --cluster create 127.0.0.1:17000 127.0.0.1:17001 127.0.0.1:17002 --cluster-yes; do sleep 1; done
         wait
     ports:
-      - "8000:8000"
-      - "8001:8001"
-      - "8002:8002"
\ No newline at end of file
+      - "17000:17000"
+      - "17001:17001"
+      - "17002:17002"
+  dragonflydb:
+    image: docker.dragonflydb.io/dragonflydb/dragonfly:v1.27.1
+    ports:
+      - "36379:6379"
+  dragonflydb-cluster:
+    image: docker.dragonflydb.io/dragonflydb/dragonfly:v1.27.1
+    ports:
+      - "37000:6379"
+    command: ["--cluster_mode", "emulated"]
diff --git a/redis_shard.go b/redis_shard.go
index 8edf2f4..c6ddcae 100644
--- a/redis_shard.go
+++ b/redis_shard.go
@@ -37,7 +37,9 @@ type RedisShard struct {
 
 var knownRedisURLPrefixes = []string{
 	"redis://",
+	"rediss://",
 	"redis+sentinel://",
+	"rediss+sentinel://",
 	"redis+cluster://",
 	"unix://",
 	"tcp://",
@@ -76,8 +78,20 @@ func optionsFromAddress(address string, options rueidis.ClientOption) (fromAddre
 
 	var addresses []string
 
+	query := u.Query()
+
+	if query.Has("sentinel_master_name") {
+		result.ClientOption.Sentinel.MasterSet = query.Get("sentinel_master_name")
+	}
+
+	isCluster := u.Scheme == "redis+cluster"
+	isSentinel := u.Scheme == "redis+sentinel" || u.Scheme == "rediss+sentinel" || result.ClientOption.Sentinel.MasterSet != ""
+	if isSentinel && result.ClientOption.Sentinel.MasterSet == "" {
+		return result, errors.New("sentinel master name must be configured for Redis Sentinel setup")
+	}
+
 	switch u.Scheme {
-	case "tcp", "redis", "redis+sentinel", "redis+cluster":
+	case "tcp", "redis", "redis+sentinel", "redis+cluster", "rediss", "rediss+sentinel":
 		addresses = []string{u.Host}
 		if u.Path != "" {
 			db, err := strconv.Atoi(strings.TrimPrefix(u.Path, "/"))
@@ -86,6 +100,14 @@ func optionsFromAddress(address string, options rueidis.ClientOption) (fromAddre
 			}
 			result.ClientOption.SelectDB = db
 		}
+		if strings.HasPrefix(u.Scheme, "rediss") {
+			if result.ClientOption.TLSConfig == nil {
+				result.ClientOption.TLSConfig = &tls.Config{}
+			}
+			if isSentinel && result.ClientOption.Sentinel.TLSConfig == nil {
+				result.ClientOption.Sentinel.TLSConfig = &tls.Config{}
+			}
+		}
 	case "unix":
 		addresses = []string{u.Path}
 		result.ClientOption.DialFn = func(s string, d *net.Dialer, c *tls.Config) (net.Conn, error) {
@@ -102,7 +124,6 @@ func optionsFromAddress(address string, options rueidis.ClientOption) (fromAddre
 		}
 	}
 
-	query := u.Query()
 	addresses = append(addresses, query["addr"]...)
 
 	if query.Has("connect_timeout") {
@@ -139,10 +160,6 @@ func optionsFromAddress(address string, options rueidis.ClientOption) (fromAddre
 		result.ClientOption.AlwaysRESP2 = val
 	}
 
-	if query.Has("sentinel_master_name") {
-		result.ClientOption.Sentinel.MasterSet = query.Get("sentinel_master_name")
-	}
-
 	if query.Has("sentinel_user") {
 		result.ClientOption.Sentinel.Username = query.Get("sentinel_user")
 	}
@@ -169,13 +186,10 @@ func optionsFromAddress(address string, options rueidis.ClientOption) (fromAddre
 		result.ReplicaClientEnabled = val
 	}
 
-	if u.Scheme == "redis+sentinel" && result.ClientOption.Sentinel.MasterSet == "" {
-		return result, errors.New("sentinel master name must be configured for Redis Sentinel setup")
-	}
-
 	result.ClientOption.InitAddress = addresses
-	result.IsCluster = u.Scheme == "redis+cluster"
-	result.IsSentinel = u.Scheme == "redis+sentinel"
+	result.IsCluster = isCluster
+	result.IsSentinel = isSentinel
+
 	return result, nil
 }
 
@@ -231,13 +245,6 @@ func NewRedisShard(_ *Node, conf RedisShardConfig) (*RedisShard, error) {
 			addressOpts.ClientOption, addressOpts.IsCluster, addressOpts.IsSentinel, addressOpts.ReplicaClientEnabled
 	}
 
-	shard := &RedisShard{
-		config:       conf,
-		isCluster:    isCluster,
-		closeCh:      make(chan struct{}),
-		finalAddress: options.InitAddress,
-	}
-
 	if isSentinel && options.Sentinel.MasterSet == "" {
 		return nil, errors.New("sentinel master name must be configured for Redis Sentinel setup")
 	}
@@ -246,6 +253,20 @@ func NewRedisShard(_ *Node, conf RedisShardConfig) (*RedisShard, error) {
 	if err != nil {
 		return nil, fmt.Errorf("error creating Redis client: %v", err)
 	}
+	if strings.Contains(strings.ToLower(fmt.Sprintf("%T", client)), "cluster") {
+		// Cluster mode is not explicitly set but client is a cluster client – thus set isCluster to true.
+		// This scenario covered with tests for our main integrations: see TestNewRedisShard.
+		// Centrifuge need to know that it's working with Redis Cluster to construct proper keys.
+		isCluster = true
+	}
+
+	shard := &RedisShard{
+		config:       conf,
+		isCluster:    isCluster,
+		closeCh:      make(chan struct{}),
+		finalAddress: options.InitAddress,
+	}
+
 	shard.client = client
 
 	if replicaClientEnabled {
@@ -269,18 +290,13 @@ type RedisShardConfig struct {
 	// - host:port
 	// - tcp://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
 	// - redis://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
+	// - rediss://[[[user]:password]@]host:port[/db][?option1=value1&optionN=valueN]
 	// - unix://[[[user]:password]@]path[?option1=value1&optionN=valueN]
-	// It's also possible to use Address with redis+sentinel:// and redis+cluster://
-	// schemes when connecting to Redis Sentinel and Redis Cluster respectively.
-	// Examples:
-	// - redis+sentinel://[[[user]:password]@]host:port?sentinel_master_name=mymaster
-	// - redis+cluster://[[[user]:password]@]host:port[?addr=host2:port2&addr=host3:port3]
-	// If you need to connect to Redis Cluster then you need to provide ClusterAddresses
-	// or must use redis+cluster:// scheme in Address.
-	// If you need to connect to Redis Sentinel then you need to provide SentinelAddresses
-	// or must use redis+sentinel:// scheme in Address.
-	// I.e. Centrifuge requires you to explicitly specify the type of Redis setup you want
-	// to connect to.
+	// It's also possible to use Address with redis+sentinel:// scheme to connect to Redis Sentinel:
+	// - redis+sentinel://[[[user]:password]@]host:port?sentinel_master_name=mymaster?addr=host2:port2&addr=host3:port3
+	// In case of using redis+sentinel://, sentinel_master_name is required and host:port points to Sentinel instance.
+	// It's also possible to connect to Redis Cluster by providing ClusterAddresses instead of Address.
+	// It's also possible to connect to Redis Sentinel by providing SentinelAddresses instead of Address.
 	Address string
 	// ClusterAddresses is a slice of seed cluster addresses to connect to.
 	// Each address should be in form of host:port. If ClusterAddresses set then



Test patch:
diff --git a/broker_redis_test.go b/broker_redis_test.go
index 04e2396..fd4839b 100644
--- a/broker_redis_test.go
+++ b/broker_redis_test.go
@@ -179,14 +179,14 @@ type historyRedisTest struct {
 
 var historyRedisTests = []historyRedisTest{
 	{"rd_single_list", false, false, 6379},
-	{"vk_single_list", false, false, 8379},
+	{"vk_single_list", false, false, 16379},
 	{"rd_single_strm", true, false, 6379},
-	{"vk_single_strm", true, false, 8379},
-	{"df_single_list", false, false, 7379},
-	{"df_single_strm", true, false, 7379},
+	{"vk_single_strm", true, false, 16379},
+	{"df_single_list", false, false, 36379},
+	{"df_single_strm", true, false, 36379},
 	{"rd_cluster_list", false, true, 7000},
 	{"rd_cluster_strm", true, true, 7000},
-	{"vk_cluster_strm", true, true, 8000},
+	{"vk_cluster_strm", true, true, 17000},
 }
 
 type noHistoryRedisTest struct {
@@ -197,8 +197,8 @@ type noHistoryRedisTest struct {
 
 var noHistoryRedisTests = []noHistoryRedisTest{
 	{"rd_single", false, 6379},
-	{"df_single", false, 7379},
-	{"vk_single", false, 8379},
+	{"df_single", false, 36379},
+	{"vk_single", false, 16379},
 	{"rd_cluster", false, 0},
 }
 
@@ -222,17 +222,6 @@ var noHistoryBenchRedisTests = func() (tests []noHistoryRedisTest) {
 	return
 }()
 
-//func excludeHistoryClusterTests(tests []historyRedisTest) []historyRedisTest {
-//	var res []historyRedisTest
-//	for _, t := range tests {
-//		if t.UseCluster {
-//			continue
-//		}
-//		res = append(res, t)
-//	}
-//	return res
-//}
-
 func excludeNoHistoryClusterTests(tests []noHistoryRedisTest) []noHistoryRedisTest {
 	var res []noHistoryRedisTest
 	for _, t := range tests {
@@ -1785,9 +1774,9 @@ var throughputTests = []throughputTest{
 	{1, 0, 0, 6379},
 	{2, 0, 0, 6379},
 	{4, 0, 0, 6379},
-	{1, 0, 0, 7379},
-	{2, 0, 0, 7379},
-	{4, 0, 0, 7379},
+	{1, 0, 0, 36379},
+	{2, 0, 0, 36379},
+	{4, 0, 0, 36379},
 }
 
 func BenchmarkPubSubThroughput(b *testing.B) {
diff --git a/presence_redis_test.go b/presence_redis_test.go
index 81819a1..34fe746 100644
--- a/presence_redis_test.go
+++ b/presence_redis_test.go
@@ -83,10 +83,10 @@ type redisPresenceTest struct {
 
 var redisPresenceTests = []redisPresenceTest{
 	{"rd_single", false, 6379},
-	{"df_single", false, 7379},
-	{"vk_single", false, 8379},
+	{"df_single", false, 36379},
+	{"vk_single", false, 16379},
 	{"rd_cluster", true, 7000},
-	{"vk_cluster", true, 8000},
+	{"vk_cluster", true, 17000},
 }
 
 func excludeClusterPresenceTests(tests []redisPresenceTest) []redisPresenceTest {
diff --git a/redis_shard_integration_test.go b/redis_shard_integration_test.go
new file mode 100644
index 0000000..2355de9
--- /dev/null
+++ b/redis_shard_integration_test.go
@@ -0,0 +1,64 @@
+//go:build integration
+
+package centrifuge
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewRedisShard(t *testing.T) {
+	testCases := []struct {
+		name            string
+		config          RedisShardConfig
+		expectedCluster bool
+	}{
+		{
+			name: "redis standalone",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:6379",
+			},
+		},
+		{
+			name: "redis cluster",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:7000",
+			},
+			expectedCluster: true,
+		},
+		{
+			name: "valkey standalone",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:16379",
+			},
+		},
+		{
+			name: "valkey cluster",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:17000",
+			},
+			expectedCluster: true,
+		},
+		{
+			name: "dragonfly standalone",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:36379",
+			},
+		},
+		{
+			name: "dragonfly cluster emulated",
+			config: RedisShardConfig{
+				Address: "redis://127.0.0.1:37000",
+			},
+			expectedCluster: true,
+		},
+	}
+	for _, tc := range testCases {
+		t.Run(tc.name, func(t *testing.T) {
+			s, err := NewRedisShard(&Node{}, tc.config)
+			require.NoError(t, err)
+			require.Equal(t, tc.expectedCluster, s.isCluster)
+		})
+	}
+}
diff --git a/redis_shard_test.go b/redis_shard_test.go
index 952c228..02e417a 100644
--- a/redis_shard_test.go
+++ b/redis_shard_test.go
@@ -170,6 +170,32 @@ func TestOptionsFromAddress(t *testing.T) {
 			},
 			expectedIsSentinel: true,
 		},
+		{
+			name:          "Redis secure URL",
+			address:       "rediss://127.0.0.1:6379",
+			inputOptions:  rueidis.ClientOption{},
+			expectedError: nil,
+			expectedOutput: rueidis.ClientOption{
+				InitAddress: []string{"127.0.0.1:6379"},
+				TLSConfig:   &tls.Config{},
+			},
+		},
+		{
+			name:    "Redis secure URL does not override explicitly set TLS config",
+			address: "rediss://127.0.0.1:6379",
+			inputOptions: rueidis.ClientOption{
+				TLSConfig: &tls.Config{
+					InsecureSkipVerify: true,
+				},
+			},
+			expectedError: nil,
+			expectedOutput: rueidis.ClientOption{
+				InitAddress: []string{"127.0.0.1:6379"},
+				TLSConfig: &tls.Config{
+					InsecureSkipVerify: true,
+				},
+			},
+		},
 	}
 
 	for _, tt := range tests {