Refactor update hooks

packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart

@@ -303,40 +303,13 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   final server = params.portServer;
   final commandPort = ReceivePort();
 
-  Timer? updateDebouncer;
-  Set<String> updatedTables = {};
+  db.throttledUpdatedTables.listen((changedTables) {
+    client.fire(UpdateNotification(changedTables));
+  });
+
   int? txId;
   Object? txError;
 
-  void maybeFireUpdates() {
-    // We keep buffering the set of updated tables until we are not
-    // in a transaction. Firing transactions inside a transaction
-    // has multiple issues:
-    // 1. Watched queries would detect changes to the underlying tables,
-    //    but the data would not be visible to queries yet.
-    // 2. It would trigger many more notifications than required.
-    //
-    // This still includes updates for transactions that are rolled back.
-    // We could handle those better at a later stage.
-
-    if (updatedTables.isNotEmpty && db.autocommit) {
-      client.fire(UpdateNotification(updatedTables));
-      updatedTables.clear();
-    }
-    updateDebouncer?.cancel();
-    updateDebouncer = null;
-  }
-
-  db.updates.listen((event) {
-    updatedTables.add(event.tableName);
-
-    // This handles two cases:
-    // 1. Update arrived after _SqliteIsolateClose (not sure if this could happen).
-    // 2. Long-running _SqliteIsolateClosure that should fire updates while running.
-    updateDebouncer ??=
-        Timer(const Duration(milliseconds: 1), maybeFireUpdates);
-  });
-
   ResultSet runStatement(_SqliteIsolateStatement data) {
     if (data.sql == 'BEGIN' || data.sql == 'BEGIN IMMEDIATE') {
       if (txId != null) {
@@ -388,8 +361,6 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
           throw sqlite.SqliteException(
               0, 'Transaction must be closed within the read or write lock');
         }
-        // We would likely have received updates by this point - fire now.
-        maybeFireUpdates();
         return null;
       case _SqliteIsolateStatement():
         return task.timeSync(
@@ -399,11 +370,7 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
           parameters: data.args,
         );
       case _SqliteIsolateClosure():
-        try {
-          return await data.cb(db);
-        } finally {
-          maybeFireUpdates();
-        }
+        return await data.cb(db);
       case _SqliteIsolateConnectionClose():
         db.dispose();
         return null;

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,60 @@
+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());
+
+      final controller = _updates[customRequest.rawSql.toDart];
+      controller?.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],
+        ));
+
+        _updates.remove(id);
+      };
+
+    return controller.stream;
+  }
+}

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

@@ -10,7 +10,7 @@ import 'package:sqlite_async/web.dart';
 import 'database.dart';
 import 'worker/worker_utils.dart';
 
-Map<String, FutureOr<WebSqlite>> webSQLiteImplementations = {};
+Map<String, FutureOr<WebSqlite>> _webSQLiteImplementations = {};
 
 /// Web implementation of [AbstractDefaultSqliteOpenFactory]
 class DefaultSqliteOpenFactory
@@ -20,13 +20,13 @@ class DefaultSqliteOpenFactory
     final cacheKey = sqliteOptions.webSqliteOptions.wasmUri +
         sqliteOptions.webSqliteOptions.workerUri;
 
-    if (webSQLiteImplementations.containsKey(cacheKey)) {
-      return webSQLiteImplementations[cacheKey]!;
+    if (_webSQLiteImplementations.containsKey(cacheKey)) {
+      return _webSQLiteImplementations[cacheKey]!;
     }
 
-    webSQLiteImplementations[cacheKey] =
+    _webSQLiteImplementations[cacheKey] =
         openWebSqlite(sqliteOptions.webSqliteOptions);
-    return webSQLiteImplementations[cacheKey]!;
+    return _webSQLiteImplementations[cacheKey]!;
   });
 
   DefaultSqliteOpenFactory(
@@ -42,6 +42,7 @@ class DefaultSqliteOpenFactory
       wasmModule: Uri.parse(sqliteOptions.webSqliteOptions.wasmUri),
       worker: Uri.parse(sqliteOptions.webSqliteOptions.workerUri),
       controller: AsyncSqliteController(),
+      handleCustomRequest: handleCustomRequest,
     );
   }
 
@@ -69,15 +70,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