Add RSocket client

Author: simolus3

connectors/supabase/src/commonMain/kotlin/com/powersync/connector/supabase/SupabaseConnector.kt

@@ -20,11 +20,14 @@ import io.github.jan.supabase.postgrest.from
 import io.github.jan.supabase.storage.BucketApi
 import io.github.jan.supabase.storage.Storage
 import io.github.jan.supabase.storage.storage
+import io.ktor.client.call.body
 import io.ktor.client.plugins.HttpSend
 import io.ktor.client.plugins.plugin
+import io.ktor.client.request.get
 import io.ktor.client.statement.bodyAsText
 import io.ktor.utils.io.InternalAPI
 import kotlinx.coroutines.flow.StateFlow
+import kotlinx.serialization.Serializable
 import kotlinx.serialization.json.Json
 
 /**
@@ -162,12 +165,17 @@ public class SupabaseConnector(
                 supabaseClient.auth.currentSessionOrNull()
                     ?: error("Could not fetch Supabase credentials")
 
-            check(session.user != null) { "No user data" }
+            @Serializable
+            class TokenResponse(
+                val token: String
+            )
+
+            val src = supabaseClient.httpClient.httpClient.get("http://localhost:6060/api/auth/token").body<TokenResponse>()
 
             // userId is for debugging purposes only
             PowerSyncCredentials(
-                endpoint = powerSyncEndpoint,
-                token = session.accessToken, // Use the access token to authenticate against PowerSync
+                endpoint = "http://localhost:8080", //powerSyncEndpoint,
+                token = src.token, //session.accessToken, // Use the access token to authenticate against PowerSync
                 userId = session.user!!.id,
             )
         }

core/build.gradle.kts

@@ -174,7 +174,8 @@ kotlin {
             implementation(libs.ktor.client.contentnegotiation)
             implementation(libs.ktor.serialization.json)
             implementation(libs.kotlinx.io)
-            implementation(libs.rsocket.client)
+            api(libs.rsocket.core)
+            implementation(libs.rsocket.transport.websocket)
             implementation(libs.kotlinx.coroutines.core)
             implementation(libs.kotlinx.datetime)
             implementation(libs.stately.concurrency)

core/src/commonMain/kotlin/com/powersync/bucket/BucketStorageImpl.kt

@@ -12,6 +12,13 @@ import com.powersync.sync.Instruction
 import com.powersync.sync.SyncDataBatch
 import com.powersync.sync.SyncLocalDatabaseResult
 import com.powersync.utils.JsonUtil
+import io.ktor.utils.io.asByteWriteChannel
+import io.ktor.utils.io.writeByteArray
+import kotlinx.io.Buffer
+import kotlinx.io.buffered
+import kotlinx.io.files.FileSystem
+import kotlinx.io.files.Path
+import kotlinx.io.files.SystemFileSystem
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.encodeToString
 
@@ -178,7 +185,17 @@ internal class BucketStorageImpl(
         return db.writeTransaction { tx ->
             logger.v { "powersync_control($op, binary payload)" }
 
-            tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
+            try {
+                tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, payload), ::handleControlResult)
+            } catch (e: Exception) {
+                println("Got control exception, writing")
+                SystemFileSystem.sink(Path("/Users/simon/failing_line.bin")).buffered().apply {
+                    write(payload)
+                    flush()
+                    close()
+                }
+                throw e
+            }
         }
     }
 }

core/src/commonMain/kotlin/com/powersync/sync/Progress.kt

@@ -90,7 +90,7 @@ public data class SyncDownloadProgress internal constructor(
             .asSequence()
             .filter { it.priority >= priority }
             .fold(0L to 0L) { (prevTarget, prevCompleted), entry ->
-                (prevTarget + entry.targetCount) to (prevCompleted + entry.sinceLast)
+                (prevTarget + entry.targetCount - entry.atLast) to (prevCompleted + entry.sinceLast)
             }
             .let { it.first.toInt() to it.second.toInt() }
 }

core/src/commonMain/kotlin/com/powersync/sync/RSocketSupport.kt

@@ -0,0 +1,111 @@
+package com.powersync.sync
+
+import com.powersync.connectors.PowerSyncCredentials
+import com.powersync.utils.JsonUtil
+import io.ktor.client.HttpClient
+import io.ktor.client.plugins.websocket.webSocketSession
+import io.ktor.http.URLBuilder
+import io.ktor.http.URLProtocol
+import io.ktor.http.takeFrom
+import io.rsocket.kotlin.core.RSocketConnector
+import io.rsocket.kotlin.payload.PayloadMimeType
+import io.rsocket.kotlin.payload.buildPayload
+import io.rsocket.kotlin.payload.data
+import io.rsocket.kotlin.payload.metadata
+import io.rsocket.kotlin.transport.RSocketClientTarget
+import io.rsocket.kotlin.transport.RSocketConnection
+import io.rsocket.kotlin.transport.RSocketTransportApi
+import io.rsocket.kotlin.transport.ktor.websocket.internal.KtorWebSocketConnection
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.currentCoroutineContext
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.emitAll
+import kotlinx.coroutines.flow.flow
+import kotlinx.coroutines.flow.flowOn
+import kotlinx.coroutines.flow.map
+import kotlinx.io.readByteArray
+import kotlinx.serialization.SerialName
+import kotlinx.serialization.Serializable
+import kotlinx.serialization.json.JsonObject
+import kotlin.coroutines.CoroutineContext
+
+@OptIn(RSocketTransportApi::class)
+internal fun HttpClient.rSocketSyncStream(
+    options: ConnectionMethod.WebSocket,
+    req: JsonObject,
+    credentials: PowerSyncCredentials,
+): Flow<ByteArray> = flow {
+    val flowContext = currentCoroutineContext()
+
+    val websocketUri = URLBuilder(credentials.endpointUri("sync/stream")).apply {
+        protocol = when (protocolOrNull) {
+            URLProtocol.HTTP -> URLProtocol.WS
+            else -> URLProtocol.WSS
+        }
+    }
+
+    // Note: We're using a custom connector here because we need to set options for each request
+    // without creating a new HTTP client each time. The recommended approach would be to add an
+    // RSocket extension to the HTTP client, but that only allows us to set the SETUP metadata for
+    // all connections (bad because we need a short-lived token in there).
+    // https://github.com/rsocket/rsocket-kotlin/issues/311
+    val target = object : RSocketClientTarget {
+        @RSocketTransportApi
+        override suspend fun connectClient(): RSocketConnection {
+            val ws = webSocketSession {
+                url.takeFrom(websocketUri)
+            }
+            return KtorWebSocketConnection(ws)
+        }
+
+        override val coroutineContext: CoroutineContext
+            get() = flowContext
+    }
+
+    val connector = RSocketConnector {
+        connectionConfig {
+            payloadMimeType = PayloadMimeType(
+                metadata = "application/json",
+                data = "application/json"
+            )
+
+            setupPayload {
+                buildPayload {
+                    data("{}")
+                    metadata(JsonUtil.json.encodeToString(ConnectionSetupMetadata(token="Bearer ${credentials.token}")))
+                }
+            }
+
+            keepAlive = options.keepAlive
+        }
+    }
+
+    val rSocket = connector.connect(target)
+    val syncStream = rSocket.requestStream(buildPayload {
+        data("{}")
+        metadata(JsonUtil.json.encodeToString(RequestStreamMetadata("/sync/stream")))
+    })
+
+    emitAll(syncStream.map { it.data.readByteArray() }.flowOn(Dispatchers.IO))
+}
+
+/**
+ * The metadata payload we need to use when connecting with RSocket.
+ *
+ * This corresponds to `RSocketContextMeta` on the sync service.
+ */
+@Serializable
+private class ConnectionSetupMetadata(
+    val token: String,
+    @SerialName("user_agent")
+    val userAgent: String = userAgent()
+)
+
+/**
+ * The metadata payload we send for the `REQUEST_STREAM` frame.
+ */
+@Serializable
+private class RequestStreamMetadata(
+    val path: String
+)

core/src/commonMain/kotlin/com/powersync/sync/SyncOptions.kt

@@ -4,25 +4,27 @@ import io.rsocket.kotlin.keepalive.KeepAlive
 import kotlin.time.Duration.Companion.seconds
 
 public class SyncOptions(
-    public val method: ConnectionMethod = ConnectionMethod.WebSocket,
+    public val method: ConnectionMethod = ConnectionMethod.Http,
 )
 
 /**
  * The connection method to use when the SDK connects to the sync service.
  */
 public sealed interface ConnectionMethod {
     /**
-     * Receive sync lines via an streamed HTTP response from the sync service.
+     * Receive sync lines via  streamed HTTP response from the sync service.
      *
      * This mode is less efficient than [WebSocket] because it doesn't support backpressure
      * properly and uses JSON instead of the more efficient BSON representation for sync lines.
+     *
+     * This is currently the default, but this will be changed once [WebSocket] support is stable.
      */
     public data object Http: ConnectionMethod
 
     /**
      * Receive binary sync lines via RSocket over a WebSocket connection.
      *
-     * This is the default mode, and recommended for most clients.
+     * This connection mode is currently experimental and requires a recent sync service to work.
      */
     public data class WebSocket(
         val keepAlive: KeepAlive = DefaultKeepAlive