Supabase error handling

CHANGELOG.md

@@ -4,6 +4,7 @@
 
 - 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.
 
 ## 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

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)

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,40 @@ 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 {
+                // Separate project because SQLDelight can't generate code in test source sets.
+                implementation(projects.integrations.sqldelightTestDatabase)
+
+                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)
     }
 }
 

integrations/supabase/src/commonIntegrationTest/kotlin/com/powersync/connector/supabase/SupabaseConnectorTest.kt

@@ -0,0 +1,67 @@
+package com.powersync.connector.supabase
+
+import com.powersync.PowerSyncDatabase
+import com.powersync.db.crud.CrudEntry
+import com.powersync.db.crud.CrudTransaction
+import com.powersync.db.schema.Column
+import com.powersync.db.schema.Schema
+import com.powersync.db.schema.Table
+import io.kotest.matchers.collections.shouldHaveSize
+import io.kotest.matchers.equals.shouldBeEqual
+import io.kotest.matchers.shouldBe
+import kotlinx.coroutines.test.runTest
+import kotlin.test.Test
+
+class SupabaseConnectorTest {
+    @Test
+    fun errorHandling() =
+        runTest {
+            val db =
+                PowerSyncDatabase.inMemory(
+                    scope = this,
+                    schema =
+                        Schema(
+                            Table(
+                                "users",
+                                listOf(
+                                    Column.text("name"),
+                                ),
+                            ),
+                        ),
+                )
+
+            try {
+                db.writeTransaction { tx ->
+                    tx.execute("INSERT INTO users (id, name) VALUES (uuid(), ?)", listOf("a"))
+                    tx.execute("INSERT INTO users (id, name) VALUES (uuid(), ?)", listOf("b"))
+                    tx.execute("INSERT INTO users (id, name) VALUES (uuid(), ?)", listOf("c"))
+                }
+
+                var calledErrorHandler = false
+                val connector =
+                    object : SupabaseConnector("", "", "") {
+                        override suspend fun uploadCrudEntry(entry: CrudEntry): Unit =
+                            throw Exception("Expected exception, failing in uploadCrudEntry")
+
+                        override suspend fun handleError(
+                            tx: CrudTransaction,
+                            entry: CrudEntry,
+                            exception: Exception,
+                            errorCode: String?,
+                        ) {
+                            calledErrorHandler = true
+
+                            tx.crud shouldHaveSize 3
+                            entry shouldBeEqual tx.crud[0]
+                            exception.message shouldBe "Expected exception, failing in uploadCrudEntry"
+                            tx.complete(null)
+                        }
+                    }
+
+                connector.uploadData(db)
+                calledErrorHandler shouldBe true
+            } finally {
+                db.close()
+            }
+        }
+}

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

@@ -5,6 +5,7 @@ import com.powersync.PowerSyncDatabase
 import com.powersync.connectors.PowerSyncBackendConnector
 import com.powersync.connectors.PowerSyncCredentials
 import com.powersync.db.crud.CrudEntry
+import com.powersync.db.crud.CrudTransaction
 import com.powersync.db.crud.UpdateType
 import com.powersync.db.runWrapped
 import io.github.jan.supabase.SupabaseClient
@@ -27,20 +28,21 @@ import io.ktor.utils.io.InternalAPI
 import kotlinx.coroutines.flow.StateFlow
 import kotlinx.serialization.json.Json
 import kotlinx.serialization.json.JsonPrimitive
+import kotlin.toString
 
 /**
  * Get a Supabase token to authenticate against the PowerSync instance.
  */
 @OptIn(SupabaseInternal::class, InternalAPI::class)
-public class SupabaseConnector(
+public open class SupabaseConnector(
     public val supabaseClient: SupabaseClient,
     public val powerSyncEndpoint: String,
     private val storageBucket: String? = null,
 ) : PowerSyncBackendConnector() {
     private val json = Json { coerceInputValues = true }
     private var errorCode: String? = null
 
-    private object PostgresFatalCodes {
+    public companion object PostgresFatalCodes {
         // Using Regex patterns for Postgres error codes
         private val FATAL_RESPONSE_CODES =
             listOf(
@@ -52,7 +54,7 @@ public class SupabaseConnector(
                 "^42501$".toRegex(),
             )
 
-        fun isFatalError(code: String): Boolean =
+        public fun isFatalError(code: String): Boolean =
             FATAL_RESPONSE_CODES.any { pattern ->
                 pattern.matches(code)
             }
@@ -172,6 +174,78 @@ public class SupabaseConnector(
             )
         }
 
+    /**
+     * Uses the PostgREST APIs to upload a given [entry] to the backend database.
+     *
+     * This method should report errors during the upload as an exception that would be caught by [uploadData].
+     */
+    public open suspend fun uploadCrudEntry(entry: CrudEntry) {
+        val table = supabaseClient.from(entry.table)
+
+        when (entry.op) {
+            UpdateType.PUT -> {
+                val data =
+                    buildMap {
+                        put("id", JsonPrimitive(entry.id))
+                        entry.opData?.jsonValues?.let { putAll(it) }
+                    }
+                table.upsert(data)
+            }
+            UpdateType.PATCH -> {
+                table.update(entry.opData!!.jsonValues) {
+                    filter {
+                        eq("id", entry.id)
+                    }
+                }
+            }
+            UpdateType.DELETE -> {
+                table.delete {
+                    filter {
+                        eq("id", entry.id)
+                    }
+                }
+            }
+        }
+    }
+
+    /**
+     * Handles an error during the upload. This method can be overridden to log errors or customize error handling.
+     *
+     * By default, it throws the rest of a transaction away when the error code indicates that this is a fatal postgres
+     * error that can't be retried. Otherwise, it rethrows the exception so that the PowerSync SDK will retry.
+     *
+     * @param tx The full [CrudTransaction] we're in the process of uploading.
+     * @param entry The [CrudEntry] for which an upload has failed.
+     * @param exception The [Exception] thrown by the Supabase client.
+     * @param [errorCode] The postgres error code, if any.
+     * @throws Exception If the upload should be retried. If this method doesn't throw, it should mark [tx] as complete
+     * by invoking [CrudTransaction.complete]. In that case, the local write would be lost.
+     */
+    public open suspend fun handleError(
+        tx: CrudTransaction,
+        entry: CrudEntry,
+        exception: Exception,
+        errorCode: String?,
+    ) {
+        if (errorCode != null && isFatalError(errorCode)) {
+            /**
+             * Instead of blocking the queue with these errors,
+             * discard the (rest of the) transaction.
+             *
+             * Note that these errors typically indicate a bug in the application.
+             * If protecting against data loss is important, save the failing records
+             * elsewhere instead of discarding, and/or notify the user.
+             */
+            Logger.e("Data upload error: ${exception.message}")
+            Logger.e("Discarding entry: $entry")
+            tx.complete(null)
+            return
+        }
+
+        Logger.e("Data upload error - retrying last entry: $entry, $exception")
+        throw exception
+    }
+
     /**
      * Upload local changes to the app backend (in this case Supabase).
      *
@@ -186,54 +260,16 @@ public class SupabaseConnector(
             try {
                 for (entry in transaction.crud) {
                     lastEntry = entry
-
-                    val table = supabaseClient.from(entry.table)
-
-                    when (entry.op) {
-                        UpdateType.PUT -> {
-                            val data =
-                                buildMap {
-                                    put("id", JsonPrimitive(entry.id))
-                                    entry.opData?.jsonValues?.let { putAll(it) }
-                                }
-                            table.upsert(data)
-                        }
-                        UpdateType.PATCH -> {
-                            table.update(entry.opData!!.jsonValues) {
-                                filter {
-                                    eq("id", entry.id)
-                                }
-                            }
-                        }
-                        UpdateType.DELETE -> {
-                            table.delete {
-                                filter {
-                                    eq("id", entry.id)
-                                }
-                            }
-                        }
-                    }
+                    uploadCrudEntry(entry)
                 }
 
                 transaction.complete(null)
             } catch (e: Exception) {
-                if (errorCode != null && PostgresFatalCodes.isFatalError(errorCode.toString())) {
-                    /**
-                     * Instead of blocking the queue with these errors,
-                     * discard the (rest of the) transaction.
-                     *
-                     * Note that these errors typically indicate a bug in the application.
-                     * If protecting against data loss is important, save the failing records
-                     * elsewhere instead of discarding, and/or notify the user.
-                     */
-                    Logger.e("Data upload error: ${e.message}")
-                    Logger.e("Discarding entry: $lastEntry")
-                    transaction.complete(null)
-                    return@runWrapped
+                if (lastEntry != null) {
+                    handleError(transaction, lastEntry, e, errorCode)
+                } else {
+                    throw e
                 }
-
-                Logger.e("Data upload error - retrying last entry: $lastEntry, $e")
-                throw e
             }
         }
     }