Move options into single class

Author: simolus3

packages/powersync_core/lib/powersync_core.dart

@@ -10,6 +10,7 @@ export 'src/exceptions.dart';
 export 'src/log.dart';
 export 'src/open_factory.dart';
 export 'src/schema.dart';
+export 'src/sync/options.dart' hide ResolvedSyncOptions;
 export 'src/sync/sync_status.dart'
     hide BucketProgress, InternalSyncDownloadProgress;
 export 'src/uuid.dart';

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/options.dart';
 import 'package:powersync_core/src/sync/streaming_sync.dart';
 import 'package:powersync_core/src/sync/sync_status.dart';
 import 'package:sqlite_async/sqlite3_common.dart';
@@ -118,10 +119,9 @@ class PowerSyncDatabaseImpl
   @internal
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
-    required Duration crudThrottleTime,
+    required SyncOptions options,
     required AbortController abort,
     required Zone asyncWorkZone,
-    Map<String, dynamic>? params,
   }) async {
     final dbRef = database.isolateConnectionFactory();
 
@@ -134,6 +134,7 @@ class PowerSyncDatabaseImpl
     SendPort? initPort;
     final hasInitPort = Completer<void>();
     final receivedIsolateExit = Completer<void>();
+    final resolved = ResolvedSyncOptions(options);
 
     Future<void> waitForShutdown() async {
       // Only complete the abortion signal after the isolate shuts down. This
@@ -175,8 +176,8 @@ class PowerSyncDatabaseImpl
         } else if (action == 'init') {
           final port = initPort = data[1] as SendPort;
           hasInitPort.complete();
-          var crudStream =
-              database.onChange(['ps_crud'], throttle: crudThrottleTime);
+          var crudStream = database
+              .onChange(['ps_crud'], throttle: resolved.crudThrottleTime);
           crudUpdateSubscription = crudStream.listen((event) {
             port.send(['update']);
           });
@@ -238,8 +239,7 @@ class PowerSyncDatabaseImpl
       _PowerSyncDatabaseIsolateArgs(
         receiveMessages.sendPort,
         dbRef,
-        retryDelay,
-        clientParams,
+        resolved,
         crudMutex.shared,
         syncMutex.shared,
       ),
@@ -282,16 +282,14 @@ class PowerSyncDatabaseImpl
 class _PowerSyncDatabaseIsolateArgs {
   final SendPort sPort;
   final IsolateConnectionFactory dbRef;
-  final Duration retryDelay;
-  final Map<String, dynamic>? parameters;
+  final ResolvedSyncOptions options;
   final SerializedMutex crudMutex;
   final SerializedMutex syncMutex;
 
   _PowerSyncDatabaseIsolateArgs(
     this.sPort,
     this.dbRef,
-    this.retryDelay,
-    this.parameters,
+    this.options,
     this.crudMutex,
     this.syncMutex,
   );
@@ -392,9 +390,8 @@ Future<void> _syncIsolate(_PowerSyncDatabaseIsolateArgs args) async {
       invalidCredentialsCallback: invalidateCredentials,
       uploadCrud: uploadCrud,
       crudUpdateTriggerStream: crudUpdateController.stream,
-      retryDelay: args.retryDelay,
+      options: args.options,
       client: http.Client(),
-      syncParameters: args.parameters,
       crudMutex: crudMutex,
       syncMutex: syncMutex,
     );

packages/powersync_core/lib/src/database/powersync_database_impl_stub.dart

@@ -6,6 +6,7 @@ import 'package:powersync_core/sqlite_async.dart';
 import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/database/powersync_db_mixin.dart';
 import 'package:powersync_core/src/open_factory/abstract_powersync_open_factory.dart';
+import '../sync/options.dart';
 import 'powersync_database.dart';
 
 import '../connector.dart';
@@ -110,12 +111,12 @@ class PowerSyncDatabaseImpl
 
   @override
   @internal
-  Future<void> connectInternal(
-      {required PowerSyncBackendConnector connector,
-      required Duration crudThrottleTime,
-      required AbortController abort,
-      required Zone asyncWorkZone,
-      Map<String, dynamic>? params}) {
+  Future<void> connectInternal({
+    required PowerSyncBackendConnector connector,
+    required AbortController abort,
+    required Zone asyncWorkZone,
+    required SyncOptions options,
+  }) {
     throw UnimplementedError();
   }
 

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -13,6 +13,7 @@ import 'package:powersync_core/src/powersync_update_notification.dart';
 import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/schema_logic.dart';
 import 'package:powersync_core/src/schema_logic.dart' as schema_logic;
+import 'package:powersync_core/src/sync/options.dart';
 import 'package:powersync_core/src/sync/sync_status.dart';
 
 mixin PowerSyncDatabaseMixin implements SqliteConnection {
@@ -72,6 +73,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Delay between retrying failed requests.
   /// Defaults to 5 seconds.
   /// Only has an effect if changed before calling [connect].
+  @Deprecated('Set option when calling connect() instead')
   Duration retryDelay = const Duration(seconds: 5);
 
   @protected
@@ -269,17 +271,30 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   ///
   /// The connection is automatically re-opened if it fails for any reason.
   ///
+  /// To set sync parameters used in your sync rules (if any), use
+  /// [SyncOptions.params]. [SyncOptions] can also be used to tune the behavior
+  /// of the sync client, see that class for more information.
+  ///
   /// Status changes are reported on [statusStream].
   Future<void> connect({
     required PowerSyncBackendConnector connector,
-    Duration crudThrottleTime = const Duration(milliseconds: 10),
-    Map<String, dynamic>? params,
+    SyncOptions? options,
+    @Deprecated('Use SyncOptions.crudThrottleTime instead')
+    Duration? crudThrottleTime,
+    @Deprecated('Use SyncOptions.params instead') Map<String, dynamic>? params,
   }) async {
     // The initialization process acquires a sync connect lock (through
     // updateSchema), so ensure the database is ready before we try to acquire
     // the lock for the connection.
     await initialize();
 
+    final resolvedOptions = SyncOptions(
+      crudThrottleTime: options?.crudThrottleTime ?? crudThrottleTime,
+      // ignore: deprecated_member_use_from_same_package
+      retryDelay: options?.retryDelay ?? retryDelay,
+      params: options?.params ?? params,
+    );
+
     clientParams = params;
     var thisConnectAborter = AbortController();
     final zone = Zone.current;
@@ -294,8 +309,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
 
       await connectInternal(
         connector: connector,
-        crudThrottleTime: crudThrottleTime,
-        params: params,
+        options: resolvedOptions,
         abort: thisConnectAborter,
         // Run follow-up async tasks in the parent zone, a new one is introduced
         // while we hold the lock (and async tasks won't hold the sync lock).
@@ -342,17 +356,13 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// [connect] method and should not be called elsewhere.
   /// This method will only be called internally when no other sync client is
   /// active, so the method should not call [disconnect] itself.
-  ///
-  /// The [crudThrottleTime] is the throttle time between CRUD operations, it
-  /// defaults to 10 milliseconds in [connect].
   @protected
   @internal
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
-    required Duration crudThrottleTime,
+    required SyncOptions options,
     required AbortController abort,
     required Zone asyncWorkZone,
-    Map<String, dynamic>? params,
   });
 
   /// Close the sync connection.

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

@@ -14,6 +14,7 @@ import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/sync/streaming_sync.dart';
 import 'package:sqlite_async/sqlite_async.dart';
 
+import '../../sync/options.dart';
 import '../../web/sync_controller.dart';
 
 /// A PowerSync managed database.
@@ -114,13 +115,11 @@ class PowerSyncDatabaseImpl
   @internal
   Future<void> connectInternal({
     required PowerSyncBackendConnector connector,
-    required Duration crudThrottleTime,
     required AbortController abort,
     required Zone asyncWorkZone,
-    Map<String, dynamic>? params,
+    required SyncOptions options,
   }) async {
-    final crudStream =
-        database.onChange(['ps_crud'], throttle: crudThrottleTime);
+    final resolved = ResolvedSyncOptions(options);
 
     final storage = BucketStorage(database);
     StreamingSync sync;
@@ -130,25 +129,25 @@ class PowerSyncDatabaseImpl
       sync = await SyncWorkerHandle.start(
         database: this,
         connector: connector,
-        crudThrottleTimeMs: crudThrottleTime.inMilliseconds,
+        options: options,
         workerUri: Uri.base.resolve('/powersync_sync.worker.js'),
-        syncParams: params,
       );
     } catch (e) {
       logger.warning(
         'Could not use shared worker for synchronization, falling back to locks.',
         e,
       );
+      final crudStream =
+          database.onChange(['ps_crud'], throttle: resolved.crudThrottleTime);
 
       sync = StreamingSyncImplementation(
         adapter: storage,
         credentialsCallback: connector.getCredentialsCached,
         invalidCredentialsCallback: connector.prefetchCredentials,
         uploadCrud: () => connector.uploadData(this),
         crudUpdateTriggerStream: crudStream,
-        retryDelay: Duration(seconds: 3),
+        options: resolved,
         client: BrowserClient(),
-        syncParameters: params,
         // Only allows 1 sync implementation to run at a time per database
         // This should be global (across tabs) when using Navigator locks.
         identifier: database.openFactory.path,

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

@@ -0,0 +1,147 @@
+import 'sync_status.dart';
+
+/// An internal instruction emitted by the sync client in the core extension in
+/// response to the Dart SDK passing sync data into the extension.
+sealed class Instruction {
+  factory Instruction.fromJson(Map<String, Object?> json) {
+    return switch (json) {
+      {'LogLine': final logLine} =>
+        LogLine.fromJson(logLine as Map<String, Object?>),
+      {'UpdateSyncStatus': final updateStatus} =>
+        UpdateSyncStatus.fromJson(updateStatus as Map<String, Object?>),
+      {'EstablishSyncStream': final establish} =>
+        EstablishSyncStream.fromJson(establish as Map<String, Object?>),
+      {'FetchCredentials': final creds} =>
+        FetchCredentials.fromJson(creds as Map<String, Object?>),
+      {'CloseSyncStream': _} => const CloseSyncStream(),
+      {'FlushFileSystem': _} => const FlushFileSystem(),
+      {'DidCompleteSync': _} => const DidCompleteSync(),
+      _ => UnknownSyncLine(json)
+    };
+  }
+}
+
+final class LogLine implements Instruction {
+  final String severity;
+  final String line;
+
+  LogLine({required this.severity, required this.line});
+
+  factory LogLine.fromJson(Map<String, Object?> json) {
+    return LogLine(
+      severity: json['severity'] as String,
+      line: json['line'] as String,
+    );
+  }
+}
+
+final class EstablishSyncStream implements Instruction {
+  final Map<String, Object?> request;
+
+  EstablishSyncStream(this.request);
+
+  factory EstablishSyncStream.fromJson(Map<String, Object?> json) {
+    return EstablishSyncStream(json['request'] as Map<String, Object?>);
+  }
+}
+
+final class UpdateSyncStatus implements Instruction {
+  final CoreSyncStatus status;
+
+  UpdateSyncStatus({required this.status});
+
+  factory UpdateSyncStatus.fromJson(Map<String, Object?> json) {
+    return UpdateSyncStatus(
+        status:
+            CoreSyncStatus.fromJson(json['status'] as Map<String, Object?>));
+  }
+}
+
+final class CoreSyncStatus implements Instruction {
+  final bool connected;
+  final bool connecting;
+  final List<SyncPriorityStatus> priorityStatus;
+  final DownloadProgress? downloading;
+
+  CoreSyncStatus({
+    required this.connected,
+    required this.connecting,
+    required this.priorityStatus,
+    required this.downloading,
+  });
+
+  factory CoreSyncStatus.fromJson(Map<String, Object?> json) {
+    return CoreSyncStatus(
+      connected: json['connected'] as bool,
+      connecting: json['connecting'] as bool,
+      priorityStatus: [
+        for (final entry in json['priority_status'] as List)
+          _priorityStatusFromJson(entry as Map<String, Object?>)
+      ],
+      downloading: switch (json['downloading']) {
+        null => null,
+        final raw as Map<String, Object?> => DownloadProgress.fromJson(raw),
+      },
+    );
+  }
+
+  static SyncPriorityStatus _priorityStatusFromJson(Map<String, Object?> json) {
+    return (
+      priority: BucketPriority(json['priority'] as int),
+      hasSynced: json['has_synced'] as bool?,
+      lastSyncedAt: switch (json['last_synced_at']) {
+        null => null,
+        final lastSyncedAt as int =>
+          DateTime.fromMillisecondsSinceEpoch(lastSyncedAt * 1000),
+      },
+    );
+  }
+}
+
+final class DownloadProgress implements Instruction {
+  final Map<String, BucketProgress> progress;
+
+  DownloadProgress(this.progress);
+
+  factory DownloadProgress.fromJson(Map<String, Object?> line) {
+    return DownloadProgress(line.map((k, v) =>
+        MapEntry(k, _bucketProgressFromJson(v as Map<String, Object?>))));
+  }
+
+  static BucketProgress _bucketProgressFromJson(Map<String, Object?> json) {
+    return (
+      priority: BucketPriority(json['priority'] as int),
+      atLast: json['at_last'] as int,
+      sinceLast: json['since_last'] as int,
+      targetCount: json['target_count'] as int,
+    );
+  }
+}
+
+final class FetchCredentials implements Instruction {
+  final bool didExpire;
+
+  FetchCredentials(this.didExpire);
+
+  factory FetchCredentials.fromJson(Map<String, Object?> line) {
+    return FetchCredentials(line['did_expire'] as bool);
+  }
+}
+
+final class CloseSyncStream implements Instruction {
+  const CloseSyncStream();
+}
+
+final class FlushFileSystem implements Instruction {
+  const FlushFileSystem();
+}
+
+final class DidCompleteSync implements Instruction {
+  const DidCompleteSync();
+}
+
+final class UnknownSyncLine implements Instruction {
+  final Map<String, Object?> source;
+
+  UnknownSyncLine(this.source);
+}