feat(realtime_client): support new postgres changes filter operators, multi-filter, column selection, and replication-ready events (#1526)

Author: filipecabaco

packages/realtime_client/lib/src/realtime_channel.dart

@@ -2,6 +2,7 @@ import 'dart:async';
 import 'dart:convert';
 import 'dart:typed_data';
 
+import 'package:collection/collection.dart';
 import 'package:http/http.dart';
 import 'package:meta/meta.dart';
 import 'package:realtime_client/realtime_client.dart';
@@ -273,6 +274,10 @@ class RealtimeChannel {
       final filter = clientPostgresBinding.filter['filter'];
       final serverPostgresFilter = serverPostgresFilters[i];
 
+      // NOTE: `select` is intentionally not part of this equality check (mirroring
+      // supabase-js), so a server that echoes `select` back in a slightly
+      // different shape does not force a spurious unsubscribe. The client
+      // binding keeps its own `select` regardless.
       if (serverPostgresFilter != null &&
           serverPostgresFilter['event'] == event &&
           serverPostgresFilter['schema'] == schema &&
@@ -369,6 +374,13 @@ class RealtimeChannel {
   ///
   /// [filter] can be used to further control which rows to listen to within the given [schema] and [table].
   ///
+  /// [filters] combines multiple [PostgresChangeFilter]s with an `AND`. Provide
+  /// either [filter] or [filters], not both.
+  ///
+  /// [select] restricts the change payload to a subset of columns instead of the
+  /// full row (reducing payload size). The listed columns must be selectable by
+  /// the subscribing role.
+  ///
   /// ```dart
   /// supabase.channel('my_channel').onPostgresChanges(
   ///     event: PostgresChangeEvent.all,
@@ -388,15 +400,29 @@ class RealtimeChannel {
     String? schema,
     String? table,
     PostgresChangeFilter? filter,
+    List<PostgresChangeFilter>? filters,
+    List<String>? select,
     required void Function(PostgresChangePayload payload) callback,
   }) {
+    assert(
+      filter == null || filters == null,
+      'Provide either `filter` or `filters`, not both.',
+    );
+
+    final allFilters = [
+      ?filter,
+      ...?filters,
+    ];
+    final filterString = allFilters.isEmpty ? null : allFilters.join(',');
+
     return onEvents(
       'postgres_changes',
       ChannelFilter(
         event: event.toRealtimeEvent(),
         schema: schema,
         table: table,
-        filter: filter?.toString(),
+        filter: filterString,
+        select: select,
       ),
       (payload, [ref]) => callback(PostgresChangePayload.fromPayload(payload)),
     );
@@ -517,7 +543,38 @@ class RealtimeChannel {
     return result;
   }
 
-  /// Sets up a listener for realtime system events for debugging purposes.
+  /// Sets up a listener for realtime `system` events.
+  ///
+  /// The [callback] receives the raw payload (typically a `Map`). To work with
+  /// it as a typed value, parse it with [RealtimeSystemPayload.fromJson].
+  ///
+  /// Opt in to the replication-ready notification with
+  /// [RealtimeChannelConfig.replicationReady] when creating the channel, then
+  /// watch for `status == 'ok'` to know the Postgres replication connection is
+  /// ready.
+  ///
+  /// ```dart
+  /// final channel = supabase.channel(
+  ///   'room1',
+  ///   opts: const RealtimeChannelConfig(replicationReady: true),
+  /// );
+  /// channel
+  ///     .onPostgresChanges(
+  ///       event: PostgresChangeEvent.all,
+  ///       schema: 'public',
+  ///       table: 'messages',
+  ///       callback: (payload) => print('Change received! $payload'),
+  ///     )
+  ///     .onSystemEvents((payload) {
+  ///       final system = RealtimeSystemPayload.fromJson(
+  ///         Map<String, dynamic>.from(payload as Map),
+  ///       );
+  ///       if (system.extension == 'system' && system.status == 'ok') {
+  ///         print('Replication connection is ready: ${system.message}');
+  ///       }
+  ///     })
+  ///     .subscribe();
+  /// ```
   RealtimeChannel onSystemEvents(
     void Function(dynamic payload) callback,
   ) {
@@ -548,7 +605,7 @@ class RealtimeChannel {
   }
 
   @internal
-  RealtimeChannel off(String type, Map<String, String> filter) {
+  RealtimeChannel off(String type, Map<String, dynamic> filter) {
     final typeLower = type.toLowerCase();
 
     _bindings[typeLower] = _bindings[typeLower]!.where((bind) {
@@ -983,13 +1040,14 @@ class RealtimeChannel {
   @internal
   bool get isLeaving => _state == ChannelStates.leaving;
 
-  static bool _isEqual(Map<String, String> obj1, Map<String, String> obj2) {
+  static bool _isEqual(Map<String, Object?> obj1, Map<String, Object?> obj2) {
     if (obj1.keys.length != obj2.keys.length) {
       return false;
     }
 
+    const equality = DeepCollectionEquality();
     for (final k in obj1.keys) {
-      if (obj1[k] != obj2[k]) {
+      if (!equality.equals(obj1[k], obj2[k])) {
         return false;
       }
     }

packages/realtime_client/lib/src/types.dart

@@ -6,7 +6,7 @@ typedef BindingCallback = void Function(dynamic payload, [dynamic ref]);
 
 class Binding {
   String type;
-  Map<String, String> filter;
+  Map<String, dynamic> filter;
   BindingCallback callback;
   String? id;
 
@@ -19,7 +19,7 @@ class Binding {
 
   Binding copyWith({
     String? type,
-    Map<String, String>? filter,
+    Map<String, dynamic>? filter,
     BindingCallback? callback,
     String? id,
   }) {
@@ -79,24 +79,31 @@ class ChannelFilter {
   final String? schema;
   final String? table;
 
-  /// For [RealtimeListenTypes.postgresChanges] it's of the format `column=filter.value` with `filter` being one of `eq, neq, lt, lte, gt, gte, in`
+  /// For [RealtimeListenTypes.postgresChanges] it's of the format `column=filter.value` with `filter` being one of `eq, neq, lt, lte, gt, gte, in, like, ilike, is, match, imatch, isdistinct`.
   ///
-  /// Only one filter can be applied
+  /// Multiple conditions can be combined with commas; they are applied as an `AND`.
+  /// Any operator can be negated with the `not.` prefix.
   final String? filter;
 
+  /// For [RealtimeListenTypes.postgresChanges], restricts the change payload to
+  /// a subset of columns instead of the full row.
+  final List<String>? select;
+
   const ChannelFilter({
     this.event,
     this.schema,
     this.table,
     this.filter,
+    this.select,
   });
 
-  Map<String, String> toMap() {
+  Map<String, dynamic> toMap() {
     return {
       if (event != null) 'event': event!,
       if (schema != null) 'schema': schema!,
       if (table != null) 'table': table!,
       if (filter != null) 'filter': filter!,
+      if (select != null) 'select': select!,
     };
   }
 }
@@ -181,13 +188,24 @@ class RealtimeChannelConfig {
   /// Defines if the channel is private or not and if RLS policies will be used to check data
   final bool private;
 
+  /// [replicationReady] instructs the server to emit a `system` event once the
+  /// Postgres replication connection backing this channel is established and
+  /// ready to stream changes.
+  ///
+  /// Listen for it with [RealtimeChannel.onSystemEvents]; the payload's
+  /// [RealtimeSystemPayload.status] is `'ok'`
+  /// (message: `'Replication connection established'`) on success or `'error'`
+  /// if the connection is not ready in time.
+  final bool replicationReady;
+
   const RealtimeChannelConfig({
     this.ack = false,
     this.self = false,
     this.replay,
     this.key = '',
     this.enabled = false,
     this.private = false,
+    this.replicationReady = false,
   });
 
   Map<String, dynamic> toMap() {
@@ -198,6 +216,9 @@ class RealtimeChannelConfig {
     if (replay != null) {
       broadcastConfig['replay'] = replay!.toMap();
     }
+    if (replicationReady) {
+      broadcastConfig['replication_ready'] = true;
+    }
 
     return {
       'config': {
@@ -212,6 +233,48 @@ class RealtimeChannelConfig {
   }
 }
 
+/// Payload of a `system` event emitted by the server.
+///
+/// Most notably, when a channel is created with
+/// [RealtimeChannelConfig.replicationReady] set to `true`, the server sends one
+/// of these once the Postgres replication connection is ready
+/// ([status] is `'ok'`) or fails to become ready in time ([status] is
+/// `'error'`).
+class RealtimeSystemPayload {
+  /// The extension that produced the message, e.g. `'system'` or
+  /// `'postgres_changes'`.
+  final String extension;
+
+  /// `'ok'` on success, `'error'` on failure.
+  final String status;
+
+  /// Human-readable description, e.g. `'Replication connection established'`.
+  final String message;
+
+  /// The channel (sub)topic the message refers to.
+  final String channel;
+
+  const RealtimeSystemPayload({
+    required this.extension,
+    required this.status,
+    required this.message,
+    required this.channel,
+  });
+
+  factory RealtimeSystemPayload.fromJson(Map<String, dynamic> json) {
+    return RealtimeSystemPayload(
+      extension: json['extension']?.toString() ?? '',
+      status: json['status']?.toString() ?? '',
+      message: json['message']?.toString() ?? '',
+      channel: json['channel']?.toString() ?? '',
+    );
+  }
+
+  @override
+  String toString() =>
+      'RealtimeSystemPayload(extension: $extension, status: $status, message: $message, channel: $channel)';
+}
+
 /// Data class that contains the Postgres change event payload.
 class PostgresChangePayload {
   final String schema;
@@ -291,6 +354,10 @@ class PostgresChangePayload {
 }
 
 /// Specifies the type of filter to be applied on realtime Postgres Change listener.
+///
+/// These mirror the PostgREST operator surface that the Realtime server
+/// evaluates for Postgres Changes. Any operator can be negated with the `not.`
+/// prefix via [PostgresChangeFilter.negate].
 enum PostgresChangeFilterType {
   /// Listens to changes where a column's value in a table equals a client-specified value.
   eq,
@@ -312,6 +379,54 @@ enum PostgresChangeFilterType {
 
   /// Listen to changes when a column's value in a table equals any of the values specified.
   inFilter,
+
+  /// Listens to changes where a column matches a case-sensitive pattern (`LIKE`).
+  ///
+  /// Use `%` and `_` as wildcards, e.g. `title=like.%foo%`.
+  like,
+
+  /// Listens to changes where a column matches a case-insensitive pattern (`ILIKE`).
+  ilike,
+
+  /// Listens to changes where a column `IS` a given value (`null`, `true`,
+  /// `false` or `unknown`), e.g. `deleted_at=is.null`.
+  isFilter,
+
+  /// Listens to changes where a column matches a POSIX regular expression (`~`).
+  match,
+
+  /// Listens to changes where a column matches a case-insensitive POSIX regular
+  /// expression (`~*`).
+  imatch,
+
+  /// Listens to changes where a column is distinct from a value (NULL-safe
+  /// inequality, `IS DISTINCT FROM`).
+  isDistinct;
+
+  /// The operator token used in the filter wire format (the part between
+  /// `column=` and `.value`). Most match [name], but a few differ because the
+  /// enum names avoid Dart reserved words / casing conventions.
+  String get token {
+    switch (this) {
+      case PostgresChangeFilterType.inFilter:
+        return 'in';
+      case PostgresChangeFilterType.isFilter:
+        return 'is';
+      case PostgresChangeFilterType.isDistinct:
+        return 'isdistinct';
+      case PostgresChangeFilterType.eq:
+      case PostgresChangeFilterType.neq:
+      case PostgresChangeFilterType.lt:
+      case PostgresChangeFilterType.lte:
+      case PostgresChangeFilterType.gt:
+      case PostgresChangeFilterType.gte:
+      case PostgresChangeFilterType.like:
+      case PostgresChangeFilterType.ilike:
+      case PostgresChangeFilterType.match:
+      case PostgresChangeFilterType.imatch:
+        return name;
+    }
+  }
 }
 
 /// {@template postgres_change_filter}
@@ -327,22 +442,42 @@ class PostgresChangeFilter {
   /// The value to perform the filter on.
   final dynamic value;
 
+  /// When `true`, the operator is negated with the `not.` prefix
+  /// (e.g. `status=not.in.(draft,archived)`, `deleted_at=not.is.null`).
+  final bool negate;
+
   /// {@macro postgres_change_filter}
   const PostgresChangeFilter({
     required this.type,
     required this.column,
     required this.value,
+    this.negate = false,
   });
 
+  /// Quotes a scalar value PostgREST-style when it contains a reserved
+  /// character (`,`, `(`, `)`, `"`, `\`) or surrounding whitespace, so the
+  /// server's filter parser doesn't misread it as a condition/list boundary.
+  /// Values without reserved characters are sent verbatim.
+  static String _serializeScalar(Object? value) {
+    final serialized = value == null ? 'null' : '$value';
+    final needsQuoting =
+        RegExp(r'[,()"\\]').hasMatch(serialized) ||
+        serialized != serialized.trim();
+    if (!needsQuoting) return serialized;
+    final escaped = serialized.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
+    return '"$escaped"';
+  }
+
   @override
   String toString() {
+    final prefix = negate ? 'not.' : '';
     if (type == PostgresChangeFilterType.inFilter) {
-      return '$column=in.(${value.map((s) {
-        final escaped = '$s'.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
-        return '"$escaped"';
-      }).join(',')})';
+      final items = (value as Iterable)
+          .map((s) => _serializeScalar(s))
+          .join(',');
+      return '$column=${prefix}in.($items)';
     }
-    return '$column=${type.name}.$value';
+    return '$column=$prefix${type.token}.${_serializeScalar(value)}';
   }
 }