Web: pass on sync parameters.

Author: rkistner

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

@@ -148,11 +148,11 @@ class PowerSyncDatabaseImpl
     // duplicating work across tabs.
     try {
       sync = await SyncWorkerHandle.start(
-        this,
-        connector,
-        crudThrottleTime.inMilliseconds,
-        Uri.base.resolve('/powersync_sync.worker.js'),
-      );
+          database: this,
+          connector: connector,
+          crudThrottleTimeMs: crudThrottleTime.inMilliseconds,
+          workerUri: Uri.base.resolve('/powersync_sync.worker.js'),
+          syncParams: params);
     } catch (e) {
       logger.warning(
         'Could not use shared worker for synchronization, falling back to locks.',

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

@@ -10,23 +10,29 @@ import '../streaming_sync.dart';
 import 'sync_worker_protocol.dart';
 
 class SyncWorkerHandle implements StreamingSync {
-  final PowerSyncDatabaseImpl _database;
-  final PowerSyncBackendConnector _connector;
-  final int _crudThrottleTimeMs;
+  final PowerSyncDatabaseImpl database;
+  final PowerSyncBackendConnector connector;
+  final int crudThrottleTimeMs;
+  final Map<String, dynamic>? syncParams;
 
   late final WorkerCommunicationChannel _channel;
 
   final StreamController<SyncStatus> _status = StreamController.broadcast();
 
-  SyncWorkerHandle._(this._database, this._connector, this._crudThrottleTimeMs,
-      MessagePort sendToWorker, SharedWorker worker) {
+  SyncWorkerHandle._(
+      {required this.database,
+      required this.connector,
+      required this.crudThrottleTimeMs,
+      required MessagePort sendToWorker,
+      required SharedWorker worker,
+      this.syncParams}) {
     _channel = WorkerCommunicationChannel(
       port: sendToWorker,
       errors: EventStreamProviders.errorEvent.forTarget(worker),
       requestHandler: (type, payload) async {
         switch (type) {
           case SyncWorkerMessageType.requestEndpoint:
-            final endpoint = await (_database.database as WebSqliteConnection)
+            final endpoint = await (database.database as WebSqliteConnection)
                 .exposeEndpoint();
 
             return (
@@ -38,18 +44,18 @@ class SyncWorkerHandle implements StreamingSync {
               [endpoint.connectPort].toJS
             );
           case SyncWorkerMessageType.uploadCrud:
-            await _connector.uploadData(_database);
+            await connector.uploadData(database);
             return (JSObject(), null);
           case SyncWorkerMessageType.invalidCredentialsCallback:
-            final credentials = await _connector.fetchCredentials();
+            final credentials = await connector.fetchCredentials();
             return (
               credentials != null
                   ? SerializedCredentials.from(credentials)
                   : null,
               null
             );
           case SyncWorkerMessageType.credentialsCallback:
-            final credentials = await _connector.getCredentialsCached();
+            final credentials = await connector.getCredentialsCached();
             return (
               credentials != null
                   ? SerializedCredentials.from(credentials)
@@ -71,13 +77,19 @@ class SyncWorkerHandle implements StreamingSync {
   }
 
   static Future<SyncWorkerHandle> start(
-      PowerSyncDatabaseImpl database,
-      PowerSyncBackendConnector connector,
-      int crudThrottleTimeMs,
-      Uri workerUri) async {
+      {required PowerSyncDatabaseImpl database,
+      required PowerSyncBackendConnector connector,
+      required int crudThrottleTimeMs,
+      required Uri workerUri,
+      Map<String, dynamic>? syncParams}) async {
     final worker = SharedWorker(workerUri.toString().toJS);
     final handle = SyncWorkerHandle._(
-        database, connector, crudThrottleTimeMs, worker.port, worker);
+        database: database,
+        connector: connector,
+        crudThrottleTimeMs: crudThrottleTimeMs,
+        sendToWorker: worker.port,
+        worker: worker,
+        syncParams: syncParams);
 
     // Make sure that the worker is working, or throw immediately.
     await handle._channel.ping();
@@ -101,6 +113,6 @@ class SyncWorkerHandle implements StreamingSync {
   @override
   Future<void> streamingSync() async {
     await _channel.startSynchronization(
-        _database.openFactory.path, _crudThrottleTimeMs);
+        database.openFactory.path, crudThrottleTimeMs, syncParams);
   }
 }

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

@@ -4,6 +4,7 @@
 library;
 
 import 'dart:async';
+import 'dart:convert';
 import 'dart:js_interop';
 
 import 'package:async/async.dart';
@@ -41,10 +42,14 @@ class _SyncWorker {
     });
   }
 
-  _SyncRunner referenceSyncTask(String databaseIdentifier,
-      int crudThrottleTimeMs, _ConnectedClient client) {
+  _SyncRunner referenceSyncTask(
+      String databaseIdentifier,
+      int crudThrottleTimeMs,
+      String? syncParamsEncoded,
+      _ConnectedClient client) {
     return _requestedSyncTasks.putIfAbsent(databaseIdentifier, () {
-      return _SyncRunner(databaseIdentifier, crudThrottleTimeMs);
+      return _SyncRunner(databaseIdentifier, crudThrottleTimeMs,
+          syncParamsEncoded == null ? null : jsonDecode(syncParamsEncoded));
     })
       ..registerClient(client);
   }
@@ -64,8 +69,8 @@ class _ConnectedClient {
         switch (type) {
           case SyncWorkerMessageType.startSynchronization:
             final request = payload as StartSynchronization;
-            _runner = _worker.referenceSyncTask(
-                request.databaseName, request.crudThrottleTimeMs, this);
+            _runner = _worker.referenceSyncTask(request.databaseName,
+                request.crudThrottleTimeMs, request.syncParamsEncoded, this);
             return (JSObject(), null);
           case SyncWorkerMessageType.abortSynchronization:
             _runner?.unregisterClient(this);
@@ -106,6 +111,7 @@ class _ConnectedClient {
 class _SyncRunner {
   final String identifier;
   final int crudThrottleTimeMs;
+  final Map<String, dynamic>? syncParams;
 
   final StreamGroup<_RunnerEvent> _group = StreamGroup();
   final StreamController<_RunnerEvent> _mainEvents = StreamController();
@@ -114,7 +120,7 @@ class _SyncRunner {
   _ConnectedClient? databaseHost;
   final connections = <_ConnectedClient>[];
 
-  _SyncRunner(this.identifier, this.crudThrottleTimeMs) {
+  _SyncRunner(this.identifier, this.crudThrottleTimeMs, this.syncParams) {
     _group.add(_mainEvents.stream);
 
     Future(() async {
@@ -227,15 +233,15 @@ class _SyncRunner {
     }
 
     sync = StreamingSyncImplementation(
-      adapter: BucketStorage(database),
-      credentialsCallback: client.channel.credentialsCallback,
-      invalidCredentialsCallback: client.channel.invalidCredentialsCallback,
-      uploadCrud: client.channel.uploadCrud,
-      crudUpdateTriggerStream: crudStream,
-      retryDelay: Duration(seconds: 3),
-      client: FetchClient(mode: RequestMode.cors),
-      identifier: identifier,
-    );
+        adapter: BucketStorage(database),
+        credentialsCallback: client.channel.credentialsCallback,
+        invalidCredentialsCallback: client.channel.invalidCredentialsCallback,
+        uploadCrud: client.channel.uploadCrud,
+        crudUpdateTriggerStream: crudStream,
+        retryDelay: Duration(seconds: 3),
+        client: FetchClient(mode: RequestMode.cors),
+        identifier: identifier,
+        syncParameters: syncParams);
     sync!.statusStream.listen((event) {
       _logger.fine('Broadcasting sync event: $event');
       for (final client in connections) {

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

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 import 'dart:js_interop';
 
 import 'package:web/web.dart';
@@ -60,15 +61,16 @@ extension type SyncWorkerMessage._(JSObject _) implements JSObject {
 
 @anonymous
 extension type StartSynchronization._(JSObject _) implements JSObject {
-  external factory StartSynchronization({
-    required String databaseName,
-    required int crudThrottleTimeMs,
-    required int requestId,
-  });
+  external factory StartSynchronization(
+      {required String databaseName,
+      required int crudThrottleTimeMs,
+      required int requestId,
+      String? syncParamsEncoded});
 
   external String get databaseName;
   external int get requestId;
   external int get crudThrottleTimeMs;
+  external String? get syncParamsEncoded;
 }
 
 @anonymous
@@ -315,15 +317,17 @@ final class WorkerCommunicationChannel {
     await _numericRequest(SyncWorkerMessageType.ping);
   }
 
-  Future<void> startSynchronization(
-      String databaseName, int crudThrottleTimeMs) async {
+  Future<void> startSynchronization(String databaseName, int crudThrottleTimeMs,
+      Map<String, dynamic>? syncParams) async {
     final (id, completion) = _newRequest();
     port.postMessage(SyncWorkerMessage(
       type: SyncWorkerMessageType.startSynchronization.name,
       payload: StartSynchronization(
           databaseName: databaseName,
           crudThrottleTimeMs: crudThrottleTimeMs,
-          requestId: id),
+          requestId: id,
+          syncParamsEncoded:
+              syncParams == null ? null : jsonEncode(syncParams)),
     ));
     await completion;
   }