Handle keepalive messages

Author: simolus3

crates/core/src/sync/interface.rs

@@ -27,6 +27,7 @@ pub enum SyncControlRequest<'a> {
 pub enum SyncEvent<'a> {
     Initialize,
     TearDown,
+    DidRefreshToken,
     TextLine { data: &'a str },
     BinaryLine { data: &'a [u8] },
 }
@@ -44,10 +45,14 @@ pub enum Instruction {
     EstablishSyncStream {
         request: StreamingSyncRequest,
     },
-    // These two are defined like this because deserializers in Kotlin can't support either an
+    FetchCredentials {
+        did_expire: bool,
+    },
+    // These are defined like this because deserializers in Kotlin can't support either an
     // object or a literal value
-    FlushFileSystem {},
     CloseSyncStream {},
+    FlushFileSystem {},
+    DidCompleteSync {},
 }
 
 #[derive(Serialize)]
@@ -129,6 +134,7 @@ pub fn register(db: *mut sqlite::sqlite3) -> Result<(), ResultCode> {
                         ));
                     },
                 }),
+                "refreshed_token" => SyncControlRequest::SyncEvent(SyncEvent::DidRefreshToken),
                 _ => {
                     return Err(SQLiteError(
                         ResultCode::MISUSE,

crates/core/src/sync/line.rs

@@ -118,9 +118,20 @@ pub enum OpType {
     REMOVE,
 }
 
-#[derive(Deserialize, Debug)]
+#[repr(transparent)]
+#[derive(Deserialize, Debug, Clone, Copy)]
 pub struct TokenExpiresIn(pub i32);
 
+impl TokenExpiresIn {
+    pub fn is_expired(self) -> bool {
+        self.0 <= 0
+    }
+
+    pub fn should_prefetch(self) -> bool {
+        !self.is_expired() && self.0 <= 30
+    }
+}
+
 impl<'a, 'de: 'a> Deserialize<'de> for OplogData<'a> {
     fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
     where

crates/core/src/sync/streaming_sync.rs

@@ -223,6 +223,10 @@ impl StreamingSyncIteration {
                 SyncEvent::TearDown => break,
                 SyncEvent::TextLine { data } => serde_json::from_str(data)?,
                 SyncEvent::BinaryLine { data } => bson::from_bytes(data)?,
+                SyncEvent::DidRefreshToken => {
+                    // Break so that the client SDK starts another iteration.
+                    break;
+                }
             };
 
             match line {
@@ -290,8 +294,9 @@ impl StreamingSyncIteration {
                         SyncLocalResult::ChangesApplied => {
                             event.instructions.push(Instruction::LogLine {
                                 severity: LogSeverity::DEBUG,
-                                line: format!("Validated checkpoint"),
+                                line: format!("Validated and applied checkpoint"),
                             });
+                            event.instructions.push(Instruction::DidCompleteSync {});
 
                             let now = self.adapter.now()?;
                             self.status.update(
@@ -347,7 +352,19 @@ impl StreamingSyncIteration {
                         .update(|s| s.track_line(&data_line), &mut event.instructions);
                     insert_bucket_operations(&self.adapter, &data_line)?;
                 }
-                SyncLine::KeepAlive(token_expires_in) => todo!(),
+                SyncLine::KeepAlive(token) => {
+                    if token.is_expired() {
+                        // Token expired already - stop the connection immediately.
+                        event
+                            .instructions
+                            .push(Instruction::FetchCredentials { did_expire: true });
+                        break;
+                    } else if token.should_prefetch() {
+                        event
+                            .instructions
+                            .push(Instruction::FetchCredentials { did_expire: false });
+                    }
+                }
             }
         }
 

crates/core/src/sync/sync_status.rs

@@ -73,15 +73,8 @@ impl DownloadSyncStatus {
         self.downloading = None;
         self.priority_status.clear();
 
-        let lowest_priority = applied
-            .buckets
-            .values()
-            .map(|bkt| bkt.priority)
-            .max()
-            .unwrap_or(BucketPriority::SENTINEL);
-
         self.priority_status.push(SyncPriorityStatus {
-            priority: lowest_priority,
+            priority: BucketPriority::SENTINEL,
             last_synced_at: Some(now),
             has_synced: Some(true),
         });