Allow subscribing to streams

simolus3 · simolus3 · commit b25c03c1f27f · 2025-07-10T18:55:12.000+02:00
diff --git a/crates/core/src/migrations.rs b/crates/core/src/migrations.rs
@@ -392,10 +392,11 @@ CREATE TABLE ps_stream_subscriptions (
   active INTEGER NOT NULL DEFAULT FALSE,
   is_default INTEGER NOT NULL DEFAULT FALSE,
   local_priority INTEGER,
-  local_params TEXT,
+  local_params TEXT NOT NULL DEFAULT 'null',
   ttl INTEGER,
   expires_at INTEGER,
-  last_synced_at INTEGER
+  last_synced_at INTEGER,
+  UNIQUE (stream_name, local_params)
 ) STRICT;
 
 INSERT INTO ps_migration(id, down_migrations) VALUES(11, json_array(
diff --git a/crates/core/src/sync/interface.rs b/crates/core/src/sync/interface.rs
@@ -15,6 +15,7 @@ use sqlite_nostd::{Connection, Context};
 use crate::error::PowerSyncError;
 use crate::schema::Schema;
 use crate::state::DatabaseState;
+use crate::sync::subscriptions::{apply_subscriptions, SubscriptionChangeRequest};
 use crate::sync::BucketPriority;
 
 use super::streaming_sync::SyncClient;
@@ -216,6 +217,11 @@ pub fn register(db: *mut sqlite::sqlite3, state: Arc<DatabaseState>) -> Result<(
                 }),
                 "refreshed_token" => SyncControlRequest::SyncEvent(SyncEvent::DidRefreshToken),
                 "completed_upload" => SyncControlRequest::SyncEvent(SyncEvent::UploadFinished),
+                "subscriptions" => {
+                    let request = serde_json::from_str(payload.text())
+                        .map_err(PowerSyncError::as_argument_error)?;
+                    return apply_subscriptions(ctx.db_handle(), request);
+                }
                 _ => {
                     return Err(PowerSyncError::argument_error("Unknown operation"));
                 }
diff --git a/crates/core/src/sync/storage_adapter.rs b/crates/core/src/sync/storage_adapter.rs
@@ -35,6 +35,7 @@ pub struct StorageAdapter {
     pub db: *mut sqlite::sqlite3,
     pub progress_stmt: ManagedStmt,
     time_stmt: ManagedStmt,
+    delete_subscription: ManagedStmt,
 }
 
 impl StorageAdapter {
@@ -47,10 +48,15 @@ impl StorageAdapter {
         // language=SQLite
         let time = db.prepare_v2("SELECT unixepoch()")?;
 
+        // language=SQLite
+        let delete_subscription =
+            db.prepare_v2("DELETE FROM ps_stream_subscriptions WHERE id = ?")?;
+
         Ok(Self {
             db,
             progress_stmt: progress,
             time_stmt: time,
+            delete_subscription,
         })
     }
 
@@ -269,6 +275,8 @@ impl StorageAdapter {
     fn read_stream_subscription(
         stmt: &ManagedStmt,
     ) -> Result<LocallyTrackedSubscription, PowerSyncError> {
+        let raw_params = stmt.column_text(5)?;
+
         Ok(LocallyTrackedSubscription {
             id: stmt.column_int64(0),
             stream_name: stmt.column_text(1)?.to_string(),
@@ -277,9 +285,11 @@ impl StorageAdapter {
             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())
-            })?,
+            local_params: if raw_params == "null" {
+                None
+            } else {
+                Some(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)))?,
             last_synced_at: column_nullable(&stmt, 8, || Ok(stmt.column_int64(8)))?,
@@ -313,6 +323,13 @@ impl StorageAdapter {
             Err(PowerSyncError::unknown_internal())
         }
     }
+
+    pub fn delete_subscription(&self, id: i64) -> Result<(), PowerSyncError> {
+        let _ = self.delete_subscription.reset();
+        self.delete_subscription.bind_int64(1, id)?;
+        self.delete_subscription.exec()?;
+        Ok(())
+    }
 }
 
 pub struct BucketInfo {
diff --git a/crates/core/src/sync/streaming_sync.rs b/crates/core/src/sync/streaming_sync.rs
@@ -573,25 +573,41 @@ impl StreamingSyncIteration {
         let mut tracked_subscriptions: Vec<LocallyTrackedSubscription> = Vec::new();
 
         // Load known subscriptions from database
-        self.adapter.iterate_local_subscriptions(|sub| {
+        self.adapter.iterate_local_subscriptions(|mut sub| {
+            // We will mark it as active again if it's part of the streams included in the
+            // checkpoint.
+            sub.active = false;
+
             tracked_subscriptions.push(sub);
         })?;
 
         // If they don't exist already, create default subscriptions included in checkpoint
         for subscription in &tracked.streams {
-            if subscription.is_default {
-                let found = tracked_subscriptions
-                    .iter()
-                    .filter(|s| s.stream_name == subscription.name && s.local_params.is_none())
-                    .next();
+            let matching_local_subscriptions = tracked_subscriptions
+                .iter_mut()
+                .filter(|s| s.stream_name == subscription.name);
+
+            let mut has_local = false;
+            for subscription in matching_local_subscriptions {
+                subscription.active = true;
+                has_local = true;
+            }
 
-                if found.is_none() {
-                    let subscription = self.adapter.create_default_subscription(subscription)?;
-                    tracked_subscriptions.push(subscription);
-                }
+            if !has_local && subscription.is_default {
+                let subscription = self.adapter.create_default_subscription(subscription)?;
+                tracked_subscriptions.push(subscription);
             }
         }
 
+        // Clean up default subscriptions that are no longer active.
+        for subscription in &tracked_subscriptions {
+            if subscription.is_default && !subscription.active {
+                self.adapter.delete_subscription(subscription.id)?;
+            }
+        }
+        tracked_subscriptions
+            .retain(|subscription| !subscription.is_default || subscription.active);
+
         debug_assert!(tracked_subscriptions.is_sorted_by_key(|s| s.id));
 
         let mut resolved: Vec<ActiveStreamSubscription> =
@@ -609,8 +625,6 @@ impl StreamingSyncIteration {
             }
         }
 
-        // TODO: Cleanup old default subscriptions?
-
         // Iterate over buckets to associate them with subscriptions
         for bucket in tracked.checkpoint.buckets.values() {
             match &bucket.subscriptions {
diff --git a/crates/core/src/sync/subscriptions.rs b/crates/core/src/sync/subscriptions.rs
@@ -3,8 +3,14 @@ use core::{cmp::Ordering, hash::Hash, time::Duration};
 use alloc::{boxed::Box, string::String};
 use serde::Deserialize;
 use serde_with::{serde_as, DurationSeconds};
+use sqlite_nostd::{self as sqlite, Connection};
 
-use crate::{sync::BucketPriority, util::JsonString};
+use crate::{
+    error::{PSResult, PowerSyncError},
+    ext::SafeManagedStmt,
+    sync::BucketPriority,
+    util::JsonString,
+};
 
 /// A key that uniquely identifies a stream subscription.
 #[derive(Debug, PartialEq, PartialOrd, Eq, Ord)]
@@ -38,16 +44,20 @@ impl LocallyTrackedSubscription {
 /// A request sent from a PowerSync SDK to alter the subscriptions managed by this client.
 #[derive(Deserialize)]
 pub enum SubscriptionChangeRequest {
+    #[serde(rename = "subscribe")]
     Subscribe(SubscribeToStream),
 }
 
 #[serde_as]
 #[derive(Deserialize)]
 pub struct SubscribeToStream {
     pub stream: String,
+    #[serde(default)]
     pub params: Option<Box<serde_json::value::RawValue>>,
     #[serde_as(as = "Option<DurationSeconds>")]
+    #[serde(default)]
     pub ttl: Option<Duration>,
+    #[serde(default)]
     pub priority: Option<BucketPriority>,
 }
 
@@ -57,3 +67,37 @@ pub struct UnsubscribeFromStream {
     pub params: Option<Box<serde_json::value::RawValue>>,
     pub immediate: bool,
 }
+
+pub fn apply_subscriptions(
+    db: *mut sqlite::sqlite3,
+    subscription: SubscriptionChangeRequest,
+) -> Result<(), PowerSyncError> {
+    match subscription {
+        SubscriptionChangeRequest::Subscribe(subscription) => {
+            let stmt = db
+                .prepare_v2("INSERT INTO ps_stream_subscriptions (stream_name, local_priority, local_params, ttl) VALUES (?, ?2, ?, ?4) ON CONFLICT DO UPDATE SET local_priority = min(coalesce(?2, local_priority), local_priority), ttl = ?4, is_default = FALSE")
+                .into_db_result(db)?;
+
+            stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;
+            match &subscription.priority {
+                Some(priority) => stmt.bind_int(2, priority.number),
+                None => stmt.bind_null(2),
+            }?;
+            stmt.bind_text(
+                3,
+                match &subscription.params {
+                    Some(params) => params.get(),
+                    None => "null",
+                },
+                sqlite::Destructor::STATIC,
+            )?;
+            match &subscription.ttl {
+                Some(ttl) => stmt.bind_int64(4, ttl.as_secs() as i64),
+                None => stmt.bind_null(4),
+            }?;
+            stmt.exec()?;
+        }
+    }
+
+    Ok(())
+}
diff --git a/dart/test/sync_stream_test.dart b/dart/test/sync_stream_test.dart
@@ -47,14 +47,19 @@ void main() {
 
     db.execute('commit');
     final [row] = result;
-    final instructions = jsonDecode(row.columnAt(0)) as List;
-    for (final instruction in instructions) {
-      if (instruction case {'UpdateSyncStatus': final status}) {
-        lastStatus = status['status']!;
+
+    final rawResult = row.columnAt(0);
+    if (rawResult is String) {
+      final instructions = jsonDecode(row.columnAt(0)) as List;
+      for (final instruction in instructions) {
+        if (instruction case {'UpdateSyncStatus': final status}) {
+          lastStatus = status['status']!;
+        }
       }
+      return instructions;
+    } else {
+      return const [];
     }
-
-    return instructions;
   }
 
   group('default streams', () {
@@ -109,5 +114,66 @@ void main() {
       final [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
       expect(stored, containsPair('last_synced_at', 1740823200));
     });
+
+    syncTest('are deleted', (_) {
+      control('start', null);
+
+      for (final stream in ['s1', 's2']) {
+        control(
+          'line_text',
+          json.encode(
+            checkpoint(
+              lastOpId: 1,
+              buckets: [
+                bucketDescription('a', subscriptions: stream, priority: 1),
+              ],
+              streams: [(stream, true)],
+            ),
+          ),
+        );
+        control(
+          'line_text',
+          json.encode(checkpointComplete(priority: 1)),
+        );
+      }
+
+      expect(
+        lastStatus,
+        containsPair(
+          'streams',
+          [containsPair('name', 's2')],
+        ),
+      );
+    });
+
+    syncTest('can be made explicit', (_) {
+      control('start', null);
+      control(
+        'line_text',
+        json.encode(
+          checkpoint(
+            lastOpId: 1,
+            buckets: [
+              bucketDescription('a', subscriptions: 'a', priority: 1),
+            ],
+            streams: [('a', true)],
+          ),
+        ),
+      );
+
+      var [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(stored, containsPair('is_default', 1));
+
+      control(
+        'subscriptions',
+        json.encode({
+          'subscribe': {'stream': 'a'},
+        }),
+      );
+
+      [stored] = db.select('SELECT * FROM ps_stream_subscriptions');
+      expect(stored, containsPair('active', 1));
+      expect(stored, containsPair('is_default', 0));
+    });
   });
 }
