feat: optionally configure stream method to use a private channel

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.
   ///
+  /// You may pass an optional [channelConfig] to configure the realtime channel to e.g., make it private.
+  ///
   /// 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,
+    RealtimeChannelConfig? channelConfig,
+  }) {
     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,
+      channelConfig: channelConfig,
     );
   }
 }

packages/supabase/lib/src/supabase_stream_builder.dart

@@ -53,6 +53,16 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
   final String _realtimeTopic;
 
+  /// Realtime channel config for the stream.
+  ///
+  /// Currently, only the [RealtimeChannelConfig.private] option affects the
+  /// `stream` method, since the `stream` method only handles postgres changes.
+  ///
+  /// Defaults to the constructor of [RealtimeChannelConfig] with its respective
+  /// default values, which means the channel will be a public channel by
+  /// default.
+  final RealtimeChannelConfig _channelConfig;
+
   RealtimeChannel? _channel;
 
   final String _schema;
@@ -89,12 +99,14 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
     required String schema,
     required String table,
     required List<String> primaryKey,
+    RealtimeChannelConfig? channelConfig,
   })  : _queryBuilder = queryBuilder,
         _realtimeTopic = realtimeTopic,
         _realtimeClient = realtimeClient,
         _schema = schema,
         _table = table,
-        _uniqueColumns = primaryKey;
+        _uniqueColumns = primaryKey,
+        _channelConfig = channelConfig ?? const RealtimeChannelConfig();
 
   /// Orders the result with the specified [column].
   ///
@@ -167,7 +179,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
       );
     }
 
-    _channel = _realtimeClient.channel(_realtimeTopic);
+    _channel = _realtimeClient.channel(_realtimeTopic, _channelConfig);
 
     _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,
+    super.channelConfig,
   });
 
   /// 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,27 @@ 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'],
+        channelConfig: const RealtimeChannelConfig(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);