Properly update subscriptions

Author: simolus3

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

@@ -1,4 +1,5 @@
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:async/async.dart';
 import 'package:logging/logging.dart';
@@ -15,6 +16,8 @@ import 'package:powersync_core/src/schema.dart';
 import 'package:powersync_core/src/schema_logic.dart';
 import 'package:powersync_core/src/schema_logic.dart' as schema_logic;
 import 'package:powersync_core/src/sync/connection_manager.dart';
+import 'package:powersync_core/src/sync/instruction.dart';
+import 'package:powersync_core/src/sync/mutable_sync_status.dart';
 import 'package:powersync_core/src/sync/options.dart';
 import 'package:powersync_core/src/sync/sync_status.dart';
 
@@ -138,53 +141,26 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
 
   Future<void> _updateHasSynced() async {
     // Query the database to see if any data has been synced.
-    final result = await database.getAll(
-      'SELECT priority, last_synced_at FROM ps_sync_state ORDER BY priority;',
+    final row = await database.get(
+      'SELECT powersync_offline_sync_status() AS r;',
     );
-    const prioritySentinel = 2147483647;
-    var hasSynced = false;
-    DateTime? lastCompleteSync;
-    final priorityStatusEntries = <SyncPriorityStatus>[];
 
-    DateTime parseDateTime(String sql) {
-      return DateTime.parse('${sql}Z').toLocal();
-    }
-
-    for (final row in result) {
-      final priority = row.columnAt(0) as int;
-      final lastSyncedAt = parseDateTime(row.columnAt(1) as String);
-
-      if (priority == prioritySentinel) {
-        hasSynced = true;
-        lastCompleteSync = lastSyncedAt;
-      } else {
-        priorityStatusEntries.add((
-          hasSynced: true,
-          lastSyncedAt: lastSyncedAt,
-          priority: BucketPriority(priority)
-        ));
-      }
-    }
+    final status = CoreSyncStatus.fromJson(
+        json.decode(row['r'] as String) as Map<String, Object?>);
 
-    if (hasSynced != currentStatus.hasSynced) {
-      final status = SyncStatus(
-        hasSynced: hasSynced,
-        lastSyncedAt: lastCompleteSync,
-        priorityStatusEntries: priorityStatusEntries,
-      );
-      setStatus(status);
-    }
+    setStatus((MutableSyncStatus()..applyFromCore(status))
+        .immutableSnapshot(setLastSynced: true));
   }
 
   /// Returns a [Future] which will resolve once at least one full sync cycle
   /// has completed (meaninng that the first consistent checkpoint has been
   /// reached across all buckets).
   ///
   /// When [priority] is null (the default), this method waits for the first
-  /// full sync checkpoint to complete. When set to a [BucketPriority] however,
+  /// full sync checkpoint to complete. When set to a [StreamPriority] however,
   /// it completes once all buckets within that priority (as well as those in
   /// higher priorities) have been synchronized at least once.
-  Future<void> waitForFirstSync({BucketPriority? priority}) async {
+  Future<void> waitForFirstSync({StreamPriority? priority}) async {
     bool matches(SyncStatus status) {
       if (priority == null) {
         return status.hasSynced == true;

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

@@ -197,7 +197,7 @@ final class ConnectionManager {
       // would be equal and don't require an event. So, check again.
       if (newStatus != currentStatus) {
         _currentStatus = newStatus;
-        _statusController.add(currentStatus);
+        _statusController.add(_currentStatus);
       }
     }
   }

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

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

@@ -97,7 +97,7 @@ final class MutableSyncStatus {
     streams = status.streams;
   }
 
-  SyncStatus immutableSnapshot() {
+  SyncStatus immutableSnapshot({bool setLastSynced = false}) {
     return SyncStatus(
       connected: connected,
       connecting: connecting,
@@ -106,7 +106,7 @@ final class MutableSyncStatus {
       downloadProgress: downloadProgress?.asSyncDownloadProgress,
       priorityStatusEntries: UnmodifiableListView(priorityStatusEntries),
       lastSyncedAt: lastSyncedAt,
-      hasSynced: null, // Stream client is not supposed to set this value.
+      hasSynced: setLastSynced ? lastSyncedAt != null : null,
       uploadError: uploadError,
       downloadError: downloadError,
       streamSubscriptions: streams,

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

@@ -135,7 +135,7 @@ class StreamingSyncImplementation implements StreamingSync {
   void updateSubscriptions(List<SubscribedStream> streams) {
     _activeSubscriptions = streams;
     if (_nonLineSyncEvents.hasListener) {
-      _nonLineSyncEvents.add(const AbortCurrentIteration());
+      _nonLineSyncEvents.add(HandleChangedSubscriptions(streams));
     }
   }
 
@@ -464,6 +464,7 @@ class StreamingSyncImplementation implements StreamingSync {
           _state.updateStatus((s) => s.setConnected());
           await handleLine(line as StreamingSyncLine);
         case UploadCompleted():
+        case HandleChangedSubscriptions():
           // Only relevant for the Rust sync implementation.
           break;
         case AbortCurrentIteration():
@@ -613,6 +614,12 @@ final class _ActiveRustStreamingIteration {
 
   _ActiveRustStreamingIteration(this.sync);
 
+  List<Object?> _encodeSubscriptions(List<SubscribedStream> subscriptions) {
+    return sync._activeSubscriptions
+        .map((s) => {'name': s.name, 'params': s.parameters})
+        .toList();
+  }
+
   Future<void> syncIteration() async {
     try {
       await _control(
@@ -621,9 +628,7 @@ final class _ActiveRustStreamingIteration {
           'parameters': sync.options.params,
           'schema': convert.json.decode(sync.schemaJson),
           'include_defaults': sync.options.includeDefaultStreams,
-          'active_streams': sync._activeSubscriptions
-              .map((s) => {'name': s.name, 'params': s.parameters})
-              .toList(),
+          'active_streams': _encodeSubscriptions(sync._activeSubscriptions),
         }),
       );
       assert(_completedStream.isCompleted, 'Should have started streaming');
@@ -673,6 +678,9 @@ final class _ActiveRustStreamingIteration {
           break loop;
         case TokenRefreshComplete():
           await _control('refreshed_token');
+        case HandleChangedSubscriptions(:final currentSubscriptions):
+          await _control('update_subscriptions',
+              convert.json.encode(_encodeSubscriptions(currentSubscriptions)));
       }
     }
   }
@@ -762,3 +770,9 @@ final class TokenRefreshComplete implements SyncEvent {
 final class AbortCurrentIteration implements SyncEvent {
   const AbortCurrentIteration();
 }
+
+final class HandleChangedSubscriptions implements SyncEvent {
+  final List<SubscribedStream> currentSubscriptions;
+
+  HandleChangedSubscriptions(this.currentSubscriptions);
+}

packages/powersync_core/test/sync/stream_test.dart

@@ -183,4 +183,32 @@ void main() {
       ),
     );
   });
+
+  test('changes subscriptions dynamically', () async {
+    await waitForConnection();
+    syncService.addKeepAlive();
+
+    final subscription = await database.syncStream('a').subscribe();
+    syncService.endCurrentListener();
+    final request = await syncService.waitForListener;
+    expect(
+      json.decode(await request.readAsString()),
+      containsPair(
+        'streams',
+        containsPair('subscriptions', [
+          {
+            'stream': 'a',
+            'parameters': null,
+            'override_priority': null,
+          },
+        ]),
+      ),
+    );
+
+    // Given that the subscription has a TTL, dropping the handle should not
+    // re-subscribe.
+    await subscription.unsubscribe();
+    await pumpEventQueue();
+    expect(syncService.controller.hasListener, isTrue);
+  });
 }