feat(realtime_client): Add support for authorized realtime channels with broadcast and presence. (#970)

* Add private channel support

* remove unused variables

* update test

* chore: add comment to _private in realtime_channel

* Update packages/realtime_client/lib/src/realtime_channel.dart

Co-authored-by: Bruno D'Luka <45696119+bdlukaa@users.noreply.github.com>

* Update packages/realtime_client/lib/src/realtime_channel.dart

Co-authored-by: Bruno D'Luka <45696119+bdlukaa@users.noreply.github.com>

* format code

---------

Co-authored-by: Bruno D'Luka <45696119+bdlukaa@users.noreply.github.com>

Author: dshukertjr

packages/realtime_client/lib/src/realtime_channel.dart

@@ -32,6 +32,9 @@ class RealtimeChannel {
   @internal
   final RealtimeClient socket;
 
+  /// Defines if the channel is private or not and if RLS policies will be used to check data
+  late final bool _private;
+
   RealtimeChannel(
     this.topic,
     this.socket, {
@@ -40,7 +43,8 @@ class RealtimeChannel {
         params = params.toMap(),
         subTopic = topic.replaceFirst(
             RegExp(r"^realtime:", caseSensitive: false), "") {
-    broadcastEndpointURL = _broadcastEndpointURL;
+    broadcastEndpointURL = '${httpEndpointURL(socket.endPoint)}/api/broadcast';
+    _private = params.private;
 
     joinPush = Push(
       this,
@@ -117,6 +121,7 @@ class RealtimeChannel {
     } else {
       final broadcast = params['config']['broadcast'];
       final presence = params['config']['presence'];
+      final isPrivate = params['config']['private'];
 
       _onError((e) {
         if (callback != null) callback(RealtimeSubscribeStatus.channelError, e);
@@ -131,6 +136,7 @@ class RealtimeChannel {
         'presence': presence,
         'postgres_changes':
             _bindings['postgres_changes']?.map((r) => r.filter).toList() ?? [],
+        'private': isPrivate == true,
       };
 
       if (socket.accessToken != null) {
@@ -483,13 +489,20 @@ class RealtimeChannel {
     }
 
     if (!canPush && type == RealtimeListenTypes.broadcast) {
-      final headers = {'Content-Type': 'application/json', ...socket.headers};
+      final headers = <String, String>{
+        'Content-Type': 'application/json',
+        if (socket.params['apikey'] != null) 'apikey': socket.params['apikey']!,
+        ...socket.headers,
+        if (socket.accessToken != null)
+          'Authorization': 'Bearer ${socket.accessToken}',
+      };
       final body = {
         'messages': [
           {
             'topic': subTopic,
             'payload': payload,
             'event': event,
+            'private': _private,
           }
         ]
       };
@@ -595,18 +608,6 @@ class RealtimeChannel {
     return completer.future;
   }
 
-  String get _broadcastEndpointURL {
-    var url = socket.endPoint;
-    url = url.replaceFirst(RegExp(r'^ws', caseSensitive: false), 'http');
-    url = url.replaceAll(
-      RegExp(r'(/socket/websocket|/socket|/websocket)/?$',
-          caseSensitive: false),
-      '',
-    );
-    url = '${url.replaceAll(RegExp(r'/+$'), '')}/api/broadcast';
-    return url;
-  }
-
   /// Overridable message hook
   ///
   /// Receives all events for specialized message handling before dispatching to the channel callbacks.

packages/realtime_client/lib/src/realtime_client.dart

@@ -51,6 +51,7 @@ class RealtimeClient {
   String? accessToken;
   List<RealtimeChannel> channels = [];
   final String endPoint;
+
   final Map<String, String> headers;
   final Map<String, dynamic> params;
   final Duration timeout;
@@ -85,6 +86,7 @@ class RealtimeClient {
   /// Initializes the Socket
   ///
   /// `endPoint` The string WebSocket endpoint, ie, "ws://example.com/socket", "wss://example.com", "/socket" (inherited host & protocol)
+  /// `httpEndpoint` The string HTTP endpoint, ie, "https://example.com", "/" (inherited host & protocol)
   /// `transport` The Websocket Transport, for example WebSocket.
   /// `timeout` The default timeout in milliseconds to trigger push timeouts.
   /// `params` The optional params to pass when connecting.

packages/realtime_client/lib/src/transformers.dart

@@ -332,3 +332,20 @@ Map<String, Map<String, dynamic>> getPayloadRecords(
 
   return records;
 }
+
+/// Converts a WebSocket URL to an HTTP URL.
+String httpEndpointURL(String socketUrl) {
+  var url = socketUrl;
+
+  // Replace 'ws' or 'wss' with 'http' or 'https' respectively
+  url = url.replaceFirst(RegExp(r'^ws', caseSensitive: false), 'http');
+
+  // Remove WebSocket-specific endings
+  url = url.replaceFirst(
+    RegExp(r'(/socket/websocket|/socket|/websocket)/?$', caseSensitive: false),
+    '',
+  );
+
+  // Remove trailing slashes
+  return url.replaceAll(RegExp(r'/+$'), '');
+}

packages/realtime_client/lib/src/types.dart

@@ -148,10 +148,14 @@ class RealtimeChannelConfig {
   /// [key] option is used to track presence payload across clients
   final String key;
 
+  /// Defines if the channel is private or not and if RLS policies will be used to check data
+  final bool private;
+
   const RealtimeChannelConfig({
     this.ack = false,
     this.self = false,
     this.key = '',
+    this.private = false,
   });
 
   Map<String, dynamic> toMap() {
@@ -164,6 +168,7 @@ class RealtimeChannelConfig {
         'presence': {
           'key': key,
         },
+        'private': private,
       }
     };
   }

packages/realtime_client/test/channel_test.dart

@@ -4,6 +4,7 @@ import 'dart:io';
 
 import 'package:realtime_client/realtime_client.dart';
 import 'package:realtime_client/src/constants.dart';
+import 'package:realtime_client/src/push.dart';
 import 'package:realtime_client/src/types.dart';
 import 'package:test/test.dart';
 
@@ -33,11 +34,31 @@ void main() {
       expect(channel.params, {
         'config': {
           'broadcast': {'ack': false, 'self': false},
-          'presence': {'key': ''}
+          'presence': {'key': ''},
+          'private': false,
         }
       });
       expect(channel.socket, socket);
     });
+
+    test('sets up joinPush object with private defined', () {
+      channel = RealtimeChannel(
+        'topic',
+        socket,
+        params: RealtimeChannelConfig(
+          private: true,
+        ),
+      );
+      final Push joinPush = channel.joinPush;
+
+      expect(joinPush.payload, {
+        'config': {
+          'broadcast': {'ack': false, 'self': false},
+          'presence': {'key': ''},
+          'private': true,
+        },
+      });
+    });
   });
 
   group('join', () {
@@ -252,7 +273,7 @@ void main() {
         params: {'apikey': 'supabaseKey'},
       );
 
-      channel = socket.channel('myTopic');
+      channel = socket.channel('myTopic', RealtimeChannelConfig(private: true));
     });
 
     tearDown(() async {
@@ -311,9 +332,11 @@ void main() {
         final body = json.decode(await utf8.decodeStream(req));
         final message = body['messages'][0];
         final payload = message['payload'];
+        final private = message['private'];
 
         expect(payload, containsPair('myKey', 'myValue'));
         expect(message, containsPair('topic', 'myTopic'));
+        expect(private, true);
 
         await req.response.close();
         break;

packages/realtime_client/test/socket_test.dart

@@ -287,7 +287,8 @@ void main() {
       expect(channel.params, {
         'config': {
           'broadcast': {'ack': false, 'self': false},
-          'presence': {'key': ''}
+          'presence': {'key': ''},
+          'private': false,
         }
       });
     });

packages/realtime_client/test/transformers_test.dart

@@ -234,4 +234,19 @@ void main() {
       expect(enrichedPayload, expectedMap);
     });
   });
+
+  group('httpEndpointURL', () {
+    test('Converts a hosted Supabase WS URL', () {
+      expect(
+        httpEndpointURL('wss://example.supabase.co/realtime/v1'),
+        equals('https://example.supabase.co/realtime/v1'),
+      );
+    });
+    test('Converts a custom domain WS URL', () {
+      expect(
+        httpEndpointURL('wss://custom-domain.com/realtime/v1'),
+        equals('https://custom-domain.com/realtime/v1'),
+      );
+    });
+  });
 }