Sync streams

CHANGELOG.md

@@ -1,8 +1,9 @@
 # Changelog
 
-## 1.6.1 (unreleased)
+## 1.7.0 (unreleased)
 
-* Update Kotlin SDK to 1.7.0.
+* Update Kotlin SDK to 1.8.0.
+* Add experimental support for [sync streams](https://docs.powersync.com/usage/sync-streams).
 
 ## 1.6.0
 

Demo/PowerSyncExample.xcodeproj/project.pbxproj

@@ -3,7 +3,7 @@
 	archiveVersion = 1;
 	classes = {
 	};
-	objectVersion = 56;
+	objectVersion = 60;
 	objects = {
 
 /* Begin PBXBuildFile section */

Demo/PowerSyncExample/Components/TodoListView.swift

@@ -1,5 +1,6 @@
 import AVFoundation
 import IdentifiedCollections
+import PowerSync
 import SwiftUI
 import SwiftUINavigation
 
@@ -11,6 +12,7 @@ struct TodoListView: View {
     @State private var error: Error?
     @State private var newTodo: NewTodo?
     @State private var editing: Bool = false
+    @State private var loadingListItems: Bool = false
 
 #if os(iOS)
     // Called when a photo has been captured. Individual widgets should register the listener
@@ -33,6 +35,10 @@ struct TodoListView: View {
                     }
                 }
             }
+
+            if (loadingListItems) {
+                ProgressView()
+            }
 
             ForEach(todos) { todo in
 #if os(iOS)
@@ -142,6 +148,22 @@ struct TodoListView: View {
                 }
             }
         }
+        .task {
+            if (Secrets.previewSyncStreams) {
+                // With sync streams, todo items are not loaded by default. We have to request them while we need them.
+                // Thanks to builtin caching, navingating to the same list multiple times does not have to fetch items again.
+                loadingListItems = true
+                do {
+                    // This will make the sync client request items from this list as long as we keep a reference to the stream subscription,
+                    // and a default TTL of one day afterwards.
+                    let streamSubscription = try await system.db.syncStream(name: "todos", params: ["list": JsonValue.string(listId)]).subscribe()
+                    try await streamSubscription.waitForFirstSync()
+                } catch {
+                    print("Error subscribing to list stream \(error)")
+                }
+                loadingListItems = false
+            }
+        }
     }
 
     func toggleCompletion(of todo: Todo) async {

Demo/PowerSyncExample/PowerSync/SystemManager.swift

@@ -78,11 +78,12 @@ final class SystemManager {
                 options: ConnectOptions(
                     clientConfiguration: SyncClientConfiguration(
                         requestLogger: SyncRequestLoggerConfiguration(
-                            requestLevel: .headers
+                            requestLevel: .all
                         ) { message in
                             self.db.logger.debug(message, tag: "SyncRequest")
                         }
-                    )
+                    ),
+                    newClientImplementation: true,
                 )
             )
             try await attachments?.startSync()

Demo/PowerSyncExample/Screens/HomeScreen.swift

@@ -8,7 +8,6 @@ struct HomeScreen: View {
 
 
     var body: some View {
-
         ListView()
             .toolbar {
                 ToolbarItem(placement: .cancellationAction) {

Demo/PowerSyncExample/Secrets.template.swift

@@ -17,4 +17,24 @@ extension Secrets {
     static var supabaseStorageBucket: String? {
         return nil
     }
+
+    static var previewSyncStreams: Bool {
+        /* 
+        Set to true to preview https://docs.powersync.com/usage/sync-streams.
+        When enabling this, also set your sync rules to the following:
+
+        streams:
+          lists:
+            query: SELECT * FROM lists WHERE owner_id = auth.user_id()
+            auto_subscribe: true
+          todos:
+            query: SELECT * FROM todos WHERE list_id = subscription.parameter('list') AND list_id IN (SELECT id FROM lists WHERE owner_id = auth.user_id())
+
+        config:
+          edition: 2
+
+        */
+
+        false
+    }
 }

Demo/PowerSyncExample/_Secrets.swift

@@ -6,6 +6,7 @@ protocol SecretsProvider {
     static var supabaseURL: URL { get }
     static var supabaseAnonKey: String { get }
     static var supabaseStorageBucket: String? { get }
+    static var previewSyncStreams: Bool { get }
 }
 
 // Default conforming type

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,24 @@ extension KotlinSyncStatusDataProtocol {
             )
         )
     }
+
+    var syncStreams: [SyncStreamStatus]? {
+        return base.syncStreams?.map(mapSyncStreamStatus)
+    }
+
+    func forStream(stream: SyncStreamDescription) -> SyncStreamStatus? {
+        var rawStatus: Optional<PowerSyncKotlin.SyncStreamStatus>
+        if let kotlinStream = stream as? any HasKotlinStreamDescription {
+            // Fast path: Reuse Kotlin stream object for lookup.
+            rawStatus = base.forStream(stream: kotlinStream.kotlinDescription)
+        } else {
+            // Custom stream description, we have to convert parameters to a Kotlin map.
+            let parameters = stream.parameters?.mapValues { $0.toValue() }
+            rawStatus = syncStatusForStream(status: base, name: stream.name, parameters: parameters)
+        }
+
+        return rawStatus.map(mapSyncStreamStatus)
+    }
 
     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 kotlinDescription: any PowerSyncKotlin.SyncStreamDescription {
+        self.stream.inner
+    }
+}
+
+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

@@ -81,27 +81,14 @@ public struct ConnectOptions: Sendable {
     ///   - retryDelay: Delay TimeInterval between retry attempts in milliseconds. Defaults to `5` seconds.
     ///   - params: Custom sync parameters to send to the server. Defaults to an empty dictionary.
     ///   - clientConfiguration: Configuration for the HTTP client used to connect to PowerSync.
+    ///   - newClientImplementation: Whether to use a new sync client implemented in Rust. Currently defaults to
+    ///   `false`, but we encourage users to try it out.
     public init(
         crudThrottle: TimeInterval = 1,
         retryDelay: TimeInterval = 5,
         params: JsonParam = [:],
-        clientConfiguration: SyncClientConfiguration? = nil
-    ) {
-        self.crudThrottle = crudThrottle
-        self.retryDelay = retryDelay
-        self.params = params
-        newClientImplementation = false
-        self.clientConfiguration = clientConfiguration
-    }
-
-    /// Initializes a ``ConnectOptions`` instance with optional values, including experimental options.
-    @_spi(PowerSyncExperimental)
-    public init(
-        crudThrottle: TimeInterval = 1,
-        retryDelay: TimeInterval = 5,
-        params: JsonParam = [:],
+        clientConfiguration: SyncClientConfiguration? = nil,
         newClientImplementation: Bool = false,
-        clientConfiguration: SyncClientConfiguration? = nil
     ) {
         self.crudThrottle = crudThrottle
         self.retryDelay = retryDelay
@@ -230,6 +217,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.