Fix bad merges

Author: simolus3

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

@@ -123,13 +123,9 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   Future<void> _updateHasSynced() async {
     // Query the database to see if any data has been synced.
     final result = await database.get('''
-      SELECT CASE
-        WHEN EXISTS (SELECT 1 FROM sqlite_master WHERE name = 'ps_sync_state')
-          THEN (SELECT json_group_array(
-            json_object('prio', priority, 'last_sync', last_synced_at)
-          ) FROM ps_sync_state ORDER BY priority)
-          ELSE powersync_last_synced_at()
-        END AS r;
+      SELECT json_group_array(
+        json_object('prio', priority, 'last_sync', last_synced_at)
+      ) AS r FROM ps_sync_state ORDER BY priority
     ''');
     final value = result['r'] as String?;
     final hasData = value != null;

packages/powersync_core/lib/src/streaming_sync.dart

@@ -444,8 +444,6 @@ class StreamingSyncImplementation implements StreamingSync {
               hasSynced: true,
             ));
           }
-
-          validatedCheckpoint = targetCheckpoint;
         case StreamingSyncCheckpointDiff():
           // TODO: It may be faster to just keep track of the diff, instead of
           // the entire checkpoint
@@ -479,12 +477,12 @@ class StreamingSyncImplementation implements StreamingSync {
         case SyncDataBatch():
           _updateStatus(downloading: true);
           await adapter.saveSyncData(line);
-        case StreamingSyncKeepalive():
-          if (line.tokenExpiresIn == 0) {
+        case StreamingSyncKeepalive(:final tokenExpiresIn):
+          if (tokenExpiresIn == 0) {
             // Token expired already - stop the connection immediately
             invalidCredentialsCallback?.call().ignore();
             break;
-          } else if (line.tokenExpiresIn <= 30) {
+          } else if (tokenExpiresIn <= 30) {
             // Token expires soon - refresh it in the background
             if (credentialsInvalidation == null &&
                 invalidCredentialsCallback != null) {
@@ -501,8 +499,7 @@ class StreamingSyncImplementation implements StreamingSync {
             }
           }
         case UnknownSyncLine(:final rawData):
-          isolateLogger.fine('Ignoring unknown sync line: $rawData');
-          break;
+          isolateLogger.fine('Unknown sync line: $rawData');
         case null: // Local ping
           if (targetCheckpoint == appliedCheckpoint) {
             _updateStatus(

packages/powersync_core/test/in_memory_sync_test.dart

@@ -102,6 +102,66 @@ void main() {
           isTrue);
     });
 
+    test('can save independent buckets in same transaction', () async {
+      final status = await waitForConnection();
+
+      syncService.addLine({
+        'checkpoint': Checkpoint(
+          lastOpId: '0',
+          writeCheckpoint: null,
+          checksums: [
+            BucketChecksum(bucket: 'a', checksum: 0, priority: 3),
+            BucketChecksum(bucket: 'b', checksum: 0, priority: 3),
+          ],
+        )
+      });
+      await expectLater(status, emits(isSyncStatus(downloading: true)));
+
+      var commits = 0;
+      raw.commits.listen((_) => commits++);
+
+      syncService
+        ..addLine({
+          'data': {
+            'bucket': 'a',
+            'data': <Map<String, Object?>>[
+              {
+                'op_id': '1',
+                'op': 'PUT',
+                'object_type': 'a',
+                'object_id': '1',
+                'checksum': 0,
+                'data': {},
+              }
+            ],
+          }
+        })
+        ..addLine({
+          'data': {
+            'bucket': 'b',
+            'data': <Map<String, Object?>>[
+              {
+                'op_id': '2',
+                'op': 'PUT',
+                'object_type': 'b',
+                'object_id': '1',
+                'checksum': 0,
+                'data': {},
+              }
+            ],
+          }
+        });
+
+      // Wait for the operations to be inserted.
+      while (raw.select('SELECT * FROM ps_oplog;').length < 2) {
+        await pumpEventQueue();
+      }
+
+      // The two buckets should have been inserted in a single transaction
+      // because the messages were received in quick succession.
+      expect(commits, 1);
+    });
+
     group('partial sync', () {
       test('updates sync state incrementally', () async {
         final status = await waitForConnection();

packages/powersync_core/test/utils/native_test_utils.dart

@@ -23,19 +23,19 @@ class TestOpenFactory extends PowerSyncOpenFactory with TestPowerSyncFactory {
     });
   }
 
+  @override
+  CommonDatabase open(SqliteOpenOptions options) {
+    applyOpenOverride();
+    return super.open(options);
+  }
+
   @override
   void enableExtension() {
     var powersyncLib = getLibraryForPlatform();
     sqlite3.ensureExtensionLoaded(SqliteExtension.inLibrary(
         DynamicLibrary.open(powersyncLib), 'sqlite3_powersync_init'));
   }
 
-  @override
-  CommonDatabase open(SqliteOpenOptions options) {
-    applyOpenOverride();
-    return super.open(options);
-  }
-
   @override
   String getLibraryForPlatform({String? path = "."}) {
     switch (Abi.current()) {