Add correct offline state

Author: simolus3

crates/core/src/sync/storage_adapter.rs

@@ -15,7 +15,7 @@ use crate::{
         interface::{RequestedStreamSubscription, StreamSubscriptionRequest},
         streaming_sync::{OwnedStreamDescription, RequestedStreamSubscriptions},
         subscriptions::{LocallyTrackedSubscription, StreamKey},
-        sync_status::SyncPriorityStatus,
+        sync_status::{ActiveStreamSubscription, DownloadSyncStatus, SyncPriorityStatus},
     },
     sync_local::{PartialSyncOperation, SyncOperation},
     util::{JsonString, column_nullable},
@@ -87,30 +87,45 @@ impl StorageAdapter {
         Ok(requests)
     }
 
-    pub fn collect_sync_state(&self) -> Result<Vec<SyncPriorityStatus>, PowerSyncError> {
-        // language=SQLite
-        let statement = self
+    pub fn offline_sync_state(&self) -> Result<DownloadSyncStatus, PowerSyncError> {
+        let priority_items = {
+            // language=SQLite
+            let statement = self
             .db
             .prepare_v2(
                 "SELECT priority, unixepoch(last_synced_at) FROM ps_sync_state ORDER BY priority",
             )
             .into_db_result(self.db)?;
 
-        let mut items = Vec::<SyncPriorityStatus>::new();
-        while statement.step()? == ResultCode::ROW {
-            let priority = BucketPriority {
-                number: statement.column_int(0),
-            };
-            let timestamp = statement.column_int64(1);
-
-            items.push(SyncPriorityStatus {
-                priority,
-                last_synced_at: Some(Timestamp(timestamp)),
-                has_synced: Some(true),
-            });
-        }
+            let mut items = Vec::<SyncPriorityStatus>::new();
+            while statement.step()? == ResultCode::ROW {
+                let priority = BucketPriority {
+                    number: statement.column_int(0),
+                };
+                let timestamp = statement.column_int64(1);
 
-        return Ok(items);
+                items.push(SyncPriorityStatus {
+                    priority,
+                    last_synced_at: Some(Timestamp(timestamp)),
+                    has_synced: Some(true),
+                });
+            }
+
+            items
+        };
+
+        let mut streams = Vec::new();
+        self.iterate_local_subscriptions(|sub| {
+            streams.push(ActiveStreamSubscription::from_local(&sub));
+        })?;
+
+        Ok(DownloadSyncStatus {
+            connected: false,
+            connecting: false,
+            priority_status: priority_items,
+            downloading: None,
+            streams,
+        })
     }
 
     pub fn delete_buckets<'a>(

crates/core/src/sync/streaming_sync.rs

@@ -784,12 +784,13 @@ impl StreamingSyncIteration {
         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 {
+
+            if !status.streams.is_empty() {
                 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 {
+                for stream in &mut status.streams {
                     if stream.is_in_priority(priority) {
                         stmt.bind_int64(1, stream.id)?;
                         if stmt.step()? == ResultCode::ROW {
@@ -822,9 +823,12 @@ impl StreamingSyncIteration {
             ));
         };
 
-        let sync_state = self.adapter.collect_sync_state()?;
+        let offline_state = self.adapter.offline_sync_state()?;
         self.status.update(
-            move |s| s.start_connecting(sync_state),
+            move |s| {
+                *s = offline_state;
+                s.start_connecting();
+            },
             &mut event.instructions,
         );
 

crates/core/src/sync/sync_status.rs

@@ -41,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: Option<Vec<ActiveStreamSubscription>>,
+    pub streams: Vec<ActiveStreamSubscription>,
 }
 
 impl DownloadSyncStatus {
@@ -58,11 +58,10 @@ impl DownloadSyncStatus {
         self.downloading = None;
     }
 
-    pub fn start_connecting(&mut self, status: Vec<SyncPriorityStatus>) {
+    pub fn start_connecting(&mut self) {
         self.connected = false;
         self.downloading = None;
         self.connecting = true;
-        self.priority_status = status;
         self.debug_assert_priority_status_is_sorted();
     }
 
@@ -82,7 +81,7 @@ impl DownloadSyncStatus {
         self.mark_connected();
 
         self.downloading = Some(progress);
-        self.streams = Some(subscriptions);
+        self.streams = subscriptions;
     }
 
     /// Increments [SyncDownloadProgress] progress for the given [DataLine].
@@ -126,7 +125,7 @@ impl Default for DownloadSyncStatus {
             connecting: false,
             downloading: None,
             priority_status: Vec::new(),
-            streams: None,
+            streams: Vec::new(),
         }
     }
 }

dart/test/goldens/simple_iteration.json

@@ -10,7 +10,7 @@
             "connecting": true,
             "priority_status": [],
             "downloading": null,
-            "streams": null
+            "streams": []
           }
         }
       },

dart/test/goldens/starting_stream.json

@@ -14,7 +14,7 @@
             "connecting": true,
             "priority_status": [],
             "downloading": null,
-            "streams": null
+            "streams": []
           }
         }
       },

dart/test/sync_test.dart

@@ -703,7 +703,9 @@ void _syncTests<T>({
                   "Checksums didn't match, failed for: a (expected 0x000004d2, got 0x000010e1 = 0x000010e1 (op) + 0x00000000 (add))")
             }
           },
-          {'CloseSyncStream': {}},
+          {
+            'CloseSyncStream': {'hide_disconnect': false}
+          },
         ],
       );