Update subscription state while offline

Author: simolus3

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

@@ -1,5 +1,4 @@
 import 'dart:async';
-import 'dart:convert';
 
 import 'package:logging/logging.dart';
 import 'package:meta/meta.dart';
@@ -15,8 +14,6 @@ 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';
 
@@ -108,7 +105,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     await _checkVersion();
     await database.execute('SELECT powersync_init()');
     await updateSchema(schema);
-    await _updateHasSynced();
+    await _connections.resolveOfflineSyncStatus();
   }
 
   /// Check that a supported version of the powersync extension is loaded.
@@ -134,23 +131,6 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     return isInitialized;
   }
 
-  Future<List<SyncStream>> get subscribedStreams {
-    throw UnimplementedError();
-  }
-
-  Future<void> _updateHasSynced() async {
-    // Query the database to see if any data has been synced.
-    final row = await database.get(
-      'SELECT powersync_offline_sync_status() AS r;',
-    );
-
-    final status = CoreSyncStatus.fromJson(
-        json.decode(row['r'] as String) as Map<String, Object?>);
-
-    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).

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

@@ -10,6 +10,8 @@ import 'package:powersync_core/src/sync/options.dart';
 import 'package:powersync_core/src/sync/stream.dart';
 import 'package:powersync_core/src/sync/sync_status.dart';
 
+import 'instruction.dart';
+import 'mutable_sync_status.dart';
 import 'streaming_sync.dart';
 
 /// A (stream name, JSON parameters) pair that uniquely identifies a stream
@@ -252,6 +254,15 @@ final class ConnectionManager {
         'priority': priority,
       },
     });
+
+    await _activeGroup.syncMutex.lock(() async {
+      if (_abortActiveSync == null) {
+        // Since we're not connected, update the offline sync status to reflect
+        // the new subscription.
+        // With a connection, the sync client would include it in its state.
+        await resolveOfflineSyncStatus();
+      }
+    });
   }
 
   Future<void> unsubscribeAll({
@@ -266,6 +277,18 @@ final class ConnectionManager {
     });
   }
 
+  Future<void> resolveOfflineSyncStatus() async {
+    final row = await db.database.get(
+      'SELECT powersync_offline_sync_status() AS r;',
+    );
+
+    final status = CoreSyncStatus.fromJson(
+        json.decode(row['r'] as String) as Map<String, Object?>);
+
+    manuallyChangeSyncStatus((MutableSyncStatus()..applyFromCore(status))
+        .immutableSnapshot(setLastSynced: true));
+  }
+
   SyncStream syncStream(String name, Map<String, Object?>? parameters) {
     return _SyncStreamImplementation(this, name, parameters);
   }

packages/powersync_core/test/sync/stream_test.dart

@@ -211,4 +211,12 @@ void main() {
     await pumpEventQueue();
     expect(syncService.controller.hasListener, isTrue);
   });
+
+  test('subscriptions update while offline', () async {
+    final stream = StreamQueue(database.statusStream);
+
+    final subscription = await database.syncStream('foo').subscribe();
+    var status = await stream.next;
+    expect(status.statusFor(subscription), isNotNull);
+  });
 }