Add shared sync worker

Author: simolus3

packages/powersync/lib/src/database/powersync_db_mixin.dart

@@ -71,11 +71,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   @protected
   Future<void> baseInit() async {
     statusStream = statusStreamController.stream;
-    updates = database.updates
-        .map((update) =>
-            PowerSyncUpdateNotification.fromUpdateNotification(update))
-        .where((update) => update.isNotEmpty)
-        .cast<UpdateNotification>();
+    updates = powerSyncUpdateNotifications(database.updates);
 
     await database.initialize();
     await database.execute('SELECT powersync_init()');
@@ -413,3 +409,12 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     return database.getAutoCommit();
   }
 }
+
+Stream<UpdateNotification> powerSyncUpdateNotifications(
+    Stream<UpdateNotification> inner) {
+  return inner
+      .map((update) =>
+          PowerSyncUpdateNotification.fromUpdateNotification(update))
+      .where((update) => update.isNotEmpty)
+      .cast<UpdateNotification>();
+}

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

@@ -15,6 +15,8 @@ import 'package:powersync/src/streaming_sync.dart';
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:powersync/src/schema_logic.dart' as schema_logic;
 
+import '../../web/sync_controller.dart';
+
 /// A PowerSync managed database.
 ///
 /// Web implementation for [PowerSyncDatabase]
@@ -93,7 +95,8 @@ class PowerSyncDatabaseImpl
       Logger? logger}) {
     final db = SqliteDatabase.withFactory(openFactory, maxReaders: 1);
     return PowerSyncDatabaseImpl.withDatabase(
-        schema: schema, logger: logger, database: db);
+        schema: schema, logger: logger, database: db)
+      ..openFactory = openFactory;
   }
 
   /// Open a PowerSyncDatabase on an existing [SqliteDatabase].
@@ -119,14 +122,15 @@ class PowerSyncDatabaseImpl
   /// The connection is automatically re-opened if it fails for any reason.
   ///
   /// Status changes are reported on [statusStream].
-  baseConnect(
-      {required PowerSyncBackendConnector connector,
-
-      /// Throttle time between CRUD operations
-      /// Defaults to 10 milliseconds.
-      required Duration crudThrottleTime,
-      required Future<void> Function() reconnect,
-      Map<String, dynamic>? params}) async {
+  baseConnect({
+    required PowerSyncBackendConnector connector,
+
+    /// Throttle time between CRUD operations
+    /// Defaults to 10 milliseconds.
+    required Duration crudThrottleTime,
+    required Future<void> Function() reconnect,
+    Map<String, dynamic>? params,
+  }) async {
     await initialize();
 
     // Disconnect if connected
@@ -135,9 +139,23 @@ class PowerSyncDatabaseImpl
 
     await isInitialized;
 
-    // TODO better multitab support
     final storage = BucketStorage(database);
-    final sync = StreamingSyncImplementation(
+    StreamingSync sync;
+    // Try using a shared worker for the synchronization implementation to avoid
+    // duplicate work across tabs.
+    try {
+      sync = await SyncWorkerHandle.start(
+        this,
+        connector,
+        Uri.base.resolve('/powersync_sync.worker.js'),
+      );
+    } catch (e) {
+      logger.warning(
+        'Could not use shared worker for synchronization, falling back to locks.',
+        e,
+      );
+
+      sync = StreamingSyncImplementation(
         adapter: storage,
         credentialsCallback: connector.getCredentialsCached,
         invalidCredentialsCallback: connector.fetchCredentials,
@@ -148,7 +166,10 @@ class PowerSyncDatabaseImpl
         syncParameters: params,
         // 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);
+        identifier: database.openFactory.path,
+      );
+    }
+
     sync.statusStream.listen((event) {
       setStatus(event);
     });

packages/powersync/lib/src/streaming_sync.dart

@@ -16,7 +16,16 @@ import 'sync_types.dart';
 /// a different value to indicate "no error".
 const _noError = Object();
 
-class StreamingSyncImplementation {
+abstract interface class StreamingSync {
+  Stream<SyncStatus> get statusStream;
+
+  Future<void> streamingSync();
+
+  /// Close any active streams.
+  Future<void> abort();
+}
+
+class StreamingSyncImplementation implements StreamingSync {
   BucketStorage adapter;
 
   final Future<PowerSyncCredentials?> Function() credentialsCallback;
@@ -28,6 +37,8 @@ class StreamingSyncImplementation {
 
   final StreamController<SyncStatus> _statusStreamController =
       StreamController<SyncStatus>.broadcast();
+
+  @override
   late final Stream<SyncStatus> statusStream;
 
   late final http.Client _client;
@@ -66,7 +77,7 @@ class StreamingSyncImplementation {
     statusStream = _statusStreamController.stream;
   }
 
-  /// Close any active streams.
+  @override
   Future<void> abort() async {
     // If streamingSync() hasn't been called yet, _abort will be null.
     var future = _abort?.abort();
@@ -95,6 +106,7 @@ class StreamingSyncImplementation {
     return _abort?.aborted ?? false;
   }
 
+  @override
   Future<void> streamingSync() async {
     try {
       _abort = AbortController();

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

@@ -0,0 +1,100 @@
+import 'dart:async';
+import 'dart:js_interop';
+
+import 'package:powersync/powersync.dart';
+import 'package:sqlite_async/web.dart';
+import 'package:web/web.dart';
+
+import '../database/web/web_powersync_database.dart';
+import '../streaming_sync.dart';
+import 'sync_worker_protocol.dart';
+
+class SyncWorkerHandle implements StreamingSync {
+  final PowerSyncDatabaseImpl _database;
+  final PowerSyncBackendConnector _connector;
+
+  late final WorkerCommunicationChannel _channel;
+
+  final StreamController<SyncStatus> _status = StreamController.broadcast();
+
+  SyncWorkerHandle._(this._database, this._connector, MessagePort sendToWorker,
+      SharedWorker worker) {
+    _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)
+                .exposeEndpoint();
+
+            return (
+              WebEndpoint(
+                databaseName: endpoint.connectName,
+                databasePort: endpoint.connectPort,
+                lockName: endpoint.lockName,
+              ),
+              [endpoint.connectPort].toJS
+            );
+          case SyncWorkerMessageType.uploadCrud:
+            await _connector.uploadData(_database);
+            return (JSObject(), null);
+          case SyncWorkerMessageType.invalidCredentialsCallback:
+            final credentials = await _connector.fetchCredentials();
+            return (
+              credentials != null
+                  ? SerializedCredentials.from(credentials)
+                  : null,
+              null
+            );
+          case SyncWorkerMessageType.credentialsCallback:
+            final credentials = await _connector.getCredentialsCached();
+            return (
+              credentials != null
+                  ? SerializedCredentials.from(credentials)
+                  : null,
+              null
+            );
+          default:
+            throw StateError('Unexpected message type $type');
+        }
+      },
+    );
+
+    _channel.events.listen((data) {
+      final (type, payload) = data;
+      if (type == SyncWorkerMessageType.notifySyncStatus) {
+        _status.add((payload as SerializedSyncStatus).asSyncStatus());
+      }
+    });
+  }
+
+  static Future<SyncWorkerHandle> start(PowerSyncDatabaseImpl database,
+      PowerSyncBackendConnector connector, Uri workerUri) async {
+    final worker = SharedWorker(workerUri.toString().toJS);
+    final handle = SyncWorkerHandle._(database, connector, worker.port, worker);
+
+    // Make sure that the worker is working, or throw immediately.
+    await handle._channel.ping();
+
+    return handle;
+  }
+
+  Future<void> close() async {
+    await abort();
+    await _channel.close();
+  }
+
+  @override
+  Future<void> abort() async {
+    await _channel.abortSynchronization();
+  }
+
+  @override
+  Stream<SyncStatus> get statusStream => _status.stream;
+
+  @override
+  Future<void> streamingSync() async {
+    await _channel.startSynchronization(_database.openFactory.path);
+  }
+}