Skip to content

Commit e9c162f

Browse files
filipecabacofilipecabaco
committed
fix: support new realtime features
* support new system message to information replication connection is ready * support new filters * support select
1 parent ae930ea commit e9c162f

6 files changed

Lines changed: 409 additions & 20 deletions

File tree

‎packages/realtime_client/lib/src/realtime_channel.dart‎

Lines changed: 63 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,7 @@ import 'dart:async';
22
import 'dart:convert';
33
import 'dart:typed_data';
44

5+
import 'package:collection/collection.dart';
56
import 'package:http/http.dart';
67
import 'package:meta/meta.dart';
78
import 'package:realtime_client/realtime_client.dart';
@@ -259,6 +260,10 @@ class RealtimeChannel {
259260
final filter = clientPostgresBinding.filter['filter'];
260261
final serverPostgresFilter = serverPostgresFilters[i];
261262

263+
// NOTE: `select` is intentionally not part of this equality check (mirroring
264+
// supabase-js), so a server that echoes `select` back in a slightly
265+
// different shape does not force a spurious unsubscribe. The client
266+
// binding keeps its own `select` regardless.
262267
if (serverPostgresFilter != null &&
263268
serverPostgresFilter['event'] == event &&
264269
serverPostgresFilter['schema'] == schema &&
@@ -342,6 +347,13 @@ class RealtimeChannel {
342347
///
343348
/// [filter] can be used to further control which rows to listen to within the given [schema] and [table].
344349
///
350+
/// [filters] combines multiple [PostgresChangeFilter]s with an `AND`. Provide
351+
/// either [filter] or [filters], not both.
352+
///
353+
/// [select] restricts the change payload to a subset of columns instead of the
354+
/// full row (reducing payload size). The listed columns must be selectable by
355+
/// the subscribing role.
356+
///
345357
/// ```dart
346358
/// supabase.channel('my_channel').onPostgresChanges(
347359
/// event: PostgresChangeEvent.all,
@@ -361,15 +373,29 @@ class RealtimeChannel {
361373
String? schema,
362374
String? table,
363375
PostgresChangeFilter? filter,
376+
List<PostgresChangeFilter>? filters,
377+
List<String>? select,
364378
required void Function(PostgresChangePayload payload) callback,
365379
}) {
380+
assert(
381+
filter == null || filters == null,
382+
'Provide either `filter` or `filters`, not both.',
383+
);
384+
385+
final allFilters = [
386+
if (filter != null) filter,
387+
...?filters,
388+
];
389+
final filterString = allFilters.isEmpty ? null : allFilters.join(',');
390+
366391
return onEvents(
367392
'postgres_changes',
368393
ChannelFilter(
369394
event: event.toRealtimeEvent(),
370395
schema: schema,
371396
table: table,
372-
filter: filter?.toString(),
397+
filter: filterString,
398+
select: select,
373399
),
374400
(payload, [ref]) => callback(PostgresChangePayload.fromPayload(payload)),
375401
);
@@ -481,7 +507,38 @@ class RealtimeChannel {
481507
return result;
482508
}
483509

484-
/// Sets up a listener for realtime system events for debugging purposes.
510+
/// Sets up a listener for realtime `system` events.
511+
///
512+
/// The [callback] receives the raw payload (typically a `Map`). To work with
513+
/// it as a typed value, parse it with [RealtimeSystemPayload.fromJson].
514+
///
515+
/// Opt in to the replication-ready notification with
516+
/// [RealtimeChannelConfig.replicationReady] when creating the channel, then
517+
/// watch for `status == 'ok'` to know the Postgres replication connection is
518+
/// ready.
519+
///
520+
/// ```dart
521+
/// final channel = supabase.channel(
522+
/// 'room1',
523+
/// opts: const RealtimeChannelConfig(replicationReady: true),
524+
/// );
525+
/// channel
526+
/// .onPostgresChanges(
527+
/// event: PostgresChangeEvent.all,
528+
/// schema: 'public',
529+
/// table: 'messages',
530+
/// callback: (payload) => print('Change received! $payload'),
531+
/// )
532+
/// .onSystemEvents((payload) {
533+
/// final system = RealtimeSystemPayload.fromJson(
534+
/// Map<String, dynamic>.from(payload as Map),
535+
/// );
536+
/// if (system.extension == 'system' && system.status == 'ok') {
537+
/// print('Replication connection is ready: ${system.message}');
538+
/// }
539+
/// })
540+
/// .subscribe();
541+
/// ```
485542
RealtimeChannel onSystemEvents(
486543
void Function(dynamic payload) callback,
487544
) {
@@ -509,7 +566,7 @@ class RealtimeChannel {
509566
}
510567

511568
@internal
512-
RealtimeChannel off(String type, Map<String, String> filter) {
569+
RealtimeChannel off(String type, Map<String, dynamic> filter) {
513570
final typeLower = type.toLowerCase();
514571

515572
_bindings[typeLower] = _bindings[typeLower]!.where((bind) {
@@ -937,13 +994,14 @@ class RealtimeChannel {
937994
@internal
938995
bool get isLeaving => _state == ChannelStates.leaving;
939996

940-
static bool _isEqual(Map<String, String> obj1, Map<String, String> obj2) {
997+
static bool _isEqual(Map<String, dynamic> obj1, Map<String, dynamic> obj2) {
941998
if (obj1.keys.length != obj2.keys.length) {
942999
return false;
9431000
}
9441001

1002+
const equality = DeepCollectionEquality();
9451003
for (final k in obj1.keys) {
946-
if (obj1[k] != obj2[k]) {
1004+
if (!equality.equals(obj1[k], obj2[k])) {
9471005
return false;
9481006
}
9491007
}

‎packages/realtime_client/lib/src/types.dart‎

Lines changed: 144 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -6,7 +6,7 @@ typedef BindingCallback = void Function(dynamic payload, [dynamic ref]);
66

77
class Binding {
88
String type;
9-
Map<String, String> filter;
9+
Map<String, dynamic> filter;
1010
BindingCallback callback;
1111
String? id;
1212

@@ -19,7 +19,7 @@ class Binding {
1919

2020
Binding copyWith({
2121
String? type,
22-
Map<String, String>? filter,
22+
Map<String, dynamic>? filter,
2323
BindingCallback? callback,
2424
String? id,
2525
}) {
@@ -78,24 +78,31 @@ class ChannelFilter {
7878
final String? schema;
7979
final String? table;
8080

81-
/// For [RealtimeListenTypes.postgresChanges] it's of the format `column=filter.value` with `filter` being one of `eq, neq, lt, lte, gt, gte, in`
81+
/// 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`.
8282
///
83-
/// Only one filter can be applied
83+
/// Multiple conditions can be combined with commas; they are applied as an `AND`.
84+
/// Any operator can be negated with the `not.` prefix.
8485
final String? filter;
8586

87+
/// For [RealtimeListenTypes.postgresChanges], restricts the change payload to
88+
/// a subset of columns instead of the full row.
89+
final List<String>? select;
90+
8691
const ChannelFilter({
8792
this.event,
8893
this.schema,
8994
this.table,
9095
this.filter,
96+
this.select,
9197
});
9298

93-
Map<String, String> toMap() {
99+
Map<String, dynamic> toMap() {
94100
return {
95101
if (event != null) 'event': event!,
96102
if (schema != null) 'schema': schema!,
97103
if (table != null) 'table': table!,
98104
if (filter != null) 'filter': filter!,
105+
if (select != null) 'select': select!,
99106
};
100107
}
101108
}
@@ -178,13 +185,24 @@ class RealtimeChannelConfig {
178185
/// Defines if the channel is private or not and if RLS policies will be used to check data
179186
final bool private;
180187

188+
/// [replicationReady] instructs the server to emit a `system` event once the
189+
/// Postgres replication connection backing this channel is established and
190+
/// ready to stream changes.
191+
///
192+
/// Listen for it with [RealtimeChannel.onSystemEvents]; the payload's
193+
/// [RealtimeSystemPayload.status] is `'ok'`
194+
/// (message: `'Replication connection established'`) on success or `'error'`
195+
/// if the connection is not ready in time.
196+
final bool replicationReady;
197+
181198
const RealtimeChannelConfig({
182199
this.ack = false,
183200
this.self = false,
184201
this.replay,
185202
this.key = '',
186203
this.enabled = false,
187204
this.private = false,
205+
this.replicationReady = false,
188206
});
189207

190208
Map<String, dynamic> toMap() {
@@ -195,6 +213,9 @@ class RealtimeChannelConfig {
195213
if (replay != null) {
196214
broadcastConfig['replay'] = replay!.toMap();
197215
}
216+
if (replicationReady) {
217+
broadcastConfig['replication_ready'] = true;
218+
}
198219

199220
return {
200221
'config': {
@@ -209,6 +230,48 @@ class RealtimeChannelConfig {
209230
}
210231
}
211232

233+
/// Payload of a `system` event emitted by the server.
234+
///
235+
/// Most notably, when a channel is created with
236+
/// [RealtimeChannelConfig.replicationReady] set to `true`, the server sends one
237+
/// of these once the Postgres replication connection is ready
238+
/// ([status] is `'ok'`) or fails to become ready in time ([status] is
239+
/// `'error'`).
240+
class RealtimeSystemPayload {
241+
/// The extension that produced the message, e.g. `'system'` or
242+
/// `'postgres_changes'`.
243+
final String extension;
244+
245+
/// `'ok'` on success, `'error'` on failure.
246+
final String status;
247+
248+
/// Human-readable description, e.g. `'Replication connection established'`.
249+
final String message;
250+
251+
/// The channel (sub)topic the message refers to.
252+
final String channel;
253+
254+
const RealtimeSystemPayload({
255+
required this.extension,
256+
required this.status,
257+
required this.message,
258+
required this.channel,
259+
});
260+
261+
factory RealtimeSystemPayload.fromJson(Map<String, dynamic> json) {
262+
return RealtimeSystemPayload(
263+
extension: json['extension']?.toString() ?? '',
264+
status: json['status']?.toString() ?? '',
265+
message: json['message']?.toString() ?? '',
266+
channel: json['channel']?.toString() ?? '',
267+
);
268+
}
269+
270+
@override
271+
String toString() =>
272+
'RealtimeSystemPayload(extension: $extension, status: $status, message: $message, channel: $channel)';
273+
}
274+
212275
/// Data class that contains the Postgres change event payload.
213276
class PostgresChangePayload {
214277
final String schema;
@@ -287,6 +350,10 @@ class PostgresChangePayload {
287350
}
288351

289352
/// Specifies the type of filter to be applied on realtime Postgres Change listener.
353+
///
354+
/// These mirror the PostgREST operator surface that the Realtime server
355+
/// evaluates for Postgres Changes. Any operator can be negated with the `not.`
356+
/// prefix via [PostgresChangeFilter.negate].
290357
enum PostgresChangeFilterType {
291358
/// Listens to changes where a column's value in a table equals a client-specified value.
292359
eq,
@@ -307,7 +374,55 @@ enum PostgresChangeFilterType {
307374
gte,
308375

309376
/// Listen to changes when a column's value in a table equals any of the values specified.
310-
inFilter;
377+
inFilter,
378+
379+
/// Listens to changes where a column matches a case-sensitive pattern (`LIKE`).
380+
///
381+
/// Use `%` and `_` as wildcards, e.g. `title=like.%foo%`.
382+
like,
383+
384+
/// Listens to changes where a column matches a case-insensitive pattern (`ILIKE`).
385+
ilike,
386+
387+
/// Listens to changes where a column `IS` a given value (`null`, `true`,
388+
/// `false` or `unknown`), e.g. `deleted_at=is.null`.
389+
isFilter,
390+
391+
/// Listens to changes where a column matches a POSIX regular expression (`~`).
392+
match,
393+
394+
/// Listens to changes where a column matches a case-insensitive POSIX regular
395+
/// expression (`~*`).
396+
imatch,
397+
398+
/// Listens to changes where a column is distinct from a value (NULL-safe
399+
/// inequality, `IS DISTINCT FROM`).
400+
isDistinct;
401+
402+
/// The operator token used in the filter wire format (the part between
403+
/// `column=` and `.value`). Most match [name], but a few differ because the
404+
/// enum names avoid Dart reserved words / casing conventions.
405+
String get token {
406+
switch (this) {
407+
case PostgresChangeFilterType.inFilter:
408+
return 'in';
409+
case PostgresChangeFilterType.isFilter:
410+
return 'is';
411+
case PostgresChangeFilterType.isDistinct:
412+
return 'isdistinct';
413+
case PostgresChangeFilterType.eq:
414+
case PostgresChangeFilterType.neq:
415+
case PostgresChangeFilterType.lt:
416+
case PostgresChangeFilterType.lte:
417+
case PostgresChangeFilterType.gt:
418+
case PostgresChangeFilterType.gte:
419+
case PostgresChangeFilterType.like:
420+
case PostgresChangeFilterType.ilike:
421+
case PostgresChangeFilterType.match:
422+
case PostgresChangeFilterType.imatch:
423+
return name;
424+
}
425+
}
311426
}
312427

313428
/// {@template postgres_change_filter}
@@ -323,22 +438,40 @@ class PostgresChangeFilter {
323438
/// The value to perform the filter on.
324439
final dynamic value;
325440

441+
/// When `true`, the operator is negated with the `not.` prefix
442+
/// (e.g. `status=not.in.(draft,archived)`, `deleted_at=not.is.null`).
443+
final bool negate;
444+
326445
/// {@macro postgres_change_filter}
327446
const PostgresChangeFilter({
328447
required this.type,
329448
required this.column,
330449
required this.value,
450+
this.negate = false,
331451
});
332452

453+
/// Quotes a scalar value PostgREST-style when it contains a reserved
454+
/// character (`,`, `(`, `)`, `"`, `\`) or surrounding whitespace, so the
455+
/// server's filter parser doesn't misread it as a condition/list boundary.
456+
/// Values without reserved characters are sent verbatim.
457+
static String _serializeScalar(dynamic value) {
458+
final serialized = value == null ? 'null' : '$value';
459+
final needsQuoting = RegExp(r'[,()"\\]').hasMatch(serialized) ||
460+
serialized != serialized.trim();
461+
if (!needsQuoting) return serialized;
462+
final escaped = serialized.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
463+
return '"$escaped"';
464+
}
465+
333466
@override
334467
String toString() {
468+
final prefix = negate ? 'not.' : '';
335469
if (type == PostgresChangeFilterType.inFilter) {
336-
return '$column=in.(${value.map((s) {
337-
final escaped = '$s'.replaceAll(r'\', r'\\').replaceAll('"', r'\"');
338-
return '"$escaped"';
339-
}).join(',')})';
470+
final items =
471+
(value as Iterable).map((s) => _serializeScalar(s)).join(',');
472+
return '$column=${prefix}in.($items)';
340473
}
341-
return '$column=${type.name}.$value';
474+
return '$column=$prefix${type.token}.${_serializeScalar(value)}';
342475
}
343476
}
344477

0 commit comments

Comments
 (0)