Two more tests

Author: simolus3

packages/powersync_core/lib/src/database/powersync_db_mixin.dart

@@ -504,6 +504,9 @@ SELECT * FROM crud_entries;
     await database.refreshSchema();
   }
 
+  /// Create a [SyncStream] instance for the given [name] and [parameters].
+  ///
+  /// Use [SyncStream.subscribe] to subscribe to the returned stream.
   SyncStream syncStream(String name, [Map<String, Object?>? parameters]) {
     return _connections.syncStream(name, parameters);
   }

packages/powersync_core/lib/src/sync/connection_manager.dart

@@ -355,6 +355,7 @@ final class _ActiveSubscription {
 
 final class _SyncStreamSubscriptionHandle implements SyncStreamSubscription {
   final _ActiveSubscription _source;
+  var _active = true;
 
   _SyncStreamSubscriptionHandle(this._source) {
     _source.refcount++;
@@ -368,9 +369,12 @@ final class _SyncStreamSubscriptionHandle implements SyncStreamSubscription {
   Map<String, Object?>? get parameters => _source.parameters;
 
   @override
-  void unsubscribe() async {
-    _finalizer.detach(this);
-    _source.decrementRefCount();
+  void unsubscribe() {
+    if (_active) {
+      _active = false;
+      _finalizer.detach(this);
+      _source.decrementRefCount();
+    }
   }
 
   @override

packages/powersync_core/lib/src/sync/options.dart

@@ -27,6 +27,10 @@ final class SyncOptions {
   /// The [SyncClientImplementation] to use.
   final SyncClientImplementation syncImplementation;
 
+  /// Whether streams that have been defined with `auto_subscribe: true` should
+  /// be synced when they don't have an explicit subscription.
+  ///
+  /// This is enabled by default.
   final bool? includeDefaultStreams;
 
   const SyncOptions({

packages/powersync_core/lib/src/sync/stream.dart

@@ -24,7 +24,7 @@ abstract interface class SyncStreamDescription {
 abstract interface class SyncSubscriptionDescription
     extends SyncStreamDescription {
   /// Whether this stream is active, meaning that the subscription has been
-  /// acknownledged by the sync serivce.
+  /// acknowledged by the sync serivce.
   bool get active;
 
   /// Whether this stream subscription is included by default, regardless of

packages/powersync_core/lib/src/sync/streaming_sync.dart

@@ -622,7 +622,8 @@ final class _ActiveRustStreamingIteration {
 
   List<Object?> _encodeSubscriptions(List<SubscribedStream> subscriptions) {
     return sync._activeSubscriptions
-        .map((s) => {'name': s.name, 'params': s.parameters})
+        .map((s) =>
+            {'name': s.name, 'params': convert.json.decode(s.parameters)})
         .toList();
   }
 

packages/powersync_core/lib/src/sync/sync_status.dart

@@ -212,11 +212,21 @@ extension InternalSyncStatusAccess on SyncStatus {
       _internalSubscriptions;
 }
 
+/// Current information about a [SyncStream] that the sync client is subscribed
+/// to.
 final class SyncStreamStatus {
+  /// If the [SyncStatus] is currently [SyncStatus.downloading], download
+  /// progress for this stream.
   final ProgressWithOperations? progress;
   final CoreActiveStreamSubscription _internal;
 
+  /// The [SyncSubscriptionDescription] providing information about the current
+  /// stream state.
   SyncSubscriptionDescription get subscription => _internal;
+
+  /// The [StreamPriority] of the current stream.
+  ///
+  /// New data on higher-priority streams can interrupt lower-priority streams.
   StreamPriority get priority => _internal.priority;
 
   SyncStreamStatus._(this._internal, SyncDownloadProgress? progress)

packages/powersync_core/test/sync/stream_test.dart

@@ -219,4 +219,45 @@ void main() {
     var status = await stream.next;
     expect(status.forStream(subscription), isNotNull);
   });
+
+  test('unsubscribing multiple times has no effect', () async {
+    final a = await database.syncStream('a').subscribe();
+    final aAgain = await database.syncStream('a').subscribe();
+    a.unsubscribe();
+    a.unsubscribe(); // Should not decrement the refcount again
+
+    // Pretend the streams are expired - they should still be requested because
+    // the core extension extends the lifetime of streams currently referenced
+    // before connecting.
+    await database.execute(
+        'UPDATE ps_stream_subscriptions SET expires_at = unixepoch() - 1000');
+
+    await waitForConnection();
+    final request = await syncService.waitForListener;
+    expect(
+      json.decode(await request.readAsString()),
+      containsPair(
+        'streams',
+        containsPair('subscriptions', isNotEmpty),
+      ),
+    );
+    aAgain.unsubscribe();
+  });
+
+  test('unsubscribeAll', () async {
+    final a = await database.syncStream('a').subscribe();
+    await database.syncStream('a').unsubscribeAll();
+
+    // Despite a being active, it should not be requested.
+    await waitForConnection();
+    final request = await syncService.waitForListener;
+    expect(
+      json.decode(await request.readAsString()),
+      containsPair(
+        'streams',
+        containsPair('subscriptions', isEmpty),
+      ),
+    );
+    a.unsubscribe();
+  });
 }