Track subscriptions

Author: simolus3

crates/core/src/migrations.rs

@@ -389,12 +389,13 @@ INSERT INTO ps_migration(id, down_migrations) VALUES (10, json_array(
 CREATE TABLE ps_stream_subscriptions (
   id NOT NULL INTEGER PRIMARY KEY,
   stream_name TEXT NOT NULL,
-  is_default INTEGER NOT NULL,
+  active INTEGER NOT NULL DEFAULT FALSE,
+  is_default INTEGER NOT NULL DEFAULT FALSE,
   local_priority INTEGER,
   local_params TEXT,
-  ttl INTEGER
+  ttl INTEGER,
+  expires_at INTEGER
 ) STRICT;
-ALTER TABLE ps_buckets ADD COLUMN from_subscriptions TEXT NOT NULL DEFAULT '[null]';
 
 INSERT INTO ps_migration(id, down_migrations) VALUES(11, json_array(
 json_object('sql', 'todo down migration'),

crates/core/src/sync/checkpoint.rs

@@ -1,7 +1,8 @@
 use alloc::{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: 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(),
         }
     }
 }

crates/core/src/sync/interface.rs

@@ -19,6 +19,8 @@ use sqlite_nostd::bindings::SQLITE_RESULT_SUBTYPE;
 use sqlite_nostd::{self as sqlite, ColumnType};
 use sqlite_nostd::{Connection, Context};
 
+use crate::sync::BucketPriority;
+
 /// Payload provided by SDKs when requesting a sync iteration.
 #[derive(Deserialize)]
 pub struct StartSyncStream {
@@ -141,6 +143,7 @@ pub struct RequestedStreamSubscription {
     pub stream: String,
     /// Parameters to make available in the stream's definition.
     pub parameters: Box<serde_json::value::RawValue>,
+    pub override_priority: Option<BucketPriority>,
     #[serde_as(as = "DisplayFromStr")]
     pub client_id: i64,
 }

crates/core/src/sync/line.rs

@@ -1,7 +1,8 @@
 use alloc::borrow::Cow;
+use alloc::string::{String, ToString};
 use alloc::vec::Vec;
 use serde::Deserialize;
-use serde::de::{IgnoredAny, VariantAccess, Visitor};
+use serde::de::{Error, IgnoredAny, VariantAccess, Visitor};
 use serde_with::{DisplayFromStr, serde_as};
 
 use super::Checksum;
@@ -82,6 +83,14 @@ pub struct Checkpoint<'a> {
     pub write_checkpoint: Option<i64>,
     #[serde(borrow)]
     pub buckets: Vec<BucketChecksum<'a>>,
+    #[serde(default, borrow)]
+    pub streams: Vec<StreamDefinition<'a>>,
+}
+
+#[derive(Deserialize, Debug)]
+pub struct StreamDefinition<'a> {
+    pub name: SyncLineStr<'a>,
+    pub is_default: bool,
 }
 
 #[serde_as]
@@ -120,14 +129,67 @@ pub struct BucketChecksum<'a> {
     pub priority: Option<BucketPriority>,
     #[serde(default)]
     pub count: Option<i64>,
-    #[serde_as(as = "Vec<Option<DisplayFromStr>>")]
     #[serde(default)]
-    pub subscriptions: Vec<Option<i64>>,
+    pub subscriptions: BucketSubscriptionReason,
     //    #[serde(default)]
     //    #[serde(deserialize_with = "deserialize_optional_string_to_i64")]
     //    pub last_op_id: Option<i64>,
 }
 
+/// The reason for why a bucket was included in a checkpoint.
+#[derive(Debug, Default, Clone)]
+pub enum BucketSubscriptionReason {
+    /// A bucket was created for all of the subscription ids we've explicitly requested in the sync
+    /// request.
+    ExplicitlySubscribed { subscriptions: Vec<i64> },
+    /// A bucket was created from a default stream.
+    IsDefault { stream_name: String },
+    /// We're talking to an older sync service not sending the reason.
+    #[default]
+    Unknown,
+}
+
+impl<'de> Deserialize<'de> for BucketSubscriptionReason {
+    fn deserialize<D>(deserializer: D) -> Result<Self, D::Error>
+    where
+        D: serde::Deserializer<'de>,
+    {
+        struct MyVisitor;
+
+        impl<'de> Visitor<'de> for MyVisitor {
+            type Value = BucketSubscriptionReason;
+
+            fn expecting(&self, formatter: &mut core::fmt::Formatter) -> core::fmt::Result {
+                write!(formatter, "a subscription reason")
+            }
+
+            fn visit_seq<A>(self, mut seq: A) -> Result<Self::Value, A::Error>
+            where
+                A: serde::de::SeqAccess<'de>,
+            {
+                let mut subscriptions = Vec::<i64>::new();
+
+                while let Some(item) = seq.next_element::<&'de str>()? {
+                    subscriptions.push(item.parse().map_err(|_| A::Error::custom("not an int"))?);
+                }
+
+                Ok(BucketSubscriptionReason::ExplicitlySubscribed { subscriptions })
+            }
+
+            fn visit_str<E>(self, v: &str) -> Result<Self::Value, E>
+            where
+                E: serde::de::Error,
+            {
+                Ok(BucketSubscriptionReason::IsDefault {
+                    stream_name: v.to_string(),
+                })
+            }
+        }
+
+        deserializer.deserialize_any(MyVisitor)
+    }
+}
+
 #[derive(Deserialize, Debug)]
 pub struct DataLine<'a> {
     #[serde(borrow)]
@@ -229,6 +291,7 @@ mod tests {
                 last_op_id: 10,
                 write_checkpoint: None,
                 buckets: _,
+                streams: _,
             })
         );
 
@@ -264,6 +327,7 @@ mod tests {
                 last_op_id: 1,
                 write_checkpoint: None,
                 buckets: _,
+                streams: _,
             })
         );
     }

crates/core/src/sync/storage_adapter.rs

@@ -2,6 +2,7 @@ use core::{assert_matches::debug_assert_matches, fmt::Display};
 
 use alloc::{string::ToString, vec::Vec};
 use serde::Serialize;
+use serde_json::value::RawValue;
 use sqlite_nostd::{self as sqlite, Connection, ManagedStmt, ResultCode};
 
 use crate::{
@@ -13,9 +14,12 @@ use crate::{
     sync::{
         checkpoint::{ChecksumMismatch, validate_checkpoint},
         interface::{RequestedStreamSubscription, StreamSubscriptionRequest},
+        streaming_sync::OwnedStreamDefinition,
+        subscriptions::LocallyTrackedSubscription,
         sync_status::SyncPriorityStatus,
     },
     sync_local::{PartialSyncOperation, SyncOperation},
+    util::{JsonString, column_nullable},
 };
 
 use super::{
@@ -288,6 +292,55 @@ impl StorageAdapter {
 
         Ok(res)
     }
+
+    fn read_stream_subscription(
+        stmt: &ManagedStmt,
+    ) -> Result<LocallyTrackedSubscription, PowerSyncError> {
+        Ok(LocallyTrackedSubscription {
+            id: stmt.column_int64(0),
+            stream_name: stmt.column_text(1)?.to_string(),
+            active: stmt.column_int(2) != 0,
+            is_default: stmt.column_int(3) != 0,
+            local_priority: column_nullable(&stmt, 4, || {
+                BucketPriority::try_from(stmt.column_int(4))
+            })?,
+            local_params: column_nullable(&stmt, 5, || {
+                JsonString::from_string(stmt.column_text(5)?.to_string())
+            })?,
+            ttl: column_nullable(&stmt, 6, || Ok(stmt.column_int64(6)))?,
+            expires_at: column_nullable(&stmt, 7, || Ok(stmt.column_int64(7)))?,
+        })
+    }
+
+    pub fn iterate_local_subscriptions<F: FnMut(LocallyTrackedSubscription) -> ()>(
+        &self,
+        mut action: F,
+    ) -> Result<(), PowerSyncError> {
+        let stmt = self
+            .db
+            .prepare_v2("SELECT * FROM ps_stream_subscriptions ORDER BY id ASC")?;
+
+        while stmt.step()? == ResultCode::ROW {
+            action(Self::read_stream_subscription(&stmt)?);
+        }
+
+        stmt.finalize()?;
+        Ok(())
+    }
+
+    pub fn create_default_subscription(
+        &self,
+        stream: &OwnedStreamDefinition,
+    ) -> Result<LocallyTrackedSubscription, PowerSyncError> {
+        let stmt = self.db.prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, active, is_default) VALUES (?, TRUE, TRUE) RETURNING *;")?;
+        stmt.bind_text(1, &stream.name, sqlite_nostd::Destructor::STATIC)?;
+
+        if stmt.step()? == ResultCode::ROW {
+            Self::read_stream_subscription(&stmt)
+        } else {
+            Err(PowerSyncError::unknown_internal())
+        }
+    }
 }
 
 pub struct BucketInfo {