Implement sync streams

Author: simolus3

Package.swift

@@ -7,7 +7,7 @@ let packageName = "PowerSync"
 
 // Set this to the absolute path of your Kotlin SDK checkout if you want to use a local Kotlin
 // build. Also see docs/LocalBuild.md for details
-let localKotlinSdkOverride: String? = nil
+let localKotlinSdkOverride: String? = "/Users/simon/src/powersync-kotlin"
 
 // Set this to the absolute path of your powersync-sqlite-core checkout if you want to use a
 // local build of the core extension.

Sources/PowerSync/Kotlin/KotlinPowerSyncDatabaseImpl.swift

@@ -329,6 +329,11 @@ final class KotlinPowerSyncDatabaseImpl: PowerSyncDatabaseProtocol,
     func close() async throws {
         try await kotlinDatabase.close()
     }
+    
+    func syncStream(name: String, params: JsonParam?) -> any SyncStream {
+        let rawStream = kotlinDatabase.syncStream(name: name, parameters: params?.mapValues { $0.toKotlinMap() });
+        return KotlinSyncStream(kotlinStream: rawStream)
+    }
 
     /// Tries to convert Kotlin PowerSyncExceptions to Swift Exceptions
     private func wrapPowerSyncException<R: Sendable>(

Sources/PowerSync/Kotlin/sync/KotlinSyncStatusData.swift

@@ -72,6 +72,27 @@ extension KotlinSyncStatusDataProtocol {
             )
         )
     }
+    
+    var syncStreams: [SyncStreamStatus]? {
+        return base.syncStreams?.map(mapSyncStreamStatus)
+    }
+
+    func forStream(stream: SyncStreamDescription) -> SyncStreamStatus? {
+        let name = stream.name
+        // To match parameters, first check if we already have access to a Kotlin map for parameters.
+        let parameters = if let kotlinStream = stream as? any HasKotlinStreamDescription {
+            // Fast path: Reuse Kotlin map
+            kotlinStream.kotlinParameters
+        } else {
+            // We don't? Ok, map to Kotlin.
+            stream.parameters?.mapValues { $0.toValue() }
+        }
+        
+        guard let kotlinStatus = syncStatusForStream(status: base, name: stream.name, parameters: parameters) else {
+            return nil
+        }
+        return mapSyncStreamStatus(kotlinStatus)
+    }
 
     private func mapPriorityStatus(_ status: PowerSyncKotlin.PriorityStatusEntry) -> PriorityStatusEntry {
         var lastSyncedAt: Date?

Sources/PowerSync/Kotlin/sync/KotlinSyncStreams.swift

@@ -0,0 +1,125 @@
+import Foundation
+import PowerSyncKotlin
+
+class KotlinStreamDescription<T: PowerSyncKotlin.SyncStreamDescription> {
+    let inner: T
+    let name: String
+    let parameters: JsonParam?
+    let kotlinParameters: [String: Any?]?
+    
+    init(inner: T) {
+        self.inner = inner
+        self.name = inner.name
+        self.kotlinParameters = inner.parameters
+        self.parameters = inner.parameters?.mapValues { JsonValue.fromValue(raw: $0) }
+    }
+}
+
+protocol HasKotlinStreamDescription {
+    associatedtype Description: PowerSyncKotlin.SyncStreamDescription
+    
+    var stream: KotlinStreamDescription<Description> { get }
+}
+
+extension HasKotlinStreamDescription {
+    var kotlinParameters: [String: Any?]? {
+        self.stream.kotlinParameters
+    }
+}
+
+class KotlinSyncStream: SyncStream, HasKotlinStreamDescription,
+// `PowerSyncKotlin.SyncStream` cannot be marked as Sendable, but is thread-safe.
+@unchecked Sendable
+{
+    let stream: KotlinStreamDescription<PowerSyncKotlin.SyncStream>
+    
+    init(kotlinStream: PowerSyncKotlin.SyncStream) {
+        self.stream = KotlinStreamDescription(inner: kotlinStream);
+    }
+    
+    var name: String {
+        stream.name
+    }
+    
+    var parameters: JsonParam? {
+        stream.parameters
+    }
+
+    func subscribe(ttl: TimeInterval?, priority: BucketPriority?) async throws -> any SyncStreamSubscription {
+        let kotlinTtl: Optional<KotlinDouble> = if let ttl {
+            KotlinDouble(value: ttl)
+        } else {
+            nil
+        }
+        let kotlinPriority: Optional<KotlinInt> = if let priority {
+            KotlinInt(value: priority.priorityCode)
+        } else {
+            nil
+        }
+
+        let kotlinSubscription = try await syncStreamSubscribeSwift(
+            stream: stream.inner,
+            ttl: kotlinTtl,
+            priority: kotlinPriority,
+        );
+        return KotlinSyncStreamSubscription(kotlinStream: kotlinSubscription)
+    }
+    
+    func unsubscribeAll() async throws {
+        try await stream.inner.unsubscribeAll()
+    }
+}
+
+class KotlinSyncStreamSubscription: SyncStreamSubscription, HasKotlinStreamDescription,
+// `PowerSyncKotlin.SyncStreamSubscription` cannot be marked as Sendable, but is thread-safe.
+@unchecked Sendable
+{
+    let stream: KotlinStreamDescription<PowerSyncKotlin.SyncStreamSubscription>
+
+    init(kotlinStream: PowerSyncKotlin.SyncStreamSubscription) {
+        self.stream = KotlinStreamDescription(inner: kotlinStream)
+    }
+    
+    var name: String {
+        stream.name
+    }
+    var parameters: JsonParam? {
+        stream.parameters
+    }
+    
+    func waitForFirstSync() async throws {
+        try await stream.inner.waitForFirstSync()
+    }
+    
+    func unsubscribe() async throws {
+        try await stream.inner.unsubscribe()
+    }
+}
+
+func mapSyncStreamStatus(_ status: PowerSyncKotlin.SyncStreamStatus) -> SyncStreamStatus {
+    let progress = status.progress.map { ProgressNumbers(source: $0) }
+    let subscription = status.subscription
+
+    return SyncStreamStatus(
+        progress: progress,
+        subscription: SyncSubscriptionDescription(
+            name: subscription.name,
+            parameters: subscription.parameters?.mapValues { JsonValue.fromValue(raw: $0) },
+            active: subscription.active,
+            isDefault: subscription.isDefault,
+            hasExplicitSubscription: subscription.hasExplicitSubscription,
+            expiresAt: subscription.expiresAt.map { Double($0.epochSeconds) },
+            lastSyncedAt: subscription.lastSyncedAt.map { Double($0.epochSeconds) }
+        )
+    )
+}
+
+struct ProgressNumbers: ProgressWithOperations {
+    let totalOperations: Int32
+    let downloadedOperations: Int32
+    
+    init(source: PowerSyncKotlin.ProgressWithOperations) {
+        self.totalOperations = source.totalOperations
+        self.downloadedOperations = source.downloadedOperations
+    }
+}

Sources/PowerSync/Protocol/PowerSyncDatabaseProtocol.swift

@@ -230,6 +230,11 @@ public protocol PowerSyncDatabaseProtocol: Queries, Sendable {
     /// Using soft clears is recommended where it's not a security issue that old data could be reconstructed from
     /// the database.
     func disconnectAndClear(clearLocal: Bool, soft: Bool) async throws
+    
+    /// Create a ``SyncStream`` instance for the given name and parameters.
+    ///
+    /// Use ``SyncStream/subscribe`` on the returned instance to subscribe to the stream.
+    func syncStream(name: String, params: JsonParam?) -> any SyncStream
 
     /// Close the database, releasing resources.
     /// Also disconnects any active connection.

Sources/PowerSync/Protocol/db/JsonParam.swift

@@ -50,6 +50,27 @@ public enum JsonValue: Codable, Sendable {
             return anyDict
         }
     }
+    
+    /// Converts a raw Swift value into a ``JsonValue``.
+    ///
+    /// The value must be one of the types returned by ``JsonValue/toValue()``.
+    static func fromValue(raw: Any?) -> Self {
+        if let string = raw as? String {
+            return Self.string(string)
+        } else if let int = raw as? Int {
+            return Self.int(int)
+        } else if let double = raw as? Double {
+            return Self.double(double)
+        } else if let bool = raw as? Bool {
+            return Self.bool(bool)
+        } else if let array = raw as? [Any?] {
+            return Self.array(array.map(fromValue))
+        } else if let object = raw as? [String: Any?] {
+            return Self.object(object.mapValues(fromValue))
+        } else {
+            return Self.null
+        }
+    }
 }
 
 /// A typealias representing a top-level JSON object with string keys and `JSONValue` values.

Sources/PowerSync/Protocol/sync/SyncStatusData.swift

@@ -47,6 +47,15 @@ public protocol SyncStatusData: Sendable {
     /// - Parameter priority: The priority for which the status is requested.
     /// - Returns: A `PriorityStatusEntry` representing the synchronization status for the given priority.
     func statusForPriority(_ priority: BucketPriority) -> PriorityStatusEntry
+    
+    /// All sync streams currently being tracked in the database.
+    ///
+    /// This returns null when the database is currently being opened and we don't have reliable information about
+    /// included streams yet.
+    var syncStreams: [SyncStreamStatus]? { get }
+
+    /// Status information for the given stream, if it's a stream that is currently tracked by the sync client.
+    func forStream(stream: SyncStreamDescription) -> SyncStreamStatus?
 }
 
 /// A protocol extending `SyncStatusData` to include flow-based updates for synchronization status.
@@ -55,3 +64,11 @@ public protocol SyncStatus: SyncStatusData, Sendable {
     /// - Returns: An `AsyncStream` that emits updates whenever the synchronization status changes.
     func asFlow() -> AsyncStream<SyncStatusData>
 }
+
+/// Current information about a ``SyncStreamSubscription``.
+public struct SyncStreamStatus {
+    /// If the sync status is currently downloading, information about download progress related to this stream.
+    let progress: ProgressWithOperations?
+    /// The ``SyncSubscriptionDescription`` providing information about the subscription.
+    let subscription: SyncSubscriptionDescription
+}

Sources/PowerSync/Protocol/sync/SyncStream.swift

@@ -0,0 +1,84 @@
+import Foundation
+
+/// Information uniquely identifying a sync stream that can be subscribed to.
+public protocol SyncStreamDescription: Sendable {
+    /// The name of the sync stream as it appeaers in the stream definition for the PowerSync service.
+    var name: String { get }
+    /// The parameters used to subscribe to the stream, if any.
+    ///
+    /// The same stream can be subscribed to multiple times with different parameters.
+    var parameters: JsonParam? { get }
+}
+
+/// A handle to a ``SyncStreamDescription`` that allows subscribing to the stream.
+///
+/// To obtain an instance of ``SyncStream``, call ``PowerSyncDatabase/syncStream``.
+public protocol SyncStream: SyncStreamDescription {
+    /// Creates a new subscription on this stream.
+    ///
+    /// As long as a subscription is active on the stream, the sync client will request it from the sync service.
+    ///
+    /// This call is generally quite cheap and can be issued frequently, e.g. when a view needing data from the stream is activated.
+    func subscribe(ttl: TimeInterval?, priority: BucketPriority?) async throws -> any SyncStreamSubscription
+    
+    /// Unsubscribes all existing subscriptions on this stream.
+    ///
+    /// This is a potentially unsafe method since it interferes with other subscriptions. A better option is to call
+    /// ``SyncStreamSubscription/unsubscribe``.
+    func unsubscribeAll() async throws
+}
+
+extension SyncStream {
+    
+    public func subscribe() async throws -> any SyncStreamSubscription {
+        return try await subscribe(ttl: nil, priority: nil)
+    }
+}
+
+/// A ``SyncStream`` that has an active subscription.
+public protocol SyncStreamSubscription: SyncStreamDescription {
+    /// An asynchronous function that completes once data on this stream has been synced.
+    func waitForFirstSync() async throws
+    /// Removes this subscription.
+    ///
+    /// Once all ``SyncStreamSubscription``s for a ``SyncStream`` have been unsubscribed, the `ttl`
+    /// for that stream thats running. When it expires without subscribing again, the stream will be evicted.
+    func unsubscribe() async throws
+}
+
+/// Information about a subscribed sync stream.
+///
+/// This includes the  ``SyncStreamDescription`` along with information about the current sync status.
+public struct SyncSubscriptionDescription: SyncStreamDescription {
+    public let name: String
+    public let parameters: JsonParam?
+    /// Whether this stream is active, meaning that the subscription has been acknowledged by the sync service.
+    public let active: Bool
+    /// Whether this stream subscription is included by default, regardless of whether the stream has explicitly
+    /// been subscribed to or not.
+    ///
+    /// Default streams are created by applying `auto_subscribe: true` in their definition on the sync service.
+    ///
+    /// It's possible for both ``SyncSubscriptionDescription/isDefault`` and
+    /// ``SyncSubscriptionDescription/hasExplicitSubscription`` to be true at the same time. This
+    /// happens when a default stream was subscribed to explicitly.
+    public let isDefault: Bool
+    /// Whether this stream has been subscribed to explicitly.
+    ///
+    /// It's possible for both ``SyncSubscriptionDescription/isDefault`` and
+    /// ``SyncSubscriptionDescription/hasExplicitSubscription`` to be true at the same time. This
+    /// happens when a default stream was subscribed to explicitly.
+    public let hasExplicitSubscription: Bool
+    /// For sync streams that have a time-to-live, the current time at which the stream would expire if not subscribed to
+    /// again.
+    public let expiresAt: TimeInterval?
+    /// If ``SyncSubscriptionDescription/hasSynced`` is true, the last time data from this stream has been synced.
+    public let lastSyncedAt: TimeInterval?
+    
+    /// Whether this stream has been synced at least once.
+    public var hasSynced: Bool {
+        get {
+            return self.expiresAt != nil
+        }
+    }
+}

Tests/PowerSyncTests/Kotlin/KotlinPowerSyncDatabaseImplTests.swift

@@ -624,4 +624,20 @@ final class KotlinPowerSyncDatabaseImplTests: XCTestCase {
         XCTAssertEqual(result[0], JoinOutput(name: "Test User", description: "task 1", comment: "comment 1"))
         XCTAssertEqual(result[1], JoinOutput(name: "Test User", description: "task 2", comment: "comment 2"))
     }
+    
+    func testSubscriptionsUpdateStateWhileOffline() async throws {
+        var streams = database.currentStatus.asFlow().makeAsyncIterator()
+        let initialStatus = await streams.next(); // Ignore initial
+        XCTAssertEqual(initialStatus?.syncStreams?.count, 0)
+        
+        // Subscribing while offline should add the stream to the subscriptions reported in the status.
+        let subscription = try await database.syncStream(name: "foo", params: ["foo": JsonValue.string("bar")]).subscribe()
+        let updatedStatus = await streams.next();
+        
+        XCTAssertEqual(updatedStatus?.syncStreams?.count, 1)
+        let status = updatedStatus?.forStream(stream: subscription)
+        XCTAssertNotNil(status)
+        
+        XCTAssertNil(status?.progress)
+    }
 }