From 4b5607625b1d3f37d6f1610ec562f7be2a79ec86 Mon Sep 17 00:00:00 2001
From: Jorge Sardina <jorge@skilldevs.com>
Date: Fri, 20 Jun 2025 12:52:04 +0200
Subject: [PATCH 01/20] Allow transforming table updates from sqlite_async.

---
 packages/drift_sqlite_async/CHANGELOG.md      |  4 ++++
 .../lib/src/connection.dart                   | 20 ++++++++++++++-----
 packages/drift_sqlite_async/pubspec.yaml      |  2 +-
 3 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/packages/drift_sqlite_async/CHANGELOG.md b/packages/drift_sqlite_async/CHANGELOG.md
index b101ac4..fa769a1 100644
--- a/packages/drift_sqlite_async/CHANGELOG.md
+++ b/packages/drift_sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.2.3
+
+- Allow transforming table updates from sqlite_async.
+
 ## 0.2.2
 
 - Fix write detection when using UPDATE/INSERT/DELETE with RETURNING in raw queries.
diff --git a/packages/drift_sqlite_async/lib/src/connection.dart b/packages/drift_sqlite_async/lib/src/connection.dart
index a1af55a..bcde1bf 100644
--- a/packages/drift_sqlite_async/lib/src/connection.dart
+++ b/packages/drift_sqlite_async/lib/src/connection.dart
@@ -15,12 +15,22 @@ import 'package:sqlite_async/sqlite_async.dart';
 class SqliteAsyncDriftConnection extends DatabaseConnection {
   late StreamSubscription _updateSubscription;
 
-  SqliteAsyncDriftConnection(SqliteConnection db, {bool logStatements = false})
-      : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
+  SqliteAsyncDriftConnection(
+    SqliteConnection db, {
+    bool logStatements = false,
+    Set<TableUpdate> Function(UpdateNotification)? transformTableUpdate,
+  }) : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
     _updateSubscription = (db as SqliteQueries).updates!.listen((event) {
-      var setUpdates = <TableUpdate>{};
-      for (var tableName in event.tables) {
-        setUpdates.add(TableUpdate(tableName));
+      final Set<TableUpdate> setUpdates;
+      // This is useful to map local table names from PowerSync that are backed by a view name
+      // which is the entity that the user interacts with.
+      if (transformTableUpdate != null) {
+        setUpdates = transformTableUpdate(event);
+      } else {
+        setUpdates = <TableUpdate>{};
+        for (var tableName in event.tables) {
+          setUpdates.add(TableUpdate(tableName));
+        }
       }
       super.streamQueries.handleTableUpdates(setUpdates);
     });
diff --git a/packages/drift_sqlite_async/pubspec.yaml b/packages/drift_sqlite_async/pubspec.yaml
index 62a64da..c17e833 100644
--- a/packages/drift_sqlite_async/pubspec.yaml
+++ b/packages/drift_sqlite_async/pubspec.yaml
@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.2
+version: 0.2.3
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.

From a2b6c4cd2226f867569b03d25318b87d6aea6fde Mon Sep 17 00:00:00 2001
From: David Martos <davidmartos96@gmail.com>
Date: Sun, 22 Jun 2025 12:35:16 +0200
Subject: [PATCH 02/20] typo

---
 packages/drift_sqlite_async/lib/src/connection.dart | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/packages/drift_sqlite_async/lib/src/connection.dart b/packages/drift_sqlite_async/lib/src/connection.dart
index bcde1bf..fa56e2e 100644
--- a/packages/drift_sqlite_async/lib/src/connection.dart
+++ b/packages/drift_sqlite_async/lib/src/connection.dart
@@ -18,14 +18,14 @@ class SqliteAsyncDriftConnection extends DatabaseConnection {
   SqliteAsyncDriftConnection(
     SqliteConnection db, {
     bool logStatements = false,
-    Set<TableUpdate> Function(UpdateNotification)? transformTableUpdate,
+    Set<TableUpdate> Function(UpdateNotification)? transformTableUpdates,
   }) : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
     _updateSubscription = (db as SqliteQueries).updates!.listen((event) {
       final Set<TableUpdate> setUpdates;
       // This is useful to map local table names from PowerSync that are backed by a view name
       // which is the entity that the user interacts with.
-      if (transformTableUpdate != null) {
-        setUpdates = transformTableUpdate(event);
+      if (transformTableUpdates != null) {
+        setUpdates = transformTableUpdates(event);
       } else {
         setUpdates = <TableUpdate>{};
         for (var tableName in event.tables) {

From eaab81ea6aacf8e8ac3da08f1d32cb6db91bdf6a Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Fri, 8 Aug 2025 11:18:17 +0200
Subject: [PATCH 03/20] chore(release): publish packages

 - sqlite_async@0.12.0
 - drift_sqlite_async@0.2.3+1
---
 CHANGELOG.md                             | 28 ++++++++++++++++++++++++
 packages/drift_sqlite_async/CHANGELOG.md |  4 ++++
 packages/drift_sqlite_async/pubspec.yaml |  4 ++--
 packages/sqlite_async/CHANGELOG.md       |  4 ++++
 packages/sqlite_async/pubspec.yaml       |  2 +-
 5 files changed, 39 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index cafde71..93a8b0d 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,34 @@
 All notable changes to this project will be documented in this file.
 See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
 
+## 2025-08-08
+
+### Changes
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`sqlite_async` - `v0.12.0`](#sqlite_async---v0120)
+ - [`drift_sqlite_async` - `v0.2.3+1`](#drift_sqlite_async---v0231)
+
+Packages with dependency updates only:
+
+> Packages listed below depend on other packages in this workspace that have had changes. Their versions have been incremented to bump the minimum dependency versions of the packages they depend upon in this project.
+
+ - `drift_sqlite_async` - `v0.2.3+1`
+
+---
+
+#### `sqlite_async` - `v0.12.0`
+
+ - Avoid large transactions creating a large internal update queue.
+
+
 ## 2025-07-29
 
 ---
diff --git a/packages/drift_sqlite_async/CHANGELOG.md b/packages/drift_sqlite_async/CHANGELOG.md
index e70cd40..7c381da 100644
--- a/packages/drift_sqlite_async/CHANGELOG.md
+++ b/packages/drift_sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.2.3+1
+
+ - Update a dependency to the latest release.
+
 ## 0.2.3
 
 - Support nested transactions.
diff --git a/packages/drift_sqlite_async/pubspec.yaml b/packages/drift_sqlite_async/pubspec.yaml
index 77f250d..b865809 100644
--- a/packages/drift_sqlite_async/pubspec.yaml
+++ b/packages/drift_sqlite_async/pubspec.yaml
@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.3
+version: 0.2.3+1
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.
@@ -15,7 +15,7 @@ environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
   drift: ">=2.28.0 <3.0.0"
-  sqlite_async: ^0.11.8
+  sqlite_async: ^0.12.0
 
 dev_dependencies:
   build_runner: ^2.4.8
diff --git a/packages/sqlite_async/CHANGELOG.md b/packages/sqlite_async/CHANGELOG.md
index 387c944..8cda2a6 100644
--- a/packages/sqlite_async/CHANGELOG.md
+++ b/packages/sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.12.0
+
+ - Avoid large transactions creating a large internal update queue.
+
 ## 0.11.8
 
 - Support nested transactions (emulated with `SAVEPOINT` statements).
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index 464bd2d..6e62e52 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.11.8
+version: 0.12.0
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: ">=3.5.0 <4.0.0"

From 0e2fa41567e96f181cd43d5bb25a3321c941c78f Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Fri, 8 Aug 2025 16:39:36 +0200
Subject: [PATCH 04/20] Make table updates a multi-subscription stream

---
 .../native_sqlite_connection_impl.dart        |   2 +-
 .../lib/src/utils/shared_utils.dart           | 123 ++++++++++--------
 .../lib/src/web/worker/worker_utils.dart      |   3 +-
 3 files changed, 74 insertions(+), 54 deletions(-)

diff --git a/packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart b/packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart
index 301d0af..e90739e 100644
--- a/packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart
+++ b/packages/sqlite_async/lib/src/native/database/native_sqlite_connection_impl.dart
@@ -303,7 +303,7 @@ Future<void> _sqliteConnectionIsolateInner(_SqliteConnectionParams params,
   final server = params.portServer;
   final commandPort = ReceivePort();
 
-  db.throttledUpdatedTables.listen((changedTables) {
+  db.updatedTables.listen((changedTables) {
     client.fire(UpdateNotification(changedTables));
   });
 
diff --git a/packages/sqlite_async/lib/src/utils/shared_utils.dart b/packages/sqlite_async/lib/src/utils/shared_utils.dart
index ff291f4..ad9a2f0 100644
--- a/packages/sqlite_async/lib/src/utils/shared_utils.dart
+++ b/packages/sqlite_async/lib/src/utils/shared_utils.dart
@@ -79,68 +79,87 @@ List<Object?> mapParameters(List<Object?> parameters) {
 }
 
 extension ThrottledUpdates on CommonDatabase {
-  /// Wraps [updatesSync] to:
+  /// An unthrottled stream of updated tables that emits on every commit.
   ///
-  ///   - 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;
+  /// A paused subscription on this stream will buffer changed tables into a
+  /// growing set instead of losing events, so this stream is simple to throttle
+  /// downstream.
+  Stream<Set<String>> get updatedTables {
+    final listeners = <_UpdateListener>[];
+    var uncommitedUpdates = <String>{};
+    var underlyingSubscriptions = <StreamSubscription<void>>[];
+
+    void handleUpdate(SqliteUpdate update) {
+      uncommitedUpdates.add(update.tableName);
+    }
+
+    void afterCommit() {
+      for (final listener in listeners) {
+        listener.notify(uncommitedUpdates);
       }
 
-      if (!autocommit) {
-        // Inside a transaction - do not fire updates
-        return;
+      uncommitedUpdates.clear();
+    }
+
+    void afterRollback() {
+      uncommitedUpdates.clear();
+    }
+
+    void addListener(_UpdateListener listener) {
+      listeners.add(listener);
+
+      if (listeners.length == 1) {
+        // First listener, start listening for raw updates on underlying
+        // database.
+        underlyingSubscriptions = [
+          updatesSync.listen(handleUpdate),
+          commits.listen((_) => afterCommit()),
+          commits.listen((_) => afterRollback())
+        ];
       }
+    }
 
-      if (pendingUpdates.isNotEmpty) {
-        controller!.add(pendingUpdates);
-        pendingUpdates = {};
+    void removeListener(_UpdateListener listener) {
+      listeners.remove(listener);
+      if (listeners.isEmpty) {
+        for (final sub in underlyingSubscriptions) {
+          sub.cancel();
+        }
       }
     }
 
-    void collectUpdate(SqliteUpdate event) {
-      pendingUpdates.add(event.tableName);
+    return Stream.multi(
+      (listener) {
+        final wrapped = _UpdateListener(listener);
+        addListener(wrapped);
 
-      updateDebouncer ??=
-          Timer(const Duration(milliseconds: 1), maybeFireUpdates);
+        listener.onCancel = () => removeListener(wrapped);
+      },
+      isBroadcast: true,
+    );
+  }
+}
+
+class _UpdateListener {
+  final MultiStreamController<Set<String>> downstream;
+  Set<String> buffered = {};
+
+  _UpdateListener(this.downstream);
+
+  void notify(Set<String> pendingUpdates) {
+    buffered.addAll(pendingUpdates);
+    if (!downstream.isPaused) {
+      downstream.add(buffered);
+      buffered = {};
     }
+  }
+}
 
-    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;
+extension StreamUtils<T> on Stream<T> {
+  Stream<T> pauseAfterEvent(Duration duration) async* {
+    await for (final event in this) {
+      yield event;
+      await Future.delayed(duration);
+    }
   }
 }
diff --git a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
index 0c3e8f7..a6dae4c 100644
--- a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
+++ b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
@@ -56,7 +56,8 @@ class AsyncSqliteDatabase extends WorkerDatabase {
   final Map<ClientConnection, _ConnectionState> _state = {};
 
   AsyncSqliteDatabase({required this.database})
-      : _updates = database.throttledUpdatedTables;
+      : _updates = database.updatedTables
+            .pauseAfterEvent(const Duration(milliseconds: 1));
 
   _ConnectionState _findState(ClientConnection connection) {
     return _state.putIfAbsent(connection, _ConnectionState.new);

From e0938c4ef0138ee7b77b5911f54fa44fe8722f58 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Fri, 8 Aug 2025 16:45:53 +0200
Subject: [PATCH 05/20] Support multiple listeners for table updates

---
 packages/sqlite_async/lib/src/utils/shared_utils.dart  |  9 ---------
 .../sqlite_async/lib/src/web/worker/worker_utils.dart  | 10 +++++-----
 2 files changed, 5 insertions(+), 14 deletions(-)

diff --git a/packages/sqlite_async/lib/src/utils/shared_utils.dart b/packages/sqlite_async/lib/src/utils/shared_utils.dart
index ad9a2f0..ae38e85 100644
--- a/packages/sqlite_async/lib/src/utils/shared_utils.dart
+++ b/packages/sqlite_async/lib/src/utils/shared_utils.dart
@@ -154,12 +154,3 @@ class _UpdateListener {
     }
   }
 }
-
-extension StreamUtils<T> on Stream<T> {
-  Stream<T> pauseAfterEvent(Duration duration) async* {
-    await for (final event in this) {
-      yield event;
-      await Future.delayed(duration);
-    }
-  }
-}
diff --git a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
index a6dae4c..059c281 100644
--- a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
+++ b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
@@ -56,8 +56,7 @@ class AsyncSqliteDatabase extends WorkerDatabase {
   final Map<ClientConnection, _ConnectionState> _state = {};
 
   AsyncSqliteDatabase({required this.database})
-      : _updates = database.updatedTables
-            .pauseAfterEvent(const Duration(milliseconds: 1));
+      : _updates = database.updatedTables;
 
   _ConnectionState _findState(ClientConnection connection) {
     return _state.putIfAbsent(connection, _ConnectionState.new);
@@ -145,12 +144,13 @@ class AsyncSqliteDatabase extends WorkerDatabase {
           state.unsubscribeUpdates();
           _registerCloseListener(state, connection);
 
-          state.updatesNotification = _updates.listen((tables) {
-            connection.customRequest(CustomDatabaseMessage(
+          late StreamSubscription<void> subscription;
+          subscription = state.updatesNotification = _updates.listen((tables) {
+            subscription.pause(connection.customRequest(CustomDatabaseMessage(
               CustomDatabaseMessageKind.notifyUpdates,
               id,
               tables.toList(),
-            ));
+            )));
           });
         } else {
           state.unsubscribeUpdates();

From b9c8d2d6a9879eda6e7293ab967c529f1cb9e9af Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Fri, 8 Aug 2025 16:48:40 +0200
Subject: [PATCH 06/20] Prepare release

---
 CHANGELOG.md                       | 5 +++++
 packages/sqlite_async/CHANGELOG.md | 4 ++++
 packages/sqlite_async/pubspec.yaml | 2 +-
 3 files changed, 10 insertions(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 93a8b0d..821f8db 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -15,6 +15,7 @@ Packages with breaking changes:
 
 Packages with other changes:
 
+ - [`sqlite_async` - `v0.12.1`](#sqlite_async---v0121)
  - [`sqlite_async` - `v0.12.0`](#sqlite_async---v0120)
  - [`drift_sqlite_async` - `v0.2.3+1`](#drift_sqlite_async---v0231)
 
@@ -26,6 +27,10 @@ Packages with dependency updates only:
 
 ---
 
+#### `sqlite_async` - `v0.12.1`
+
+- Fix distributing updates from shared worker.
+
 #### `sqlite_async` - `v0.12.0`
 
  - Avoid large transactions creating a large internal update queue.
diff --git a/packages/sqlite_async/CHANGELOG.md b/packages/sqlite_async/CHANGELOG.md
index 8cda2a6..342adca 100644
--- a/packages/sqlite_async/CHANGELOG.md
+++ b/packages/sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.12.1
+
+- Fix distributing updates from shared worker.
+
 ## 0.12.0
 
  - Avoid large transactions creating a large internal update queue.
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index 6e62e52..4b45d57 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.12.0
+version: 0.12.1
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: ">=3.5.0 <4.0.0"

From 6705d13335df92b8f548a182e53a097a91c16e37 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Fri, 8 Aug 2025 16:57:24 +0200
Subject: [PATCH 07/20] Add tests

---
 .../lib/src/utils/shared_utils.dart           |  7 +++
 .../sqlite_async/test/native/watch_test.dart  | 46 +++++++++++++++++++
 2 files changed, 53 insertions(+)

diff --git a/packages/sqlite_async/lib/src/utils/shared_utils.dart b/packages/sqlite_async/lib/src/utils/shared_utils.dart
index ae38e85..542611e 100644
--- a/packages/sqlite_async/lib/src/utils/shared_utils.dart
+++ b/packages/sqlite_async/lib/src/utils/shared_utils.dart
@@ -133,6 +133,7 @@ extension ThrottledUpdates on CommonDatabase {
         final wrapped = _UpdateListener(listener);
         addListener(wrapped);
 
+        listener.onResume = wrapped.addPending;
         listener.onCancel = () => removeListener(wrapped);
       },
       isBroadcast: true,
@@ -149,6 +150,12 @@ class _UpdateListener {
   void notify(Set<String> pendingUpdates) {
     buffered.addAll(pendingUpdates);
     if (!downstream.isPaused) {
+      addPending();
+    }
+  }
+
+  void addPending() {
+    if (buffered.isNotEmpty) {
       downstream.add(buffered);
       buffered = {};
     }
diff --git a/packages/sqlite_async/test/native/watch_test.dart b/packages/sqlite_async/test/native/watch_test.dart
index 4e4fb83..0a97b17 100644
--- a/packages/sqlite_async/test/native/watch_test.dart
+++ b/packages/sqlite_async/test/native/watch_test.dart
@@ -7,6 +7,7 @@ import 'dart:math';
 
 import 'package:sqlite3/common.dart';
 import 'package:sqlite_async/sqlite_async.dart';
+import 'package:sqlite_async/src/utils/shared_utils.dart';
 import 'package:test/test.dart';
 
 import '../utils/test_utils_impl.dart';
@@ -31,6 +32,51 @@ void main() {
       return db;
     });
 
+    test('raw update notifications', () async {
+      final factory = await testUtils.testFactory(path: path);
+      final db = factory
+          .openDB(SqliteOpenOptions(primaryConnection: true, readOnly: false));
+
+      db.execute('CREATE TABLE a (bar INTEGER);');
+      db.execute('CREATE TABLE b (bar INTEGER);');
+      final events = <Set<String>>[];
+      final subscription = db.updatedTables.listen(events.add);
+
+      db.execute('insert into a default values');
+      expect(events, isEmpty); // should be async
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a'});
+
+      db.execute('begin');
+      db.execute('insert into a default values');
+      db.execute('insert into b default values');
+      await pumpEventQueue();
+      expect(events, isEmpty); // should only trigger on commit
+      db.execute('commit');
+
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a', 'b'});
+
+      db.execute('begin');
+      db.execute('insert into a default values');
+      db.execute('rollback');
+      expect(events, isEmpty);
+      await pumpEventQueue();
+      expect(events, isEmpty); // should ignore cancelled transactions
+
+      // Should still listen during pause, and dispatch on resume
+      subscription.pause();
+      db.execute('insert into a default values');
+      await pumpEventQueue();
+      expect(events, isEmpty);
+
+      subscription.resume();
+      await pumpEventQueue();
+      expect(events.removeLast(), {'a'});
+
+      subscription.pause();
+    });
+
     test('watch in isolate', () async {
       final db = await testUtils.setupDatabase(path: path);
       await createTables(db);

From d76435962555b05f472ae0d2aa2621f0ead38004 Mon Sep 17 00:00:00 2001
From: David Martos <davidmartos96@gmail.com>
Date: Fri, 19 Sep 2025 12:11:17 +0200
Subject: [PATCH 08/20] update doc

---
 packages/drift_sqlite_async/lib/src/connection.dart | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/packages/drift_sqlite_async/lib/src/connection.dart b/packages/drift_sqlite_async/lib/src/connection.dart
index fa56e2e..6c34ed4 100644
--- a/packages/drift_sqlite_async/lib/src/connection.dart
+++ b/packages/drift_sqlite_async/lib/src/connection.dart
@@ -15,6 +15,8 @@ import 'package:sqlite_async/sqlite_async.dart';
 class SqliteAsyncDriftConnection extends DatabaseConnection {
   late StreamSubscription _updateSubscription;
 
+  /// [transformTableUpdates] is useful to map local table names from PowerSync that are backed by a view name
+  /// which is the entity that the user interacts with.
   SqliteAsyncDriftConnection(
     SqliteConnection db, {
     bool logStatements = false,
@@ -22,8 +24,6 @@ class SqliteAsyncDriftConnection extends DatabaseConnection {
   }) : super(SqliteAsyncQueryExecutor(db, logStatements: logStatements)) {
     _updateSubscription = (db as SqliteQueries).updates!.listen((event) {
       final Set<TableUpdate> setUpdates;
-      // This is useful to map local table names from PowerSync that are backed by a view name
-      // which is the entity that the user interacts with.
       if (transformTableUpdates != null) {
         setUpdates = transformTableUpdates(event);
       } else {

From 09b841ba9b34a1e309504808e399695239e08c17 Mon Sep 17 00:00:00 2001
From: David Martos <davidmartos96@gmail.com>
Date: Fri, 19 Sep 2025 14:14:10 +0200
Subject: [PATCH 09/20] add test

---
 .../drift_sqlite_async/test/basic_test.dart   | 40 +++++++++++++++++++
 .../test/generated/database.dart              |  2 +
 2 files changed, 42 insertions(+)

diff --git a/packages/drift_sqlite_async/test/basic_test.dart b/packages/drift_sqlite_async/test/basic_test.dart
index 339cc04..dd7a159 100644
--- a/packages/drift_sqlite_async/test/basic_test.dart
+++ b/packages/drift_sqlite_async/test/basic_test.dart
@@ -11,6 +11,7 @@ import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
 import './utils/test_utils.dart';
+import 'generated/database.dart';
 
 class EmptyDatabase extends GeneratedDatabase {
   EmptyDatabase(super.executor);
@@ -245,4 +246,43 @@ INSERT INTO test_data(description) VALUES('test data');
       expect(row, isEmpty);
     });
   });
+
+  test('transform table updates', () async {
+    final path = dbPath();
+    await cleanDb(path: path);
+
+    final db = await setupDatabase(path: path);
+    final connection = SqliteAsyncDriftConnection(
+      db,
+      // tables with the local_ prefix are mapped to the name without the prefix
+      transformTableUpdates: (event) {
+        final updates = <TableUpdate>{};
+
+        for (final originalTableName in event.tables) {
+          final effectiveName = originalTableName.startsWith("local_")
+              ? originalTableName.substring(6)
+              : originalTableName;
+          updates.add(TableUpdate(effectiveName));
+        }
+
+        return updates;
+      },
+    );
+
+    // Create table with a different name than drift. (Mimicking a table name backed by a view in PowerSync with the optional sync strategy)
+    await db.execute(
+      'CREATE TABLE local_todos(id INTEGER PRIMARY KEY AUTOINCREMENT, description TEXT)',
+    );
+
+    final dbu = TodoDatabase.fromSqliteAsyncConnection(connection);
+
+    final tableUpdatesFut =
+        dbu.tableUpdates(TableUpdateQuery.onTableName("todos")).first;
+
+    // This insert will trigger the sqlite_async "updates" stream
+    await db.execute("INSERT INTO local_todos(description) VALUES('Test 1')");
+
+    expect(await tableUpdatesFut.timeout(const Duration(seconds: 2)),
+        {TableUpdate("todos")});
+  });
 }
diff --git a/packages/drift_sqlite_async/test/generated/database.dart b/packages/drift_sqlite_async/test/generated/database.dart
index 928c7dd..b8a5109 100644
--- a/packages/drift_sqlite_async/test/generated/database.dart
+++ b/packages/drift_sqlite_async/test/generated/database.dart
@@ -15,6 +15,8 @@ class TodoItems extends Table {
 @DriftDatabase(tables: [TodoItems])
 class TodoDatabase extends _$TodoDatabase {
   TodoDatabase(SqliteConnection db) : super(SqliteAsyncDriftConnection(db));
+  
+  TodoDatabase.fromSqliteAsyncConnection(SqliteAsyncDriftConnection super.conn);
 
   @override
   int get schemaVersion => 1;

From 0a3d31e3b686ae23673f458eef1e6afb8658acf4 Mon Sep 17 00:00:00 2001
From: David Martos <davidmartos96@gmail.com>
Date: Thu, 25 Sep 2025 19:17:54 +0200
Subject: [PATCH 10/20] Update basic_test.dart

Remove legacy skip. It now supports nested transactions
---
 packages/drift_sqlite_async/test/basic_test.dart | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/drift_sqlite_async/test/basic_test.dart b/packages/drift_sqlite_async/test/basic_test.dart
index dd7a159..cac6f55 100644
--- a/packages/drift_sqlite_async/test/basic_test.dart
+++ b/packages/drift_sqlite_async/test/basic_test.dart
@@ -183,7 +183,7 @@ void main() {
             {'description': 'Test 1'},
             {'description': 'Test 3'}
           ]));
-    }, skip: 'sqlite_async does not support nested transactions');
+    });
 
     test('Concurrent select', () async {
       var completer1 = Completer<void>();

From 1252190e3522de429b751ecaadeb301137cb3d68 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Mon, 29 Sep 2025 16:07:25 +0200
Subject: [PATCH 11/20] Run CI for pull requests

---
 .github/workflows/test.yaml | 9 ++++++---
 1 file changed, 6 insertions(+), 3 deletions(-)

diff --git a/.github/workflows/test.yaml b/.github/workflows/test.yaml
index cb1814b..0e0d34b 100644
--- a/.github/workflows/test.yaml
+++ b/.github/workflows/test.yaml
@@ -3,13 +3,15 @@ name: Test
 on:
   push:
     branches:
-      - "**"
+      - "*"
+  pull_request:
 
 jobs:
   build:
     runs-on: ubuntu-latest
+    if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository)  
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v5
       - uses: dart-lang/setup-dart@v1
 
       - name: Install Melos
@@ -29,6 +31,7 @@ jobs:
 
   test:
     runs-on: ubuntu-latest
+    if: github.event_name == 'push' || (github.event_name == 'pull_request' && github.event.pull_request.head.repo.full_name != github.repository)  
     strategy:
       fail-fast: false
       matrix:
@@ -49,7 +52,7 @@ jobs:
             sqlite_url: "https://www.sqlite.org/2022/sqlite-autoconf-3380000.tar.gz"
             dart_sdk: stable
     steps:
-      - uses: actions/checkout@v3
+      - uses: actions/checkout@v5
       - uses: dart-lang/setup-dart@v1
         with:
           sdk: ${{ matrix.dart_sdk }}

From 0234d4fa469702f2709d6933ba199ffac9a19941 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Mon, 29 Sep 2025 16:07:40 +0200
Subject: [PATCH 12/20] Format

---
 packages/drift_sqlite_async/test/generated/database.dart | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/drift_sqlite_async/test/generated/database.dart b/packages/drift_sqlite_async/test/generated/database.dart
index b8a5109..4e03600 100644
--- a/packages/drift_sqlite_async/test/generated/database.dart
+++ b/packages/drift_sqlite_async/test/generated/database.dart
@@ -15,7 +15,7 @@ class TodoItems extends Table {
 @DriftDatabase(tables: [TodoItems])
 class TodoDatabase extends _$TodoDatabase {
   TodoDatabase(SqliteConnection db) : super(SqliteAsyncDriftConnection(db));
-  
+
   TodoDatabase.fromSqliteAsyncConnection(SqliteAsyncDriftConnection super.conn);
 
   @override

From b97be616b5ab5095480cf71848a74d812ee74fac Mon Sep 17 00:00:00 2001
From: David Martos <davidmartos96@gmail.com>
Date: Tue, 7 Oct 2025 16:26:06 +0100
Subject: [PATCH 13/20] Get all Sqlite connections in the pool (#101)

---
 .../lib/src/common/sqlite_database.dart       |   8 +
 .../src/impl/single_connection_database.dart  |   8 +
 .../lib/src/impl/stub_sqlite_database.dart    |   8 +
 .../src/native/database/connection_pool.dart  |  70 ++++++++
 .../database/native_sqlite_database.dart      |   8 +
 .../sqlite_async/lib/src/web/database.dart    |   8 +
 .../src/web/database/web_sqlite_database.dart |   8 +
 packages/sqlite_async/test/basic_test.dart    |  44 +++++
 .../sqlite_async/test/native/basic_test.dart  | 152 ++++++++++++++++++
 9 files changed, 314 insertions(+)

diff --git a/packages/sqlite_async/lib/src/common/sqlite_database.dart b/packages/sqlite_async/lib/src/common/sqlite_database.dart
index 3cb12bb..3201135 100644
--- a/packages/sqlite_async/lib/src/common/sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/common/sqlite_database.dart
@@ -39,6 +39,14 @@ mixin SqliteDatabaseMixin implements SqliteConnection, SqliteQueries {
   ///
   /// Use this to access the database in background isolates.
   IsolateConnectionFactory isolateConnectionFactory();
+
+  /// Locks all underlying connections making up this database, and gives [block] access to all of them at once.
+  /// This can be useful to run the same statement on all connections. For instance,
+  /// ATTACHing a database, that is expected to be available in all connections.
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block);
 }
 
 /// A SQLite database instance.
diff --git a/packages/sqlite_async/lib/src/impl/single_connection_database.dart b/packages/sqlite_async/lib/src/impl/single_connection_database.dart
index 4cd3144..7ca4357 100644
--- a/packages/sqlite_async/lib/src/impl/single_connection_database.dart
+++ b/packages/sqlite_async/lib/src/impl/single_connection_database.dart
@@ -57,4 +57,12 @@ final class SingleConnectionDatabase
     return connection.writeLock(callback,
         lockTimeout: lockTimeout, debugContext: debugContext);
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return writeLock((_) => block(connection, []));
+  }
 }
diff --git a/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart b/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
index 29db641..ee254f3 100644
--- a/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
@@ -64,4 +64,12 @@ class SqliteDatabaseImpl
   Future<bool> getAutoCommit() {
     throw UnimplementedError();
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    throw UnimplementedError();
+  }
 }
diff --git a/packages/sqlite_async/lib/src/native/database/connection_pool.dart b/packages/sqlite_async/lib/src/native/database/connection_pool.dart
index 9521b34..8dab27e 100644
--- a/packages/sqlite_async/lib/src/native/database/connection_pool.dart
+++ b/packages/sqlite_async/lib/src/native/database/connection_pool.dart
@@ -31,6 +31,8 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
 
   final MutexImpl mutex;
 
+  int _runningWithAllConnectionsCount = 0;
+
   @override
   bool closed = false;
 
@@ -88,6 +90,14 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       return;
     }
 
+    if (_availableReadConnections.isEmpty &&
+        _runningWithAllConnectionsCount > 0) {
+      // Wait until [withAllConnections] is done. Otherwise we could spawn a new
+      // reader while the user is configuring all the connections,
+      // e.g. a global open factory configuration shared across all connections.
+      return;
+    }
+
     var nextItem = _queue.removeFirst();
     while (nextItem.completer.isCompleted) {
       // This item already timed out - try the next one if available
@@ -232,6 +242,66 @@ class SqliteConnectionPool with SqliteQueries implements SqliteConnection {
       await connection.refreshSchema();
     }
   }
+
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) async {
+    try {
+      _runningWithAllConnectionsCount++;
+
+      final blockCompleter = Completer<T>();
+      final (write, reads) = await _lockAllConns<T>(blockCompleter);
+
+      try {
+        final res = await block(write, reads);
+        blockCompleter.complete(res);
+        return res;
+      } catch (e, st) {
+        blockCompleter.completeError(e, st);
+        rethrow;
+      }
+    } finally {
+      _runningWithAllConnectionsCount--;
+
+      // Continue processing any pending read requests that may have been queued while
+      // the block was running.
+      Timer.run(_nextRead);
+    }
+  }
+
+  /// Locks all connections, returning the acquired contexts.
+  /// We pass a completer that would be called after the locks are taken.
+  Future<(SqliteWriteContext, List<SqliteReadContext>)> _lockAllConns<T>(
+      Completer<T> lockCompleter) async {
+    final List<Completer<SqliteReadContext>> readLockedCompleters = [];
+    final Completer<SqliteWriteContext> writeLockedCompleter = Completer();
+
+    // Take the write lock
+    writeLock((ctx) {
+      writeLockedCompleter.complete(ctx);
+      return lockCompleter.future;
+    });
+
+    // Take all the read locks
+    for (final readConn in _allReadConnections) {
+      final completer = Completer<SqliteReadContext>();
+      readLockedCompleters.add(completer);
+
+      readConn.readLock((ctx) {
+        completer.complete(ctx);
+        return lockCompleter.future;
+      });
+    }
+
+    // Wait after all locks are taken
+    final [writer as SqliteWriteContext, ...readers] = await Future.wait([
+      writeLockedCompleter.future,
+      ...readLockedCompleters.map((e) => e.future)
+    ]);
+
+    return (writer, readers);
+  }
 }
 
 typedef ReadCallback<T> = Future<T> Function(SqliteReadContext tx);
diff --git a/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart b/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
index 7bea111..22cacf3 100644
--- a/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
@@ -171,4 +171,12 @@ class SqliteDatabaseImpl
   Future<void> refreshSchema() {
     return _pool.refreshSchema();
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return _pool.withAllConnections(block);
+  }
 }
diff --git a/packages/sqlite_async/lib/src/web/database.dart b/packages/sqlite_async/lib/src/web/database.dart
index cfaf987..f2dc998 100644
--- a/packages/sqlite_async/lib/src/web/database.dart
+++ b/packages/sqlite_async/lib/src/web/database.dart
@@ -171,6 +171,14 @@ class WebDatabase
     await isInitialized;
     return _database.fileSystem.flush();
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return writeLock((_) => block(this, []));
+  }
 }
 
 final class _UnscopedContext extends UnscopedContext {
diff --git a/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart b/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
index c6d1b75..69f01ab 100644
--- a/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
@@ -178,4 +178,12 @@ class SqliteDatabaseImpl
   Future<WebDatabaseEndpoint> exposeEndpoint() async {
     return await _connection.exposeEndpoint();
   }
+
+  @override
+  Future<T> withAllConnections<T>(
+      Future<T> Function(
+              SqliteWriteContext writer, List<SqliteReadContext> readers)
+          block) {
+    return writeLock((_) => block(_connection, []));
+  }
 }
diff --git a/packages/sqlite_async/test/basic_test.dart b/packages/sqlite_async/test/basic_test.dart
index 6a315da..e2914b3 100644
--- a/packages/sqlite_async/test/basic_test.dart
+++ b/packages/sqlite_async/test/basic_test.dart
@@ -7,6 +7,7 @@ import 'utils/test_utils_impl.dart';
 
 final testUtils = TestUtils();
 const _isDart2Wasm = bool.fromEnvironment('dart.tool.dart2wasm');
+const _isWeb = identical(0, 0.0) || _isDart2Wasm;
 
 void main() {
   group('Shared Basic Tests', () {
@@ -301,6 +302,49 @@ void main() {
         'Web locks are managed with a shared worker, which does not support timeouts',
       )
     });
+
+    test('with all connections', () async {
+      final maxReaders = _isWeb ? 0 : 3;
+
+      final db = SqliteDatabase.withFactory(
+        await testUtils.testFactory(path: path),
+        maxReaders: maxReaders,
+      );
+      await db.initialize();
+      await createTables(db);
+
+      // Warm up to spawn the max readers
+      await Future.wait([for (var i = 0; i < 10; i++) db.get('SELECT $i')]);
+
+      bool finishedWithAllConns = false;
+
+      late Future<void> readsCalledWhileWithAllConnsRunning;
+
+      final parentZone = Zone.current;
+      await db.withAllConnections((writer, readers) async {
+        expect(readers.length, maxReaders);
+
+        // Run some reads during the block that they should run after the block finishes and releases
+        // all locks
+        // Need a root zone here to avoid recursive lock errors.
+        readsCalledWhileWithAllConnsRunning =
+            Future(parentZone.bindCallback(() async {
+          await Future.wait(
+            [1, 2, 3, 4, 5, 6, 7, 8].map((i) async {
+              await db.readLock((c) async {
+                expect(finishedWithAllConns, isTrue);
+                await Future.delayed(const Duration(milliseconds: 100));
+              });
+            }),
+          );
+        }));
+
+        await Future.delayed(const Duration(milliseconds: 200));
+        finishedWithAllConns = true;
+      });
+
+      await readsCalledWhileWithAllConnsRunning;
+    });
   });
 }
 
diff --git a/packages/sqlite_async/test/native/basic_test.dart b/packages/sqlite_async/test/native/basic_test.dart
index dec1fed..3f348e6 100644
--- a/packages/sqlite_async/test/native/basic_test.dart
+++ b/packages/sqlite_async/test/native/basic_test.dart
@@ -2,12 +2,16 @@
 library;
 
 import 'dart:async';
+import 'dart:io';
 import 'dart:math';
 
+import 'package:collection/collection.dart';
+import 'package:path/path.dart' show join;
 import 'package:sqlite3/common.dart' as sqlite;
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:test/test.dart';
 
+import '../utils/abstract_test_utils.dart';
 import '../utils/test_utils_impl.dart';
 
 final testUtils = TestUtils();
@@ -100,6 +104,126 @@ void main() {
       print("${DateTime.now()} done");
     });
 
+    test('prevent opening new readers while in withAllConnections', () async {
+      final sharedStateDir = Directory.systemTemp.createTempSync();
+      addTearDown(() => sharedStateDir.deleteSync(recursive: true));
+
+      final File sharedStateFile =
+          File(join(sharedStateDir.path, 'shared-state.txt'));
+
+      sharedStateFile.writeAsStringSync('initial');
+
+      final db = SqliteDatabase.withFactory(
+          _TestSqliteOpenFactoryWithSharedStateFile(
+              path: path, sharedStateFilePath: sharedStateFile.path),
+          maxReaders: 3);
+      await db.initialize();
+      await createTables(db);
+
+      // The writer saw 'initial' in the file when opening the connection
+      expect(
+        await db
+            .writeLock((c) => c.get('SELECT file_contents_on_open() AS state')),
+        {'state': 'initial'},
+      );
+
+      final withAllConnectionsCompleter = Completer<void>();
+
+      final withAllConnsFut = db.withAllConnections((writer, readers) async {
+        expect(readers.length, 0); // No readers yet
+
+        // Simulate some work until the file is updated
+        await Future.delayed(const Duration(milliseconds: 200));
+        sharedStateFile.writeAsStringSync('updated');
+
+        await withAllConnectionsCompleter.future;
+      });
+
+      // Start a reader that gets the contents of the shared file
+      bool readFinished = false;
+      final someReadFut =
+          db.get('SELECT file_contents_on_open() AS state', []).then((r) {
+        readFinished = true;
+        return r;
+      });
+
+      // The withAllConnections should prevent the reader from opening
+      await Future.delayed(const Duration(milliseconds: 100));
+      expect(readFinished, isFalse);
+
+      // Free all the locks
+      withAllConnectionsCompleter.complete();
+      await withAllConnsFut;
+
+      final readerInfo = await someReadFut;
+      expect(readFinished, isTrue);
+      // The read should see the updated value in the file. This checks
+      // that a reader doesn't spawn while running withAllConnections
+      expect(readerInfo, {'state': 'updated'});
+    });
+
+    test('with all connections', () async {
+      final maxReaders = 3;
+
+      final db = SqliteDatabase.withFactory(
+        await testUtils.testFactory(path: path),
+        maxReaders: maxReaders,
+      );
+      await db.initialize();
+      await createTables(db);
+
+      Future<sqlite.Row> readWithRandomDelay(
+          SqliteReadContext ctx, int id) async {
+        return await ctx.get(
+            'SELECT ? as i, test_sleep(?) as sleep, test_connection_name() as connection',
+            [id, 5 + Random().nextInt(10)]);
+      }
+
+      // Warm up to spawn the max readers
+      await Future.wait(
+        [1, 2, 3, 4, 5, 6, 7, 8].map((i) => readWithRandomDelay(db, i)),
+      );
+
+      bool finishedWithAllConns = false;
+
+      late Future<void> readsCalledWhileWithAllConnsRunning;
+
+      print("${DateTime.now()} start");
+      await db.withAllConnections((writer, readers) async {
+        expect(readers.length, maxReaders);
+
+        // Run some reads during the block that they should run after the block finishes and releases
+        // all locks
+        readsCalledWhileWithAllConnsRunning = Future.wait(
+          [1, 2, 3, 4, 5, 6, 7, 8].map((i) async {
+            final r = await db.readLock((c) async {
+              expect(finishedWithAllConns, isTrue);
+              return await readWithRandomDelay(c, i);
+            });
+            print(
+                "${DateTime.now()} After withAllConnections, started while running $r");
+          }),
+        );
+
+        await Future.wait([
+          writer.execute(
+              "INSERT OR REPLACE INTO test_data(id, description) SELECT ? as i, test_sleep(?) || ' ' || test_connection_name() || ' 1 ' || datetime() as connection RETURNING *",
+              [
+                123,
+                5 + Random().nextInt(20)
+              ]).then((value) =>
+              print("${DateTime.now()} withAllConnections writer done $value")),
+          ...readers
+              .mapIndexed((i, r) => readWithRandomDelay(r, i).then((results) {
+                    print(
+                        "${DateTime.now()} withAllConnections readers done $results");
+                  }))
+        ]);
+      }).then((_) => finishedWithAllConns = true);
+
+      await readsCalledWhileWithAllConnsRunning;
+    });
+
     test('read-only transactions', () async {
       final db = await testUtils.setupDatabase(path: path);
       await createTables(db);
@@ -379,3 +503,31 @@ class _InvalidPragmaOnOpenFactory extends DefaultSqliteOpenFactory {
     ];
   }
 }
+
+class _TestSqliteOpenFactoryWithSharedStateFile
+    extends TestDefaultSqliteOpenFactory {
+  final String sharedStateFilePath;
+
+  _TestSqliteOpenFactoryWithSharedStateFile(
+      {required super.path, required this.sharedStateFilePath});
+
+  @override
+  sqlite.CommonDatabase open(SqliteOpenOptions options) {
+    final File sharedStateFile = File(sharedStateFilePath);
+    final String sharedState = sharedStateFile.readAsStringSync();
+
+    final db = super.open(options);
+
+    // Function to return the contents of the shared state file at the time of opening
+    // so that we know at which point the factory was called.
+    db.createFunction(
+      functionName: 'file_contents_on_open',
+      argumentCount: const sqlite.AllowedArgumentCount(0),
+      function: (args) {
+        return sharedState;
+      },
+    );
+
+    return db;
+  }
+}

From 21df6da19d3e69762047a638c1e6fef8b710e713 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Mon, 13 Oct 2025 17:25:46 +0200
Subject: [PATCH 14/20] chore(release): publish packages

 - sqlite_async@0.12.2
 - drift_sqlite_async@0.2.5
---
 CHANGELOG.md                             | 26 ++++++++++++++++++++++++
 packages/drift_sqlite_async/CHANGELOG.md |  4 ++++
 packages/drift_sqlite_async/pubspec.yaml |  4 ++--
 packages/sqlite_async/CHANGELOG.md       |  4 ++++
 packages/sqlite_async/pubspec.yaml       |  2 +-
 5 files changed, 37 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 821f8db..5181152 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,32 @@
 All notable changes to this project will be documented in this file.
 See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
 
+## 2025-10-13
+
+### Changes
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`sqlite_async` - `v0.12.2`](#sqlite_async---v0122)
+ - [`drift_sqlite_async` - `v0.2.5`](#drift_sqlite_async---v025)
+
+---
+
+#### `sqlite_async` - `v0.12.2`
+
+ - Add `withAllConnections` method to run statements on all connections in the pool.
+
+#### `drift_sqlite_async` - `v0.2.5`
+
+ - Allow customizing update notifications from `sqlite_async`.
+
+
 ## 2025-08-08
 
 ### Changes
diff --git a/packages/drift_sqlite_async/CHANGELOG.md b/packages/drift_sqlite_async/CHANGELOG.md
index 8dd5e5a..bd36dd1 100644
--- a/packages/drift_sqlite_async/CHANGELOG.md
+++ b/packages/drift_sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.2.5
+
+ - Allow customizing update notifications from `sqlite_async`.
+
 ## 0.2.4
 
 - Allow transforming table updates from sqlite_async.
diff --git a/packages/drift_sqlite_async/pubspec.yaml b/packages/drift_sqlite_async/pubspec.yaml
index 06e2963..bd2b701 100644
--- a/packages/drift_sqlite_async/pubspec.yaml
+++ b/packages/drift_sqlite_async/pubspec.yaml
@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.4
+version: 0.2.5
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.
@@ -15,7 +15,7 @@ environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
   drift: ">=2.28.0 <3.0.0"
-  sqlite_async: ^0.12.0
+  sqlite_async: ^0.12.2
 
 dev_dependencies:
   build_runner: ^2.4.8
diff --git a/packages/sqlite_async/CHANGELOG.md b/packages/sqlite_async/CHANGELOG.md
index 342adca..1bbc939 100644
--- a/packages/sqlite_async/CHANGELOG.md
+++ b/packages/sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.12.2
+
+ - Add `withAllConnections` method to run statements on all connections in the pool.
+
 ## 0.12.1
 
 - Fix distributing updates from shared worker.
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index 4b45d57..3d5130d 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.12.1
+version: 0.12.2
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: ">=3.5.0 <4.0.0"

From 5d147a7e74ac39c655cc1980411734e8823ce5c8 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Thu, 30 Oct 2025 17:43:37 +0100
Subject: [PATCH 15/20] Update to latest lints

---
 .../lib/src/common/isolate_connection_factory.dart        | 4 ++--
 .../sqlite_async/lib/src/common/port_channel_native.dart  | 4 ++--
 .../sqlite_async/lib/src/common/port_channel_stub.dart    | 4 ++--
 packages/sqlite_async/lib/src/common/sqlite_database.dart | 2 +-
 .../sqlite_async/lib/src/impl/stub_sqlite_database.dart   | 2 +-
 .../lib/src/native/database/native_sqlite_database.dart   | 2 +-
 .../lib/src/native/native_isolate_connection_factory.dart | 2 +-
 .../sqlite_async/lib/src/native/native_isolate_mutex.dart | 2 +-
 packages/sqlite_async/lib/src/sqlite_migrations.dart      | 8 ++++----
 .../lib/src/web/database/web_sqlite_database.dart         | 2 +-
 packages/sqlite_async/pubspec.yaml                        | 4 ++--
 packages/sqlite_async/test/watch_test.dart                | 2 +-
 12 files changed, 19 insertions(+), 19 deletions(-)

diff --git a/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart b/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
index 90dc9b1..959c38b 100644
--- a/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
+++ b/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
@@ -30,8 +30,8 @@ abstract class IsolateConnectionFactory<Database extends sqlite.CommonDatabase>
   SerializedPortClient get upstreamPort;
 
   factory IsolateConnectionFactory(
-      {required openFactory,
-      required mutex,
+      {required dynamic /* platform-specific type */ openFactory,
+      required /* platform-specific type */ mutex,
       required SerializedPortClient upstreamPort}) {
     return IsolateConnectionFactoryImpl(
         openFactory: openFactory,
diff --git a/packages/sqlite_async/lib/src/common/port_channel_native.dart b/packages/sqlite_async/lib/src/common/port_channel_native.dart
index bb8318e..c9f8e6b 100644
--- a/packages/sqlite_async/lib/src/common/port_channel_native.dart
+++ b/packages/sqlite_async/lib/src/common/port_channel_native.dart
@@ -128,7 +128,7 @@ class ParentPortClient implements PortClient {
     _close();
   }
 
-  tieToIsolate(Isolate isolate) {
+  void tieToIsolate(Isolate isolate) {
     _isolateDebugName = isolate.debugName;
     isolate.addErrorListener(_errorPort.sendPort);
     isolate.addOnExitListener(_receivePort.sendPort, response: _closeMessage);
@@ -207,7 +207,7 @@ class RequestPortServer {
 
   RequestPortServer(this.port);
 
-  open(Future<Object?> Function(Object? message) handle) {
+  PortServer open(Future<Object?> Function(Object? message) handle) {
     return PortServer.forSendPort(port, handle);
   }
 }
diff --git a/packages/sqlite_async/lib/src/common/port_channel_stub.dart b/packages/sqlite_async/lib/src/common/port_channel_stub.dart
index 6c6e5cc..d5c50ae 100644
--- a/packages/sqlite_async/lib/src/common/port_channel_stub.dart
+++ b/packages/sqlite_async/lib/src/common/port_channel_stub.dart
@@ -53,7 +53,7 @@ class ParentPortClient implements PortClient {
     _stub();
   }
 
-  tieToIsolate(Isolate isolate) {
+  void tieToIsolate(Isolate isolate) {
     _stub();
   }
 }
@@ -97,7 +97,7 @@ class RequestPortServer {
 
   RequestPortServer(this.port);
 
-  open(Future<Object?> Function(Object? message) handle) {
+  PortServer open(Future<Object?> Function(Object? message) handle) {
     _stub();
   }
 }
diff --git a/packages/sqlite_async/lib/src/common/sqlite_database.dart b/packages/sqlite_async/lib/src/common/sqlite_database.dart
index 3201135..a73828a 100644
--- a/packages/sqlite_async/lib/src/common/sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/common/sqlite_database.dart
@@ -70,7 +70,7 @@ abstract class SqliteDatabase
   ///
   /// A maximum of [maxReaders] concurrent read transactions are allowed.
   factory SqliteDatabase(
-      {required path,
+      {required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       SqliteOptions options = const SqliteOptions.defaults()}) {
     return SqliteDatabaseImpl(
diff --git a/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart b/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
index ee254f3..cc106a7 100644
--- a/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/impl/stub_sqlite_database.dart
@@ -20,7 +20,7 @@ class SqliteDatabaseImpl
   int maxReaders;
 
   factory SqliteDatabaseImpl(
-      {required path,
+      {required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       SqliteOptions options = const SqliteOptions.defaults()}) {
     throw UnimplementedError();
diff --git a/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart b/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
index 22cacf3..85b0dd0 100644
--- a/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/native/database/native_sqlite_database.dart
@@ -54,7 +54,7 @@ class SqliteDatabaseImpl
   ///
   /// A maximum of [maxReaders] concurrent read transactions are allowed.
   factory SqliteDatabaseImpl(
-      {required path,
+      {required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       SqliteOptions options = const SqliteOptions.defaults()}) {
     final factory =
diff --git a/packages/sqlite_async/lib/src/native/native_isolate_connection_factory.dart b/packages/sqlite_async/lib/src/native/native_isolate_connection_factory.dart
index 915d2ad..5cbb1a7 100644
--- a/packages/sqlite_async/lib/src/native/native_isolate_connection_factory.dart
+++ b/packages/sqlite_async/lib/src/native/native_isolate_connection_factory.dart
@@ -75,7 +75,7 @@ class _IsolateUpdateListener {
     return controller.stream;
   }
 
-  close() {
+  void close() {
     client.fire(UnsubscribeToUpdates(port.sendPort));
     controller.close();
     port.close();
diff --git a/packages/sqlite_async/lib/src/native/native_isolate_mutex.dart b/packages/sqlite_async/lib/src/native/native_isolate_mutex.dart
index 7d82f68..9d1a212 100644
--- a/packages/sqlite_async/lib/src/native/native_isolate_mutex.dart
+++ b/packages/sqlite_async/lib/src/native/native_isolate_mutex.dart
@@ -159,7 +159,7 @@ class SharedMutex implements Mutex {
     }, zoneValues: {this: true});
   }
 
-  _unlock() {
+  void _unlock() {
     client.fire(const _UnlockMessage());
   }
 
diff --git a/packages/sqlite_async/lib/src/sqlite_migrations.dart b/packages/sqlite_async/lib/src/sqlite_migrations.dart
index a0c9991..7f2e5d8 100644
--- a/packages/sqlite_async/lib/src/sqlite_migrations.dart
+++ b/packages/sqlite_async/lib/src/sqlite_migrations.dart
@@ -25,7 +25,7 @@ class SqliteMigrations {
 
   SqliteMigrations({this.migrationTable = "_migrations"});
 
-  add(SqliteMigration migration) {
+  void add(SqliteMigration migration) {
     assert(
         migrations.isEmpty || migrations.last.toVersion < migration.toVersion);
 
@@ -48,7 +48,7 @@ class SqliteMigrations {
   }
 
   /// The current version as specified by the migrations.
-  get version {
+  int get version {
     return migrations.isEmpty ? 0 : migrations.last.toVersion;
   }
 
@@ -68,7 +68,7 @@ class SqliteMigrations {
     }
   }
 
-  _validateCreateDatabase() {
+  void _validateCreateDatabase() {
     if (createDatabase != null) {
       if (createDatabase!.downMigration != null) {
         throw MigrationError("createDatabase may not contain down migrations");
@@ -229,7 +229,7 @@ class SqliteDownMigration {
   SqliteDownMigration({required this.toVersion});
 
   /// Add an statement to execute to downgrade the database version.
-  add(String sql, [List<Object?>? params]) {
+  void add(String sql, [List<Object?>? params]) {
     _statements.add(_SqliteMigrationStatement(sql, params ?? []));
   }
 }
diff --git a/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart b/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
index 69f01ab..e129afa 100644
--- a/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
+++ b/packages/sqlite_async/lib/src/web/database/web_sqlite_database.dart
@@ -56,7 +56,7 @@ class SqliteDatabaseImpl
   ///
   /// A maximum of [maxReaders] concurrent read transactions are allowed.
   factory SqliteDatabaseImpl(
-      {required path,
+      {required String path,
       int maxReaders = SqliteDatabase.defaultMaxReaders,
       SqliteOptions options = const SqliteOptions.defaults()}) {
     final factory =
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index 3d5130d..a5589d8 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -23,8 +23,8 @@ dependencies:
 dev_dependencies:
   build_runner: ^2.4.14
   build_web_compilers: ^4.1.1
-  build_test: ^2.2.3
-  lints: ^5.0.0
+  build_test: ^3.5.0
+  lints: ^6.0.0
   test: ^1.21.0
   test_api: ^0.7.0
   glob: ^2.1.1
diff --git a/packages/sqlite_async/test/watch_test.dart b/packages/sqlite_async/test/watch_test.dart
index dc28add..1597aa0 100644
--- a/packages/sqlite_async/test/watch_test.dart
+++ b/packages/sqlite_async/test/watch_test.dart
@@ -8,7 +8,7 @@ import 'utils/test_utils_impl.dart';
 
 final testUtils = TestUtils();
 
-createTables(SqliteDatabase db) async {
+Future<void> createTables(SqliteDatabase db) async {
   await db.writeTransaction((tx) async {
     await tx.execute(
         'CREATE TABLE assets(id INTEGER PRIMARY KEY AUTOINCREMENT, make TEXT, customer_id INTEGER)');

From 71785c15c22be33e568808fd31e246d0cb982490 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Thu, 30 Oct 2025 17:44:09 +0100
Subject: [PATCH 16/20] make dynamic explicit

---
 .../sqlite_async/lib/src/common/isolate_connection_factory.dart | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart b/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
index 959c38b..87e6176 100644
--- a/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
+++ b/packages/sqlite_async/lib/src/common/isolate_connection_factory.dart
@@ -31,7 +31,7 @@ abstract class IsolateConnectionFactory<Database extends sqlite.CommonDatabase>
 
   factory IsolateConnectionFactory(
       {required dynamic /* platform-specific type */ openFactory,
-      required /* platform-specific type */ mutex,
+      required dynamic /* platform-specific type */ mutex,
       required SerializedPortClient upstreamPort}) {
     return IsolateConnectionFactoryImpl(
         openFactory: openFactory,

From e5965f4c47bc2e0f8ce58ef0583f69fa8f936fba Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Thu, 30 Oct 2025 17:07:55 +0100
Subject: [PATCH 17/20] Update sqlite3_web, simplify

---
 .../sqlite_async/lib/src/web/database.dart    | 220 ++++++++----------
 .../sqlite_async/lib/src/web/protocol.dart    |   6 +-
 .../lib/src/web/worker/worker_utils.dart      |  44 +---
 packages/sqlite_async/pubspec.yaml            |   4 +-
 4 files changed, 104 insertions(+), 170 deletions(-)

diff --git a/packages/sqlite_async/lib/src/web/database.dart b/packages/sqlite_async/lib/src/web/database.dart
index f2dc998..23424fc 100644
--- a/packages/sqlite_async/lib/src/web/database.dart
+++ b/packages/sqlite_async/lib/src/web/database.dart
@@ -1,11 +1,9 @@
 import 'dart:async';
 import 'dart:developer';
 import 'dart:js_interop';
-import 'dart:js_interop_unsafe';
 
 import 'package:sqlite3/common.dart';
 import 'package:sqlite3_web/sqlite3_web.dart';
-import 'package:sqlite3_web/protocol_utils.dart' as proto;
 import 'package:sqlite_async/sqlite_async.dart';
 import 'package:sqlite_async/src/utils/profiler.dart';
 import 'package:sqlite_async/src/web/database/broadcast_updates.dart';
@@ -97,24 +95,17 @@ class WebDatabase
   @override
   Future<T> readLock<T>(Future<T> Function(SqliteReadContext tx) callback,
       {Duration? lockTimeout, String? debugContext}) async {
-    if (_mutex case var mutex?) {
-      return await mutex.lock(timeout: lockTimeout, () {
-        return ScopedReadContext.assumeReadLock(
-            _UnscopedContext(this), callback);
-      });
-    } else {
-      // No custom mutex, coordinate locks through shared worker.
-      await _database.customRequest(
-          CustomDatabaseMessage(CustomDatabaseMessageKind.requestSharedLock));
-
-      try {
-        return await ScopedReadContext.assumeReadLock(
-            _UnscopedContext(this), callback);
-      } finally {
-        await _database.customRequest(
-            CustomDatabaseMessage(CustomDatabaseMessageKind.releaseLock));
-      }
-    }
+    // Since there is only a single physical connection per database on the web,
+    // we can't enable concurrent readers to a writer. Even supporting multiple
+    // readers alone isn't safe, since these readers could start read
+    // transactions where we need to block other tabs from sending `BEGIN` and
+    // `COMMIT` statements if they were to start their own transactions.
+    return _lockInternal(
+      (unscoped) => ScopedReadContext.assumeReadLock(unscoped, callback),
+      lockTimeout: lockTimeout,
+      debugContext: debugContext,
+      flush: false,
+    );
   }
 
   @override
@@ -122,47 +113,67 @@ class WebDatabase
       Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout,
       bool? flush}) {
-    return writeLock((writeContext) {
-      return ScopedWriteContext.assumeWriteLock(
-        _UnscopedContext(this),
-        (ctx) async {
-          return await ctx.writeTransaction(callback);
-        },
-      );
-    },
-        debugContext: 'writeTransaction()',
-        lockTimeout: lockTimeout,
-        flush: flush);
+    return _lockInternal(
+      (context) {
+        return ScopedWriteContext.assumeWriteLock(
+          context,
+          (ctx) async {
+            return await ctx.writeTransaction(callback);
+          },
+        );
+      },
+      flush: flush ?? true,
+      lockTimeout: lockTimeout,
+    );
   }
 
   @override
   Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
       {Duration? lockTimeout, String? debugContext, bool? flush}) async {
+    return await _lockInternal(
+      (unscoped) {
+        return ScopedWriteContext.assumeWriteLock(unscoped, callback);
+      },
+      flush: flush ?? true,
+      debugContext: debugContext,
+      lockTimeout: lockTimeout,
+    );
+  }
+
+  Future<T> _lockInternal<T>(
+    Future<T> Function(_UnscopedContext) callback, {
+    required bool flush,
+    Duration? lockTimeout,
+    String? debugContext,
+  }) async {
     if (_mutex case var mutex?) {
       return await mutex.lock(timeout: lockTimeout, () async {
-        final context = _UnscopedContext(this);
+        final context = _UnscopedContext(this, null);
         try {
-          return await ScopedWriteContext.assumeWriteLock(context, callback);
+          return await callback(context);
         } finally {
-          if (flush != false) {
+          if (flush) {
             await this.flush();
           }
         }
       });
     } else {
-      // No custom mutex, coordinate locks through shared worker.
-      await _database.customRequest(CustomDatabaseMessage(
-          CustomDatabaseMessageKind.requestExclusiveLock));
-      final context = _UnscopedContext(this);
-      try {
-        return await ScopedWriteContext.assumeWriteLock(context, callback);
-      } finally {
-        if (flush != false) {
-          await this.flush();
+      final abortTrigger = switch (lockTimeout) {
+        null => null,
+        final duration => Future.delayed(duration),
+      };
+
+      return await _database.requestLock(abortTrigger: abortTrigger,
+          (token) async {
+        final context = _UnscopedContext(this, token);
+        try {
+          return await callback(context);
+        } finally {
+          if (flush) {
+            await this.flush();
+          }
         }
-        await _database.customRequest(
-            CustomDatabaseMessage(CustomDatabaseMessageKind.releaseLock));
-      }
+      });
     }
   }
 
@@ -184,9 +195,20 @@ class WebDatabase
 final class _UnscopedContext extends UnscopedContext {
   final WebDatabase _database;
 
+  /// If this context is scoped to a lock on the database, the [LockToken] from
+  /// `package:sqlite3_web`.
+  ///
+  /// This token needs to be passed to queries to run them. While a lock token
+  /// exists on the database, all queries not passing that token are blocked.
+  final LockToken? _lock;
+
   final TimelineTask? _task;
 
-  _UnscopedContext(this._database)
+  /// Whether statements should be rejected if the database is not in an
+  /// autocommit state.
+  bool _checkInTransaction = false;
+
+  _UnscopedContext(this._database, this._lock)
       : _task = _database.profileQueries ? TimelineTask() : null;
 
   @override
@@ -213,8 +235,15 @@ final class _UnscopedContext extends UnscopedContext {
       sql: sql,
       parameters: parameters,
       () async {
-        return await wrapSqliteException(
-            () => _database._database.select(sql, parameters));
+        return await wrapSqliteException(() async {
+          final result = await _database._database.select(
+            sql,
+            parameters: parameters,
+            token: _lock,
+            checkInTransaction: _checkInTransaction,
+          );
+          return result.result;
+        });
       },
     );
   }
@@ -234,8 +263,15 @@ final class _UnscopedContext extends UnscopedContext {
   @override
   Future<ResultSet> execute(String sql, [List<Object?> parameters = const []]) {
     return _task.timeAsync('execute', sql: sql, parameters: parameters, () {
-      return wrapSqliteException(
-          () => _database._database.select(sql, parameters));
+      return wrapSqliteException(() async {
+        final result = await _database._database.select(
+          sql,
+          parameters: parameters,
+          token: _lock,
+          checkInTransaction: _checkInTransaction,
+        );
+        return result.result;
+      });
     });
   }
 
@@ -246,7 +282,12 @@ final class _UnscopedContext extends UnscopedContext {
         for (final set in parameterSets) {
           // use execute instead of select to avoid transferring rows from the
           // worker to this context.
-          await _database._database.execute(sql, set);
+          await _database._database.execute(
+            sql,
+            parameters: set,
+            token: _lock,
+            checkInTransaction: _checkInTransaction,
+          );
         }
       });
     });
@@ -256,75 +297,7 @@ final class _UnscopedContext extends UnscopedContext {
   UnscopedContext interceptOutermostTransaction() {
     // All execute calls done in the callback will be checked for the
     // autocommit state
-    return _ExclusiveTransactionContext(_database);
-  }
-}
-
-final class _ExclusiveTransactionContext extends _UnscopedContext {
-  _ExclusiveTransactionContext(super._database);
-
-  Future<ResultSet> _executeInternal(
-      String sql, List<Object?> parameters) async {
-    // Operations inside transactions are executed with custom requests
-    // in order to verify that the connection does not have autocommit enabled.
-    // The worker will check if autocommit = true before executing the SQL.
-    // An exception will be thrown if autocommit is enabled.
-    // The custom request which does the above will return the ResultSet as a formatted
-    // JavaScript object. This is the converted into a Dart ResultSet.
-    return await wrapSqliteException(() async {
-      var res = await _database._database.customRequest(CustomDatabaseMessage(
-              CustomDatabaseMessageKind.executeInTransaction, sql, parameters))
-          as JSObject;
-
-      if (res.has('format') && (res['format'] as JSNumber).toDartInt == 2) {
-        // Newer workers use a serialization format more efficient than dartify().
-        return proto.deserializeResultSet(res['r'] as JSObject);
-      }
-
-      var result = Map<String, dynamic>.from(res.dartify() as Map);
-      final columnNames = [
-        for (final entry in result['columnNames']) entry as String
-      ];
-      final rawTableNames = result['tableNames'];
-      final tableNames = rawTableNames != null
-          ? [
-              for (final entry in (rawTableNames as List<Object?>))
-                entry as String
-            ]
-          : null;
-
-      final rows = <List<Object?>>[];
-      for (final row in (result['rows'] as List<Object?>)) {
-        final dartRow = <Object?>[];
-
-        for (final column in (row as List<Object?>)) {
-          dartRow.add(column);
-        }
-
-        rows.add(dartRow);
-      }
-      final resultSet = ResultSet(columnNames, tableNames, rows);
-      return resultSet;
-    });
-  }
-
-  @override
-  Future<ResultSet> execute(String sql,
-      [List<Object?> parameters = const []]) async {
-    return _task.timeAsync('execute', sql: sql, parameters: parameters, () {
-      return _executeInternal(sql, parameters);
-    });
-  }
-
-  @override
-  Future<void> executeBatch(
-      String sql, List<List<Object?>> parameterSets) async {
-    return _task.timeAsync('executeBatch', sql: sql, () async {
-      for (final set in parameterSets) {
-        await _database._database.customRequest(CustomDatabaseMessage(
-            CustomDatabaseMessageKind.executeBatchInTransaction, sql, set));
-      }
-    });
+    return _UnscopedContext(_database, _lock).._checkInTransaction = true;
   }
 }
 
@@ -337,6 +310,11 @@ Future<T> wrapSqliteException<T>(Future<T> Function() callback) async {
       throw serializedCause;
     }
 
+    if (ex.message.contains('Database is not in a transaction')) {
+      throw SqliteException(
+          0, "Transaction rolled back by earlier statement. Cannot execute.");
+    }
+
     // Older versions of package:sqlite_web reported SqliteExceptions as strings
     // only.
     if (ex.toString().contains('SqliteException')) {
diff --git a/packages/sqlite_async/lib/src/web/protocol.dart b/packages/sqlite_async/lib/src/web/protocol.dart
index d17c06b..9fcbb57 100644
--- a/packages/sqlite_async/lib/src/web/protocol.dart
+++ b/packages/sqlite_async/lib/src/web/protocol.dart
@@ -6,12 +6,8 @@ import 'dart:js_interop';
 import 'package:sqlite3_web/protocol_utils.dart' as proto;
 
 enum CustomDatabaseMessageKind {
-  requestSharedLock,
-  requestExclusiveLock,
-  releaseLock,
-  lockObtained,
+  ok,
   getAutoCommit,
-  executeInTransaction,
   executeBatchInTransaction,
   updateSubscriptionManagement,
   notifyUpdates,
diff --git a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
index 059c281..164c8ff 100644
--- a/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
+++ b/packages/sqlite_async/lib/src/web/worker/worker_utils.dart
@@ -1,6 +1,5 @@
 import 'dart:async';
 import 'dart:js_interop';
-import 'dart:js_interop_unsafe';
 
 import 'package:meta/meta.dart';
 import 'package:mutex/mutex.dart';
@@ -62,12 +61,6 @@ class AsyncSqliteDatabase extends WorkerDatabase {
     return _state.putIfAbsent(connection, _ConnectionState.new);
   }
 
-  void _markHoldsMutex(ClientConnection connection) {
-    final state = _findState(connection);
-    state.holdsMutex = true;
-    _registerCloseListener(state, connection);
-  }
-
   void _registerCloseListener(
       _ConnectionState state, ClientConnection connection) {
     if (!state.hasOnCloseListener) {
@@ -87,44 +80,11 @@ class AsyncSqliteDatabase extends WorkerDatabase {
     final message = request as CustomDatabaseMessage;
 
     switch (message.kind) {
-      case CustomDatabaseMessageKind.requestSharedLock:
-        await mutex.acquireRead();
-        _markHoldsMutex(connection);
-      case CustomDatabaseMessageKind.requestExclusiveLock:
-        await mutex.acquireWrite();
-        _markHoldsMutex(connection);
-      case CustomDatabaseMessageKind.releaseLock:
-        _findState(connection).holdsMutex = false;
-        mutex.release();
-      case CustomDatabaseMessageKind.lockObtained:
+      case CustomDatabaseMessageKind.ok:
       case CustomDatabaseMessageKind.notifyUpdates:
         throw UnsupportedError('This is a response, not a request');
       case CustomDatabaseMessageKind.getAutoCommit:
         return database.autocommit.toJS;
-      case CustomDatabaseMessageKind.executeInTransaction:
-        final sql = message.rawSql.toDart;
-        final hasTypeInfo = message.typeInfo.isDefinedAndNotNull;
-        final parameters = proto.deserializeParameters(
-            message.rawParameters, message.typeInfo);
-        if (database.autocommit) {
-          throw SqliteException(0,
-              "Transaction rolled back by earlier statement. Cannot execute: $sql");
-        }
-
-        var res = database.select(sql, parameters);
-        if (hasTypeInfo) {
-          // If the client is sending a request that has parameters with type
-          // information, it will also support a newer serialization format for
-          // result sets.
-          return JSObject()
-            ..['format'] = 2.toJS
-            ..['r'] = proto.serializeResultSet(res);
-        } else {
-          var dartMap = resultSetToMap(res);
-          var jsObject = dartMap.jsify();
-          return jsObject;
-        }
-
       case CustomDatabaseMessageKind.executeBatchInTransaction:
         final sql = message.rawSql.toDart;
         final parameters = proto.deserializeParameters(
@@ -157,7 +117,7 @@ class AsyncSqliteDatabase extends WorkerDatabase {
         }
     }
 
-    return CustomDatabaseMessage(CustomDatabaseMessageKind.lockObtained);
+    return CustomDatabaseMessage(CustomDatabaseMessageKind.ok);
   }
 
   Map<String, dynamic> resultSetToMap(ResultSet resultSet) {
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index a5589d8..3be8dd9 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -12,8 +12,8 @@ topics:
   - flutter
 
 dependencies:
-  sqlite3: ^2.9.0
-  sqlite3_web: ^0.3.2
+  sqlite3: ^2.9.4
+  sqlite3_web: ^0.4.0
   async: ^2.10.0
   collection: ^1.17.0
   mutex: ^3.1.0

From 07ace505ff850f2410d4fff3f6562bbbde5e8559 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Thu, 30 Oct 2025 17:23:12 +0100
Subject: [PATCH 18/20] Skip second mutex for OPFS implementations

---
 .../lib/src/web/web_sqlite_open_factory.dart    | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/packages/sqlite_async/lib/src/web/web_sqlite_open_factory.dart b/packages/sqlite_async/lib/src/web/web_sqlite_open_factory.dart
index 513b1f2..29248d5 100644
--- a/packages/sqlite_async/lib/src/web/web_sqlite_open_factory.dart
+++ b/packages/sqlite_async/lib/src/web/web_sqlite_open_factory.dart
@@ -63,10 +63,19 @@ class DefaultSqliteOpenFactory
     final workers = await _initialized;
     final connection = await connectToWorker(workers, path);
 
-    // When the database is accessed through a shared worker, we implement
-    // mutexes over custom messages sent through the shared worker. In other
-    // cases, we need to implement a mutex locally.
-    final mutex = connection.access == AccessMode.throughSharedWorker
+    // When the database is hosted in a shared worker, we don't need a local
+    // mutex since that worker will hand out leases for us.
+    // Additionally, package:sqlite3_web uses navigator locks internally for
+    // OPFS databases.
+    // Technically, the only other implementation (IndexedDB in a local context
+    // or a dedicated worker) is inherently unsafe to use across tabs. But
+    // wrapping those in a mutex and flushing the file system helps a little bit
+    // (still something we're trying to avoid).
+    final hasSqliteWebMutex =
+        connection.access == AccessMode.throughSharedWorker ||
+            connection.storage == StorageMode.opfs;
+
+    final mutex = hasSqliteWebMutex
         ? null
         : MutexImpl(identifier: path); // Use the DB path as a mutex identifier
 

From 9d21581defc03809724e353b17fb15275b9d1ee6 Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Tue, 4 Nov 2025 09:12:15 +0100
Subject: [PATCH 19/20] chore(release): publish packages

 - sqlite_async@0.13.0
 - drift_sqlite_async@0.2.6
---
 CHANGELOG.md                             | 25 ++++++++++++++++++++++++
 packages/drift_sqlite_async/CHANGELOG.md |  4 ++++
 packages/drift_sqlite_async/pubspec.yaml |  4 ++--
 packages/sqlite_async/CHANGELOG.md       |  4 ++++
 packages/sqlite_async/pubspec.yaml       |  2 +-
 5 files changed, 36 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 5181152..e84e261 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -3,6 +3,31 @@
 All notable changes to this project will be documented in this file.
 See [Conventional Commits](https://conventionalcommits.org) for commit guidelines.
 
+## 2025-11-04
+
+### Changes
+
+---
+
+Packages with breaking changes:
+
+ - There are no breaking changes in this release.
+
+Packages with other changes:
+
+ - [`sqlite_async` - `v0.13.0`](#sqlite_async---v0130)
+ - [`drift_sqlite_async` - `v0.2.6`](#drift_sqlite_async---v026)
+
+---
+
+#### `sqlite_async` - `v0.13.0`
+
+ - Update sqlite3_web to 0.4.0
+
+#### `drift_sqlite_async` - `v0.2.6`
+
+- Support latest `sqlite_async`.
+
 ## 2025-10-13
 
 ### Changes
diff --git a/packages/drift_sqlite_async/CHANGELOG.md b/packages/drift_sqlite_async/CHANGELOG.md
index bd36dd1..bd780e9 100644
--- a/packages/drift_sqlite_async/CHANGELOG.md
+++ b/packages/drift_sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.2.6
+
+- Support latest `sqlite_async`.
+
 ## 0.2.5
 
  - Allow customizing update notifications from `sqlite_async`.
diff --git a/packages/drift_sqlite_async/pubspec.yaml b/packages/drift_sqlite_async/pubspec.yaml
index bd2b701..dc19fe7 100644
--- a/packages/drift_sqlite_async/pubspec.yaml
+++ b/packages/drift_sqlite_async/pubspec.yaml
@@ -1,5 +1,5 @@
 name: drift_sqlite_async
-version: 0.2.5
+version: 0.2.6
 homepage: https://github.com/powersync-ja/sqlite_async.dart
 repository: https://github.com/powersync-ja/sqlite_async.dart
 description: Use Drift with a sqlite_async database, allowing both to be used in the same application.
@@ -15,7 +15,7 @@ environment:
   sdk: ">=3.0.0 <4.0.0"
 dependencies:
   drift: ">=2.28.0 <3.0.0"
-  sqlite_async: ^0.12.2
+  sqlite_async: ^0.13.0
 
 dev_dependencies:
   build_runner: ^2.4.8
diff --git a/packages/sqlite_async/CHANGELOG.md b/packages/sqlite_async/CHANGELOG.md
index 1bbc939..d833558 100644
--- a/packages/sqlite_async/CHANGELOG.md
+++ b/packages/sqlite_async/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 0.13.0
+
+ - Update sqlite3_web to 0.4.0
+
 ## 0.12.2
 
  - Add `withAllConnections` method to run statements on all connections in the pool.
diff --git a/packages/sqlite_async/pubspec.yaml b/packages/sqlite_async/pubspec.yaml
index 3be8dd9..1ddd431 100644
--- a/packages/sqlite_async/pubspec.yaml
+++ b/packages/sqlite_async/pubspec.yaml
@@ -1,6 +1,6 @@
 name: sqlite_async
 description: High-performance asynchronous interface for SQLite on Dart and Flutter.
-version: 0.12.2
+version: 0.13.0
 repository: https://github.com/powersync-ja/sqlite_async.dart
 environment:
   sdk: ">=3.5.0 <4.0.0"

From b8b9a657b1983a854227c661be6f9c16a96556ca Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Tue, 4 Nov 2025 09:36:43 +0100
Subject: [PATCH 20/20] Document breaking worker change

---
 CHANGELOG.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index e84e261..0342a48 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -11,7 +11,7 @@ See [Conventional Commits](https://conventionalcommits.org) for commit guideline
 
 Packages with breaking changes:
 
- - There are no breaking changes in this release.
+ - `sqlite_async` uses version 0.4.x of `sqlite3_web`, which requires a new worker.
 
 Packages with other changes: