Refactor update streams on web

Author: simolus3

packages/sqlite_async/lib/src/utils/shared_utils.dart

@@ -1,6 +1,8 @@
 import 'dart:async';
 import 'dart:convert';
 
+import 'package:sqlite3/common.dart';
+
 import '../sqlite_connection.dart';
 
 Future<T> internalReadTransaction<T>(SqliteReadContext ctx,
@@ -75,3 +77,70 @@ Object? mapParameter(Object? parameter) {
 List<Object?> mapParameters(List<Object?> parameters) {
   return [for (var p in parameters) mapParameter(p)];
 }
+
+extension ThrottledUpdates on CommonDatabase {
+  /// Wraps [updatesSync] to:
+  ///
+  ///   - Not fire in transactions.
+  ///   - Fire asynchronously.
+  ///   - Only report table names, which are buffered to avoid duplicates.
+  Stream<Set<String>> get throttledUpdatedTables {
+    StreamController<Set<String>>? controller;
+    var pendingUpdates = <String>{};
+    var paused = false;
+
+    Timer? updateDebouncer;
+
+    void maybeFireUpdates() {
+      updateDebouncer?.cancel();
+      updateDebouncer = null;
+
+      if (paused) {
+        // Continue collecting updates, but don't fire any
+        return;
+      }
+
+      if (!autocommit) {
+        // Inside a transaction - do not fire updates
+        return;
+      }
+
+      if (pendingUpdates.isNotEmpty) {
+        controller!.add(pendingUpdates);
+        pendingUpdates = {};
+      }
+    }
+
+    void collectUpdate(SqliteUpdate event) {
+      pendingUpdates.add(event.tableName);
+
+      updateDebouncer ??=
+          Timer(const Duration(milliseconds: 1), maybeFireUpdates);
+    }
+
+    StreamSubscription? txSubscription;
+    StreamSubscription? sourceSubscription;
+
+    controller = StreamController(onListen: () {
+      txSubscription = commits.listen((_) {
+        maybeFireUpdates();
+      }, onError: (error) {
+        controller?.addError(error);
+      });
+
+      sourceSubscription = updatesSync.listen(collectUpdate, onError: (error) {
+        controller?.addError(error);
+      });
+    }, onPause: () {
+      paused = true;
+    }, onResume: () {
+      paused = false;
+      maybeFireUpdates();
+    }, onCancel: () {
+      txSubscription?.cancel();
+      sourceSubscription?.cancel();
+    });
+
+    return controller.stream;
+  }
+}

packages/sqlite_async/lib/src/web/database.dart

@@ -21,6 +21,9 @@ class WebDatabase
   final Mutex? _mutex;
   final bool profileQueries;
 
+  @override
+  final Stream<UpdateNotification> updates;
+
   /// For persistent databases that aren't backed by a shared worker, we use
   /// web broadcast channels to forward local update events to other tabs.
   final BroadcastUpdates? broadcastUpdates;
@@ -32,6 +35,7 @@ class WebDatabase
     this._database,
     this._mutex, {
     required this.profileQueries,
+    required this.updates,
     this.broadcastUpdates,
   });
 
@@ -113,10 +117,6 @@ class WebDatabase
     }
   }
 
-  @override
-  Stream<UpdateNotification> get updates =>
-      _database.updates.map((event) => UpdateNotification({event.tableName}));
-
   @override
   Future<T> writeTransaction<T>(
       Future<T> Function(SqliteWriteContext tx) callback,

packages/sqlite_async/lib/src/web/protocol.dart

@@ -13,6 +13,8 @@ enum CustomDatabaseMessageKind {
   getAutoCommit,
   executeInTransaction,
   executeBatchInTransaction,
+  updateSubscriptionManagement,
+  notifyUpdates,
 }
 
 extension type CustomDatabaseMessage._raw(JSObject _) implements JSObject {

packages/sqlite_async/lib/src/web/update_notifications.dart

@@ -0,0 +1,57 @@
+import 'dart:async';
+import 'dart:js_interop';
+
+import 'package:sqlite3_web/sqlite3_web.dart';
+
+import '../update_notification.dart';
+import 'protocol.dart';
+
+/// Utility to request a stream of update notifications from the worker.
+///
+/// Because we want to debounce update notifications on the worker, we're using
+/// custom requests instead of the default [Database.updates] stream.
+///
+/// Clients send a message to the worker to subscribe or unsubscribe, providing
+/// an id for the subscription. The worker distributes update notifications with
+/// custom requests to the client, which [handleRequest] distributes to the
+/// original streams.
+final class UpdateNotificationStreams {
+  var _idCounter = 0;
+  final Map<String, StreamController<UpdateNotification>> _updates = {};
+
+  Future<JSAny?> handleRequest(JSAny? request) async {
+    final customRequest = request as CustomDatabaseMessage;
+    if (customRequest.kind == CustomDatabaseMessageKind.notifyUpdates) {
+      final notification = UpdateNotification(customRequest.rawParameters.toDart
+          .map((e) => (e as JSString).toDart)
+          .toSet());
+
+      _updates[customRequest.rawSql.toDart]?.add(notification);
+    }
+
+    return null;
+  }
+
+  Stream<UpdateNotification> updatesFor(Database database) {
+    final id = (_idCounter++).toString();
+    final controller = _updates[id] = StreamController();
+
+    controller
+      ..onListen = () {
+        database.customRequest(CustomDatabaseMessage(
+          CustomDatabaseMessageKind.updateSubscriptionManagement,
+          id,
+          [true],
+        ));
+      }
+      ..onCancel = () {
+        database.customRequest(CustomDatabaseMessage(
+          CustomDatabaseMessageKind.updateSubscriptionManagement,
+          id,
+          [false],
+        ));
+      };
+
+    return controller.stream;
+  }
+}

packages/sqlite_async/lib/src/web/web_sqlite_open_factory.dart

@@ -69,15 +69,19 @@ class DefaultSqliteOpenFactory
         ? null
         : MutexImpl(identifier: path); // Use the DB path as a mutex identifier
 
-    BroadcastUpdates? updates;
+    BroadcastUpdates? broadcastUpdates;
     if (connection.access != AccessMode.throughSharedWorker &&
         connection.storage != StorageMode.inMemory) {
-      updates = BroadcastUpdates(path);
+      broadcastUpdates = BroadcastUpdates(path);
     }
 
-    return WebDatabase(connection.database, options.mutex ?? mutex,
-        broadcastUpdates: updates,
-        profileQueries: sqliteOptions.profileQueries);
+    return WebDatabase(
+      connection.database,
+      options.mutex ?? mutex,
+      broadcastUpdates: broadcastUpdates,
+      profileQueries: sqliteOptions.profileQueries,
+      updates: updatesFor(connection.database),
+    );
   }
 
   @override