feat: support more realtime filters and multiple filters for stream

packages/supabase/lib/src/supabase_query_builder.dart

@@ -23,24 +23,43 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder<dynamic> {
          url: Uri.parse(url),
        );
 
-  /// 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].
+  /// 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/subscribing-to-database-changes#enable-postgres-changes).
+  /// 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.
+  /// 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 realtime 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.
+  /// 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.
+  /// It handles the life cycle 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.
+  /// Make sure to provide `onError` and `onDone` callbacks to [Stream.listen]
+  /// to handle errors and completion of the stream.
   /// The stream gets closed when the realtime connection is closed.
   ///
+  /// Be aware of the following limitations when using streams:
+  /// - When using filters like `eq` only realtime updates matching the filter
+  ///   will be received. Therefore, an update that changes a record to no
+  ///   longer match the filter will not be received, and the record will remain
+  ///   in the stream.
+  /// - By default, for DELETE events only the primary key columns can be used.
+  ///   Refer to the documentation about [receiving old records](https://supabase.com/docs/guides/realtime/postgres-changes?queryGroups=language&language=dart#receiving-old-records).
+  ///
   /// ```dart
   /// supabase.from('chats').stream(primaryKey: ['id']).listen(_onChatsReceived);
   /// ```
   ///
-  /// `eq`, `neq`, `lt`, `lte`, `gt` or `gte` and `order`, `limit` filter are available to limit the data being queried.
+  /// `eq`, `neq`, `lt`, `lte`, `gt` `gte`, `like`, `ilike`, `match`, `imatch`,
+  /// or `isDistinct` and `order`, `limit` filter are available to limit the
+  /// data being queried.
   ///
   /// ```dart
   /// supabase.from('chats').stream(primaryKey: ['id']).eq('room_id','123').order('created_at').limit(20).listen(_onChatsReceived);

packages/supabase/lib/src/supabase_stream_builder.dart

@@ -58,9 +58,8 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   /// Contains the combined data of postgrest and realtime to emit as stream.
   SupabaseStreamEvent _streamData = [];
 
-  /// `eq` filter used for both postgrest and realtime
-  // ignore: avoid-unassigned-fields
-  _StreamPostgrestFilter? _streamFilter;
+  /// Filters to be applied to the stream
+  final List<_StreamPostgrestFilter> _streamFilter = [];
 
   /// Which column to order by and whether it's ascending
   _Order? _orderBy;
@@ -147,16 +146,16 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
   }
 
   void _getStreamData() {
-    final currentStreamFilter = _streamFilter;
     _streamData = [];
-    PostgresChangeFilter? realtimeFilter;
-    if (currentStreamFilter != null) {
-      realtimeFilter = PostgresChangeFilter(
-        type: currentStreamFilter.type,
-        column: currentStreamFilter.column,
-        value: currentStreamFilter.value,
-      );
-    }
+    List<PostgresChangeFilter> realtimeFilter = _streamFilter
+        .map(
+          (filter) => PostgresChangeFilter(
+            column: filter.column,
+            type: filter.type,
+            value: filter.value,
+          ),
+        )
+        .toList();
 
     _channel = _realtimeClient.channel(
       _realtimeTopic,
@@ -170,7 +169,7 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
           event: PostgresChangeEvent.all,
           schema: _schema,
           table: _table,
-          filter: realtimeFilter,
+          filters: realtimeFilter,
           callback: (payload) {
             switch (payload.eventType) {
               case PostgresChangeEvent.insert:
@@ -226,48 +225,59 @@ class SupabaseStreamBuilder extends Stream<SupabaseStreamEvent> {
 
   Future<void> _getPostgrestData() async {
     PostgrestFilterBuilder<PostgrestList> query = _queryBuilder.select();
-    if (_streamFilter != null) {
-      query = switch (_streamFilter!.type) {
+    for (final filter in _streamFilter) {
+      query = switch (filter.type) {
         PostgresChangeFilterType.eq => query.eq(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.neq => query.neq(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.lt => query.lt(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.lte => query.lte(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.gt => query.gt(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.gte => query.gte(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
         ),
         PostgresChangeFilterType.inFilter => query.inFilter(
-          _streamFilter!.column,
-          _streamFilter!.value,
+          filter.column,
+          filter.value,
+        ),
+        PostgresChangeFilterType.like => query.like(
+          filter.column,
+          filter.value,
+        ),
+        PostgresChangeFilterType.ilike => query.ilike(
+          filter.column,
+          filter.value,
+        ),
+        PostgresChangeFilterType.match => query.matchRegex(
+          filter.column,
+          filter.value,
+        ),
+        PostgresChangeFilterType.imatch => query.imatchRegex(
+          filter.column,
+          filter.value,
+        ),
+        PostgresChangeFilterType.isFilter => query.isFilter(
+          filter.column,
+          filter.value,
         ),
-        // These operators are only reachable through the realtime
-        // `onPostgresChanges` API, not through `.stream()`'s filter builder,
-        // so they can never be set on `_streamFilter`. Guard the exhaustive
-        // switch defensively in case that ever changes.
-        PostgresChangeFilterType.like ||
-        PostgresChangeFilterType.ilike ||
-        PostgresChangeFilterType.isFilter ||
-        PostgresChangeFilterType.match ||
-        PostgresChangeFilterType.imatch ||
-        PostgresChangeFilterType.isDistinct => throw UnsupportedError(
-          'The "${_streamFilter!.type.name}" filter is not supported by '
-          '`.stream()`. Use one of eq, neq, lt, lte, gt, gte or inFilter.',
+        PostgresChangeFilterType.isDistinct => query.isDistinct(
+          filter.column,
+          filter.value,
         ),
       };
     }