feat: optionally configure stream method to use a private channel (#1357)

* feat: optionally configure stream method to use a private channel

close #1311

* test: add missing mock tests

* refactor: move to dedicated argument in stream method

Author: Vinzent03

packages/supabase/lib/src/supabase_query_builder.dart

@@ -25,10 +25,12 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
 
   /// Combines the current state of your table from PostgREST with changes from the realtime server to return real-time data from your table as a [Stream].
   ///
-  /// Realtime is disabled by default for new tables. You can turn it on by [managing replication](https://supabase.com/docs/guides/realtime/extensions/postgres-changes#replication-setup).
+  /// Realtime is disabled by default for new tables. You can turn it on by [managing replication](https://supabase.com/docs/guides/realtime/subscribing-to-database-changes#enable-postgres-changes).
   ///
   /// Pass the list of primary key column names to [primaryKey], which will be used to update and delete the proper records internally as the stream receives real-time updates.
   ///
+  /// The underlying [RealtimeChannel] is public by default. Set [private] to `true` to make it private, which requires additional RLS policies to be set up. See https://supabase.com/docs/guides/realtime/authorization for more details.
+  ///
   /// It handles the lifecycle of the realtime connection and automatically refetches data from PostgREST when needed.
   ///
   /// Make sure to provide `onError` and `onDone` callbacks to [Stream.listen] to handle errors and completion of the stream.
@@ -43,7 +45,10 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
   /// ```dart
   /// supabase.from('chats').stream(primaryKey: ['id']).eq('room_id','123').order('created_at').limit(20).listen(_onChatsReceived);
   /// ```
-  SupabaseStreamFilterBuilder stream({required List<String> primaryKey}) {
+  SupabaseStreamFilterBuilder stream({
+    required List<String> primaryKey,
+    bool private = false,
+  }) {
     assert(primaryKey.isNotEmpty, 'Please specify primary key column(s).');
     return SupabaseStreamFilterBuilder(
       queryBuilder: this,
@@ -52,6 +57,7 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder {
       schema: _schema,
       table: _table,
       primaryKey: primaryKey,
+      private: private,
     );
   }
 }

packages/supabase/lib/src/supabase_stream_builder.dart

@@ -53,6 +53,10 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
   final String _realtimeTopic;
 
+  /// Whether the underlying [_channel] should be initialized as private
+  /// or not. Default is false, which means the channel is public.
+  final bool _private;
+
   RealtimeChannel? _channel;
 
   final String _schema;
@@ -89,12 +93,14 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
     required String schema,
     required String table,
     required List<String> primaryKey,
+    required bool private,
   })  : _queryBuilder = queryBuilder,
         _realtimeTopic = realtimeTopic,
         _realtimeClient = realtimeClient,
         _schema = schema,
         _table = table,
-        _uniqueColumns = primaryKey;
+        _uniqueColumns = primaryKey,
+        _private = private;
 
   /// Orders the result with the specified [column].
   ///
@@ -167,7 +173,12 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
       );
     }
 
-    _channel = _realtimeClient.channel(_realtimeTopic);
+    _channel = _realtimeClient.channel(
+      _realtimeTopic,
+      RealtimeChannelConfig(
+        private: _private,
+      ),
+    );
 
     _channel!
         .onPostgresChanges(

packages/supabase/lib/src/supabase_stream_filter_builder.dart

@@ -8,6 +8,7 @@ class SupabaseStreamFilterBuilder extends SupabaseStreamBuilder {
     required super.schema,
     required super.table,
     required super.primaryKey,
+    required super.private,
   });
 
   /// Filters the results where [column] equals [value].

packages/supabase/test/mock_test.dart

@@ -23,6 +23,7 @@ void main() {
   Future<void> handleRequests(
     HttpServer server, {
     String? expectedFilter,
+    bool? expectedPrivate,
   }) async {
     await for (final HttpRequest request in server) {
       final headers = request.headers;
@@ -113,8 +114,9 @@ void main() {
           final requestJson = jsonDecode(request);
           final topic = requestJson['topic'];
           final ref = requestJson["ref"];
+          final event = requestJson['event'];
 
-          if (requestJson["event"] == "phx_leave") {
+          if (event == 'phx_leave') {
             listeners.remove(topic);
             return;
           }
@@ -126,10 +128,15 @@ void main() {
           final String? realtimeFilter = requestJson['payload']['config']
                   ['postgres_changes']
               .first['filter'];
+          final bool isPrivate =
+              requestJson['payload']['config']['private'] as bool;
 
           if (expectedFilter != null) {
             expect(realtimeFilter, expectedFilter);
           }
+          if (expectedPrivate != null) {
+            expect(isPrivate, expectedPrivate);
+          }
 
           final replyString = jsonEncode({
             'event': 'phx_reply',
@@ -682,6 +689,25 @@ void main() {
     });
   });
 
+  group('stream() channel config', () {
+    test('forwards channelConfig.private=true to realtime join payload', () {
+      handleRequests(mockServer, expectedPrivate: true);
+
+      final stream =
+          supabase.from('todos').stream(primaryKey: ['id'], private: true);
+
+      expect(stream, emits(isList));
+    });
+
+    test('uses default private=false when channelConfig is omitted', () {
+      handleRequests(mockServer, expectedPrivate: false);
+
+      final stream = supabase.from('todos').stream(primaryKey: ['id']);
+
+      expect(stream, emits(isList));
+    });
+  });
+
   group('Deprecated execute method', () {
     test('should work with deprecated execute method', () {
       handleRequests(mockServer);