diff --git a/packages/supabase/lib/src/supabase_query_builder.dart b/packages/supabase/lib/src/supabase_query_builder.dart index a8d6b48ef..e957d1074 100644 --- a/packages/supabase/lib/src/supabase_query_builder.dart +++ b/packages/supabase/lib/src/supabase_query_builder.dart @@ -23,24 +23,43 @@ class SupabaseQueryBuilder extends PostgrestQueryBuilder { 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); diff --git a/packages/supabase/lib/src/supabase_stream_builder.dart b/packages/supabase/lib/src/supabase_stream_builder.dart index 0247bf5ae..b5f5724d3 100644 --- a/packages/supabase/lib/src/supabase_stream_builder.dart +++ b/packages/supabase/lib/src/supabase_stream_builder.dart @@ -58,9 +58,8 @@ class SupabaseStreamBuilder extends Stream { /// 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 { } void _getStreamData() { - final currentStreamFilter = _streamFilter; _streamData = []; - PostgresChangeFilter? realtimeFilter; - if (currentStreamFilter != null) { - realtimeFilter = PostgresChangeFilter( - type: currentStreamFilter.type, - column: currentStreamFilter.column, - value: currentStreamFilter.value, - ); - } + List 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 { 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 { Future _getPostgrestData() async { PostgrestFilterBuilder 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, ), }; } diff --git a/packages/supabase/lib/src/supabase_stream_filter_builder.dart b/packages/supabase/lib/src/supabase_stream_filter_builder.dart index adff3a650..52c3406b7 100644 --- a/packages/supabase/lib/src/supabase_stream_filter_builder.dart +++ b/packages/supabase/lib/src/supabase_stream_filter_builder.dart @@ -13,113 +13,184 @@ class SupabaseStreamFilterBuilder extends SupabaseStreamBuilder { /// Filters the results where [column] equals [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).eq('name', 'Supabase'); /// ``` - SupabaseStreamBuilder eq(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder eq(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.eq, column: column, value: value, - ); + )); return this; } /// Filters the results where [column] does not equal [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).neq('name', 'Supabase'); /// ``` - SupabaseStreamBuilder neq(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder neq(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.neq, column: column, value: value, - ); + )); return this; } /// Filters the results where [column] is less than [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).lt('likes', 100); /// ``` - SupabaseStreamBuilder lt(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder lt(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.lt, column: column, value: value, - ); + )); return this; } /// Filters the results where [column] is less than or equal to [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).lte('likes', 100); /// ``` - SupabaseStreamBuilder lte(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder lte(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.lte, column: column, value: value, - ); + )); return this; } /// Filters the results where [column] is greater than [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).gt('likes', '100'); /// ``` - SupabaseStreamBuilder gt(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder gt(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.gt, column: column, value: value, - ); + )); return this; } /// Filters the results where [column] is greater than or equal to [value]. /// - /// Only one filter can be applied to `.stream()`. - /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).gte('likes', 100); /// ``` - SupabaseStreamBuilder gte(String column, Object value) { - _streamFilter = ( + SupabaseStreamFilterBuilder gte(String column, Object value) { + _streamFilter.add(( type: PostgresChangeFilterType.gte, column: column, value: value, - ); + )); return this; } - /// Filters the results where [column] is included in [value]. - /// - /// Only one filter can be applied to `.stream()`. + /// Filters the results where [column] is included in [values]. /// /// ```dart /// supabase.from('users').stream(primaryKey: ['id']).inFilter('name', ['Andy', 'Amy', 'Terry']); /// ``` - SupabaseStreamBuilder inFilter(String column, List values) { - _streamFilter = ( + SupabaseStreamFilterBuilder inFilter(String column, List values) { + _streamFilter.add(( type: PostgresChangeFilterType.inFilter, column: column, value: values, - ); + )); + return this; + } + + /// Filters the results where [column] matches the [pattern] case-sensitive. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).like('title', '%foo%'); + /// ``` + SupabaseStreamFilterBuilder like(String column, String pattern) { + _streamFilter.add(( + type: PostgresChangeFilterType.like, + column: column, + value: pattern, + )); + return this; + } + + /// Filters the results where [column] matches the [pattern] case-insensitive. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).ilike('title', '%foo%'); + /// ``` + SupabaseStreamFilterBuilder ilike(String column, String pattern) { + _streamFilter.add(( + type: PostgresChangeFilterType.ilike, + column: column, + value: pattern, + )); + return this; + } + + /// Filters the results where [column] matches the POSIX [regex] case-sensitive. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).match('slug', r'^post-\d+$'); + /// ``` + SupabaseStreamFilterBuilder match(String column, String regex) { + _streamFilter.add(( + type: PostgresChangeFilterType.match, + column: column, + value: regex, + )); + return this; + } + + /// Filters the results where [column] matches the POSIX [regex] case-insensitive. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).imatch('slug', r'^post-\d+$'); + /// ``` + SupabaseStreamFilterBuilder imatch(String column, String regex) { + _streamFilter.add(( + type: PostgresChangeFilterType.imatch, + column: column, + value: regex, + )); + return this; + } + + /// Filters the results where [column] is `null`, `true` or `false`. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).isFilter('data', null); + /// ``` + SupabaseStreamFilterBuilder isFilter(String column, bool? value) { + _streamFilter.add(( + type: PostgresChangeFilterType.isFilter, + column: column, + value: value, + )); + return this; + } + + /// Filters the results where [column] is not equal to [value] treating `null + /// as a distinct value. + /// + /// ```dart + /// supabase.from('users').stream(primaryKey: ['id']).isDistinct('age', null); + /// ``` + SupabaseStreamFilterBuilder isDistinct(String column, Object? value) { + _streamFilter.add(( + type: PostgresChangeFilterType.isDistinct, + column: column, + value: value, + )); return this; } }