Add offline sync state helper function

Author: simolus3

crates/core/src/sync/interface.rs

@@ -4,9 +4,11 @@ use core::ffi::{c_int, c_void};
 use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
 use crate::constants::SUBTYPE_JSON;
+use crate::create_sqlite_text_fn;
 use crate::error::PowerSyncError;
 use crate::schema::Schema;
 use crate::state::DatabaseState;
+use crate::sync::storage_adapter::StorageAdapter;
 use crate::sync::subscriptions::{StreamKey, apply_subscriptions};
 use alloc::borrow::Cow;
 use alloc::boxed::Box;
@@ -285,5 +287,33 @@ pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(
         Some(destroy),
     )?;
 
+    db.create_function_v2(
+        "powersync_offline_sync_status",
+        0,
+        sqlite::UTF8 | sqlite::DIRECTONLY | SQLITE_RESULT_SUBTYPE,
+        None,
+        Some(powersync_offline_sync_status),
+        None,
+        None,
+        None,
+    )?;
+
     Ok(())
 }
+
+fn powersync_offline_sync_status_impl(
+    ctx: *mut sqlite::context,
+    _args: &[*mut sqlite::value],
+) -> Result<String, PowerSyncError> {
+    let adapter = StorageAdapter::new(ctx.db_handle())?;
+    let state = adapter.offline_sync_state()?;
+    let serialized = serde_json::to_string(&state).map_err(PowerSyncError::internal)?;
+
+    Ok(serialized)
+}
+
+create_sqlite_text_fn!(
+    powersync_offline_sync_status,
+    powersync_offline_sync_status_impl,
+    "powersync_offline_sync_status"
+);

dart/test/sync_stream_test.dart

@@ -591,4 +591,61 @@ void main() {
     expect(instructions,
         contains(containsPair('CloseSyncStream', {'hide_disconnect': true})));
   });
+
+  syncTest('persists stream state', (_) {
+    control(
+      'subscriptions',
+      json.encode({
+        'subscribe': {
+          'stream': {'name': 'a'},
+        }
+      }),
+    );
+
+    control(
+      'start',
+      json.encode({
+        'active_streams': [
+          {'name': 'a'}
+        ]
+      }),
+    );
+    control(
+      'line_text',
+      json.encode(
+        checkpoint(
+          lastOpId: 1,
+          buckets: [
+            bucketDescription(
+              'a',
+              subscriptions: [
+                {'sub': 0}
+              ],
+              priority: 1,
+            )
+          ],
+          streams: [stream('a', false)],
+        ),
+      ),
+    );
+    control('line_text', json.encode(checkpointComplete()));
+
+    final [row] = db.select('select powersync_offline_sync_status();');
+    expect(
+        json.decode(row[0]),
+        containsPair('streams', [
+          {
+            'name': 'a',
+            'parameters': null,
+            // not persisted, only needed for download progress
+            'associated_buckets': [],
+            'priority': null, // same
+            'active': true,
+            'is_default': false,
+            'has_explicit_subscription': true,
+            'expires_at': 1740909600,
+            'last_synced_at': 1740823200
+          }
+        ]));
+  });
 }

dart/test/sync_test.dart

@@ -317,6 +317,22 @@ void _syncTests<T>({
         ),
       ),
     );
+
+    final [row] = db.select('select powersync_offline_sync_status();');
+    expect(json.decode(row[0]), {
+      'connected': false,
+      'connecting': false,
+      'priority_status': [
+        {'priority': 2, 'last_synced_at': 1740823800, 'has_synced': true},
+        {
+          'priority': 2147483647,
+          'last_synced_at': 1740823200,
+          'has_synced': true
+        }
+      ],
+      'downloading': null,
+      'streams': [],
+    });
   });
 
   test('clearing database clears sync status', () {