Start with some unit tests

Author: simolus3

packages/powersync_core/lib/powersync_core.dart

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

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

@@ -7,7 +7,6 @@ import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/database/active_instances.dart';
 import 'package:powersync_core/src/database/powersync_db_mixin.dart';
 import 'package:powersync_core/src/sync/options.dart';
-import 'package:powersync_core/src/sync/stream.dart';
 
 import 'streaming_sync.dart';
 
@@ -189,6 +188,7 @@ final class ConnectionManager {
         hasSynced: status.lastSyncedAt != null
             ? true
             : status.hasSynced ?? currentStatus.hasSynced,
+        streamSubscriptions: status.internalSubscriptions,
       );
 
       // If the absence of hasSynced was the only difference, the new states
@@ -226,7 +226,7 @@ final class ConnectionManager {
 
   Future<void> _subscriptionsCommand(Object? command) async {
     await db.writeTransaction((tx) {
-      return db.execute(
+      return tx.execute(
         'SELECT powersync_control(?, ?)',
         ['subscriptions', json.encode(command)],
       );

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

@@ -85,7 +85,7 @@ final class CoreSyncStatus {
         null => null,
         final raw as Map<String, Object?> => DownloadProgress.fromJson(raw),
       },
-      streams: (json['stream'] as List<Object?>?)
+      streams: (json['streams'] as List<Object?>?)
           ?.map((e) =>
               CoreActiveStreamSubscription.fromJson(e as Map<String, Object?>))
           .toList(),

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

@@ -16,7 +16,7 @@ final class MutableSyncStatus {
 
   InternalSyncDownloadProgress? downloadProgress;
   List<SyncPriorityStatus> priorityStatusEntries = const [];
-  List<CoreActiveStreamSubscription> streams = const [];
+  List<CoreActiveStreamSubscription>? streams;
 
   DateTime? lastSyncedAt;
 
@@ -53,9 +53,9 @@ final class MutableSyncStatus {
           hasSynced: true,
           lastSyncedAt: now,
           priority: maxBy(
-            applied.checksums.map((cs) => BucketPriority(cs.priority)),
+            applied.checksums.map((cs) => StreamPriority(cs.priority)),
             (priority) => priority,
-            compare: BucketPriority.comparator,
+            compare: StreamPriority.comparator,
           )!,
         )
     ];
@@ -92,8 +92,9 @@ final class MutableSyncStatus {
       final downloading => InternalSyncDownloadProgress(downloading.buckets),
     };
     lastSyncedAt = status.priorityStatus
-        .firstWhereOrNull((s) => s.priority == BucketPriority.fullSyncPriority)
+        .firstWhereOrNull((s) => s.priority == StreamPriority.fullSyncPriority)
         ?.lastSyncedAt;
+    streams = status.streams;
   }
 
   SyncStatus immutableSnapshot() {
@@ -108,6 +109,7 @@ final class MutableSyncStatus {
       hasSynced: null, // Stream client is not supposed to set this value.
       uploadError: uploadError,
       downloadError: downloadError,
+      streamSubscriptions: streams,
     );
   }
 }

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

@@ -47,6 +47,7 @@ final class SyncOptions {
       retryDelay: retryDelay,
       params: params ?? this.params,
       syncImplementation: syncImplementation,
+      includeDefaultStreams: includeDefaultStreams,
     );
   }
 }
@@ -106,11 +107,14 @@ extension type ResolvedSyncOptions(SyncOptions source) {
       crudThrottleTime: other.crudThrottleTime ?? crudThrottleTime,
       retryDelay: other.retryDelay ?? retryDelay,
       params: other.params ?? params,
+      includeDefaultStreams:
+          other.includeDefaultStreams ?? includeDefaultStreams,
     );
 
     final didChange = !_mapEquality.equals(newOptions.params, params) ||
         newOptions.crudThrottleTime != crudThrottleTime ||
-        newOptions.retryDelay != retryDelay;
+        newOptions.retryDelay != retryDelay ||
+        newOptions.includeDefaultStreams != includeDefaultStreams;
     return (ResolvedSyncOptions(newOptions), didChange);
   }
 

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

@@ -123,8 +123,11 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
   factory CoreActiveStreamSubscription.fromJson(Map<String, Object?> json) {
     return CoreActiveStreamSubscription._(
       name: json['name'] as String,
-      parameters: json['parameters'] as Map<String, Object?>,
-      priority: StreamPriority(json['priority'] as int),
+      parameters: json['parameters'] as Map<String, Object?>?,
+      priority: switch (json['priority'] as int?) {
+        final prio? => StreamPriority(prio),
+        null => StreamPriority.fullSyncPriority,
+      },
       associatedBuckets: (json['associated_buckets'] as List).cast(),
       active: json['active'] as bool,
       isDefault: json['is_default'] as bool,

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

@@ -208,6 +208,12 @@ final class SyncStatus {
   static const _mapEquality = MapEquality<Object?, Object?>();
 }
 
+@internal
+extension InternalSyncStatusAccess on SyncStatus {
+  List<CoreActiveStreamSubscription>? get internalSubscriptions =>
+      _internalSubscriptions;
+}
+
 final class SyncStreamStatus {
   final ProgressWithOperations? progress;
   final CoreActiveStreamSubscription _internal;
@@ -276,7 +282,7 @@ class UploadQueueStats {
 /// Per-bucket download progress information.
 @internal
 typedef BucketProgress = ({
-  BucketPriority priority,
+  StreamPriority priority,
   int atLast,
   int sinceLast,
   int targetCount,

packages/powersync_core/lib/src/web/sync_worker_protocol.dart

@@ -92,7 +92,7 @@ extension type StartSynchronization._(JSObject _) implements JSObject {
 }
 
 @anonymous
-extension type UpdateSubscriptions.__(JSObject _inner) implements JSObject {
+extension type UpdateSubscriptions._raw(JSObject _inner) implements JSObject {
   external factory UpdateSubscriptions._({
     required int requestId,
     required JSArray content,

packages/powersync_core/test/bucket_storage_test.dart renamed to packages/powersync_core/test/sync/bucket_storage_test.dart

@@ -4,8 +4,9 @@ import 'package:powersync_core/src/sync/protocol.dart';
 import 'package:sqlite_async/sqlite3_common.dart';
 import 'package:test/test.dart';
 
-import 'utils/abstract_test_utils.dart';
-import 'utils/test_utils_impl.dart';
+import '../utils/abstract_test_utils.dart';
+import '../utils/test_utils_impl.dart';
+import 'utils.dart';
 
 final testUtils = TestUtils();
 
@@ -39,11 +40,6 @@ const removeAsset1_4 = OplogEntry(
 const removeAsset1_5 = OplogEntry(
     opId: '5', op: OpType.remove, rowType: 'assets', rowId: 'O1', checksum: 5);
 
-BucketChecksum checksum(
-    {required String bucket, required int checksum, int priority = 1}) {
-  return BucketChecksum(bucket: bucket, priority: priority, checksum: checksum);
-}
-
 SyncDataBatch syncDataBatch(List<SyncBucketData> data) {
   return SyncDataBatch(data);
 }

packages/powersync_core/test/sync/in_memory_sync_test.dart

@@ -10,11 +10,11 @@ import 'package:shelf/shelf.dart';
 import 'package:shelf_router/shelf_router.dart';
 import 'package:test/test.dart';
 
-import '../bucket_storage_test.dart';
 import '../server/sync_server/in_memory_sync_server.dart';
 import '../utils/abstract_test_utils.dart';
 import '../utils/in_memory_http.dart';
 import '../utils/test_utils_impl.dart';
+import 'utils.dart';
 
 void main() {
   _declareTests(
@@ -968,51 +968,3 @@ void _declareTests(String name, SyncOptions options, bool bson) {
     });
   });
 }
-
-TypeMatcher<SyncStatus> isSyncStatus({
-  Object? downloading,
-  Object? connected,
-  Object? connecting,
-  Object? hasSynced,
-  Object? downloadProgress,
-}) {
-  var matcher = isA<SyncStatus>();
-  if (downloading != null) {
-    matcher = matcher.having((e) => e.downloading, 'downloading', downloading);
-  }
-  if (connected != null) {
-    matcher = matcher.having((e) => e.connected, 'connected', connected);
-  }
-  if (connecting != null) {
-    matcher = matcher.having((e) => e.connecting, 'connecting', connecting);
-  }
-  if (hasSynced != null) {
-    matcher = matcher.having((e) => e.hasSynced, 'hasSynced', hasSynced);
-  }
-  if (downloadProgress != null) {
-    matcher = matcher.having(
-        (e) => e.downloadProgress, 'downloadProgress', downloadProgress);
-  }
-
-  return matcher;
-}
-
-TypeMatcher<SyncDownloadProgress> isSyncDownloadProgress({
-  required Object progress,
-  Map<BucketPriority, Object> priorities = const {},
-}) {
-  var matcher =
-      isA<SyncDownloadProgress>().having((e) => e, 'untilCompletion', progress);
-  priorities.forEach((priority, expected) {
-    matcher = matcher.having(
-        (e) => e.untilPriority(priority), 'untilPriority($priority)', expected);
-  });
-
-  return matcher;
-}
-
-TypeMatcher<ProgressWithOperations> progress(int completed, int total) {
-  return isA<ProgressWithOperations>()
-      .having((e) => e.downloadedOperations, 'completed', completed)
-      .having((e) => e.totalOperations, 'total', total);
-}