Add sync stream test

Author: simolus3

core/build.gradle.kts

@@ -235,6 +235,8 @@ kotlin {
             implementation(libs.test.coroutines)
             implementation(libs.test.turbine)
             implementation(libs.kermit.test)
+            implementation(libs.ktor.client.mock)
+            implementation(libs.test.turbine)
         }
 
         // We're putting the native libraries into our JAR, so integration tests for the JVM can run as part of the unit

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

@@ -6,6 +6,10 @@ import kotlin.jvm.JvmInline
 @JvmInline
 @Serializable
 public value class BucketPriority(private val priorityCode: Int): Comparable<BucketPriority> {
+    init {
+        require(priorityCode >= 0)
+    }
+
     override fun compareTo(other: BucketPriority): Int {
         return other.priorityCode.compareTo(priorityCode)
     }

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

@@ -10,8 +10,9 @@ import com.powersync.bucket.WriteCheckpointResponse
 import com.powersync.connectors.PowerSyncBackendConnector
 import com.powersync.db.crud.CrudEntry
 import com.powersync.utils.JsonUtil
-import io.ktor.client.HttpClient
+import io.ktor.client.*
 import io.ktor.client.call.body
+import io.ktor.client.engine.*
 import io.ktor.client.plugins.HttpTimeout
 import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
 import io.ktor.client.plugins.timeout
@@ -45,6 +46,7 @@ internal class SyncStream(
     private val retryDelayMs: Long = 5000L,
     private val logger: Logger,
     private val params: JsonObject,
+    httpEngine: HttpClientEngine? = null,
 ) {
     private var isUploadingCrud = AtomicBoolean(false)
 
@@ -55,12 +57,25 @@ internal class SyncStream(
 
     private var clientId: String? = null
 
-    private val httpClient: HttpClient =
-        HttpClient {
+    private val httpClient: HttpClient
+
+    init {
+        fun HttpClientConfig<*>.configureClient() {
             install(HttpTimeout)
             install(ContentNegotiation)
         }
 
+        httpClient = if (httpEngine == null) {
+            HttpClient {
+                configureClient()
+            }
+        } else {
+            HttpClient(httpEngine) {
+                configureClient()
+            }
+        }
+    }
+
     fun invalidateCredentials() {
         connector.invalidateCredentials()
     }
@@ -239,16 +254,18 @@ internal class SyncStream(
             )
 
         streamingSyncRequest(req).collect { value ->
-            state = handleInstruction(value, state)
+            val line = JsonUtil.json.decodeFromString<SyncLine>(value)
+            state = handleInstruction(line, value, state)
         }
 
         return state
     }
 
     private suspend fun handleInstruction(
+        line: SyncLine,
         jsonString: String,
         state: SyncStreamState,
-    ): SyncStreamState = when (val line = JsonUtil.json.decodeFromString<SyncLine>(jsonString)) {
+    ): SyncStreamState = when (line) {
         is SyncLine.FullCheckpoint -> handleStreamingSyncCheckpoint(line, state)
         is SyncLine.CheckpointDiff -> handleStreamingSyncCheckpointDiff(line, state)
         is SyncLine.CheckpointComplete -> handleStreamingSyncCheckpointComplete(state)

core/src/commonTest/kotlin/com/powersync/sync/SyncStreamTest.kt

@@ -1,27 +1,40 @@
 package com.powersync.sync
 
+import app.cash.turbine.turbineScope
 import co.touchlab.kermit.Logger
 import co.touchlab.kermit.Severity
 import co.touchlab.kermit.TestConfig
 import co.touchlab.kermit.TestLogWriter
+import com.powersync.bucket.*
+import com.powersync.bucket.BucketChecksum
 import com.powersync.bucket.BucketStorage
+import com.powersync.bucket.Checkpoint
+import com.powersync.bucket.OplogEntry
 import com.powersync.connectors.PowerSyncBackendConnector
 import com.powersync.connectors.PowerSyncCredentials
 import com.powersync.db.crud.CrudEntry
 import com.powersync.db.crud.UpdateType
+import com.powersync.testutils.MockSyncService
+import com.powersync.testutils.waitFor
+import com.powersync.utils.JsonUtil
+import dev.mokkery.*
 import dev.mokkery.answering.returns
-import dev.mokkery.everySuspend
-import dev.mokkery.mock
-import dev.mokkery.verify
+import dev.mokkery.matcher.any
+import dev.mokkery.verify.VerifyMode.Companion.order
+import io.ktor.client.engine.mock.*
+import kotlinx.coroutines.channels.Channel
 import kotlinx.coroutines.delay
+import kotlinx.coroutines.flow.receiveAsFlow
 import kotlinx.coroutines.launch
 import kotlinx.coroutines.test.runTest
 import kotlinx.coroutines.withTimeout
+import kotlinx.serialization.encodeToString
 import kotlinx.serialization.json.JsonObject
 import kotlin.test.BeforeTest
 import kotlin.test.Test
 import kotlin.test.assertContains
 import kotlin.test.assertEquals
+import kotlin.time.Duration.Companion.seconds
 
 @OptIn(co.touchlab.kermit.ExperimentalKermitApi::class)
 class SyncStreamTest {
@@ -39,11 +52,33 @@ class SyncStreamTest {
                 logWriterList = listOf(testLogWriter),
             ),
         )
+    private val assertNoHttpEngine = MockEngine { request ->
+        error("Unexpected HTTP request: $request")
+    }
 
     @BeforeTest
     fun setup() {
-        bucketStorage = mock<BucketStorage>()
-        connector = mock<PowerSyncBackendConnector>()
+        bucketStorage = mock<BucketStorage> {
+            everySuspend { getClientId() } returns "test-client-id"
+            everySuspend { getBucketStates() } returns emptyList()
+            everySuspend { removeBuckets(any()) } returns Unit
+            everySuspend { setTargetCheckpoint(any()) } returns Unit
+            everySuspend { saveSyncData(any()) } returns Unit
+            everySuspend { syncLocalDatabase(any(), any()) } returns SyncLocalDatabaseResult(
+                ready = true,
+                checkpointValid = true,
+                checkpointFailures = emptyList()
+            )
+        }
+        connector =
+            mock<PowerSyncBackendConnector> {
+                everySuspend { getCredentialsCached() } returns
+                        PowerSyncCredentials(
+                            token = "test-token",
+                            userId = "test-user",
+                            endpoint = "https://test.com",
+                        )
+            }
     }
 
     @Test
@@ -58,6 +93,7 @@ class SyncStreamTest {
                 SyncStream(
                     bucketStorage = bucketStorage,
                     connector = connector,
+                    httpEngine = assertNoHttpEngine,
                     uploadCrud = {},
                     logger = logger,
                     params = JsonObject(emptyMap()),
@@ -92,6 +128,7 @@ class SyncStreamTest {
                 SyncStream(
                     bucketStorage = bucketStorage,
                     connector = connector,
+                    httpEngine = assertNoHttpEngine,
                     uploadCrud = { },
                     retryDelayMs = 10,
                     logger = logger,
@@ -126,20 +163,11 @@ class SyncStreamTest {
                     everySuspend { getBucketStates() } returns emptyList()
                 }
 
-            connector =
-                mock<PowerSyncBackendConnector> {
-                    everySuspend { getCredentialsCached() } returns
-                        PowerSyncCredentials(
-                            token = "test-token",
-                            userId = "test-user",
-                            endpoint = "https://test.com",
-                        )
-                }
-
             syncStream =
                 SyncStream(
                     bucketStorage = bucketStorage,
                     connector = connector,
+                    httpEngine = assertNoHttpEngine,
                     uploadCrud = { },
                     retryDelayMs = 10,
                     logger = logger,
@@ -166,4 +194,103 @@ class SyncStreamTest {
             // Clean up
             job.cancel()
         }
+
+    @Test
+    fun testPartialSync() = runTest {
+        // TODO: It would be neat if we could use in-memory sqlite instances instead of mocking everything
+        // Revisit https://github.com/powersync-ja/powersync-kotlin/pull/117/files at some point
+        val syncLines = Channel<SyncLine>()
+        val client = MockSyncService.client(this, syncLines.receiveAsFlow())
+
+        syncStream = SyncStream(
+            bucketStorage = bucketStorage,
+            connector = connector,
+            httpEngine = client,
+            uploadCrud = { },
+            retryDelayMs = 10,
+            logger = logger,
+            params = JsonObject(emptyMap()),
+        )
+
+        val job = launch { syncStream.streamingSync() }
+        var operationId = 1
+
+        suspend fun pushData(priority: Int) {
+            val id = operationId++
+
+            syncLines.send(SyncLine.SyncDataBucket(
+                bucket = "prio$priority",
+                data = listOf(OplogEntry(
+                    checksum = (priority + 10).toLong(),
+                    data = JsonUtil.json.encodeToString(mapOf("foo" to "bar")),
+                    op = OpType.PUT,
+                    opId = id.toString(),
+                    rowId = "prio$priority",
+                    rowType = "customers"
+                )),
+                after = null,
+                nextAfter = null,
+            ))
+        }
+
+        turbineScope(timeout=10.0.seconds) {
+            val turbine = syncStream.status.asFlow().testIn(this)
+            turbine.waitFor { it.connected }
+            resetCalls(bucketStorage)
+
+            // Start a sync flow
+            syncLines.send(SyncLine.FullCheckpoint(Checkpoint(
+                lastOpId = "4",
+                checksums = buildList {
+                    for (priority in 0..3) {
+                        add(BucketChecksum(
+                            bucket = "prio$priority",
+                            priority = BucketPriority(priority),
+                            checksum = 10 + priority,
+                        ))
+                    }
+                }
+            )))
+
+            // Emit a partial sync complete for each priority but the last.
+            for (priorityNo in 0..<3) {
+                val priority = BucketPriority(priorityNo)
+                pushData(priorityNo)
+                syncLines.send(SyncLine.CheckpointPartiallyComplete(
+                    lastOpId = operationId.toString(),
+                    priority = priority,
+                ))
+
+                turbine.waitFor { it.priorityStatusFor(priority).hasSynced == true }
+
+                verifySuspend(order) {
+                    if (priorityNo == 0) {
+                        bucketStorage.removeBuckets(any())
+                        bucketStorage.setTargetCheckpoint(any())
+                    }
+
+                    bucketStorage.saveSyncData(any())
+                    bucketStorage.syncLocalDatabase(any(), priority)
+                }
+            }
+
+            // Then complete the sync
+            pushData(3)
+            syncLines.send(SyncLine.CheckpointComplete(
+                lastOpId = operationId.toString(),
+            ))
+
+            turbine.waitFor { it.hasSynced == true }
+            verifySuspend {
+                bucketStorage.saveSyncData(any())
+                bucketStorage.syncLocalDatabase(any(), null)
+            }
+
+            turbine.cancel()
+        }
+
+        verifyNoMoreCalls(bucketStorage)
+        job.cancel()
+        syncLines.close()
+    }
 }

core/src/commonTest/kotlin/com/powersync/testutils/MockSyncService.kt

@@ -0,0 +1,54 @@
+package com.powersync.testutils
+
+import app.cash.turbine.ReceiveTurbine
+import com.powersync.sync.SyncLine
+import com.powersync.sync.SyncStatusData
+import com.powersync.utils.JsonUtil
+import io.ktor.client.engine.*
+import io.ktor.client.engine.mock.*
+import io.ktor.client.request.*
+import io.ktor.utils.io.*
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.launch
+import kotlinx.serialization.encodeToString
+
+internal class MockSyncService private constructor(private val scope: CoroutineScope, private val lines: Flow<SyncLine>) {
+    private fun handleRequest(scope: MockRequestHandleScope, request: HttpRequestData): HttpResponseData {
+        return if (request.url.encodedPath == "/sync/stream") {
+            val channel = ByteChannel(autoFlush = true)
+            this.scope.launch {
+                lines.collect {
+                    val serializedLine = JsonUtil.json.encodeToString(it)
+                    channel.writeStringUtf8("$serializedLine\n")
+                }
+            }
+
+            scope.respond(channel)
+        } else {
+            scope.respondBadRequest()
+        }
+    }
+
+    companion object {
+        fun client(scope: CoroutineScope, lines: Flow<SyncLine>): HttpClientEngine {
+            val service = MockSyncService(scope, lines)
+            return MockEngine { request ->
+                service.handleRequest(this, request)
+            }
+        }
+    }
+}
+
+public suspend inline fun ReceiveTurbine<SyncStatusData>.waitFor(matcher: (SyncStatusData) -> Boolean) {
+    while (true) {
+        val item = awaitItem()
+        if (matcher(item)) {
+            break
+        }
+
+        item.anyError?.let {
+            error("Unexpected error in $item")
+        }
+    }
+}

gradle/libs.versions.toml

@@ -72,6 +72,7 @@ ktor-client-core = { module = "io.ktor:ktor-client-core", version.ref = "ktor" }
 ktor-client-ios = { module = "io.ktor:ktor-client-darwin", version.ref = "ktor" }
 ktor-client-okhttp = { module = "io.ktor:ktor-client-okhttp", version.ref = "ktor" }
 ktor-client-contentnegotiation = { module = "io.ktor:ktor-client-content-negotiation", version.ref = "ktor" }
+ktor-client-mock = { module = "io.ktor:ktor-client-mock", version.ref = "ktor" }
 ktor-serialization-json = { module = "io.ktor:ktor-serialization-kotlinx-json", version.ref = "ktor" }
 kotlinx-coroutines-core = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-core", version.ref = "coroutines" }