Add protocol changes

Author: simolus3

crates/core/src/migrations.rs

@@ -386,12 +386,13 @@ INSERT INTO ps_migration(id, down_migrations) VALUES (10, json_array(
 
     if current_version < 11 && target_version >= 11 {
         let stmt = "\
-CREATE TABLE ps_streams (
+CREATE TABLE ps_stream_subscriptions (
   id NOT NULL INTEGER PRIMARY KEY,
-  definition_name TEXT NOT NULL,
+  stream_name TEXT NOT NULL,
   is_default INTEGER NOT NULL,
   local_priority INTEGER,
-  local_params TEXT
+  local_params TEXT,
+  ttl INTEGER
 ) STRICT;
 ALTER TABLE ps_buckets ADD COLUMN derived_from INTEGER REFERENCES ps_streams (id);
 

crates/core/src/sync/interface.rs

@@ -7,6 +7,7 @@ use crate::constants::SUBTYPE_JSON;
 use crate::error::PowerSyncError;
 use crate::schema::Schema;
 use crate::state::DatabaseState;
+use crate::util::serialize_i64_to_string;
 use alloc::borrow::Cow;
 use alloc::boxed::Box;
 use alloc::rc::Rc;
@@ -108,6 +109,22 @@ pub struct StreamingSyncRequest {
     pub parameters: Option<serde_json::Map<String, serde_json::Value>>,
 }
 
+#[derive(Serialize)]
+pub struct StreamSubscriptionRequest {
+    pub include_defaults: bool,
+    pub subscriptions: Vec<RequestedStreamSubscription>,
+}
+
+#[derive(Serialize)]
+pub struct RequestedStreamSubscription {
+    /// The name of the sync stream to subscribe to.
+    pub stream: String,
+    /// Parameters to make available in the stream's definition.
+    pub parameters: Box<serde_json::value::RawValue>,
+    #[serde(serialize_with = "serialize_i64_to_string")]
+    pub client_id: i64,
+}
+
 #[derive(Serialize)]
 pub struct BucketRequest {
     pub name: String,

crates/core/src/sync/line.rs

@@ -119,11 +119,60 @@ pub struct BucketChecksum<'a> {
     pub priority: Option<BucketPriority>,
     #[serde(default)]
     pub count: Option<i64>,
+    #[serde(
+        default,
+        deserialize_with = "BucketChecksum::deserialize_subscriptions"
+    )]
+    pub subscriptions: Option<Vec<i64>>,
     //    #[serde(default)]
     //    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
     //    pub last_op_id: Option<i64>,
 }
 
+impl BucketChecksum<'_> {
+    fn deserialize_subscriptions<'de, D: serde::Deserializer<'de>>(
+        deserializer: D,
+    ) -> Result<Option<Vec<i64>>, D::Error> {
+        struct MyVisitor;
+
+        impl<'de> Visitor<'de> for MyVisitor {
+            type Value = Option<Vec<i64>>;
+
+            fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
+                write!(formatter, "optional list of subscriptions")
+            }
+
+            fn visit_none<E>(self) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                Ok(None)
+            }
+
+            fn visit_some<D>(self, deserializer: D) -> Result<Self::Value, D::Error>
+            where
+                D: serde::Deserializer<'de>,
+            {
+                deserializer.deserialize_seq(self)
+            }
+
+            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: serde::de::SeqAccess<'de>,
+            {
+                let mut result: Vec<i64> = Vec::new();
+                while let Some(element) = seq.next_element::<&'de str>()? {
+                    result.push(element.parse::<i64>().map_err(serde::de::Error::custom)?);
+                }
+
+                Ok(Some(result))
+            }
+        }
+
+        deserializer.deserialize_option(MyVisitor)
+    }
+}
+
 #[derive(Deserialize, Debug)]
 pub struct DataLine<'a> {
     #[serde(borrow)]

crates/core/src/util.rs

@@ -60,29 +60,19 @@ pub fn quote_identifier_prefixed(prefix: &str, name: &str) -> String {
     return format!("\"{:}{:}\"", prefix, name.replace("\"", "\"\""));
 }
 
+pub fn serialize_i64_to_string<'de, S: serde::Serializer>(
+    value: &i64,
+    serializer: S,
+) -> Result<S::Ok, S::Error> {
+    serializer.collect_str(value)
+}
+
 pub fn deserialize_string_to_i64<'de, D>(deserializer: D) -> Result<i64, D::Error>
 where
     D: serde::Deserializer<'de>,
 {
-    struct ValueVisitor;
-
-    impl<'de> Visitor<'de> for ValueVisitor {
-        type Value = i64;
-
-        fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
-            formatter.write_str("a string representation of a number")
-        }
-
-        fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
-        where
-            E: serde::de::Error,
-        {
-            v.parse::<i64>().map_err(serde::de::Error::custom)
-        }
-    }
-
-    // Using a custom visitor here to avoid an intermediate string allocation
-    deserializer.deserialize_str(ValueVisitor)
+    let value: &'de str = serde::Deserialize::deserialize(deserializer)?;
+    value.parse::<i64>().map_err(serde::de::Error::custom)
 }
 
 pub fn deserialize_optional_string_to_i64<'de, D>(deserializer: D) -> Result<Option<i64>, D::Error>