Fix crud upload triggering to only occur on ps_crud changes.

Author: rkistner

packages/powersync/lib/src/database/native/native_powersync_database.dart

@@ -143,7 +143,7 @@ class PowerSyncDatabaseImpl
     await isInitialized;
     final dbRef = database.isolateConnectionFactory();
     ReceivePort rPort = ReceivePort();
-    StreamSubscription? updateSubscription;
+    StreamSubscription? crudUpdateSubscription;
     rPort.listen((data) async {
       if (data is List) {
         String action = data[0];
@@ -160,9 +160,9 @@ class PowerSyncDatabaseImpl
           });
         } else if (action == 'init') {
           SendPort port = data[1];
-          var throttled =
-              UpdateNotification.throttleStream(updates, crudThrottleTime);
-          updateSubscription = throttled.listen((event) {
+          var crudStream =
+              database.onChange(['ps_crud'], throttle: crudThrottleTime);
+          crudUpdateSubscription = crudStream.listen((event) {
             port.send(['update']);
           });
           disconnector.onAbort.then((_) {
@@ -179,7 +179,7 @@ class PowerSyncDatabaseImpl
           // Clear status apart from lastSyncedAt
           setStatus(SyncStatus(lastSyncedAt: currentStatus.lastSyncedAt));
           rPort.close();
-          updateSubscription?.cancel();
+          crudUpdateSubscription?.cancel();
         } else if (action == 'log') {
           LogRecord record = data[1];
           logger.log(
@@ -281,7 +281,7 @@ Future<void> _powerSyncDatabaseIsolate(
     _PowerSyncDatabaseIsolateArgs args) async {
   final sPort = args.sPort;
   ReceivePort rPort = ReceivePort();
-  StreamController updateController = StreamController.broadcast();
+  StreamController<String> crudUpdateController = StreamController.broadcast();
   final upstreamDbClient = args.dbRef.upstreamPort.open();
 
   CommonDatabase? db;
@@ -292,14 +292,14 @@ Future<void> _powerSyncDatabaseIsolate(
     if (message is List) {
       String action = message[0];
       if (action == 'update') {
-        updateController.add('update');
+        crudUpdateController.add('update');
       } else if (action == 'close') {
         // The SyncSqliteConnection uses this mutex
         // It needs to be closed before killing the isolate
         // in order to free the mutex for other operations.
         await mutex.close();
         db?.dispose();
-        updateController.close();
+        crudUpdateController.close();
         upstreamDbClient.close();
         // Abort any open http requests, and wait for it to be closed properly
         await openedStreamingSync?.abort();
@@ -349,7 +349,7 @@ Future<void> _powerSyncDatabaseIsolate(
         credentialsCallback: loadCredentials,
         invalidCredentialsCallback: invalidateCredentials,
         uploadCrud: uploadCrud,
-        updateStream: updateController.stream,
+        crudUpdateTriggerStream: crudUpdateController.stream,
         retryDelay: args.retryDelay,
         client: http.Client(),
         syncParameters: args.parameters);

packages/powersync/lib/src/database/web/web_powersync_database.dart

@@ -135,14 +135,17 @@ class PowerSyncDatabaseImpl
 
     await isInitialized;
 
+    final crudStream =
+        database.onChange(['ps_crud'], throttle: crudThrottleTime);
+
     // TODO better multitab support
     final storage = BucketStorage(database);
     final sync = StreamingSyncImplementation(
         adapter: storage,
         credentialsCallback: connector.getCredentialsCached,
         invalidCredentialsCallback: connector.fetchCredentials,
         uploadCrud: () => connector.uploadData(this),
-        updateStream: updates,
+        crudUpdateTriggerStream: crudStream,
         retryDelay: Duration(seconds: 3),
         client: FetchClient(mode: RequestMode.cors),
         syncParameters: params,

packages/powersync/lib/src/streaming_sync.dart

@@ -27,7 +27,7 @@ class StreamingSyncImplementation {
 
   final Future<void> Function() uploadCrud;
 
-  final Stream updateStream;
+  final Stream crudUpdateTriggerStream;
 
   final StreamController<SyncStatus> _statusStreamController =
       StreamController<SyncStatus>.broadcast();
@@ -59,7 +59,7 @@ class StreamingSyncImplementation {
       required this.credentialsCallback,
       this.invalidCredentialsCallback,
       required this.uploadCrud,
-      required this.updateStream,
+      required this.crudUpdateTriggerStream,
       required this.retryDelay,
       this.syncParameters,
       required http.Client client,
@@ -155,7 +155,7 @@ class StreamingSyncImplementation {
   Future<void> crudLoop() async {
     await uploadAllCrud();
 
-    await for (var _ in updateStream) {
+    await for (var _ in crudUpdateTriggerStream) {
       if (_abort?.aborted == true) {
         break;
       }
@@ -169,11 +169,11 @@ class StreamingSyncImplementation {
       CrudEntry? checkedCrudItem;
 
       while (true) {
-        _updateStatus(uploading: true);
         try {
           // This is the first item in the FIFO CRUD queue.
           CrudEntry? nextCrudItem = await adapter.nextCrudItem();
           if (nextCrudItem != null) {
+            _updateStatus(uploading: true);
             if (nextCrudItem.clientId == checkedCrudItem?.clientId) {
               // This will force a higher log level than exceptions which are caught here.
               isolateLogger.warning(