Update last_synced_at for subscriptions

Author: simolus3

crates/core/src/sync/streaming_sync.rs

@@ -31,7 +31,7 @@ use crate::{
         sync_status::Timestamp,
     },
 };
-use sqlite_nostd::{self as sqlite};
+use sqlite_nostd::{self as sqlite, Connection, ResultCode};
 
 use super::{
     interface::{Instruction, LogSeverity, StreamingSyncRequest, SyncControlRequest, SyncEvent},
@@ -307,9 +307,7 @@ impl StreamingSyncIteration {
                     ));
                 };
                 let target = &checkpoint.checkpoint;
-                let result =
-                    self.adapter
-                        .sync_local(&self.state, target, None, &self.options.schema)?;
+                let result = self.sync_local(target, None)?;
 
                 match result {
                     SyncLocalResult::ChecksumFailure(checkpoint_result) => {
@@ -352,12 +350,7 @@ impl StreamingSyncIteration {
                         "Received checkpoint complete without previous checkpoint",
                     ));
                 };
-                let result = self.adapter.sync_local(
-                    &self.state,
-                    &target.checkpoint,
-                    Some(priority),
-                    &self.options.schema,
-                )?;
+                let result = self.sync_local(&target.checkpoint, Some(priority))?;
 
                 match result {
                     SyncLocalResult::ChecksumFailure(checkpoint_result) => {
@@ -503,12 +496,7 @@ impl StreamingSyncIteration {
                     .map_err(|e| PowerSyncError::sync_protocol_error("invalid binary line", e))?,
                 SyncEvent::UploadFinished => {
                     if let Some(checkpoint) = self.validated_but_not_applied.take() {
-                        let result = self.adapter.sync_local(
-                            &self.state,
-                            &checkpoint,
-                            None,
-                            &self.options.schema,
-                        )?;
+                        let result = self.sync_local(&checkpoint, None)?;
 
                         match result {
                             SyncLocalResult::ChangesApplied => {
@@ -629,17 +617,13 @@ impl StreamingSyncIteration {
                         if let Ok(index) =
                             tracked_subscriptions.binary_search_by_key(subscription_id, |s| s.id)
                         {
-                            resolved[index]
-                                .associated_buckets
-                                .push(bucket.bucket.clone());
+                            resolved[index].mark_associated_with_bucket(&bucket);
                         }
                     }
                 }
                 BucketSubscriptionReason::IsDefault { stream_name } => {
                     if let Some(index) = default_stream_subscriptions.get(stream_name.as_str()) {
-                        resolved[*index]
-                            .associated_buckets
-                            .push(bucket.bucket.clone());
+                        resolved[*index].mark_associated_with_bucket(&bucket);
                     }
                 }
                 BucketSubscriptionReason::Unknown => {}
@@ -649,6 +633,41 @@ impl StreamingSyncIteration {
         Ok(resolved)
     }
 
+    /// Performs a partial or a complete local sync.
+    fn sync_local(
+        &self,
+        target: &OwnedCheckpoint,
+        priority: Option<BucketPriority>,
+    ) -> Result<SyncLocalResult, PowerSyncError> {
+        let result =
+            self.adapter
+                .sync_local(&self.state, target, priority, &self.options.schema)?;
+
+        if matches!(&result, SyncLocalResult::ChangesApplied) {
+            // Update affected stream subscriptions to mark them as synced.
+            let mut status = self.status.inner().borrow_mut();
+            if let Some(ref mut streams) = status.streams {
+                let stmt = self.adapter.db.prepare_v2(
+                    "UPDATE ps_stream_subscriptions SET last_synced_at = unixepoch() WHERE id = ? RETURNING last_synced_at",
+                )?;
+
+                for stream in streams {
+                    if stream.is_in_priority(priority) {
+                        stmt.bind_int64(1, stream.id)?;
+                        if stmt.step()? == ResultCode::ROW {
+                            let timestamp = Timestamp(stmt.column_int64(0));
+                            stream.last_synced_at = Some(timestamp);
+                        }
+
+                        stmt.reset()?;
+                    }
+                }
+            }
+        }
+
+        Ok(result)
+    }
+
     /// Prepares a sync iteration by handling the initial [SyncEvent::Initialize].
     ///
     /// This prepares a [StreamingSyncRequest] by fetching local sync state and the requested bucket

crates/core/src/sync/sync_status.rs

@@ -1,14 +1,18 @@
 use alloc::{boxed::Box, collections::btree_map::BTreeMap, rc::Rc, string::String, vec::Vec};
 use core::{
     cell::RefCell,
+    cmp::min,
     hash::{BuildHasher, Hash},
 };
 use rustc_hash::FxBuildHasher;
 use serde::Serialize;
 use sqlite_nostd::ResultCode;
 
 use crate::{
-    sync::{storage_adapter::StorageAdapter, subscriptions::LocallyTrackedSubscription},
+    sync::{
+        checkpoint::OwnedBucketChecksum, storage_adapter::StorageAdapter,
+        subscriptions::LocallyTrackedSubscription,
+    },
     util::JsonString,
 };
 
@@ -37,7 +41,7 @@ pub struct DownloadSyncStatus {
     /// When a download is active (that is, a `checkpoint` or `checkpoint_diff` line has been
     /// received), information about how far the download has progressed.
     pub downloading: Option<SyncDownloadProgress>,
-    pub streams: Vec<ActiveStreamSubscription>,
+    pub streams: Option<Vec<ActiveStreamSubscription>>,
 }
 
 impl DownloadSyncStatus {
@@ -78,7 +82,7 @@ impl DownloadSyncStatus {
         self.mark_connected();
 
         self.downloading = Some(progress);
-        self.streams = subscriptions;
+        self.streams = Some(subscriptions);
     }
 
     /// Increments [SyncDownloadProgress] progress for the given [DataLine].
@@ -122,7 +126,7 @@ impl Default for DownloadSyncStatus {
             connecting: false,
             downloading: None,
             priority_status: Vec::new(),
-            streams: Vec::new(),
+            streams: None,
         }
     }
 }
@@ -140,6 +144,10 @@ impl SyncStatusContainer {
         }
     }
 
+    pub fn inner(&self) -> &Rc<RefCell<DownloadSyncStatus>> {
+        &self.status
+    }
+
     /// Invokes a function to update the sync status, then emits an [Instruction::UpdateSyncStatus]
     /// if the function did indeed change the status.
     pub fn update<F: FnOnce(&mut DownloadSyncStatus) -> ()>(
@@ -265,9 +273,12 @@ impl SyncDownloadProgress {
 
 #[derive(Serialize, Hash)]
 pub struct ActiveStreamSubscription {
+    #[serde(skip)]
+    pub id: i64,
     pub name: String,
     pub parameters: Option<Box<JsonString>>,
     pub associated_buckets: Vec<String>,
+    pub priority: Option<BucketPriority>,
     pub active: bool,
     pub is_default: bool,
     pub expires_at: Option<Timestamp>,
@@ -277,13 +288,30 @@ pub struct ActiveStreamSubscription {
 impl ActiveStreamSubscription {
     pub fn from_local(local: &LocallyTrackedSubscription) -> Self {
         Self {
+            id: local.id,
             name: local.stream_name.clone(),
             parameters: local.local_params.clone(),
             is_default: local.is_default,
+            priority: None,
             associated_buckets: Vec::new(),
             active: local.active,
             expires_at: local.expires_at.clone().map(|e| Timestamp(e)),
             last_synced_at: local.last_synced_at.map(|e| Timestamp(e)),
         }
     }
+
+    pub fn mark_associated_with_bucket(&mut self, bucket: &OwnedBucketChecksum) {
+        self.associated_buckets.push(bucket.bucket.clone());
+        self.priority = Some(match self.priority {
+            None => bucket.priority,
+            Some(prio) => min(prio, bucket.priority),
+        });
+    }
+
+    pub fn is_in_priority(&self, prio: Option<BucketPriority>) -> bool {
+        match prio {
+            None => true,
+            Some(prio) => self.priority >= Some(prio),
+        }
+    }
 }

dart/test/sync_stream_test.dart

@@ -58,7 +58,7 @@ void main() {
   }
 
   group('default streams', () {
-    test('are created on-demand', () {
+    syncTest('are created on-demand', (_) {
       control('start', null);
       control(
         'line_text',
@@ -86,11 +86,28 @@ void main() {
               'active': true,
               'is_default': true,
               'expires_at': null,
-              'last_synced_at': null
+              'last_synced_at': null,
+              'priority': 1,
             }
           ],
         ),
       );
+
+      control(
+        'line_text',
+        json.encode(checkpointComplete(priority: 1)),
+      );
+
+      expect(
+        lastStatus,
+        containsPair(
+          'streams',
+          [containsPair('last_synced_at', 1740823200)],
+        ),
+      );
+
+      final [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(stored, containsPair('last_synced_at', 1740823200));
     });
   });
 }

dart/test/sync_test.dart

@@ -126,13 +126,7 @@ void _syncTests<T>({
   }
 
   List<Object?> pushCheckpointComplete({int? priority, String lastOpId = '1'}) {
-    return syncLine({
-      priority == null ? 'checkpoint_complete' : 'partial_checkpoint_complete':
-          {
-        'last_op_id': lastOpId,
-        if (priority != null) 'priority': priority,
-      },
-    });
+    return syncLine(checkpointComplete(priority: priority, lastOpId: lastOpId));
   }
 
   ResultSet fetchRows() {

dart/test/utils/test_utils.dart

@@ -21,6 +21,16 @@ Object checkpoint({
   };
 }
 
+/// Creates a `checkpoint_complete` or `partial_checkpoint_complete` line.
+Object checkpointComplete({int? priority, String lastOpId = '1'}) {
+  return {
+    priority == null ? 'checkpoint_complete' : 'partial_checkpoint_complete': {
+      'last_op_id': lastOpId,
+      if (priority != null) 'priority': priority,
+    },
+  };
+}
+
 Object bucketDescription(
   String name, {
   int checksum = 0,