Start migrating to new model

Author: simolus3

packages/powersync_core/lib/powersync_core.dart

@@ -10,5 +10,5 @@ export 'src/exceptions.dart';
 export 'src/log.dart';
 export 'src/open_factory.dart';
 export 'src/schema.dart';
-export 'src/sync_status.dart' hide InternalSyncDownloadProgress;
+export 'src/sync_status.dart' hide BucketProgress, InternalSyncDownloadProgress;
 export 'src/uuid.dart';

packages/powersync_core/lib/src/bucket_storage.dart

@@ -29,8 +29,6 @@ class BucketStorage {
     return await _internalDb.execute(query, parameters);
   }
 
-  void startSession() {}
-
   Future<List<BucketState>> getBucketStates() async {
     final rows = await select(
         'SELECT name as bucket, cast(last_op as TEXT) as op_id FROM ps_buckets WHERE pending_delete = 0 AND name != \'\$local\'');

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

@@ -40,6 +40,7 @@ final class MutableSyncStatus {
 
   void applyCheckpointReached(Checkpoint applied) {
     downloading = false;
+    downloadProgress = null;
     downloadError = null;
     final now = lastSyncedAt = DateTime.now();
     priorityStatusEntries = [

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

@@ -285,8 +285,6 @@ class StreamingSyncImplementation implements StreamingSync {
 
   Future<bool> streamingSyncIteration(
       {AbortController? abortController}) async {
-    adapter.startSession();
-
     var (bucketRequests, bucketMap) = await _collectLocalBucketState();
 
     Checkpoint? targetCheckpoint;

packages/powersync_core/lib/src/sync_status.dart

@@ -216,26 +216,32 @@ class UploadQueueStats {
   }
 }
 
+/// Per-bucket download progress information.
+@internal
+typedef BucketProgress = ({
+  BucketPriority priority,
+  int atLast,
+  int sinceLast,
+  int targetCount,
+});
+
 @internal
 final class InternalSyncDownloadProgress {
-  final Map<BucketPriority, int> downloaded;
-  final Map<BucketPriority, int> target;
+  final Map<String, BucketProgress> buckets;
 
   final int _totalDownloaded;
   final int _totalTarget;
 
-  InternalSyncDownloadProgress(this.downloaded, this.target)
-      : _totalDownloaded = target.values.sum,
-        _totalTarget = target.values.sum;
+  InternalSyncDownloadProgress(this.buckets)
+      : _totalDownloaded = buckets.values.map((e) => e.sinceLast).sum,
+        _totalTarget = buckets.values.map((e) => e.targetCount - e.atLast).sum;
 
   factory InternalSyncDownloadProgress.fromZero(Checkpoint target) {
     final targetOps = target.checksums.groupFoldBy<BucketPriority, int>(
       (cs) => BucketPriority(cs.priority),
       (prev, cs) => (prev ?? 0) + (cs.count ?? 0),
     );
     final downloaded = targetOps.map((k, v) => MapEntry(k, 0));
-
-    return InternalSyncDownloadProgress(downloaded, targetOps);
   }
 
   static InternalSyncDownloadProgress ofPublic(SyncDownloadProgress public) {
@@ -250,29 +256,26 @@ final class InternalSyncDownloadProgress {
         .sum;
   }
 
-  InternalSyncDownloadProgress incrementDownloaded(
-      List<(BucketPriority, int)> opsInPriority) {
-    var downloadedOps = {...downloaded};
-
-    for (final (priority, addedOps) in opsInPriority) {
-      assert(downloaded.containsKey(priority));
-      assert(target.containsKey(priority));
-
-      downloadedOps[priority] =
-          max(downloadedOps[priority]! + addedOps, target[priority]!);
+  InternalSyncDownloadProgress incrementDownloaded(SyncDataBatch batch) {
+    final newBucketStates = Map.of(buckets);
+    for (final dataForBucket in batch.buckets) {
+      final previous = newBucketStates[dataForBucket.bucket]!;
+      newBucketStates[dataForBucket.bucket] = (
+        priority: previous.priority,
+        atLast: previous.atLast,
+        sinceLast: previous.sinceLast,
+        targetCount: previous.targetCount,
+      );
     }
 
-    return InternalSyncDownloadProgress(downloadedOps, target);
+    return InternalSyncDownloadProgress(newBucketStates);
   }
 
   SyncDownloadProgress get asSyncDownloadProgress =>
       SyncDownloadProgress._(this);
 
   @override
-  int get hashCode => Object.hash(
-        _mapEquality.hash(downloaded),
-        _mapEquality.hash(target),
-      );
+  int get hashCode => _mapEquality.hash(buckets);
 
   @override
   bool operator ==(Object other) {
@@ -281,8 +284,7 @@ final class InternalSyncDownloadProgress {
         // them first helps find a difference faster.
         _totalDownloaded == other._totalDownloaded &&
         _totalTarget == other._totalTarget &&
-        _mapEquality.equals(downloaded, other.downloaded) &&
-        _mapEquality.equals(target, other.target);
+        _mapEquality.equals(buckets, other.buckets);
   }
 
   static const _mapEquality = MapEquality<Object?, Object?>();