From 836dec5ff7dbb097707e615e25dd10f8e314868d Mon Sep 17 00:00:00 2001
From: Simon Binder <simon@journeyapps.com>
Date: Wed, 15 Jan 2025 14:43:51 +0100
Subject: [PATCH] Avoid emitting duplicate sync status events

---
 packages/powersync_core/CHANGELOG.md          |  4 ++++
 .../lib/src/database/powersync_db_mixin.dart  | 17 ++++++++++----
 packages/powersync_core/pubspec.yaml          |  2 +-
 .../test/powersync_shared_test.dart           | 23 +++++++++++++++++++
 4 files changed, 40 insertions(+), 6 deletions(-)

diff --git a/packages/powersync_core/CHANGELOG.md b/packages/powersync_core/CHANGELOG.md
index c70b5525..8d0ee49b 100644
--- a/packages/powersync_core/CHANGELOG.md
+++ b/packages/powersync_core/CHANGELOG.md
@@ -1,3 +1,7 @@
+## 1.1.1-dev
+
+- Fix `statusStream` emitting the same sync status multiple times.
+
 ## 1.1.0
 
  - Increase limit on number of columns per table to 1999.
diff --git a/packages/powersync_core/lib/src/database/powersync_db_mixin.dart b/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
index bc359616..a5fb93a5 100644
--- a/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
+++ b/packages/powersync_core/lib/src/database/powersync_db_mixin.dart
@@ -148,17 +148,23 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   }
 
   @protected
+  @visibleForTesting
   void setStatus(SyncStatus status) {
     if (status != currentStatus) {
-      currentStatus = status.copyWith(
-          // Note that currently the streaming sync implementation will never set hasSynced.
-          // lastSyncedAt implies that syncing has completed at some point (hasSynced = true).
-          // The previous values of hasSynced should be preserved here.
+      // Note that currently the streaming sync implementation will never set hasSynced.
+      // lastSyncedAt implies that syncing has completed at some point (hasSynced = true).
+      // The previous values of hasSynced should be preserved here.
+      final newStatus = status.copyWith(
           hasSynced: status.lastSyncedAt != null
               ? true
               : status.hasSynced ?? currentStatus.hasSynced,
           lastSyncedAt: status.lastSyncedAt ?? currentStatus.lastSyncedAt);
-      statusStreamController.add(currentStatus);
+      // If the absence of hasSync was the only difference, the new states would be equal
+      // and don't require an event. So, check again.
+      if (newStatus != currentStatus) {
+        currentStatus = newStatus;
+        statusStreamController.add(currentStatus);
+      }
     }
   }
 
@@ -181,6 +187,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
     await disconnect();
     // Now we can close the database
     await database.close();
+    await statusStreamController.close();
   }
 
   /// Connect to the PowerSync service, and keep the databases in sync.
diff --git a/packages/powersync_core/pubspec.yaml b/packages/powersync_core/pubspec.yaml
index cf972f47..b2f91b01 100644
--- a/packages/powersync_core/pubspec.yaml
+++ b/packages/powersync_core/pubspec.yaml
@@ -1,5 +1,5 @@
 name: powersync_core
-version: 1.1.0
+version: 1.1.1-dev
 homepage: https://powersync.com
 repository: https://github.com/powersync-ja/powersync.dart
 description: PowerSync Dart SDK - sync engine for building local-first apps.
diff --git a/packages/powersync_core/test/powersync_shared_test.dart b/packages/powersync_core/test/powersync_shared_test.dart
index 26b458d8..34405eb6 100644
--- a/packages/powersync_core/test/powersync_shared_test.dart
+++ b/packages/powersync_core/test/powersync_shared_test.dart
@@ -1,3 +1,4 @@
+import 'package:powersync_core/powersync_core.dart';
 import 'package:sqlite_async/mutex.dart';
 import 'package:test/test.dart';
 import 'package:uuid/parsing.dart';
@@ -89,5 +90,27 @@ void main() {
       // Check that it is a valid uuid
       UuidParsing.parseAsByteList(id);
     });
+
+    test('does not emit duplicate sync status events', () async {
+      final db = await testUtils.setupPowerSync(path: path);
+      expectLater(
+        db.statusStream,
+        emitsInOrder(
+          [
+            // Manual setStatus call. hasSynced set to true because lastSyncedAt is set
+            isA<SyncStatus>().having((e) => e.hasSynced, 'hasSynced', true),
+            // Closing the database emits a disconnected status
+            isA<SyncStatus>().having((e) => e.connected, 'connected', false),
+            emitsDone
+          ],
+        ),
+      );
+
+      final status = SyncStatus(connected: true, lastSyncedAt: DateTime.now());
+      db.setStatus(status);
+      db.setStatus(status); // Should not re-emit!
+
+      await db.close();
+    });
   });
 }