Support unsubscribing

Author: simolus3

crates/core/src/sync/storage_adapter.rs

@@ -266,9 +266,11 @@ impl StorageAdapter {
         self.delete_outdated_subscriptions()?;
 
         let mut subscriptions: Vec<RequestedStreamSubscription> = Vec::new();
+        // We have an explicit subscription iff ttl is not null. Checking is_default is not enough,
+        // because a stream can both be a default stream and have an explicit subscription.
         let stmt = self
             .db
-            .prepare_v2("SELECT * FROM ps_stream_subscriptions WHERE NOT is_default;")?;
+            .prepare_v2("SELECT * FROM ps_stream_subscriptions WHERE ttl IS NOT NULL;")?;
 
         while let ResultCode::ROW = stmt.step()? {
             let subscription = Self::read_stream_subscription(&stmt)?;

crates/core/src/sync/subscriptions.rs

@@ -53,6 +53,8 @@ impl LocallyTrackedSubscription {
 pub enum SubscriptionChangeRequest {
     #[serde(rename = "subscribe")]
     Subscribe(SubscribeToStream),
+    #[serde(rename = "unsubscribe")]
+    Unsubscribe(UnsubscribeFromStream),
 }
 
 #[serde_as]
@@ -104,6 +106,21 @@ pub fn apply_subscriptions(
             }?;
             stmt.exec()?;
         }
+        SubscriptionChangeRequest::Unsubscribe(subscription) => {
+            let stmt = db
+                .prepare_v2("UPDATE ps_stream_subscriptions SET ttl = NULL WHERE stream_name = ? AND local_params = ?")
+                .into_db_result(db)?;
+            stmt.bind_text(1, &subscription.stream, sqlite::Destructor::STATIC)?;
+            stmt.bind_text(
+                2,
+                match &subscription.params {
+                    Some(params) => params.get(),
+                    None => "null",
+                },
+                sqlite::Destructor::STATIC,
+            )?;
+            stmt.exec()?;
+        }
     }
 
     Ok(())

dart/test/sync_stream_test.dart

@@ -175,6 +175,65 @@ void main() {
       expect(stored, containsPair('active', 1));
       expect(stored, containsPair('is_default', 0));
     });
+  });
+
+  group('explicit subscriptions', () {
+    syncTest('unsubscribe', (_) {
+      db.execute(
+          'INSERT INTO ps_stream_subscriptions (stream_name, ttl) VALUES (?, ?);',
+          ['my_stream', 3600]);
+
+      var startInstructions = control('start', null);
+      expect(
+        startInstructions,
+        contains(
+          containsPair(
+            'EstablishSyncStream',
+            containsPair(
+              'request',
+              containsPair(
+                'streams',
+                {
+                  'include_defaults': true,
+                  'subscriptions': isNotEmpty,
+                },
+              ),
+            ),
+          ),
+        ),
+      );
+      control('stop', null);
+
+      control(
+        'subscriptions',
+        json.encode({
+          'unsubscribe': {
+            'stream': 'my_stream',
+            'params': null,
+            'immediate': false,
+          }
+        }),
+      );
+      startInstructions = control('start', null);
+      expect(
+        startInstructions,
+        contains(
+          containsPair(
+            'EstablishSyncStream',
+            containsPair(
+              'request',
+              containsPair(
+                'streams',
+                {
+                  'include_defaults': true,
+                  'subscriptions': isEmpty,
+                },
+              ),
+            ),
+          ),
+        ),
+      );
+    });
 
     syncTest('ttl', (controller) {
       db.execute(