Merge pull request simolus3#275 from powersync-ja/sqlite3-web-without-workers

`sqlite3_web`: Support opening databases without workers

Author: simolus3

sqlite3_web/CHANGELOG.md

@@ -1,6 +1,7 @@
 ## 0.2.1
 
 - Add `WebSqlite.deleteDatabase` to delete databases.
+- Support opening databases without workers.
 
 ## 0.2.0
 

sqlite3_web/lib/src/client.dart

@@ -227,15 +227,19 @@ final class WorkerConnection extends ProtocolChannel {
 final class DatabaseClient implements WebSqlite {
   final Uri workerUri;
   final Uri wasmUri;
+  final DatabaseController _localController;
 
   final Lock _startWorkersLock = Lock();
   bool _startedWorkers = false;
   WorkerConnection? _connectionToDedicated;
   WorkerConnection? _connectionToShared;
   WorkerConnection? _connectionToDedicatedInShared;
+
+  WorkerConnection? _connectionToLocal;
+
   final Set<MissingBrowserFeature> _missingFeatures = {};
 
-  DatabaseClient(this.workerUri, this.wasmUri);
+  DatabaseClient(this.workerUri, this.wasmUri, this._localController);
 
   Future<void> startWorkers() {
     return _startWorkersLock.synchronized(() async {
@@ -244,36 +248,52 @@ final class DatabaseClient implements WebSqlite {
       }
       _startedWorkers = true;
 
-      if (globalContext.has('Worker')) {
-        final dedicated = Worker(
+      await _startDedicated();
+      await _startShared();
+    });
+  }
+
+  Future<void> _startDedicated() async {
+    if (globalContext.has('Worker')) {
+      final Worker dedicated;
+      try {
+        dedicated = Worker(
           workerUri.toString().toJS,
           WorkerOptions(name: 'sqlite3_worker'),
         );
-
-        final (endpoint, channel) = await createChannel();
-        ConnectRequest(endpoint: endpoint, requestId: 0)
-            .sendToWorker(dedicated);
-
-        _connectionToDedicated =
-            WorkerConnection(channel.injectErrorsFrom(dedicated));
-      } else {
+      } on Object {
         _missingFeatures.add(MissingBrowserFeature.dedicatedWorkers);
+        return;
       }
 
-      if (globalContext.has('SharedWorker')) {
-        final shared = SharedWorker(workerUri.toString().toJS);
-        shared.port.start();
+      final (endpoint, channel) = await createChannel();
+      ConnectRequest(endpoint: endpoint, requestId: 0).sendToWorker(dedicated);
 
-        final (endpoint, channel) = await createChannel();
-        ConnectRequest(endpoint: endpoint, requestId: 0)
-            .sendToPort(shared.port);
+      _connectionToDedicated =
+          WorkerConnection(channel.injectErrorsFrom(dedicated));
+    } else {
+      _missingFeatures.add(MissingBrowserFeature.dedicatedWorkers);
+    }
+  }
 
-        _connectionToShared =
-            WorkerConnection(channel.injectErrorsFrom(shared));
-      } else {
+  Future<void> _startShared() async {
+    if (globalContext.has('SharedWorker')) {
+      final SharedWorker shared;
+      try {
+        shared = SharedWorker(workerUri.toString().toJS);
+        shared.port.start();
+      } on Object {
         _missingFeatures.add(MissingBrowserFeature.sharedWorkers);
+        return;
       }
-    });
+
+      final (endpoint, channel) = await createChannel();
+      ConnectRequest(endpoint: endpoint, requestId: 0).sendToPort(shared.port);
+
+      _connectionToShared = WorkerConnection(channel.injectErrorsFrom(shared));
+    } else {
+      _missingFeatures.add(MissingBrowserFeature.sharedWorkers);
+    }
   }
 
   Future<WorkerConnection> _connectToDedicatedInShared() {
@@ -291,6 +311,22 @@ final class DatabaseClient implements WebSqlite {
     });
   }
 
+  Future<WorkerConnection> _connectToLocal() async {
+    return _startWorkersLock.synchronized(() async {
+      if (_connectionToLocal case final conn?) {
+        return conn;
+      }
+
+      final local = Local();
+      final (endpoint, channel) = await createChannel();
+      WorkerRunner(_localController, environment: local).handleRequests();
+      local
+          .addTopLevelMessage(ConnectRequest(requestId: 0, endpoint: endpoint));
+
+      return _connectionToLocal = WorkerConnection(channel);
+    });
+  }
+
   @override
   Future<void> deleteDatabase(
       {required String name, required StorageMode storage}) async {
@@ -310,6 +346,7 @@ final class DatabaseClient implements WebSqlite {
 
     final existing = <ExistingDatabase>{};
     final available = <(StorageMode, AccessMode)>[];
+    var workersReportedIndexedDbSupport = false;
 
     if (_connectionToDedicated case final connection?) {
       final response = await connection.sendRequest(
@@ -327,6 +364,8 @@ final class DatabaseClient implements WebSqlite {
       if (result.canUseIndexedDb) {
         available
             .add((StorageMode.indexedDb, AccessMode.throughDedicatedWorker));
+
+        workersReportedIndexedDbSupport = true;
       } else {
         _missingFeatures.add(MissingBrowserFeature.indexedDb);
       }
@@ -363,6 +402,7 @@ final class DatabaseClient implements WebSqlite {
       final result = CompatibilityResult.fromJS(response.response as JSObject);
 
       if (result.canUseIndexedDb) {
+        workersReportedIndexedDbSupport = true;
         available.add((StorageMode.indexedDb, AccessMode.throughSharedWorker));
       } else {
         _missingFeatures.add(MissingBrowserFeature.indexedDb);
@@ -383,6 +423,12 @@ final class DatabaseClient implements WebSqlite {
       }
     }
 
+    available.add((StorageMode.inMemory, AccessMode.inCurrentContext));
+    if (workersReportedIndexedDbSupport || await checkIndexedDbSupport()) {
+      // If the workers can use IndexedDb, so can we.
+      available.add((StorageMode.indexedDb, AccessMode.inCurrentContext));
+    }
+
     return FeatureDetectionResult(
       missingFeatures: _missingFeatures.toList(),
       existingDatabases: existing.toList(),
@@ -425,7 +471,8 @@ final class DatabaseClient implements WebSqlite {
         connection = _connectionToDedicated!;
         shared = false;
       case AccessMode.inCurrentContext:
-        throw UnimplementedError('todo: Open database locally');
+        connection = await _connectToLocal();
+        shared = false;
     }
 
     final response = await connection.sendRequest(

sqlite3_web/lib/src/database.dart

@@ -9,6 +9,9 @@ import 'worker.dart';
 
 /// A controller responsible for opening databases in the worker.
 abstract base class DatabaseController {
+  /// Constant base constructor.
+  const DatabaseController();
+
   /// Loads a wasm module from the given [uri] with the specified [headers].
   Future<WasmSqlite3> loadWasmModule(Uri uri,
       {Map<String, String>? headers}) async {
@@ -199,11 +202,17 @@ abstract class WebSqlite {
 
   /// Opens a [WebSqlite] instance by connecting to the given [worker] and
   /// using the [wasmModule] url to load sqlite3.
+  ///
+  /// The [controller] is used when connecting to a sqlite3 database without
+  /// using workers. It should typically be the same implementation as the one
+  /// passed to [workerEntrypoint].
   static WebSqlite open({
     required Uri worker,
     required Uri wasmModule,
+    DatabaseController? controller,
   }) {
-    return DatabaseClient(worker, wasmModule);
+    return DatabaseClient(
+        worker, wasmModule, controller ?? const _DefaultDatabaseController());
   }
 
   /// Connects to an endpoint previously obtained with [Database.additionalConnection].
@@ -218,7 +227,37 @@ abstract class WebSqlite {
   /// was called. This limitation does not exist for databases hosted by shared
   /// workers.
   static Future<Database> connectToPort(SqliteWebEndpoint endpoint) {
-    final client = DatabaseClient(Uri.base, Uri.base);
+    final client =
+        DatabaseClient(Uri.base, Uri.base, const _DefaultDatabaseController());
     return client.connectToExisting(endpoint);
   }
 }
+
+final class _DefaultDatabaseController extends DatabaseController {
+  const _DefaultDatabaseController();
+
+  @override
+  Future<JSAny?> handleCustomRequest(
+      ClientConnection connection, JSAny? request) {
+    throw UnimplementedError();
+  }
+
+  @override
+  Future<WorkerDatabase> openDatabase(
+      WasmSqlite3 sqlite3, String path, String vfs) async {
+    return _DefaultWorkerDatabase(sqlite3.open(path, vfs: vfs));
+  }
+}
+
+final class _DefaultWorkerDatabase extends WorkerDatabase {
+  @override
+  final CommonDatabase database;
+
+  _DefaultWorkerDatabase(this.database);
+
+  @override
+  Future<JSAny?> handleCustomRequest(
+      ClientConnection connection, JSAny? request) {
+    throw UnimplementedError();
+  }
+}

sqlite3_web/lib/src/worker.dart

@@ -105,6 +105,27 @@ final class Shared extends WorkerEnvironment {
   }
 }
 
+/// A fake worker environment running in the same context as the main
+/// application.
+///
+/// This allows using a communication channel based on message ports regardless
+/// of where the database is hosted. While that adds overhead, a local
+/// environment is only used as a fallback if workers are unavailable.
+final class Local extends WorkerEnvironment {
+  final StreamController<Message> _messages = StreamController();
+
+  Local() : super._();
+
+  void addTopLevelMessage(Message message) {
+    _messages.add(message);
+  }
+
+  @override
+  Stream<Message> get topLevelRequests {
+    return _messages.stream;
+  }
+}
+
 /// A database opened by a client.
 final class _ConnectionDatabase {
   final DatabaseState database;
@@ -429,7 +450,8 @@ final class WorkerRunner {
   /// a shared context that can use synchronous JS APIs.
   Worker? _innerWorker;
 
-  WorkerRunner(this._controller) : _environment = WorkerEnvironment();
+  WorkerRunner(this._controller, {WorkerEnvironment? environment})
+      : _environment = environment ?? WorkerEnvironment();
 
   void handleRequests() async {
     await for (final message in _environment.topLevelRequests) {

sqlite3_web/test/integration_test.dart

@@ -45,8 +45,13 @@ enum Browser {
     final available = <(StorageMode, AccessMode)>{};
     for (final storage in StorageMode.values) {
       for (final access in AccessMode.values) {
-        if (access != AccessMode.inCurrentContext &&
-            !unsupportedImplementations.contains((storage, access))) {
+        if (access == AccessMode.inCurrentContext &&
+            storage == StorageMode.opfs) {
+          // OPFS access is only available in workers.
+          continue;
+        }
+
+        if (!unsupportedImplementations.contains((storage, access))) {
           available.add((storage, access));
         }
       }

sqlite3_web/web/main.dart

@@ -6,6 +6,8 @@ import 'dart:typed_data';
 
 import 'package:sqlite3_web/sqlite3_web.dart';
 
+import 'controller.dart';
+
 final sqlite3WasmUri = Uri.parse('sqlite3.wasm');
 final workerUri = Uri.parse('worker.dart.js');
 const databaseName = 'database';
@@ -127,6 +129,7 @@ WebSqlite initializeSqlite() {
   return webSqlite ??= WebSqlite.open(
     worker: workerUri,
     wasmModule: sqlite3WasmUri,
+    controller: ExampleController(isInWorker: false),
   );
 }