Map stream status from core

Author: simolus3

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

@@ -16,6 +16,8 @@ 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';
 
+import '../sync/stream.dart';
+
 mixin PowerSyncDatabaseMixin implements SqliteConnection {
   /// Schema used for the local database.
   Schema get schema;
@@ -140,6 +142,10 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     return isInitialized;
   }
 
+  Future<List<SyncStreamSubscription>> get activeSubscriptions {
+    throw UnimplementedError();
+  }
+
   Future<void> _updateHasSynced() async {
     // Query the database to see if any data has been synced.
     final result = await database.getAll(

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

@@ -1,3 +1,4 @@
+import 'stream.dart';
 import 'sync_status.dart';
 
 /// An internal instruction emitted by the sync client in the core extension in
@@ -62,12 +63,14 @@ final class CoreSyncStatus {
   final bool connecting;
   final List<SyncPriorityStatus> priorityStatus;
   final DownloadProgress? downloading;
+  final List<CoreActiveStreamSubscription>? streams;
 
   CoreSyncStatus({
     required this.connected,
     required this.connecting,
     required this.priorityStatus,
     required this.downloading,
+    required this.streams,
   });
 
   factory CoreSyncStatus.fromJson(Map<String, Object?> json) {
@@ -82,6 +85,10 @@ final class CoreSyncStatus {
         null => null,
         final raw as Map<String, Object?> => DownloadProgress.fromJson(raw),
       },
+      streams: (json['stream'] as List<Object?>?)
+          ?.map((e) =>
+              CoreActiveStreamSubscription.fromJson(e as Map<String, Object?>))
+          .toList(),
     );
   }
 

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

@@ -3,6 +3,7 @@ import 'dart:async';
 import 'package:collection/collection.dart';
 
 import 'instruction.dart';
+import 'stream.dart';
 import 'sync_status.dart';
 import 'bucket_storage.dart';
 import 'protocol.dart';
@@ -15,6 +16,7 @@ final class MutableSyncStatus {
 
   InternalSyncDownloadProgress? downloadProgress;
   List<SyncPriorityStatus> priorityStatusEntries = const [];
+  List<CoreActiveStreamSubscription> streams = const [];
 
   DateTime? lastSyncedAt;
 

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

@@ -1,3 +1,5 @@
+import 'package:meta/meta.dart';
+
 import 'sync_status.dart';
 
 abstract interface class SyncStreamDescription {
@@ -10,7 +12,7 @@ abstract interface class SyncSubscriptionDefinition
   bool get active;
   DateTime? get expiresAt;
   bool get hasSynced;
-  DateTime? lastSyncedAt;
+  DateTime? get lastSyncedAt;
 }
 
 abstract interface class SyncStream extends SyncStreamDescription {
@@ -26,3 +28,56 @@ abstract interface class SyncStreamSubscription
   Future<void> waitForFirstSync();
   Future<void> unsubscribe({bool immediately = false});
 }
+
+/// An `ActiveStreamSubscription` as part of the sync status in Rust.
+@internal
+final class CoreActiveStreamSubscription implements SyncSubscriptionDefinition {
+  @override
+  final String name;
+  @override
+  final Map<String, Object?>? parameters;
+  final BucketPriority priority;
+  final List<String> associatedBuckets;
+  @override
+  final bool active;
+  final bool isDefault;
+  @override
+  final DateTime? expiresAt;
+  @override
+  final DateTime? lastSyncedAt;
+
+  @override
+  bool get hasSynced => lastSyncedAt != null;
+
+  CoreActiveStreamSubscription._({
+    required this.name,
+    required this.parameters,
+    required this.priority,
+    required this.associatedBuckets,
+    required this.active,
+    required this.isDefault,
+    required this.expiresAt,
+    required this.lastSyncedAt,
+  });
+
+  factory CoreActiveStreamSubscription.fromJson(Map<String, Object?> json) {
+    return CoreActiveStreamSubscription._(
+      name: json['name'] as String,
+      parameters: json['parameters'] as Map<String, Object?>,
+      priority: BucketPriority(json['priority'] as int),
+      associatedBuckets: (json['associated_buckets'] as List).cast(),
+      active: json['active'] as bool,
+      isDefault: json['is_default'] as bool,
+      expiresAt: switch (json['expires_at']) {
+        null => null,
+        final timestamp as int =>
+          DateTime.fromMillisecondsSinceEpoch(timestamp * 1000),
+      },
+      lastSyncedAt: switch (json['last_synced_at']) {
+        null => null,
+        final timestamp as int =>
+          DateTime.fromMillisecondsSinceEpoch(timestamp * 1000),
+      },
+    );
+  }
+}

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

@@ -2,6 +2,7 @@ import 'dart:math';
 
 import 'package:collection/collection.dart';
 import 'package:meta/meta.dart';
+import '../database/powersync_database.dart';
 
 import 'bucket_storage.dart';
 import 'protocol.dart';
@@ -55,7 +56,7 @@ final class SyncStatus {
 
   final List<SyncPriorityStatus> priorityStatusEntries;
 
-  final List<SyncStreamStatus>? activeSubscriptions;
+  final List<CoreActiveStreamSubscription>? _internalSubscriptions;
 
   const SyncStatus({
     this.connected = false,
@@ -68,8 +69,8 @@ final class SyncStatus {
     this.downloadError,
     this.uploadError,
     this.priorityStatusEntries = const [],
-    this.activeSubscriptions,
-  });
+    List<CoreActiveStreamSubscription>? streamSubscriptions,
+  }) : _internalSubscriptions = streamSubscriptions;
 
   @override
   bool operator ==(Object other) {
@@ -82,8 +83,10 @@ final class SyncStatus {
         other.uploadError == uploadError &&
         other.lastSyncedAt == lastSyncedAt &&
         other.hasSynced == hasSynced &&
-        _statusEquality.equals(
+        _listEquality.equals(
             other.priorityStatusEntries, priorityStatusEntries) &&
+        _listEquality.equals(
+            other._internalSubscriptions, _internalSubscriptions) &&
         other.downloadProgress == downloadProgress);
   }
 
@@ -114,6 +117,18 @@ final class SyncStatus {
     );
   }
 
+  /// All sync streams currently being tracked in this subscription.
+  ///
+  /// This returns null when the sync stream is currently being opened and we
+  /// don't have reliable information about all included streams yet (in that
+  /// state, [PowerSyncDatabase.activeSubscriptions] can still be used to
+  /// resolve known subscriptions locally).
+  Iterable<SyncStreamStatus>? get activeSubscriptions {
+    return _internalSubscriptions?.map((subscription) {
+      return SyncStreamStatus._(subscription, downloadProgress);
+    });
+  }
+
   /// Get the current [downloadError] or [uploadError].
   Object? get anyError {
     return downloadError ?? uploadError;
@@ -153,6 +168,21 @@ final class SyncStatus {
     );
   }
 
+  /// If the [stream] appears in [activeSubscriptions], returns the current
+  /// status for that stream.
+  SyncStreamStatus? statusFor(SyncStreamDescription stream) {
+    final raw = _internalSubscriptions?.firstWhereOrNull(
+      (e) =>
+          e.name == stream.name &&
+          _mapEquality.equals(e.parameters, stream.parameters),
+    );
+
+    if (raw == null) {
+      return null;
+    }
+    return SyncStreamStatus._(raw, downloadProgress);
+  }
+
   @override
   int get hashCode {
     return Object.hash(
@@ -163,8 +193,9 @@ final class SyncStatus {
       uploadError,
       downloadError,
       lastSyncedAt,
-      _statusEquality.hash(priorityStatusEntries),
+      _listEquality.hash(priorityStatusEntries),
       downloadProgress,
+      _listEquality.hash(_internalSubscriptions),
     );
   }
 
@@ -173,21 +204,20 @@ final class SyncStatus {
     return "SyncStatus<connected: $connected connecting: $connecting downloading: $downloading (progress: $downloadProgress) 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?!)
-  static const _statusEquality = ListEquality<Object?>();
+  static const _listEquality = ListEquality<Object?>();
+  static const _mapEquality = MapEquality<Object?, Object?>();
 }
 
 final class SyncStreamStatus {
-  final SyncSubscriptionDefinition subscription;
-  final BucketPriority priority;
+  final ProgressWithOperations? progress;
+  final CoreActiveStreamSubscription _internal;
 
-  final bool isDefault;
-  final ProgressWithOperations progress;
+  SyncSubscriptionDefinition get subscription => _internal;
+  BucketPriority get priority => _internal.priority;
+  bool get isDefault => _internal.isDefault;
 
-  @internal
-  SyncStreamStatus(
-      this.subscription, this.priority, this.isDefault, this.progress);
+  SyncStreamStatus._(this._internal, SyncDownloadProgress? progress)
+      : progress = progress?._internal._forStream(_internal);
 }
 
 /// The priority of a PowerSync bucket.
@@ -304,13 +334,23 @@ 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) {
-    final (total, downloaded) =
-        buckets.values.where((e) => e.priority >= priority).fold(
+    final (total, downloaded) = buckets.values
+        .where((e) => e.priority >= priority)
+        .fold((0, 0), _addProgress);
+
+    return ProgressWithOperations._(total, downloaded);
+  }
+
+  ProgressWithOperations _forStream(CoreActiveStreamSubscription subscription) {
+    final (total, downloaded) = subscription.associatedBuckets.fold(
       (0, 0),
-      (prev, entry) {
-        final downloaded = entry.sinceLast;
-        final total = entry.targetCount - entry.atLast;
-        return (prev.$1 + total, prev.$2 + downloaded);
+      (prev, bucket) {
+        final foundProgress = buckets[bucket];
+        if (foundProgress == null) {
+          return prev;
+        }
+
+        return _addProgress(prev, foundProgress);
       },
     );
 
@@ -356,6 +396,12 @@ final class InternalSyncDownloadProgress extends ProgressWithOperations {
   }
 
   static const _mapEquality = MapEquality<Object?, Object?>();
+
+  (int, int) _addProgress((int, int) prev, BucketProgress entry) {
+    final downloaded = entry.sinceLast;
+    final total = entry.targetCount - entry.atLast;
+    return (prev.$1 + total, prev.$2 + downloaded);
+  }
 }
 
 /// Information about a progressing download.