Sync streams (#112)

* Start adding support for sync streams

* Add protocol changes

* Use serde_with

* Start with subscription logic

* Start tracking subscriptions

* Track subscriptions

* Test handling default streams

* Update last_synced_at for subscriptions

* Allow subscribing to streams

* Expire subscriptions after TTL

* Support unsubscribing

* Delete outdated subscriptions

* Include default ttl

* New protocol format

* Fix tests

* More stream management tests

* Remove immediate parameter when unsubscribing

* Increase expires_at only when subscribing again

* Implement new protocol format

* Report errors

* Update TTL behavior

* Refresh on keepalive

* Instruction to update expiry

* Add correct offline state

* Add offline sync state helper function

* Simplify error reporting

* Improve comment

* Use set for associated buckets

* Compute progress in core extension

Author: simolus3

Cargo.lock

@@ -23,6 +23,7 @@ const_format = "0.2.34"
 futures-lite = { version = "2.6.0", default-features = false, features = ["alloc"] }
 rustc-hash = { version = "2.1", default-features = false }
 thiserror = { version = "2", default-features = false }
+serde_with = { version = "3.14.0", default-features = false, features = ["alloc", "macros"] }
 
 [dependencies.uuid]
 version = "1.4.1"

crates/core/Cargo.toml

@@ -12,7 +12,7 @@ use crate::error::{PSResult, PowerSyncError};
 use crate::fix_data::apply_v035_fix;
 use crate::sync::BucketPriority;
 
-pub const LATEST_VERSION: i32 = 10;
+pub const LATEST_VERSION: i32 = 11;
 
 pub fn powersync_migrate(
     ctx: *mut sqlite::context,
@@ -384,5 +384,28 @@ INSERT INTO ps_migration(id, down_migrations) VALUES (10, json_array(
             .into_db_result(local_db)?;
     }
 
+    if current_version < 11 && target_version >= 11 {
+        let stmt = "\
+CREATE TABLE ps_stream_subscriptions (
+  id INTEGER NOT NULL PRIMARY KEY,
+  stream_name TEXT NOT NULL,
+  active INTEGER NOT NULL DEFAULT FALSE,
+  is_default INTEGER NOT NULL DEFAULT FALSE,
+  local_priority INTEGER,
+  local_params TEXT NOT NULL DEFAULT 'null',
+  ttl INTEGER,
+  expires_at INTEGER,
+  last_synced_at INTEGER,
+  UNIQUE (stream_name, local_params)
+) STRICT;
+
+INSERT INTO ps_migration(id, down_migrations) VALUES(11, json_array(
+json_object('sql', 'DROP TABLE ps_stream_subscriptions'),
+json_object('sql', 'DELETE FROM ps_migration WHERE id >= 11')
+));
+";
+        local_db.exec_safe(stmt).into_db_result(local_db)?;
+    }
+
     Ok(())
 }

crates/core/src/migrations.rs

@@ -1,7 +1,8 @@
-use alloc::{string::String, vec::Vec};
+use alloc::{rc::Rc, string::String, vec::Vec};
 use num_traits::Zero;
 
-use crate::sync::{BucketPriority, Checksum, line::BucketChecksum};
+use crate::sync::line::{BucketChecksum, BucketSubscriptionReason};
+use crate::sync::{BucketPriority, Checksum};
 use sqlite_nostd::{self as sqlite, Connection, ResultCode};
 
 /// A structure cloned from [BucketChecksum]s with an owned bucket name instead of one borrowed from
@@ -12,6 +13,7 @@ pub struct OwnedBucketChecksum {
     pub checksum: Checksum,
     pub priority: BucketPriority,
     pub count: Option<i64>,
+    pub subscriptions: Rc<Vec<BucketSubscriptionReason>>,
 }
 
 impl OwnedBucketChecksum {
@@ -30,6 +32,7 @@ impl From<&'_ BucketChecksum<'_>> for OwnedBucketChecksum {
             checksum: value.checksum,
             priority: value.priority.unwrap_or(BucketPriority::FALLBACK),
             count: value.count,
+            subscriptions: value.subscriptions.clone(),
         }
     }
 }