Use new progress information

Author: simolus3

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

@@ -93,7 +93,7 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
   @override
   final Map<String, Object?>? parameters;
   final StreamPriority priority;
-  final List<String> associatedBuckets;
+  final ({int total, int downloaded}) progress;
   @override
   final bool active;
   @override
@@ -112,7 +112,7 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
     required this.name,
     required this.parameters,
     required this.priority,
-    required this.associatedBuckets,
+    required this.progress,
     required this.active,
     required this.isDefault,
     required this.hasExplicitSubscription,
@@ -128,7 +128,7 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
         final prio? => StreamPriority(prio),
         null => StreamPriority.fullSyncPriority,
       },
-      associatedBuckets: (json['associated_buckets'] as List).cast(),
+      progress: _progressFromJson(json['progress'] as Map<String, Object?>),
       active: json['active'] as bool,
       isDefault: json['is_default'] as bool,
       hasExplicitSubscription: json['has_explicit_subscription'] as bool,
@@ -150,7 +150,10 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
       'name': name,
       'parameters': parameters,
       'priority': priority.priorityNumber,
-      'associated_buckets': associatedBuckets,
+      'progress': {
+        'total': progress.total,
+        'downloaded': progress.downloaded,
+      },
       'active': active,
       'is_default': isDefault,
       'has_explicit_subscription': hasExplicitSubscription,
@@ -164,4 +167,9 @@ final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
       }
     };
   }
+
+  static ({int total, int downloaded}) _progressFromJson(
+      Map<String, Object?> json) {
+    return (total: json['total'] as int, downloaded: json['downloaded'] as int);
+  }
 }

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

@@ -148,9 +148,9 @@ final class SyncStatus {
   /// information extracted from the lower priority `2` since each partial sync
   /// in priority `2` necessarily includes a consistent view over data in
   /// priority `1`.
-  SyncPriorityStatus statusForPriority(BucketPriority priority) {
+  SyncPriorityStatus statusForPriority(StreamPriority priority) {
     assert(priorityStatusEntries.isSortedByCompare(
-        (e) => e.priority, BucketPriority.comparator));
+        (e) => e.priority, StreamPriority.comparator));
 
     for (final known in priorityStatusEntries) {
       // Lower-priority buckets are synchronized after higher-priority buckets,
@@ -309,7 +309,7 @@ final class InternalSyncDownloadProgress extends ProgressWithOperations {
       final sinceLast = savedProgress?.sinceLast ?? 0;
 
       buckets[bucket.bucket] = (
-        priority: BucketPriority._(bucket.priority),
+        priority: StreamPriority._(bucket.priority),
         atLast: atLast,
         sinceLast: sinceLast,
         targetCount: bucket.count ?? 0,
@@ -324,7 +324,7 @@ final class InternalSyncDownloadProgress extends ProgressWithOperations {
           return InternalSyncDownloadProgress({
             for (final bucket in target.checksums)
               bucket.bucket: (
-                priority: BucketPriority(bucket.priority),
+                priority: StreamPriority(bucket.priority),
                 atLast: 0,
                 sinceLast: 0,
                 targetCount: knownCount,
@@ -343,7 +343,7 @@ final class InternalSyncDownloadProgress extends ProgressWithOperations {
 
   /// Sums the total target and completed operations for all buckets up until
   /// the given [priority] (inclusive).
-  ProgressWithOperations untilPriority(BucketPriority priority) {
+  ProgressWithOperations untilPriority(StreamPriority priority) {
     final (total, downloaded) = buckets.values
         .where((e) => e.priority >= priority)
         .fold((0, 0), _addProgress);
@@ -352,18 +352,7 @@ final class InternalSyncDownloadProgress extends ProgressWithOperations {
   }
 
   ProgressWithOperations _forStream(CoreActiveStreamSubscription subscription) {
-    final (total, downloaded) = subscription.associatedBuckets.fold(
-      (0, 0),
-      (prev, bucket) {
-        final foundProgress = buckets[bucket];
-        if (foundProgress == null) {
-          return prev;
-        }
-
-        return _addProgress(prev, foundProgress);
-      },
-    );
-
+    final (:total, :downloaded) = subscription.progress;
     return ProgressWithOperations._(total, downloaded);
   }
 
@@ -475,7 +464,7 @@ extension type SyncDownloadProgress._(InternalSyncDownloadProgress _internal)
   /// The returned [ProgressWithOperations] tracks the target amount of
   /// operations that need to be downloaded in total and how many of them have
   /// already been received.
-  ProgressWithOperations untilPriority(BucketPriority priority) {
+  ProgressWithOperations untilPriority(StreamPriority priority) {
     return _internal.untilPriority(priority);
   }
 }