Fix tests

Author: simolus3

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

@@ -123,6 +123,7 @@ extension ThrottledUpdates on CommonDatabase {
 
     controller = StreamController(onListen: () {
       txSubscription = commits.listen((_) {
+        print('did commit');
         maybeFireUpdates();
       }, onError: (error) {
         controller?.addError(error);

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

@@ -26,7 +26,8 @@ final class UpdateNotificationStreams {
           .map((e) => (e as JSString).toDart)
           .toSet());
 
-      _updates[customRequest.rawSql.toDart]?.add(notification);
+      final controller = _updates[customRequest.rawSql.toDart];
+      controller?.add(notification);
     }
 
     return null;
@@ -50,6 +51,8 @@ final class UpdateNotificationStreams {
           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,
     );
   }
 

packages/sqlite_async/lib/web.dart

@@ -29,15 +29,15 @@ typedef WebDatabaseEndpoint = ({
   String? lockName,
 });
 
+final UpdateNotificationStreams _updateStreams = UpdateNotificationStreams();
+
 /// An additional interface for [SqliteOpenFactory] exposing additional
 /// functionality that is only relevant when compiling to the web.
 ///
 /// The [DefaultSqliteOpenFactory] class implements this interface only when
 /// compiling for the web.
 abstract mixin class WebSqliteOpenFactory
     implements SqliteOpenFactory<CommonDatabase> {
-  final UpdateNotificationStreams _updateStreams = UpdateNotificationStreams();
-
   /// Handles a custom request sent from the worker to the client.
   Future<JSAny?> handleCustomRequest(JSAny? request) {
     return _updateStreams.handleRequest(request);

scripts/sqlite3_wasm_download.dart

@@ -4,7 +4,7 @@ library;
 import 'dart:io';
 
 final sqliteUrl =
-    'https://github.com/simolus3/sqlite3.dart/releases/download/sqlite3-2.4.3/sqlite3.wasm';
+    'https://github.com/simolus3/sqlite3.dart/releases/download/sqlite3-2.8.0/sqlite3.wasm';
 
 void main() async {
   // Create assets directory if it doesn't exist