Sqlite Prototype

.github/workflows/test.yaml

@@ -3,7 +3,7 @@ name: Test
 on:
   push:
     branches:
-      - "**"
+      - '**'
 
 jobs:
   build:
@@ -57,10 +57,10 @@ jobs:
       - name: Install SQLite
         run: |
           ./scripts/install_sqlite.sh ${{ matrix.sqlite_version }} ${{ matrix.sqlite_url }}
-          mkdir -p assets && curl -LJ https://github.com/simolus3/sqlite3.dart/releases/download/sqlite3-2.3.0/sqlite3.wasm -o assets/sqlite3.wasm
+          mkdir -p assets && curl -LJ https://github.com/simolus3/sqlite3.dart/releases/download/sqlite3-2.4.3/sqlite3.wasm -o assets/sqlite3.wasm
 
       - name: Compile WebWorker
-        run: dart compile js -o assets/db_worker.js -O0 lib/src/web/worker/drift_worker.dart
+        run: dart compile js -o assets/db_worker.js -O0 lib/src/web/worker/worker.dart
 
       - name: Run Tests
         run: |

DEVELOPING.md

@@ -0,0 +1,9 @@
+# Developing Instructions
+
+## Testing
+
+Running tests for the `web` platform requires some preparation to be executed. The `sqlite3.wasm` and `db_worker.js` files need to be available in the Git ignored `./assets` folder.
+
+See the [test action](./.github/workflows/test.yaml) for the latest steps.
+
+On your local machine run the commands from the `Install SQLite`, `Compile WebWorker` and `Run Tests` steps.

lib/sqlite3_wasm.dart

@@ -0,0 +1,5 @@
+/// Re-exports [sqlite3 WASM](https://pub.dev/packages/sqlite3) to expose sqlite3 without
+/// adding it as a direct dependency.
+library;
+
+export 'package:sqlite3/wasm.dart';

lib/sqlite3_web.dart

@@ -0,0 +1,5 @@
+/// Re-exports [sqlite3_web](https://pub.dev/packages/sqlite3_web) to expose sqlite3_web without
+/// adding it as a direct dependency.
+library;
+
+export 'package:sqlite3_web/sqlite3_web.dart';

lib/sqlite3_web_worker.dart

@@ -0,0 +1,3 @@
+// Items required for extending custom web workers
+export './src/web/worker/worker_utils.dart';
+export './src/web/protocol.dart';

lib/src/common/sqlite_database.dart

@@ -24,9 +24,6 @@ mixin SqliteDatabaseMixin implements SqliteConnection, SqliteQueries {
   @override
   Stream<UpdateNotification> get updates;
 
-  final StreamController<UpdateNotification> updatesController =
-      StreamController.broadcast();
-
   @protected
   Future<void> get isInitialized;
 

lib/src/native/database/native_sqlite_connection_impl.dart

@@ -42,6 +42,7 @@ class SqliteConnectionImpl
       this.readOnly = false,
       bool primary = false})
       : _writeMutex = mutex {
+    isInitialized = _isolateClient.ready;
     this.upstreamPort = upstreamPort ?? listenForEvents();
     // Accept an incoming stream of updates, or expose one if not given.
     this.updates = updates ?? updatesController.stream;
@@ -88,7 +89,6 @@ class SqliteConnectionImpl
           paused: true);
       _isolateClient.tieToIsolate(_isolate);
       _isolate.resume(_isolate.pauseCapability!);
-      isInitialized = _isolateClient.ready;
       await _isolateClient.ready;
     });
   }

lib/src/native/database/native_sqlite_database.dart

@@ -40,6 +40,9 @@ class SqliteDatabaseImpl
   late final SqliteConnectionImpl _internalConnection;
   late final SqliteConnectionPool _pool;
 
+  final StreamController<UpdateNotification> updatesController =
+      StreamController.broadcast();
+
   /// Open a SqliteDatabase.
   ///
   /// Only a single SqliteDatabase per [path] should be opened at a time.

lib/src/web/database.dart

@@ -0,0 +1,282 @@
+import 'dart:async';
+import 'dart:js_interop';
+
+import 'package:sqlite3/common.dart';
+import 'package:sqlite3_web/sqlite3_web.dart';
+import 'package:sqlite_async/mutex.dart';
+import 'package:sqlite_async/sqlite_async.dart';
+import 'package:sqlite_async/src/common/sqlite_database.dart';
+import 'package:sqlite_async/src/sqlite_connection.dart';
+import 'package:sqlite_async/src/sqlite_queries.dart';
+import 'package:sqlite_async/src/update_notification.dart';
+import 'package:sqlite_async/src/utils/shared_utils.dart';
+import 'protocol.dart';
+
+class WebDatabase
+    with SqliteQueries, SqliteDatabaseMixin
+    implements SqliteDatabase {
+  final Database _database;
+  final Mutex? _mutex;
+
+  @override
+  bool closed = false;
+
+  WebDatabase(this._database, this._mutex);
+
+  @override
+  Future<void> close() async {
+    await _database.dispose();
+    closed = true;
+  }
+
+  @override
+  Future<bool> getAutoCommit() async {
+    final response = await _database.customRequest(
+        CustomDatabaseMessage(CustomDatabaseMessageKind.getAutoCommit));
+    return (response as JSBoolean?)?.toDart ?? false;
+  }
+
+  @override
+  Future<void> initialize() {
+    return Future.value();
+  }
+
+  @override
+  Future<void> get isInitialized => initialize();
+
+  @override
+
+  /// Not relevant for web.
+  Never isolateConnectionFactory() {
+    throw UnimplementedError();
+  }
+
+  @override
+
+  /// Not supported on web. There is only 1 connection.
+  int get maxReaders => throw UnimplementedError();
+
+  @override
+
+  /// Not relevant for web.
+  Never get openFactory => throw UnimplementedError();
+
+  @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(() async {
+        final context = _SharedContext(this);
+        try {
+          return await callback(context);
+        } finally {
+          context.markClosed();
+        }
+      });
+    } else {
+      // No custom mutex, coordinate locks through shared worker.
+      await _database.customRequest(
+          CustomDatabaseMessage(CustomDatabaseMessageKind.requestSharedLock));
+
+      try {
+        return await callback(_SharedContext(this));
+      } finally {
+        await _database.customRequest(
+            CustomDatabaseMessage(CustomDatabaseMessageKind.releaseLock));
+      }
+    }
+  }
+
+  @override
+  Stream<UpdateNotification> get updates =>
+      _database.updates.map((event) => UpdateNotification({event.tableName}));
+
+  @override
+  Future<T> writeTransaction<T>(
+      Future<T> Function(SqliteWriteContext tx) callback,
+      {Duration? lockTimeout}) {
+    return writeLock(
+        (writeContext) =>
+            internalWriteTransaction(writeContext, (context) async {
+              // All execute calls done in the callback will be checked for the
+              // autocommit state
+              return callback(_ExclusiveTransactionContext(this, writeContext));
+            }),
+        debugContext: 'writeTransaction()',
+        lockTimeout: lockTimeout);
+  }
+
+  @override
+
+  /// Internal writeLock which intercepts transaction context's to verify auto commit is not active
+  Future<T> writeLock<T>(Future<T> Function(SqliteWriteContext tx) callback,
+      {Duration? lockTimeout, String? debugContext}) async {
+    if (_mutex case var mutex?) {
+      return await mutex.lock(() async {
+        final context = _ExclusiveContext(this);
+        try {
+          return await callback(context);
+        } finally {
+          context.markClosed();
+        }
+      });
+    } else {
+      // No custom mutex, coordinate locks through shared worker.
+      await _database.customRequest(CustomDatabaseMessage(
+          CustomDatabaseMessageKind.requestExclusiveLock));
+      final context = _ExclusiveContext(this);
+      try {
+        return await callback(context);
+      } finally {
+        context.markClosed();
+        await _database.customRequest(
+            CustomDatabaseMessage(CustomDatabaseMessageKind.releaseLock));
+      }
+    }
+  }
+}
+
+class _SharedContext implements SqliteReadContext {
+  final WebDatabase _database;
+  bool _contextClosed = false;
+
+  _SharedContext(this._database);
+
+  @override
+  bool get closed => _contextClosed || _database.closed;
+
+  @override
+  Future<T> computeWithDatabase<T>(
+      Future<T> Function(CommonDatabase db) compute) {
+    // Can't be implemented: The database may live on another worker.
+    throw UnimplementedError();
+  }
+
+  @override
+  Future<Row> get(String sql, [List<Object?> parameters = const []]) async {
+    final results = await getAll(sql, parameters);
+    return results.single;
+  }
+
+  @override
+  Future<ResultSet> getAll(String sql,
+      [List<Object?> parameters = const []]) async {
+    return await wrapSqliteException(
+        () => _database._database.select(sql, parameters));
+  }
+
+  @override
+  Future<bool> getAutoCommit() async {
+    return _database.getAutoCommit();
+  }
+
+  @override
+  Future<Row?> getOptional(String sql,
+      [List<Object?> parameters = const []]) async {
+    final results = await getAll(sql, parameters);
+    return results.singleOrNull;
+  }
+
+  void markClosed() {
+    _contextClosed = true;
+  }
+}
+
+class _ExclusiveContext extends _SharedContext implements SqliteWriteContext {
+  _ExclusiveContext(super.database);
+
+  @override
+  Future<ResultSet> execute(String sql,
+      [List<Object?> parameters = const []]) async {
+    return wrapSqliteException(
+        () => _database._database.select(sql, parameters));
+  }
+
+  @override
+  Future<void> executeBatch(
+      String sql, List<List<Object?>> parameterSets) async {
+    return wrapSqliteException(() async {
+      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);
+      }
+    });
+  }
+}
+
+class _ExclusiveTransactionContext extends _ExclusiveContext {
+  SqliteWriteContext baseContext;
+  _ExclusiveTransactionContext(super.database, this.baseContext);
+
+  @override
+  bool get closed => baseContext.closed;
+
+  @override
+  Future<ResultSet> execute(String sql,
+      [List<Object?> parameters = const []]) 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));
+      var result =
+          Map<String, dynamic>.from((res as JSObject).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<void> executeBatch(
+      String sql, List<List<Object?>> parameterSets) async {
+    return await wrapSqliteException(() async {
+      for (final set in parameterSets) {
+        await _database._database.customRequest(CustomDatabaseMessage(
+            CustomDatabaseMessageKind.executeBatchInTransaction, sql, set));
+      }
+      return;
+    });
+  }
+}
+
+/// Throws SqliteException if the Remote Exception is a SqliteException
+Future<T> wrapSqliteException<T>(Future<T> Function() callback) async {
+  try {
+    return await callback();
+  } on RemoteException catch (ex) {
+    if (ex.toString().contains('SqliteException')) {
+      RegExp regExp = RegExp(r'SqliteException\((\d+)\)');
+      // Drift wraps these in remote errors
+      throw SqliteException(
+          int.parse(regExp.firstMatch(ex.message)?.group(1) ?? '0'),
+          ex.message);
+    }
+    rethrow;
+  }
+}