Refactor sync client and options

Author: simolus3

demos/benchmarks/lib/powersync.dart

@@ -88,13 +88,15 @@ Future<String> getDatabasePath() async {
 
 var currentConnector = BackendConnector();
 
+const options = SyncOptions(
+  params: {'size_bucket': AppConfig.sizeBucket},
+  crudThrottleTime: Duration(milliseconds: 1),
+);
+
 Future<void> resync() async {
   await db.disconnectAndClear();
   timer.start(db);
-  db.connect(
-      connector: currentConnector,
-      params: {'size_bucket': AppConfig.sizeBucket},
-      crudThrottleTime: const Duration(milliseconds: 1));
+  db.connect(connector: currentConnector, options: options);
 }
 
 Future<void> openDatabase() async {
@@ -106,8 +108,5 @@ Future<void> openDatabase() async {
   BenchmarkItem.updateItemBenchmarks();
 
   timer.start(db);
-  db.connect(
-      connector: currentConnector,
-      params: {'size_bucket': AppConfig.sizeBucket},
-      crudThrottleTime: const Duration(milliseconds: 1));
+  db.connect(connector: currentConnector, options: options);
 }

packages/powersync_core/lib/src/database/native/native_powersync_database.dart

@@ -15,6 +15,7 @@ import 'package:powersync_core/src/log_internal.dart';
 import 'package:powersync_core/src/open_factory/abstract_powersync_open_factory.dart';
 import 'package:powersync_core/src/open_factory/native/native_open_factory.dart';
 import 'package:powersync_core/src/schema.dart';
+import 'package:powersync_core/src/sync/internal_connector.dart';
 import 'package:powersync_core/src/sync/options.dart';
 import 'package:powersync_core/src/sync/streaming_sync.dart';
 import 'package:powersync_core/src/sync/sync_status.dart';
@@ -162,16 +163,21 @@ class PowerSyncDatabaseImpl
     Future<void> handleMessage(Object? data) async {
       if (data is List) {
         String action = data[0] as String;
-        if (action == "getCredentials") {
+        if (action == "getCredentialsCached") {
           await (data[1] as PortCompleter).handle(() async {
             final token = await connector.getCredentialsCached();
             logger.fine('Credentials: $token');
             return token;
           });
-        } else if (action == "invalidateCredentials") {
+        } else if (action == "prefetchCredentials") {
           logger.fine('Refreshing credentials');
+          final invalidate = data[2] as bool;
+
           await (data[1] as PortCompleter).handle(() async {
-            await connector.prefetchCredentials();
+            if (invalidate) {
+              connector.invalidateCredentials();
+            }
+            return await connector.prefetchCredentials();
           });
         } else if (action == 'init') {
           final port = initPort = data[1] as SendPort;
@@ -360,15 +366,16 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     sPort.send(['log', copy]);
   });
 
-  Future<PowerSyncCredentials?> loadCredentials() async {
+  Future<PowerSyncCredentials?> getCredentialsCached() async {
     final r = IsolateResult<PowerSyncCredentials?>();
-    sPort.send(['getCredentials', r.completer]);
+    sPort.send(['getCredentialsCached', r.completer]);
     return r.future;
   }
 
-  Future<void> invalidateCredentials() async {
-    final r = IsolateResult<void>();
-    sPort.send(['invalidateCredentials', r.completer]);
+  Future<PowerSyncCredentials?> prefetchCredentials(
+      {required bool invalidate}) async {
+    final r = IsolateResult<PowerSyncCredentials?>();
+    sPort.send(['prefetchCredentials', r.completer, invalidate]);
     return r.future;
   }
 
@@ -386,9 +393,11 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     final storage = BucketStorage(connection);
     final sync = StreamingSyncImplementation(
       adapter: storage,
-      credentialsCallback: loadCredentials,
-      invalidCredentialsCallback: invalidateCredentials,
-      uploadCrud: uploadCrud,
+      connector: InternalConnector(
+        getCredentialsCached: getCredentialsCached,
+        prefetchCredentials: prefetchCredentials,
+        uploadCrud: uploadCrud,
+      ),
       crudUpdateTriggerStream: crudUpdateController.stream,
       options: args.options,
       client: http.Client(),
@@ -426,6 +435,6 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
     // This should be rare - any uncaught error is a bug. And in most cases,
     // it should occur after the database is already open.
     await shutdown();
-    throw error;
+    Error.throwWithStackTrace(error, stack);
   });
 }

packages/powersync_core/lib/src/database/web/web_powersync_database.dart

@@ -11,6 +11,7 @@ import 'package:powersync_core/src/log.dart';
 import 'package:powersync_core/src/open_factory/abstract_powersync_open_factory.dart';
 import 'package:powersync_core/src/open_factory/web/web_open_factory.dart';
 import 'package:powersync_core/src/schema.dart';
+import 'package:powersync_core/src/sync/internal_connector.dart';
 import 'package:powersync_core/src/sync/streaming_sync.dart';
 import 'package:sqlite_async/sqlite_async.dart';
 
@@ -142,9 +143,7 @@ class PowerSyncDatabaseImpl
 
       sync = StreamingSyncImplementation(
         adapter: storage,
-        credentialsCallback: connector.getCredentialsCached,
-        invalidCredentialsCallback: connector.prefetchCredentials,
-        uploadCrud: () => connector.uploadData(this),
+        connector: InternalConnector.wrap(connector, this),
         crudUpdateTriggerStream: crudStream,
         options: resolved,
         client: BrowserClient(),

packages/powersync_core/lib/src/sync/instruction.dart

@@ -0,0 +1,92 @@
+import 'package:meta/meta.dart';
+
+import '../connector.dart';
+import '../database/powersync_database.dart';
+
+/// A view over a backend connector that does not require a reference to the
+/// PowerSync database.
+@internal
+abstract interface class InternalConnector {
+  /// Fetch or return cached credentials.
+  Future<PowerSyncCredentials?> getCredentialsCached();
+
+  /// Ask the backend connector to fetch a new set of credentials.
+  ///
+  /// [invalidate] describes whether the current ([getCredentialsCached])
+  /// credentials are already invalid, or whether this call is a pre-fetch.
+  ///
+  /// A call to [getCredentialsCached] after this future completes should return
+  /// the same credentials.
+  Future<PowerSyncCredentials?> prefetchCredentials({bool invalidate = false});
+
+  /// Requests the connector to upload a crud batch to the backend.
+  Future<void> uploadCrud();
+
+  const factory InternalConnector({
+    required Future<PowerSyncCredentials?> Function() getCredentialsCached,
+    required Future<PowerSyncCredentials?> Function({required bool invalidate})
+        prefetchCredentials,
+    required Future<void> Function() uploadCrud,
+  }) = _CallbackConnector;
+
+  factory InternalConnector.wrap(
+      PowerSyncBackendConnector connector, PowerSyncDatabase db) {
+    return _WrapConnector(connector, db);
+  }
+}
+
+final class _WrapConnector implements InternalConnector {
+  final PowerSyncBackendConnector connector;
+  final PowerSyncDatabase database;
+
+  _WrapConnector(this.connector, this.database);
+
+  @override
+  Future<PowerSyncCredentials?> getCredentialsCached() async {
+    return connector.getCredentialsCached();
+  }
+
+  @override
+  Future<PowerSyncCredentials?> prefetchCredentials({bool invalidate = false}) {
+    if (invalidate) {
+      connector.invalidateCredentials();
+    }
+    return connector.prefetchCredentials();
+  }
+
+  @override
+  Future<void> uploadCrud() {
+    return connector.uploadData(database);
+  }
+}
+
+final class _CallbackConnector implements InternalConnector {
+  final Future<PowerSyncCredentials?> Function() _getCredentialsCached;
+  final Future<PowerSyncCredentials?> Function({required bool invalidate})
+      _prefetchCredentials;
+  final Future<void> Function() _uploadCrud;
+
+  const _CallbackConnector({
+    required Future<PowerSyncCredentials?> Function() getCredentialsCached,
+    required Future<PowerSyncCredentials?> Function({required bool invalidate})
+        prefetchCredentials,
+    required Future<void> Function() uploadCrud,
+  })  : _getCredentialsCached = getCredentialsCached,
+        _prefetchCredentials = prefetchCredentials,
+        _uploadCrud = uploadCrud;
+
+  @override
+  Future<PowerSyncCredentials?> getCredentialsCached() {
+    return _getCredentialsCached();
+  }
+
+  @override
+  Future<PowerSyncCredentials?> prefetchCredentials({bool invalidate = false}) {
+    return _prefetchCredentials(invalidate: invalidate);
+  }
+
+  @override
+  Future<void> uploadCrud() {
+    return _uploadCrud();
+  }
+}