Start tracking subscriptions

Author: simolus3

crates/core/src/migrations.rs

@@ -394,11 +394,11 @@ CREATE TABLE ps_stream_subscriptions (
   local_params TEXT,
   ttl INTEGER
 ) STRICT;
-ALTER TABLE ps_buckets ADD COLUMN derived_from INTEGER REFERENCES ps_streams (id);
+ALTER TABLE ps_buckets ADD COLUMN from_subscriptions TEXT NOT NULL DEFAULT '[null]';
 
-INSERT INTO ps_migration(id, down_migrations) VALUES(9, json_array(
+INSERT INTO ps_migration(id, down_migrations) VALUES(11, json_array(
 json_object('sql', 'todo down migration'),
-json_object('sql', 'DELETE FROM ps_migration WHERE id >= 10')
+json_object('sql', 'DELETE FROM ps_migration WHERE id >= 11')
 ));
 ";
         local_db.exec_safe(stmt)?;

crates/core/src/sync/interface.rs

@@ -20,13 +20,31 @@ use super::streaming_sync::SyncClient;
 use super::sync_status::DownloadSyncStatus;
 
 /// Payload provided by SDKs when requesting a sync iteration.
-#[derive(Default, Deserialize)]
+#[derive(Deserialize)]
 pub struct StartSyncStream {
     /// Bucket parameters to include in the request when opening a sync stream.
     #[serde(default)]
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
     #[serde(default)]
     pub schema: Schema,
+    #[serde(default = "StartSyncStream::include_defaults_by_default")]
+    pub include_defaults: bool,
+}
+
+impl StartSyncStream {
+    pub const fn include_defaults_by_default() -> bool {
+        true
+    }
+}
+
+impl Default for StartSyncStream {
+    fn default() -> Self {
+        Self {
+            parameters: Default::default(),
+            schema: Default::default(),
+            include_defaults: Self::include_defaults_by_default(),
+        }
+    }
 }
 
 /// A request sent from a client SDK to the [SyncClient] with a `powersync_control` invocation.
@@ -107,6 +125,7 @@ pub struct StreamingSyncRequest {
     pub binary_data: bool,
     pub client_id: String,
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
+    pub streams: StreamSubscriptionRequest,
 }
 
 #[derive(Serialize)]

crates/core/src/sync/line.rs

@@ -120,9 +120,9 @@ pub struct BucketChecksum<'a> {
     pub priority: Option<BucketPriority>,
     #[serde(default)]
     pub count: Option<i64>,
-    #[serde_as(as = "Option<Vec<DisplayFromStr>>")]
+    #[serde_as(as = "Vec<Option<DisplayFromStr>>")]
     #[serde(default)]
-    pub subscriptions: Option<Vec<i64>>,
+    pub subscriptions: Vec<Option<i64>>,
     //    #[serde(default)]
     //    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
     //    pub last_op_id: Option<i64>,

crates/core/src/sync/storage_adapter.rs

@@ -10,7 +10,10 @@ use crate::{
     operations::delete_bucket,
     schema::Schema,
     state::DatabaseState,
-    sync::checkpoint::{validate_checkpoint, ChecksumMismatch},
+    sync::{
+        checkpoint::{validate_checkpoint, ChecksumMismatch},
+        interface::{RequestedStreamSubscription, StreamSubscriptionRequest},
+    },
     sync_local::{PartialSyncOperation, SyncOperation},
 };
 
@@ -239,6 +242,18 @@ impl StorageAdapter {
         }
     }
 
+    pub fn collect_subscription_requests(
+        &self,
+        include_defaults: bool,
+    ) -> Result<StreamSubscriptionRequest, PowerSyncError> {
+        let mut subscriptions: Vec<RequestedStreamSubscription> = Vec::new();
+
+        Ok(StreamSubscriptionRequest {
+            include_defaults,
+            subscriptions,
+        })
+    }
+
     pub fn now(&self) -> Result<Timestamp, ResultCode> {
         self.time_stmt.step()?;
         let res = Timestamp(self.time_stmt.column_int64(0));

crates/core/src/sync/streaming_sync.rs

@@ -589,6 +589,9 @@ impl StreamingSyncIteration {
             binary_data: true,
             client_id: client_id(self.db)?,
             parameters: self.options.parameters.take(),
+            streams: self
+                .adapter
+                .collect_subscription_requests(self.options.include_defaults)?,
         };
 
         event

crates/core/src/sync/subscriptions.rs

@@ -1,6 +1,6 @@
 use core::time::Duration;
 
-use alloc::string::String;
+use alloc::{boxed::Box, string::String};
 use serde::Deserialize;
 use serde_with::{serde_as, DurationSeconds};
 
@@ -16,7 +16,7 @@ pub enum SubscriptionChangeRequest {
 #[derive(Deserialize)]
 pub struct SubscribeToStream {
     pub stream: String,
-    pub params: Option<serde_json::value::RawValue>,
+    pub params: Option<Box<serde_json::value::RawValue>>,
     #[serde_as(as = "Option<DurationSeconds>")]
     pub ttl: Option<Duration>,
     pub priority: Option<BucketPriority>,
@@ -25,6 +25,6 @@ pub struct SubscribeToStream {
 #[derive(Deserialize)]
 pub struct UnsubscribeFromStream {
     pub stream: String,
-    pub params: Option<serde_json::value::RawValue>,
+    pub params: Option<Box<serde_json::value::RawValue>>,
     pub immediate: bool,
 }