feat: Make snapshots optional with cursor-based filtering (HYP-148)

rust/hyperstack-sdk/src/connection.rs

@@ -22,6 +22,16 @@ pub enum ConnectionCommand {
     Disconnect,
 }
 
+/// Options for subscribing to a view with specific parameters
+#[derive(Debug, Clone, Default)]
+pub struct SubscriptionOptions {
+    pub take: Option<u32>,
+    pub skip: Option<u32>,
+    pub with_snapshot: Option<bool>,
+    pub after: Option<String>,
+    pub snapshot_limit: Option<usize>,
+}
+
 struct ConnectionManagerInner {
     #[allow(dead_code)]
     url: String,
@@ -63,24 +73,26 @@ impl ConnectionManager {
     }
 
     pub async fn ensure_subscription(&self, view: &str, key: Option<&str>) {
-        self.ensure_subscription_with_opts(view, key, None, None)
+        self.ensure_subscription_with_opts(view, key, SubscriptionOptions::default())
             .await
     }
 
     pub async fn ensure_subscription_with_opts(
         &self,
         view: &str,
         key: Option<&str>,
-        take: Option<u32>,
-        skip: Option<u32>,
+        opts: SubscriptionOptions,
     ) {
         let sub = Subscription {
             view: view.to_string(),
             key: key.map(|s| s.to_string()),
             partition: None,
             filters: None,
-            take,
-            skip,
+            take: opts.take,
+            skip: opts.skip,
+            with_snapshot: opts.with_snapshot,
+            after: opts.after,
+            snapshot_limit: opts.snapshot_limit,
         };
 
         if !self.inner.subscriptions.read().await.contains(&sub) {
@@ -196,6 +208,9 @@ fn spawn_connection_loop(
                                             filters: None,
                                             take: None,
                                             skip: None,
+                                            with_snapshot: None,
+                                            after: None,
+                                            snapshot_limit: None,
                                         };
                                         subscriptions.write().await.remove(&sub);
                                         let client_msg = ClientMessage::Unsubscribe(unsub);

rust/hyperstack-sdk/src/frame.rs

@@ -81,6 +81,9 @@ pub struct Frame {
     pub data: serde_json::Value,
     #[serde(default)]
     pub append: Vec<String>,
+    /// Sequence cursor for ordering and resume capability
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub seq: Option<String>,
 }
 
 impl Frame {

rust/hyperstack-sdk/src/stream.rs

@@ -1,4 +1,4 @@
-use crate::connection::ConnectionManager;
+use crate::connection::{ConnectionManager, SubscriptionOptions};
 use crate::frame::Operation;
 use crate::store::{SharedStore, StoreUpdate};
 use futures_util::Stream;
@@ -179,6 +179,9 @@ enum EntityStreamState<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     },
     Active {
         inner: BroadcastStream<StoreUpdate>,
@@ -246,6 +249,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> EntityStream<T> {
             subscription_key,
             None,
             None,
+            None,
+            None,
+            None,
         )
     }
 
@@ -259,6 +265,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> EntityStream<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     ) -> Self {
         Self {
             state: EntityStreamState::Lazy {
@@ -268,6 +277,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> EntityStream<T> {
                 subscription_key,
                 take,
                 skip,
+                with_snapshot,
+                after,
+                snapshot_limit,
             },
             view: entity_name,
             key_filter,
@@ -313,6 +325,9 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for EntityStre
                         subscription_key,
                         take,
                         skip,
+                        with_snapshot,
+                        after,
+                        snapshot_limit,
                     } = std::mem::replace(&mut this.state, EntityStreamState::Invalid)
                     else {
                         unreachable!()
@@ -326,7 +341,14 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for EntityStre
                     let view = subscription_view.clone();
                     let key = subscription_key.clone();
                     let fut = Box::pin(async move {
-                        conn.ensure_subscription_with_opts(&view, key.as_deref(), take, skip)
+                        let opts = SubscriptionOptions {
+                            take,
+                            skip,
+                            with_snapshot,
+                            after,
+                            snapshot_limit,
+                        };
+                        conn.ensure_subscription_with_opts(&view, key.as_deref(), opts)
                             .await;
                     });
 
@@ -429,6 +451,9 @@ enum RichEntityStreamState<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     },
     Active {
         inner: BroadcastStream<StoreUpdate>,
@@ -481,6 +506,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> RichEntityStream<T> {
             subscription_key,
             None,
             None,
+            None,
+            None,
+            None,
         )
     }
 
@@ -494,6 +522,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> RichEntityStream<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     ) -> Self {
         Self {
             state: RichEntityStreamState::Lazy {
@@ -503,6 +534,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> RichEntityStream<T> {
                 subscription_key,
                 take,
                 skip,
+                with_snapshot,
+                after,
+                snapshot_limit,
             },
             view: entity_name,
             key_filter,
@@ -527,6 +561,9 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for RichEntity
                         subscription_key,
                         take,
                         skip,
+                        with_snapshot,
+                        after,
+                        snapshot_limit,
                     } = std::mem::replace(&mut this.state, RichEntityStreamState::Invalid)
                     else {
                         unreachable!()
@@ -540,7 +577,14 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for RichEntity
                     let view = subscription_view.clone();
                     let key = subscription_key.clone();
                     let fut = Box::pin(async move {
-                        conn.ensure_subscription_with_opts(&view, key.as_deref(), take, skip)
+                        let opts = SubscriptionOptions {
+                            take,
+                            skip,
+                            with_snapshot,
+                            after,
+                            snapshot_limit,
+                        };
+                        conn.ensure_subscription_with_opts(&view, key.as_deref(), opts)
                             .await;
                     });
 
@@ -890,6 +934,9 @@ enum UseStreamState<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     },
     Active {
         inner: BroadcastStream<StoreUpdate>,
@@ -942,6 +989,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> UseStream<T> {
             subscription_key,
             None,
             None,
+            None,
+            None,
+            None,
         )
     }
 
@@ -955,6 +1005,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> UseStream<T> {
         subscription_key: Option<String>,
         take: Option<u32>,
         skip: Option<u32>,
+        with_snapshot: Option<bool>,
+        after: Option<String>,
+        snapshot_limit: Option<usize>,
     ) -> Self {
         Self {
             state: UseStreamState::Lazy {
@@ -964,6 +1017,9 @@ impl<T: DeserializeOwned + Clone + Send + 'static> UseStream<T> {
                 subscription_key,
                 take,
                 skip,
+                with_snapshot,
+                after,
+                snapshot_limit,
             },
             view: entity_name,
             key_filter,
@@ -1012,6 +1068,9 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for UseStream<
                         subscription_key,
                         take,
                         skip,
+                        with_snapshot,
+                        after,
+                        snapshot_limit,
                     } = std::mem::replace(&mut this.state, UseStreamState::Invalid)
                     else {
                         unreachable!()
@@ -1024,7 +1083,14 @@ impl<T: DeserializeOwned + Clone + Send + Unpin + 'static> Stream for UseStream<
                     let view = subscription_view.clone();
                     let key = subscription_key.clone();
                     let fut = Box::pin(async move {
-                        conn.ensure_subscription_with_opts(&view, key.as_deref(), take, skip)
+                        let opts = SubscriptionOptions {
+                            take,
+                            skip,
+                            with_snapshot,
+                            after,
+                            snapshot_limit,
+                        };
+                        conn.ensure_subscription_with_opts(&view, key.as_deref(), opts)
                             .await;
                     });
 

rust/hyperstack-sdk/src/subscription.rs

@@ -10,6 +10,7 @@ pub enum ClientMessage {
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
 pub struct Subscription {
     pub view: String,
     #[serde(skip_serializing_if = "Option::is_none")]
@@ -22,6 +23,15 @@ pub struct Subscription {
     pub take: Option<u32>,
     #[serde(skip_serializing_if = "Option::is_none")]
     pub skip: Option<u32>,
+    /// Whether to include initial snapshot (defaults to true for backward compatibility)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub with_snapshot: Option<bool>,
+    /// Cursor for resuming from a specific point (_seq value)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub after: Option<String>,
+    /// Maximum number of entities to include in snapshot (pagination hint)
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub snapshot_limit: Option<usize>,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -67,6 +77,9 @@ impl Subscription {
             filters: None,
             take: None,
             skip: None,
+            with_snapshot: None,
+            after: None,
+            snapshot_limit: None,
         }
     }
 
@@ -90,6 +103,24 @@ impl Subscription {
         self
     }
 
+    /// Set whether to include the initial snapshot (defaults to true)
+    pub fn with_snapshot(mut self, with_snapshot: bool) -> Self {
+        self.with_snapshot = Some(with_snapshot);
+        self
+    }
+
+    /// Set the cursor to resume from (for reconnecting and getting only newer data)
+    pub fn after(mut self, cursor: impl Into<String>) -> Self {
+        self.after = Some(cursor.into());
+        self
+    }
+
+    /// Set the maximum number of entities to include in the snapshot
+    pub fn with_snapshot_limit(mut self, limit: usize) -> Self {
+        self.snapshot_limit = Some(limit);
+        self
+    }
+
     pub fn sub_key(&self) -> String {
         let filters_str = self
             .filters