Increment counters when receiving sync lines

Author: simolus3

packages/powersync_core/lib/powersync_core.dart

@@ -10,6 +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, OperationCounter;
+export 'src/sync_status.dart' hide InternalSyncDownloadProgress;
 export 'src/uuid.dart';

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

@@ -58,6 +58,7 @@ final class MutableSyncStatus {
 
   void applyCheckpointStarted(Checkpoint target) {
     downloading = true;
+    // TODO: Include pending ops from interrupted download, if any...
     downloadProgress = InternalSyncDownloadProgress.fromZero(target);
   }
 
@@ -66,6 +67,18 @@ final class MutableSyncStatus {
     uploadError = error;
   }
 
+  void applyBatchReceived(
+      Map<String, BucketDescription?> currentBuckets, SyncDataBatch batch) {
+    downloading = true;
+    if (downloadProgress case final previousProgress?) {
+      downloadProgress = previousProgress.incrementDownloaded([
+        for (final bucket in batch.buckets)
+          if (currentBuckets[bucket.bucket] case final knownBucket?)
+            (BucketPriority(knownBucket.priority), bucket.data.length),
+      ]);
+    }
+  }
+
   SyncStatus immutableSnapsot() {
     return SyncStatus(
       connected: connected,

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

@@ -391,6 +391,9 @@ class StreamingSyncImplementation implements StreamingSync {
           await adapter.removeBuckets(diff.removedBuckets);
           adapter.setTargetCheckpoint(targetCheckpoint);
         case SyncDataBatch():
+          // TODO: This increments the counters before actually saving sync
+          // data. Might be fine though?
+          _state.updateStatus((s) => s.applyBatchReceived(bucketMap, line));
           _state.updateStatus((s) => s.downloading = true);
           await adapter.saveSyncData(line);
         case StreamingSyncKeepalive(:final tokenExpiresIn):

packages/powersync_core/lib/src/sync_status.dart

@@ -1,3 +1,5 @@
+import 'dart:math';
+
 import 'package:collection/collection.dart';
 import 'package:meta/meta.dart';
 
@@ -161,11 +163,11 @@ final class SyncStatus {
   String toString() {
     return "SyncStatus<connected: $connected connecting: $connecting downloading: $downloading uploading: $uploading lastSyncedAt: $lastSyncedAt, hasSynced: $hasSynced, error: $anyError>";
   }
-}
 
-// This should be a ListEquality<SyncPriorityStatus>, but that appears to
-// cause weird type errors with DDC (but only after hot reloads?!)
-const _statusEquality = ListEquality<Object?>();
+  // This should be a ListEquality<SyncPriorityStatus>, but that appears to
+  // cause weird type errors with DDC (but only after hot reloads?!)
+  static const _statusEquality = ListEquality<Object?>();
+}
 
 /// The priority of a PowerSync bucket.
 extension type const BucketPriority._(int priorityNumber) {
@@ -214,34 +216,24 @@ class UploadQueueStats {
   }
 }
 
-@internal
-typedef OperationCounter = ({BucketPriority priority, int opCount});
-
 @internal
 final class InternalSyncDownloadProgress {
-  final List<OperationCounter> downloaded;
-  final List<OperationCounter> target;
+  final Map<BucketPriority, int> downloaded;
+  final Map<BucketPriority, int> target;
 
   final int _totalDownloaded;
   final int _totalTarget;
 
   InternalSyncDownloadProgress(this.downloaded, this.target)
-      : _totalDownloaded = downloaded.map((e) => e.opCount).sum,
-        _totalTarget = target.map((e) => e.opCount).sum;
+      : _totalDownloaded = target.values.sum,
+        _totalTarget = target.values.sum;
 
   factory InternalSyncDownloadProgress.fromZero(Checkpoint target) {
-    final totalOpsPerPriority =
-        target.checksums.groupFoldBy<BucketPriority, int>(
+    final targetOps = target.checksums.groupFoldBy<BucketPriority, int>(
       (cs) => BucketPriority(cs.priority),
       (prev, cs) => (prev ?? 0) + (cs.count ?? 0),
     );
-    final downloaded = [
-      for (final involvedPriority in totalOpsPerPriority.keys)
-        (priority: involvedPriority, opCount: 0),
-    ];
-    final targetOps = totalOpsPerPriority.entries
-        .map((e) => (priority: e.key, opCount: e.value))
-        .toList();
+    final downloaded = targetOps.map((k, v) => MapEntry(k, 0));
 
     return InternalSyncDownloadProgress(downloaded, targetOps);
   }
@@ -251,20 +243,35 @@ final class InternalSyncDownloadProgress {
   }
 
   static int sumInPriority(
-      List<OperationCounter> counters, BucketPriority priority) {
-    return counters
-        .where((e) => e.priority >= priority)
-        .map((e) => e.opCount)
+      Map<BucketPriority, int> counters, BucketPriority priority) {
+    return counters.entries
+        .where((e) => e.key >= priority)
+        .map((e) => e.value)
         .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]!);
+    }
+
+    return InternalSyncDownloadProgress(downloadedOps, target);
+  }
+
   SyncDownloadProgress get asSyncDownloadProgress =>
       SyncDownloadProgress._(this);
 
   @override
   int get hashCode => Object.hash(
-        _statusEquality.hash(downloaded),
-        _statusEquality.hash(target),
+        _mapEquality.hash(downloaded),
+        _mapEquality.hash(target),
       );
 
   @override
@@ -274,9 +281,11 @@ final class InternalSyncDownloadProgress {
         // them first helps find a difference faster.
         _totalDownloaded == other._totalDownloaded &&
         _totalTarget == other._totalTarget &&
-        _statusEquality.equals(downloaded, other.downloaded) &&
-        _statusEquality.equals(target, other.target);
+        _mapEquality.equals(downloaded, other.downloaded) &&
+        _mapEquality.equals(target, other.target);
   }
+
+  static const _mapEquality = MapEquality<Object?, Object?>();
 }
 
 /// Provides realtime progress about how PowerSync is downloading rows.

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

@@ -153,22 +153,8 @@ extension type SerializedOperationCounter._(JSObject _) implements JSObject {
     required int opCount,
   });
 
-  factory SerializedOperationCounter.fromDart(OperationCounter progress) {
-    return SerializedOperationCounter(
-      priority: progress.priority.priorityNumber,
-      opCount: progress.opCount,
-    );
-  }
-
   external JSNumber get priority;
   external JSNumber get opCount;
-
-  OperationCounter get toDart {
-    return (
-      priority: BucketPriority(priority.toDartInt),
-      opCount: opCount.toDartInt
-    );
-  }
 }
 
 @anonymous
@@ -197,15 +183,22 @@ extension type SerializedDownloadProgress._(JSObject _) implements JSObject {
   }
 
   static JSArray<SerializedOperationCounter> _serializeCounters(
-      List<OperationCounter> counters) {
+      Map<BucketPriority, int> counters) {
     return [
-      for (final entry in counters) SerializedOperationCounter.fromDart(entry)
+      for (final MapEntry(:key, :value) in counters.entries)
+        SerializedOperationCounter(
+          priority: key.priorityNumber,
+          opCount: value,
+        )
     ].toJS;
   }
 
-  static List<OperationCounter> _deserializeCounters(
+  static Map<BucketPriority, int> _deserializeCounters(
       JSArray<SerializedOperationCounter> counters) {
-    return [for (final entry in counters.toDart) entry.toDart];
+    return {
+      for (final entry in counters.toDart)
+        BucketPriority(entry.priority.toDartInt): entry.opCount.toDartInt,
+    };
   }
 }