Supabase error handling

CHANGELOG.md

@@ -4,6 +4,8 @@
 
 - Add `PowerSyncDatabase.inMemory` to create an in-memory SQLite database with PowerSync.
   This may be useful for testing.
+- The Supabase connector can now be subclassed to customize how rows are uploaded and how errors are handled.
+- Experimental support for sync streams.
 
 ## 1.6.1
 

README.md

@@ -26,18 +26,14 @@ and API documentation [here](https://powersync-ja.github.io/powersync-kotlin/).
 
     - This is the Kotlin Multiplatform SDK implementation.
 
-- [connectors](./connectors/)
-
-    - [SupabaseConnector.kt](./connectors/supabase/src/commonMain/kotlin/com/powersync/connector/supabase/SupabaseConnector.kt) An example connector implementation for Supabase (Postgres). The backend
-      connector provides the connection between your application backend and the PowerSync managed database. It is used to:
-        1. Retrieve a token to connect to the PowerSync service.
-        2. Apply local changes on your backend application server (and from there, to your backend database).
-
 - [integrations](./integrations/)
   - [room](./integrations/room/README.md): Allows using the [Room database library](https://developer.android.com/jetpack/androidx/releases/room) with PowerSync, making it easier to run typed queries on the database.
   - [sqldelight](./integrations/sqldelight/README.md): Allows using [SQLDelight](https://sqldelight.github.io/sqldelight)
     with PowerSync, also enabling typed statements on the database.
-
+  - [SupabaseConnector.kt](./integrations/supabase/src/commonMain/kotlin/com/powersync/connector/supabase/SupabaseConnector.kt) An example connector implementation for Supabase (Postgres). The backend
+    connector provides the connection between your application backend and the PowerSync managed database. It is used to:
+      1. Retrieve a token to connect to the PowerSync service.
+      2. Apply local changes on your backend application server (and from there, to your backend database).
 
 ## Demo Apps / Example Projects
 

build.gradle.kts

@@ -68,10 +68,10 @@ tasks.getByName<Delete>("clean") {
 // Merges individual module docs into a single HTML output
 dependencies {
     dokka(project(":core:"))
-    dokka(project(":connectors:supabase"))
     dokka(project(":compose:"))
     dokka(project(":integrations:room"))
     dokka(project(":integrations:sqldelight"))
+    dokka(project(":integrations:supabase"))
 }
 
 dokka {

core/src/commonIntegrationTest/kotlin/com/powersync/DatabaseTest.kt

@@ -1,7 +1,5 @@
 package com.powersync
 
-import androidx.sqlite.SQLiteConnection
-import androidx.sqlite.execSQL
 import app.cash.turbine.test
 import app.cash.turbine.turbineScope
 import co.touchlab.kermit.ExperimentalKermitApi

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

@@ -163,9 +163,10 @@ class SyncStreamTest : AbstractSyncTest(true) {
                 requestedSyncStreams.clear()
 
                 val subscription = database.syncStream("a").subscribe()
+                waitForSyncLinesChannelClosed()
 
                 // Adding the subscription should reconnect
-                turbine.waitFor { it.connected && !it.downloading }
+                turbine.waitFor { it.connected }
                 requestedSyncStreams shouldHaveSingleElement {
                     val streams = it.jsonObject["streams"]!!.jsonObject
                     val subscriptions = streams["subscriptions"]!!.jsonArray

core/src/commonIntegrationTest/kotlin/com/powersync/testutils/TestUtils.kt

@@ -23,10 +23,12 @@ import io.ktor.client.HttpClient
 import io.ktor.client.engine.mock.toByteArray
 import io.ktor.http.ContentType
 import kotlinx.coroutines.channels.Channel
+import kotlinx.coroutines.suspendCancellableCoroutine
 import kotlinx.coroutines.test.TestScope
 import kotlinx.coroutines.test.runTest
 import kotlinx.io.files.Path
 import kotlinx.serialization.json.JsonElement
+import kotlin.coroutines.resume
 
 expect val factory: DatabaseDriverFactory
 
@@ -102,6 +104,23 @@ internal class ActiveDatabaseTest(
 
     var connector = TestConnector()
 
+    suspend fun waitForSyncLinesChannelClosed() {
+        suspendCancellableCoroutine { continuation ->
+            var cancelled = false
+            continuation.invokeOnCancellation {
+                cancelled = true
+            }
+
+            syncLines.invokeOnClose {
+                if (!cancelled) {
+                    continuation.resume(Unit)
+                }
+
+                syncLines = Channel()
+            }
+        }
+    }
+
     fun openDatabase(schema: Schema = Schema(UserRow.table)): PowerSyncDatabaseImpl {
         logger.d { "Opening database $databaseName in directory $testDirectory" }
         val db =
@@ -123,7 +142,7 @@ internal class ActiveDatabaseTest(
     fun createSyncClient(): HttpClient {
         val engine =
             MockSyncService(
-                lines = syncLines,
+                lines = { syncLines },
                 generateCheckpoint = { checkpointResponse() },
                 syncLinesContentType = { syncLinesContentType },
                 trackSyncRequest = {

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

@@ -38,7 +38,7 @@ import kotlinx.serialization.json.JsonElement
  */
 @OptIn(LegacySyncImplementation::class)
 internal class MockSyncService(
-    private val lines: ReceiveChannel<Any>,
+    private val lines: () -> ReceiveChannel<Any>,
     private val syncLinesContentType: () -> ContentType,
     private val generateCheckpoint: () -> WriteCheckpointResponse,
     private val trackSyncRequest: suspend (HttpRequestData) -> Unit,
@@ -60,12 +60,11 @@ internal class MockSyncService(
             trackSyncRequest(data)
             val job =
                 scope.writer {
-                    lines.consume {
+                    lines().consume {
                         while (true) {
                             // Wait for a downstream listener being ready before requesting a sync line
                             channel.awaitFreeSpace()
-                            val line = receive()
-                            when (line) {
+                            when (val line = receive()) {
                                 is SyncLine -> {
                                     val serializedLine = JsonUtil.json.encodeToString(line)
                                     channel.writeStringUtf8("$serializedLine\n")

demos/android-supabase-todolist/build.gradle.kts

@@ -127,7 +127,7 @@ dependencies {
     // When adopting the PowerSync dependencies into your project, use the latest version available at
     // https://central.sonatype.com/artifact/com.powersync/core
     implementation(projects.core) // "com.powersync:core:latest.release"
-    implementation(projects.connectors.supabase) // "com.powersync:connector-supabase:latest.release"
+    implementation(projects.integrations.supabase) // "com.powersync:connector-supabase:latest.release"
     implementation(projects.compose) // "com.powersync:compose:latest.release"
     implementation(libs.uuid)
     implementation(libs.kermit)

demos/supabase-todolist/androidBackgroundSync/build.gradle.kts

@@ -44,7 +44,7 @@ android {
 dependencies {
     // When copying this example, use the the current version available
     // at: https://central.sonatype.com/artifact/com.powersync/connector-supabase
-    implementation(projects.connectors.supabase) // "com.powersync:connector-supabase"
+    implementation(projects.integrations.supabase) // "com.powersync:connector-supabase"
 
     implementation(projects.demos.supabaseTodolist.shared)
 

demos/supabase-todolist/shared/build.gradle.kts

@@ -42,7 +42,7 @@ kotlin {
             // When copying this example, use the current version available
             // at: https://central.sonatype.com/artifact/com.powersync/core
             api(projects.core) // "com.powersync:core"
-            implementation(projects.connectors.supabase) // "com.powersync:connector-supabase"
+            implementation(projects.integrations.supabase) // "com.powersync:connector-supabase"
             implementation(projects.compose) // "com.powersync:compose"
             implementation(libs.uuid)
             implementation(compose.runtime)

gradle.properties

@@ -19,7 +19,7 @@ development=true
 RELEASE_SIGNING_ENABLED=true
 # Library config
 GROUP=com.powersync
-LIBRARY_VERSION=1.6.1
+LIBRARY_VERSION=1.7.0
 GITHUB_REPO=https://github.com/powersync-ja/powersync-kotlin.git
 # POM
 POM_URL=https://github.com/powersync-ja/powersync-kotlin/

integrations/room/src/commonIntegrationTest/kotlin/com/powersync/integrations/room/PowerSyncRoomTest.kt

@@ -2,7 +2,9 @@ package com.powersync.integrations.room
 
 import androidx.sqlite.driver.bundled.BundledSQLiteDriver
 import app.cash.turbine.turbineScope
+import co.touchlab.kermit.CommonWriter
 import co.touchlab.kermit.Logger
+import co.touchlab.kermit.Severity
 import co.touchlab.kermit.loggerConfigInit
 import com.powersync.PowerSyncDatabase
 import com.powersync.db.getString
@@ -28,14 +30,14 @@ class PowerSyncRoomTest {
 
     @AfterTest
     fun tearDown() {
+        logger.i { "Closing Room database" }
         database.close()
     }
 
     @Test
     fun roomWritePowerSyncRead() =
         runTest {
             database.userDao().create(User(id = "test", name = "Test user"))
-            val logger = Logger(loggerConfigInit())
 
             val powersync =
                 PowerSyncDatabase.opened(
@@ -61,7 +63,6 @@ class PowerSyncRoomTest {
     @Test
     fun roomWritePowerSyncWatch() =
         runTest {
-            val logger = Logger(loggerConfigInit())
             val pool = RoomConnectionPool(database, TestDatabase.schema)
 
             val powersync =
@@ -88,12 +89,13 @@ class PowerSyncRoomTest {
                 turbine.awaitItem() shouldHaveSize 1
                 turbine.cancel()
             }
+
+            powersync.close()
         }
 
     @Test
     fun powersyncWriteRoomRead() =
         runTest {
-            val logger = Logger(loggerConfigInit())
             val pool = RoomConnectionPool(database, TestDatabase.schema)
 
             val powersync =
@@ -108,12 +110,12 @@ class PowerSyncRoomTest {
             database.userDao().getAll() shouldHaveSize 0
             powersync.execute("insert into user values (uuid(), ?)", listOf("PowerSync user"))
             database.userDao().getAll() shouldHaveSize 1
+            powersync.close()
         }
 
     @Test
     fun powersyncWriteRoomWatch() =
         runTest {
-            val logger = Logger(loggerConfigInit())
             val pool = RoomConnectionPool(database, TestDatabase.schema)
 
             val powersync =
@@ -133,5 +135,11 @@ class PowerSyncRoomTest {
                 turbine.awaitItem() shouldHaveSize 1
                 turbine.cancel()
             }
+
+            powersync.close()
         }
+
+    companion object {
+        private val logger = Logger(loggerConfigInit(CommonWriter()))
+    }
 }

integrations/room/src/commonIntegrationTest/kotlin/com/powersync/integrations/room/TestDatabase.kt

@@ -37,7 +37,7 @@ interface UserDao {
     suspend fun delete(user: User)
 }
 
-@Database(entities = [User::class], version = 1)
+@Database(entities = [User::class], version = 1, exportSchema = false)
 @ConstructedBy(TestDatabaseConstructor::class)
 abstract class TestDatabase : RoomDatabase() {
     abstract fun userDao(): UserDao

integrations/room/src/commonMain/kotlin/com/powersync/integrations/room/RoomConnectionPool.kt

@@ -5,6 +5,7 @@ import androidx.room.Transactor
 import androidx.room.execSQL
 import androidx.room.useReaderConnection
 import androidx.room.useWriterConnection
+import androidx.sqlite.SQLiteException
 import androidx.sqlite.SQLiteStatement
 import com.powersync.db.driver.SQLiteConnectionLease
 import com.powersync.db.driver.SQLiteConnectionPool
@@ -73,7 +74,18 @@ public class RoomConnectionPool(
         db.getCoroutineScope().launch {
             val tables = schema.rawTables.map { it.name }.toTypedArray()
             db.invalidationTracker.createFlow(*tables, emitInitialState = false).collect {
-                transferPendingRoomUpdatesToPowerSync()
+                try {
+                    transferPendingRoomUpdatesToPowerSync()
+                } catch (e: SQLiteException) {
+                    // It can happen that we get an update shortly before the database is closed. Since this is
+                    // asynchronous, we'd then be using the database in a closed state, which fails. We handle that by
+                    // stopping the flow collection.
+                    if (e.message == "Connection pool is closed") {
+                        return@collect
+                    }
+
+                    throw e
+                }
             }
         }
     }
@@ -91,7 +103,7 @@ public class RoomConnectionPool(
                 val changed =
                     it.usePrepared("SELECT powersync_update_hooks('get')") { stmt ->
                         check(stmt.step())
-                        json.decodeFromString<Set<String>>(stmt.getText(0))
+                        Json.decodeFromString<Set<String>>(stmt.getText(0))
                     }
 
                 val userTables =
@@ -114,10 +126,6 @@ public class RoomConnectionPool(
     override suspend fun close() {
         // Noop, Room database managed independently
     }
-
-    private companion object {
-        val json = Json {}
-    }
 }
 
 private class RoomTransactionLease(

connectors/README.md → integrations/supabase/README.md

@@ -1,16 +1,12 @@
-# PowerSync Backend Connectors
+# PowerSync Supabase Connector
 
-Convenience implementations of backend connectors that provide the connection between your application backend and the PowerSync managed database.
+Convenience implementation of a backend connector that provide the connection between your application backend and the PowerSync managed database
+by delegating to Supabase.
 
 It is used to:
 1. Retrieve a token to connect to the PowerSync service.
 2. Apply local changes on your backend application server (and from there, to your backend database).
 
+The connector is fairly basic, and also serves as an example for getting started.
 
-## Provided Connectors
-
-### Supabase (Postgres)
-
-A basic implementation of a PowerSync Backend Connector for Supabase, that serves as getting started example.
-
-See a step-by-step tutorial for connecting to Supabase, [here](https://docs.powersync.com/integration-guides/supabase-+-powersync).
+See a step-by-step tutorial for connecting to Supabase, [here](https://docs.powersync.com/integration-guides/supabase-+-powersync).

connectors/supabase/build.gradle.kts → integrations/supabase/build.gradle.kts

@@ -7,6 +7,7 @@ plugins {
     alias(libs.plugins.android.library)
     alias(libs.plugins.kotlinter)
     id("com.powersync.plugins.sonatype")
+    id("com.powersync.plugins.sharedbuild")
     id("dokka-convention")
 }
 
@@ -22,15 +23,37 @@ kotlin {
     }
 
     explicitApi()
+    applyDefaultHierarchyTemplate()
 
     sourceSets {
         commonMain.dependencies {
-            api(project(":core"))
+            api(projects.core)
             implementation(libs.kotlinx.coroutines.core)
             implementation(libs.supabase.client)
             api(libs.supabase.auth)
             api(libs.supabase.storage)
         }
+
+        val commonIntegrationTest by creating {
+            dependsOn(commonTest.get())
+
+            dependencies {
+                implementation(libs.kotlin.test)
+                implementation(libs.kotlinx.io)
+                implementation(libs.test.turbine)
+                implementation(libs.test.coroutines)
+                implementation(libs.test.kotest.assertions)
+
+                implementation(libs.sqldelight.coroutines)
+            }
+        }
+
+        // The PowerSync SDK links the core extension, so we can just run tests as-is.
+        jvmTest.get().dependsOn(commonIntegrationTest)
+
+        // We have special setup in this build configuration to make these tests link the PowerSync extension, so they
+        // can run integration tests along with the executable for unit testing.
+        nativeTest.orNull?.dependsOn(commonIntegrationTest)
     }
 }