Track subscriptions across requests

Author: simolus3

packages/powersync_core/lib/src/database/web/web_powersync_database.dart

@@ -143,6 +143,7 @@ class PowerSyncDatabaseImpl
         connector: connector,
         options: options.source,
         workerUri: Uri.base.resolve('/powersync_sync.worker.js'),
+        subscriptions: initiallyActiveStreams,
       );
     } catch (e) {
       logger.warning(
@@ -159,6 +160,7 @@ class PowerSyncDatabaseImpl
         crudUpdateTriggerStream: crudStream,
         options: options,
         client: BrowserClient(),
+        activeSubscriptions: initiallyActiveStreams,
         // Only allows 1 sync implementation to run at a time per database
         // This should be global (across tabs) when using Navigator locks.
         identifier: database.openFactory.path,
@@ -170,7 +172,10 @@ class PowerSyncDatabaseImpl
     });
     sync.streamingSync();
 
+    final subscriptions = activeStreams.listen(sync.updateSubscriptions);
+
     abort.onAbort.then((_) async {
+      subscriptions.cancel();
       await sync.abort();
       abort.completeAbort();
     }).ignore();

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -30,6 +30,8 @@ abstract interface class StreamingSync {
 
   /// Close any active streams.
   Future<void> abort();
+
+  void updateSubscriptions(List<SubscribedStream> streams);
 }
 
 @internal
@@ -129,6 +131,7 @@ class StreamingSyncImplementation implements StreamingSync {
     return _abort?.aborted ?? false;
   }
 
+  @override
   void updateSubscriptions(List<SubscribedStream> streams) {
     _activeSubscriptions = streams;
     if (_nonLineSyncEvents.hasListener) {

packages/powersync_core/lib/src/web/sync_controller.dart

@@ -15,6 +15,7 @@ class SyncWorkerHandle implements StreamingSync {
   final PowerSyncBackendConnector connector;
   final SyncOptions options;
   late final WorkerCommunicationChannel _channel;
+  List<SubscribedStream> subscriptions;
 
   final StreamController<SyncStatus> _status = StreamController.broadcast();
 
@@ -24,6 +25,7 @@ class SyncWorkerHandle implements StreamingSync {
     required this.options,
     required MessagePort sendToWorker,
     required SharedWorker worker,
+    required this.subscriptions,
   }) {
     _channel = WorkerCommunicationChannel(
       port: sendToWorker,
@@ -81,6 +83,7 @@ class SyncWorkerHandle implements StreamingSync {
     required PowerSyncBackendConnector connector,
     required Uri workerUri,
     required SyncOptions options,
+    required List<SubscribedStream> subscriptions,
   }) async {
     final worker = SharedWorker(workerUri.toString().toJS);
     final handle = SyncWorkerHandle._(
@@ -89,6 +92,7 @@ class SyncWorkerHandle implements StreamingSync {
       connector: connector,
       sendToWorker: worker.port,
       worker: worker,
+      subscriptions: subscriptions,
     );
 
     // Make sure that the worker is working, or throw immediately.
@@ -116,6 +120,13 @@ class SyncWorkerHandle implements StreamingSync {
       database.database.openFactory.path,
       ResolvedSyncOptions(options),
       database.schema,
+      subscriptions,
     );
   }
+
+  @override
+  void updateSubscriptions(List<SubscribedStream> streams) {
+    subscriptions = streams;
+    _channel.updateSubscriptions(streams);
+  }
 }

packages/powersync_core/lib/src/web/sync_worker.dart

@@ -8,6 +8,7 @@ import 'dart:convert';
 import 'dart:js_interop';
 
 import 'package:async/async.dart';
+import 'package:collection/collection.dart';
 import 'package:http/browser_client.dart';
 import 'package:logging/logging.dart';
 import 'package:powersync_core/powersync_core.dart';
@@ -45,15 +46,20 @@ class _SyncWorker {
     });
   }
 
-  _SyncRunner referenceSyncTask(String databaseIdentifier, SyncOptions options,
-      String schemaJson, _ConnectedClient client) {
+  _SyncRunner referenceSyncTask(
+      String databaseIdentifier,
+      SyncOptions options,
+      String schemaJson,
+      List<SubscribedStream> subscriptions,
+      _ConnectedClient client) {
     return _requestedSyncTasks.putIfAbsent(databaseIdentifier, () {
       return _SyncRunner(databaseIdentifier);
     })
       ..registerClient(
         client,
         options,
         schemaJson,
+        subscriptions,
       );
   }
 }
@@ -90,13 +96,20 @@ class _ConnectedClient {
               },
             );
 
-            _runner = _worker.referenceSyncTask(request.databaseName,
-                recoveredOptions, request.schemaJson, this);
+            _runner = _worker.referenceSyncTask(
+              request.databaseName,
+              recoveredOptions,
+              request.schemaJson,
+              request.subscriptions?.toDart ?? const [],
+              this,
+            );
             return (JSObject(), null);
           case SyncWorkerMessageType.abortSynchronization:
             _runner?.disconnectClient(this);
             _runner = null;
             return (JSObject(), null);
+          case SyncWorkerMessageType.updateSubscriptions:
+            return (JSObject(), null);
           default:
             throw StateError('Unexpected message type $type');
         }
@@ -137,9 +150,10 @@ class _SyncRunner {
   final StreamGroup<_RunnerEvent> _group = StreamGroup();
   final StreamController<_RunnerEvent> _mainEvents = StreamController();
 
-  StreamingSync? sync;
+  StreamingSyncImplementation? sync;
   _ConnectedClient? databaseHost;
-  final connections = <_ConnectedClient>[];
+  final connections = <_ConnectedClient, List<SubscribedStream>>{};
+  List<SubscribedStream> currentStreams = [];
 
   _SyncRunner(this.identifier) {
     _group.add(_mainEvents.stream);
@@ -152,8 +166,9 @@ class _SyncRunner {
                 :final client,
                 :final options,
                 :final schemaJson,
+                :final subscriptions,
               ):
-              connections.add(client);
+              connections[client] = subscriptions;
               final (newOptions, reconnect) = this.options.applyFrom(options);
               this.options = newOptions;
               this.schemaJson = schemaJson;
@@ -165,6 +180,8 @@ class _SyncRunner {
                 sync?.abort();
                 sync = null;
                 await _requestDatabase(client);
+              } else {
+                reindexSubscriptions();
               }
             case _RemoveConnection(:final client):
               connections.remove(client);
@@ -191,6 +208,12 @@ class _SyncRunner {
               } else {
                 await _requestDatabase(newHost);
               }
+            case _ClientSubscriptionsChanged(
+                :final client,
+                :final subscriptions
+              ):
+              connections[client] = subscriptions;
+              reindexSubscriptions();
           }
         } catch (e, s) {
           _logger.warning('Error handling $event', e, s);
@@ -199,12 +222,22 @@ class _SyncRunner {
     });
   }
 
+  /// Updates [currentStreams] to the union of values in [connections].
+  void reindexSubscriptions() {
+    final before = currentStreams.toSet();
+    final after = connections.values.flattenedToSet;
+    if (!const SetEquality<SubscribedStream>().equals(before, after)) {
+      currentStreams = after.toList();
+      sync?.updateSubscriptions(currentStreams);
+    }
+  }
+
   /// Pings all current [connections], removing those that don't answer in 5s
   /// (as they are likely closed tabs as well).
   ///
   /// Returns the first client that responds (without waiting for others).
   Future<_ConnectedClient?> _collectActiveClients() async {
-    final candidates = connections.toList();
+    final candidates = connections.keys.toList();
     if (candidates.isEmpty) {
       return null;
     }
@@ -269,6 +302,7 @@ class _SyncRunner {
       );
     }
 
+    currentStreams = connections.values.flattenedToSet.toList();
     sync = StreamingSyncImplementation(
       adapter: WebBucketStorage(database),
       schemaJson: client._runner!.schemaJson,
@@ -283,20 +317,21 @@ class _SyncRunner {
       options: options,
       client: BrowserClient(),
       identifier: identifier,
+      activeSubscriptions: currentStreams,
     );
     sync!.statusStream.listen((event) {
       _logger.fine('Broadcasting sync event: $event');
-      for (final client in connections) {
+      for (final client in connections.keys) {
         client.channel.notify(SyncWorkerMessageType.notifySyncStatus,
             SerializedSyncStatus.from(event));
       }
     });
     sync!.streamingSync();
   }
 
-  void registerClient(
-      _ConnectedClient client, SyncOptions options, String schemaJson) {
-    _mainEvents.add(_AddConnection(client, options, schemaJson));
+  void registerClient(_ConnectedClient client, SyncOptions options,
+      String schemaJson, List<SubscribedStream> subscriptions) {
+    _mainEvents.add(_AddConnection(client, options, schemaJson, subscriptions));
   }
 
   /// Remove a client, disconnecting if no clients remain..
@@ -308,6 +343,11 @@ class _SyncRunner {
   void disconnectClient(_ConnectedClient client) {
     _mainEvents.add(_DisconnectClient(client));
   }
+
+  void updateClientSubscriptions(
+      _ConnectedClient client, List<SubscribedStream> subscriptions) {
+    _mainEvents.add(_ClientSubscriptionsChanged(client, subscriptions));
+  }
 }
 
 sealed class _RunnerEvent {}
@@ -316,8 +356,10 @@ final class _AddConnection implements _RunnerEvent {
   final _ConnectedClient client;
   final SyncOptions options;
   final String schemaJson;
+  final List<SubscribedStream> subscriptions;
 
-  _AddConnection(this.client, this.options, this.schemaJson);
+  _AddConnection(
+      this.client, this.options, this.schemaJson, this.subscriptions);
 }
 
 final class _RemoveConnection implements _RunnerEvent {
@@ -332,6 +374,13 @@ final class _DisconnectClient implements _RunnerEvent {
   _DisconnectClient(this.client);
 }
 
+final class _ClientSubscriptionsChanged implements _RunnerEvent {
+  final _ConnectedClient client;
+  final List<SubscribedStream> subscriptions;
+
+  _ClientSubscriptionsChanged(this.client, this.subscriptions);
+}
+
 final class _ActiveDatabaseClosed implements _RunnerEvent {
   const _ActiveDatabaseClosed();
 }

packages/powersync_core/lib/src/web/sync_worker_protocol.dart

@@ -9,6 +9,7 @@ import 'package:web/web.dart';
 
 import '../connector.dart';
 import '../log.dart';
+import '../sync/streaming_sync.dart';
 import '../sync/sync_status.dart';
 
 /// Names used in [SyncWorkerMessage]
@@ -20,6 +21,9 @@ enum SyncWorkerMessageType {
   /// If parameters change, the sync worker reconnects.
   startSynchronization,
 
+  /// Update the active subscriptions that this client is interested in.
+  updateSubscriptions,
+
   /// The [SyncWorkerMessage.payload] for the request is a numeric id, the
   /// response can be anything (void).
   /// This disconnects immediately, even if other clients are still open.
@@ -74,6 +78,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
     required String implementationName,
     required String schemaJson,
     String? syncParamsEncoded,
+    UpdateSubscriptions? subscriptions,
   });
 
   external String get databaseName;
@@ -83,6 +88,36 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
   external String? get implementationName;
   external String get schemaJson;
   external String? get syncParamsEncoded;
+  external UpdateSubscriptions? get subscriptions;
+}
+
+@anonymous
+extension type UpdateSubscriptions.__(JSObject _inner) implements JSObject {
+  external factory UpdateSubscriptions._({
+    required int requestId,
+    required JSArray content,
+  });
+
+  factory UpdateSubscriptions(int requestId, List<SubscribedStream> streams) {
+    return UpdateSubscriptions._(
+      requestId: requestId,
+      content: streams
+          .map((e) => <JSString>[e.name.toJS, e.parameters.toJS].toJS)
+          .toList()
+          .toJS,
+    );
+  }
+
+  external int get requestId;
+  external JSArray get content;
+
+  List<SubscribedStream> get toDart {
+    return content.toDart.map((e) {
+      final [name, parameters] = (e as JSArray<JSString>).toDart;
+
+      return (name: name.toDart, parameters: parameters.toDart);
+    }).toList();
+  }
 }
 
 @anonymous
@@ -339,6 +374,8 @@ final class WorkerCommunicationChannel {
           return;
         case SyncWorkerMessageType.startSynchronization:
           requestId = (message.payload as StartSynchronization).requestId;
+        case SyncWorkerMessageType.updateSubscriptions:
+          requestId = (message.payload as UpdateSubscriptions).requestId;
         case SyncWorkerMessageType.requestEndpoint:
         case SyncWorkerMessageType.abortSynchronization:
         case SyncWorkerMessageType.credentialsCallback:
@@ -413,7 +450,11 @@ final class WorkerCommunicationChannel {
   }
 
   Future<void> startSynchronization(
-      String databaseName, ResolvedSyncOptions options, Schema schema) async {
+    String databaseName,
+    ResolvedSyncOptions options,
+    Schema schema,
+    List<SubscribedStream> streams,
+  ) async {
     final (id, completion) = _newRequest();
     port.postMessage(SyncWorkerMessage(
       type: SyncWorkerMessageType.startSynchronization.name,
@@ -428,11 +469,22 @@ final class WorkerCommunicationChannel {
           null => null,
           final params => jsonEncode(params),
         },
+        subscriptions: UpdateSubscriptions(-1, streams),
       ),
     ));
     await completion;
   }
 
+  Future<void> updateSubscriptions(List<SubscribedStream> streams) async {
+    final (id, completion) = _newRequest();
+    port.postMessage(SyncWorkerMessage(
+      type: SyncWorkerMessageType.updateSubscriptions.name,
+      payload: UpdateSubscriptions(id, streams),
+    ));
+
+    await completion;
+  }
+
   Future<void> abortSynchronization() async {
     await _numericRequest(SyncWorkerMessageType.abortSynchronization);
   }