Initial API

Author: simolus3

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

@@ -191,14 +191,7 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
       }
     }
 
-    if (matches(currentStatus)) {
-      return;
-    }
-    await for (final result in statusStream) {
-      if (matches(result)) {
-        break;
-      }
-    }
+    return _connections.firstStatusMatching(matches);
   }
 
   @protected
@@ -514,6 +507,10 @@ mixin PowerSyncDatabaseMixin implements SqliteConnection {
   Future<void> refreshSchema() async {
     await database.refreshSchema();
   }
+
+  SyncStream syncStream(String name, [Map<String, Object?>? parameters]) {
+    return _connections.syncStream(name, parameters);
+  }
 }
 
 Stream<UpdateNotification> powerSyncUpdateNotifications(

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

@@ -1,36 +1,39 @@
 import 'dart:async';
+import 'dart:convert';
 
 import 'package:meta/meta.dart';
 import 'package:powersync_core/powersync_core.dart';
 import 'package:powersync_core/src/abort_controller.dart';
 import 'package:powersync_core/src/database/active_instances.dart';
 import 'package:powersync_core/src/database/powersync_db_mixin.dart';
 import 'package:powersync_core/src/sync/options.dart';
+import 'package:powersync_core/src/sync/stream.dart';
 
 @internal
 final class ConnectionManager {
   final PowerSyncDatabaseMixin db;
-  final StreamController<SyncStatus> _statusController = StreamController();
+  final ActiveDatabaseGroup _activeGroup;
 
+  final StreamController<SyncStatus> _statusController = StreamController();
   SyncStatus _currentStatus =
       const SyncStatus(connected: false, lastSyncedAt: null);
 
   SyncStatus get currentStatus => _currentStatus;
   Stream<SyncStatus> get statusStream => _statusController.stream;
 
-  final ActiveDatabaseGroup _activeGroup;
-
-  ConnectionManager(this.db) : _activeGroup = db.group;
-
   /// The abort controller for the current sync iteration.
   ///
   /// null when disconnected, present when connecting or connected.
   ///
   /// The controller must only be accessed from within a critical section of the
   /// sync mutex.
-  @protected
   AbortController? _abortActiveSync;
 
+  /// Only to be called in the sync mutex.
+  Future<void> Function()? _connectWithLastOptions;
+
+  ConnectionManager(this.db) : _activeGroup = db.group;
+
   void checkNotConnected() {
     if (_abortActiveSync != null) {
       throw StateError('Cannot update schema while connected');
@@ -55,12 +58,39 @@ final class ConnectionManager {
   Future<void> disconnect() async {
     // Also wrap this in the sync mutex to ensure there's no race between us
     // connecting and disconnecting.
-    await _activeGroup.syncConnectMutex.lock(_abortCurrentSync);
+    await _activeGroup.syncConnectMutex.lock(() async {
+      await _abortCurrentSync();
+      _connectWithLastOptions = null;
+    });
 
     manuallyChangeSyncStatus(
         SyncStatus(connected: false, lastSyncedAt: currentStatus.lastSyncedAt));
   }
 
+  Future<void> firstStatusMatching(bool Function(SyncStatus) predicate) async {
+    if (predicate(currentStatus)) {
+      return;
+    }
+    await for (final result in statusStream) {
+      if (predicate(result)) {
+        break;
+      }
+    }
+  }
+
+  Future<void> reconnect() async {
+    // Also wrap this in the sync mutex to ensure there's no race between us
+    // connecting and disconnecting.
+    await _activeGroup.syncConnectMutex.lock(() async {
+      if (_connectWithLastOptions case final activeSync?) {
+        await _abortCurrentSync();
+        assert(_abortActiveSync == null);
+
+        await activeSync();
+      }
+    });
+  }
+
   Future<void> connect({
     required PowerSyncBackendConnector connector,
     required ResolvedSyncOptions options,
@@ -112,6 +142,7 @@ final class ConnectionManager {
       // Disconnect a previous sync client, if one is active.
       await _abortCurrentSync();
       assert(_abortActiveSync == null);
+      _connectWithLastOptions = connectWithSyncLock;
 
       // Install the abort controller for this particular connect call, allowing
       // it to be disconnected.
@@ -150,7 +181,158 @@ final class ConnectionManager {
     }
   }
 
+  Future<void> _subscriptionsCommand(Object? command) async {
+    await db.writeTransaction((tx) {
+      return db.execute(
+        'SELECT powersync_control(?, ?)',
+        ['subscriptions', json.encode(command)],
+      );
+    });
+
+    await reconnect();
+  }
+
+  Future<void> subscribe({
+    required String stream,
+    required Object? parameters,
+    Duration? ttl,
+    BucketPriority? priority,
+  }) async {
+    await _subscriptionsCommand({
+      'subscribe': {
+        'stream': stream,
+        'params': parameters,
+        'ttl': ttl?.inSeconds,
+        'priority': priority,
+      },
+    });
+  }
+
+  Future<void> unsubscribe({
+    required String stream,
+    required Object? parameters,
+    required bool immediate,
+  }) async {
+    await _subscriptionsCommand({
+      'unsubscribe': {
+        'stream': stream,
+        'params': parameters,
+        'immediate': immediate,
+      },
+    });
+  }
+
+  Future<SyncStreamSubscription?> resolveCurrent(
+      String name, Map<String, Object?>? parameters) async {
+    final row = await db.getOptional(
+      'SELECT stream_name, active, is_default, local_priority, local_params, expires_at, last_synced_at FROM ps_stream_subscriptions WHERE stream_name = ? AND local_params = ?',
+      [name, json.encode(parameters)],
+    );
+
+    if (row == null) {
+      return null;
+    }
+
+    return _SyncStreamSubscription(
+      this,
+      name: name,
+      parameters:
+          json.decode(row['local_params'] as String) as Map<String, Object?>?,
+      active: row['active'] != 0,
+      expiresAt: switch (row['expires_at']) {
+        null => null,
+        final expiresAt as int =>
+          DateTime.fromMicrosecondsSinceEpoch(expiresAt * 1000),
+      },
+      hasSynced: row['has_synced'] != 0,
+      lastSyncedAt: switch (row['last_synced_at']) {
+        null => null,
+        final lastSyncedAt as int =>
+          DateTime.fromMicrosecondsSinceEpoch(lastSyncedAt * 1000),
+      },
+    );
+  }
+
+  SyncStream syncStream(String name, Map<String, Object?>? parameters) {
+    return _SyncStreamImplementation(this, name, parameters);
+  }
+
   void close() {
     _statusController.close();
   }
 }
+
+final class _SyncStreamImplementation implements SyncStream {
+  @override
+  final String name;
+
+  @override
+  final Map<String, Object?>? parameters;
+
+  final ConnectionManager _connections;
+
+  _SyncStreamImplementation(this._connections, this.name, this.parameters);
+
+  @override
+  Future<SyncStreamSubscription?> get current {
+    return _connections.resolveCurrent(name, parameters);
+  }
+
+  @override
+  Future<void> subscribe(
+      {Duration? ttl,
+      BucketPriority? priority,
+      Map<String, Object?>? parameters}) async {
+    await _connections.subscribe(
+      stream: name,
+      parameters: parameters,
+      ttl: ttl,
+      priority: priority,
+    );
+  }
+}
+
+final class _SyncStreamSubscription implements SyncStreamSubscription {
+  final ConnectionManager _connections;
+
+  @override
+  final String name;
+  @override
+  final Map<String, Object?>? parameters;
+
+  @override
+  final bool active;
+  @override
+  final DateTime? expiresAt;
+  @override
+  final bool hasSynced;
+  @override
+  final DateTime? lastSyncedAt;
+
+  _SyncStreamSubscription(
+    this._connections, {
+    required this.name,
+    required this.parameters,
+    required this.active,
+    required this.expiresAt,
+    required this.hasSynced,
+    required this.lastSyncedAt,
+  });
+
+  @override
+  Future<void> unsubscribe({bool immediately = false}) async {
+    await _connections.unsubscribe(
+        stream: name, parameters: parameters, immediate: immediately);
+  }
+
+  @override
+  Future<void> waitForFirstSync() async {
+    if (hasSynced) {
+      return;
+    }
+    return _connections.firstStatusMatching((status) {
+      final currentProgress = status.statusFor(this);
+      return currentProgress?.subscription.hasSynced ?? false;
+    });
+  }
+}

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

@@ -21,6 +21,8 @@ abstract interface class SyncStream extends SyncStreamDescription {
     BucketPriority? priority,
     Map<String, Object?>? parameters,
   });
+
+  Future<SyncStreamSubscription?> get current;
 }
 
 abstract interface class SyncStreamSubscription