Compile :common project for the web

.gitignore

@@ -23,3 +23,4 @@ Pods/
 dialect/bin
 .build
 .vscode
+kotlin-js-store

common/build.gradle.kts

@@ -155,7 +155,8 @@ kotlin {
     powersyncTargets(
         android = {
             namespace = "com.powersync.common"
-        }
+        },
+        web = true
     )
 
     targets.withType<KotlinNativeTarget> {
@@ -195,8 +196,17 @@ kotlin {
             dependsOn(commonTest.get())
         }
 
-        val commonJava by creating {
+        val commonNonWeb by creating {
             dependsOn(commonMain.get())
+
+            dependencies {
+                implementation(libs.rsocket.core)
+                implementation(libs.rsocket.transport.websocket)
+            }
+        }
+
+        val commonJava by creating {
+            dependsOn(commonNonWeb)
         }
 
         commonMain.configure {
@@ -206,6 +216,7 @@ kotlin {
 
             dependencies {
                 api(libs.androidx.sqlite.sqlite)
+                implementation(libs.androidx.sqlite.async)
 
                 implementation(libs.uuid)
                 implementation(libs.kotlin.stdlib)
@@ -215,8 +226,6 @@ kotlin {
                 implementation(libs.kotlinx.coroutines.core)
                 implementation(libs.kotlinx.datetime)
                 implementation(libs.stately.concurrency)
-                implementation(libs.rsocket.core)
-                implementation(libs.rsocket.transport.websocket)
                 api(libs.ktor.client.core)
                 api(libs.kermit)
             }
@@ -233,6 +242,10 @@ kotlin {
             dependsOn(commonJava)
         }
 
+        nativeMain {
+            dependsOn(commonNonWeb)
+        }
+
         commonTest.dependencies {
             implementation(projects.internal.testutils)
             implementation(libs.kotlin.test)

common/src/commonMain/kotlin/com/powersync/PowerSyncDatabase.kt

@@ -10,18 +10,15 @@ import com.powersync.db.Queries
 import com.powersync.db.crud.CrudBatch
 import com.powersync.db.crud.CrudTransaction
 import com.powersync.db.driver.SQLiteConnectionPool
-import com.powersync.db.driver.SingleConnectionPool
 import com.powersync.db.schema.Schema
 import com.powersync.sync.SyncOptions
 import com.powersync.sync.SyncStatus
 import com.powersync.sync.SyncStream
 import com.powersync.utils.JsonParam
-import com.powersync.utils.generateLogger
 import kotlinx.coroutines.CoroutineScope
 import kotlinx.coroutines.flow.Flow
 import kotlinx.coroutines.flow.firstOrNull
 import kotlin.coroutines.cancellation.CancellationException
-import kotlin.native.HiddenFromObjC
 
 /**
  * A PowerSync managed database.
@@ -265,33 +262,6 @@ public interface PowerSyncDatabase : Queries {
             return openedWithGroup(pool, scope, schema, logger, group)
         }
 
-        /**
-         * Creates a PowerSync database backed by a single in-memory database connection opened from the
-         * [InMemoryConnectionFactory].
-         *
-         * This can be useful for writing tests relying on PowerSync databases.
-         */
-        public fun openInMemory(
-            factory: InMemoryConnectionFactory,
-            schema: Schema,
-            scope: CoroutineScope,
-            logger: Logger? = null,
-        ): PowerSyncDatabase {
-            val logger = generateLogger(logger)
-            // Since this returns a fresh in-memory database every time, use a fresh group to avoid warnings about the
-            // same database being opened multiple times.
-            val collection =
-                ActiveDatabaseGroup.GroupsCollection().referenceDatabase(logger, "test")
-
-            return openedWithGroup(
-                SingleConnectionPool(factory.openInMemoryConnection()),
-                scope,
-                schema,
-                logger,
-                collection,
-            )
-        }
-
         internal fun openedWithGroup(
             pool: SQLiteConnectionPool,
             scope: CoroutineScope,

common/src/commonMain/kotlin/com/powersync/bucket/BucketStorage.kt

@@ -18,11 +18,11 @@ internal interface BucketStorage {
 
     suspend fun nextCrudItem(): CrudEntry?
 
-    fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry?
+    suspend fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry?
 
     suspend fun hasCrud(): Boolean
 
-    fun hasCrud(transaction: PowerSyncTransaction): Boolean
+    suspend fun hasCrud(transaction: PowerSyncTransaction): Boolean
 
     fun mapCrudEntry(row: SqlCursor): CrudEntry
 

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

@@ -33,8 +33,8 @@ internal class BucketStorageImpl(
 
     override suspend fun nextCrudItem(): CrudEntry? = db.getOptional(sql = nextCrudQuery, mapper = ::mapCrudEntry)
 
-    override fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry? =
-        transaction.getOptional(sql = nextCrudQuery, mapper = ::mapCrudEntry)
+    override suspend fun nextCrudItem(transaction: PowerSyncTransaction): CrudEntry? =
+        transaction.getOptionalAsync(sql = nextCrudQuery, mapper = ::mapCrudEntry)
 
     private val nextCrudQuery = "SELECT id, tx_id, data FROM ${InternalTable.CRUD} ORDER BY id ASC LIMIT 1"
 
@@ -52,8 +52,8 @@ internal class BucketStorageImpl(
         return res == 1L
     }
 
-    override fun hasCrud(transaction: PowerSyncTransaction): Boolean {
-        val res = transaction.getOptional(sql = hasCrudQuery, mapper = hasCrudMapper)
+    override suspend fun hasCrud(transaction: PowerSyncTransaction): Boolean {
+        val res = transaction.getOptionalAsync(sql = hasCrudQuery, mapper = hasCrudMapper)
         return res == 1L
     }
 
@@ -81,14 +81,14 @@ internal class BucketStorageImpl(
 
         logger.i { "[updateLocalTarget] Updating target to checkpoint $opId" }
 
-        return db.writeTransaction { tx ->
+        return db.writeTransactionAsync { tx ->
             if (hasCrud(tx)) {
                 logger.w { "[updateLocalTarget] ps crud is not empty" }
-                return@writeTransaction false
+                return@writeTransactionAsync false
             }
 
             val seqAfter =
-                tx.getOptional("SELECT seq FROM main.sqlite_sequence WHERE name = '${InternalTable.CRUD}'") {
+                tx.getOptionalAsync("SELECT seq FROM main.sqlite_sequence WHERE name = '${InternalTable.CRUD}'") {
                     it.getLong(0)!!
                 }
                     ?: // assert isNotEmpty
@@ -97,15 +97,15 @@ internal class BucketStorageImpl(
             if (seqAfter != seqBefore) {
                 logger.d("seqAfter != seqBefore seqAfter: $seqAfter seqBefore: $seqBefore")
                 // New crud data may have been uploaded since we got the checkpoint. Abort.
-                return@writeTransaction false
+                return@writeTransactionAsync false
             }
 
-            tx.execute(
+            tx.executeAsync(
                 "UPDATE ${InternalTable.BUCKETS} SET target_op = CAST(? as INTEGER) WHERE name='\$local'",
                 listOf(opId),
             )
 
-            return@writeTransaction true
+            return@writeTransactionAsync true
         }
     }
 
@@ -138,10 +138,10 @@ internal class BucketStorageImpl(
     }
 
     override suspend fun control(args: PowerSyncControlArguments): List<Instruction> =
-        db.writeTransaction { tx ->
+        db.writeTransactionAsync { tx ->
             logger.v { "powersync_control: $args" }
 
             val (op: String, data: Any?) = args.sqlArguments
-            tx.get("SELECT powersync_control(?, ?) AS r", listOf(op, data), ::handleControlResult)
+            tx.getAsync("SELECT powersync_control(?, ?) AS r", listOf(op, data), ::handleControlResult)
         }
 }

common/src/commonMain/kotlin/com/powersync/db/PowerSyncDatabaseImpl.kt

@@ -14,8 +14,10 @@ import com.powersync.db.crud.CrudRow
 import com.powersync.db.crud.CrudTransaction
 import com.powersync.db.driver.SQLiteConnectionLease
 import com.powersync.db.driver.SQLiteConnectionPool
+import com.powersync.db.internal.ConnectionContext
 import com.powersync.db.internal.InternalDatabaseImpl
 import com.powersync.db.internal.InternalTable
+import com.powersync.db.internal.PowerSyncTransaction
 import com.powersync.db.internal.PowerSyncVersion
 import com.powersync.db.schema.Schema
 import com.powersync.sync.CoreSyncStatus
@@ -111,8 +113,8 @@ internal class PowerSyncDatabaseImpl(
         checkVersion(powerSyncVersion)
         logger.d { "PowerSyncVersion: $powerSyncVersion" }
 
-        internalDb.writeTransaction { tx ->
-            tx.getOptional("SELECT powersync_init()") {}
+        internalDb.writeTransactionAsync { tx ->
+            tx.getOptionalAsync("SELECT powersync_init()") {}
         }
 
         updateSchemaInternal(schema)
@@ -393,26 +395,6 @@ internal class PowerSyncDatabaseImpl(
             emitAll(internalDb.watch(sql, parameters, throttleMs, mapper))
         }
 
-    override suspend fun <R> readLock(callback: ThrowableLockCallback<R>): R {
-        waitReady()
-        return internalDb.readLock(callback)
-    }
-
-    override suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R {
-        waitReady()
-        return internalDb.writeTransaction(callback)
-    }
-
-    override suspend fun <R> writeLock(callback: ThrowableLockCallback<R>): R {
-        waitReady()
-        return internalDb.writeLock(callback)
-    }
-
-    override suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R {
-        waitReady()
-        return internalDb.writeTransaction(callback)
-    }
-
     override suspend fun execute(
         sql: String,
         parameters: List<Any?>?,
@@ -425,19 +407,19 @@ internal class PowerSyncDatabaseImpl(
         lastTransactionId: Int,
         writeCheckpoint: String?,
     ) {
-        internalDb.writeTransaction { transaction ->
-            transaction.execute(
+        internalDb.writeTransactionAsync { transaction ->
+            transaction.executeAsync(
                 "DELETE FROM ps_crud WHERE id <= ?",
                 listOf(lastTransactionId.toLong()),
             )
 
             if (writeCheckpoint != null && !bucketStorage.hasCrud(transaction)) {
-                transaction.execute(
+                transaction.executeAsync(
                     "UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name='\$local'",
                     listOf(writeCheckpoint),
                 )
             } else {
-                transaction.execute(
+                transaction.executeAsync(
                     "UPDATE ps_buckets SET target_op = CAST(? AS INTEGER) WHERE name='\$local'",
                     listOf(bucketStorage.getMaxOpId()),
                 )
@@ -483,8 +465,8 @@ internal class PowerSyncDatabaseImpl(
             flags = flags or 2 // MASK_SOFT_CLEAR
         }
 
-        internalDb.writeTransaction { tx ->
-            tx.getOptional("SELECT powersync_clear(?)", listOf(flags)) {}
+        internalDb.writeTransactionAsync { tx ->
+            tx.getOptionalAsync("SELECT powersync_clear(?)", listOf(flags)) {}
         }
         currentStatus.update { copy(lastSyncedAt = null, hasSynced = false) }
     }

common/src/commonMain/kotlin/com/powersync/db/Queries.kt

@@ -5,6 +5,8 @@ import com.powersync.PowerSyncException
 import com.powersync.db.driver.SQLiteConnectionLease
 import com.powersync.db.internal.ConnectionContext
 import com.powersync.db.internal.PowerSyncTransaction
+import com.powersync.db.internal.asContext
+import com.powersync.db.internal.runTransaction
 import kotlinx.coroutines.flow.Flow
 import kotlin.coroutines.cancellation.CancellationException
 import kotlin.native.HiddenFromObjC
@@ -42,7 +44,7 @@ public interface Queries {
     public suspend fun execute(
         sql: String,
         parameters: List<Any?>? = listOf(),
-    ): Long
+    ): Long = writeLockAsync { connection -> connection.executeAsync(sql, parameters) }
 
     /**
      * Executes a read-only (SELECT) query and returns a single result.
@@ -59,7 +61,7 @@ public interface Queries {
         sql: String,
         parameters: List<Any?>? = listOf(),
         mapper: (SqlCursor) -> RowType,
-    ): RowType
+    ): RowType = readLockAsync { connection -> connection.getAsync(sql, parameters, mapper) }
 
     /**
      * Executes a read-only (SELECT) query and returns all results.
@@ -76,7 +78,7 @@ public interface Queries {
         sql: String,
         parameters: List<Any?>? = listOf(),
         mapper: (SqlCursor) -> RowType,
-    ): List<RowType>
+    ): List<RowType> = readLockAsync { connection -> connection.getAllAsync(sql, parameters, mapper) }
 
     /**
      * Executes a read-only (SELECT) query and returns a single optional result.
@@ -93,7 +95,7 @@ public interface Queries {
         sql: String,
         parameters: List<Any?>? = listOf(),
         mapper: (SqlCursor) -> RowType,
-    ): RowType?
+    ): RowType? = readLockAsync { connection -> connection.getOptionalAsync(sql, parameters, mapper) }
 
     /**
      * Returns a [Flow] that emits whenever the source tables are modified.
@@ -144,7 +146,15 @@ public interface Queries {
      * @throws CancellationException If the operation is cancelled.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public suspend fun <R> writeLock(callback: ThrowableLockCallback<R>): R
+    public suspend fun <R> writeLock(callback: ThrowableLockCallback<R>): R =
+        useConnection(readOnly = false) { connection ->
+            callback.execute(connection.asContext())
+        }
+
+    public suspend fun <R> writeLockAsync(callback: suspend (tx: ConnectionContext) -> R): R =
+        useConnection(readOnly = false) { connection ->
+            callback(connection.asContext())
+        }
 
     /**
      * Opens a read-write transaction.
@@ -159,7 +169,15 @@ public interface Queries {
      * @throws CancellationException If the operation is cancelled.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R
+    public suspend fun <R> writeTransaction(callback: ThrowableTransactionCallback<R>): R =
+        useConnection(readOnly = false) { connection ->
+            connection.runTransaction { callback.execute(it) }
+        }
+
+    public suspend fun <R> writeTransactionAsync(callback: suspend (tx: PowerSyncTransaction) -> R): R =
+        useConnection(readOnly = false) { connection ->
+            connection.runTransaction { callback(it) }
+        }
 
     /**
      * Takes a read lock without starting a transaction.
@@ -172,7 +190,15 @@ public interface Queries {
      * @throws CancellationException If the operation is cancelled.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public suspend fun <R> readLock(callback: ThrowableLockCallback<R>): R
+    public suspend fun <R> readLock(callback: ThrowableLockCallback<R>): R =
+        useConnection(readOnly = true) { connection ->
+            callback.execute(connection.asContext())
+        }
+
+    public suspend fun <R> readLockAsync(callback: suspend (tx: ConnectionContext) -> R): R =
+        useConnection(readOnly = true) { connection ->
+            callback(connection.asContext())
+        }
 
     /**
      * Opens a read-only transaction.
@@ -185,7 +211,15 @@ public interface Queries {
      * @throws CancellationException If the operation is cancelled.
      */
     @Throws(PowerSyncException::class, CancellationException::class)
-    public suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R
+    public suspend fun <R> readTransaction(callback: ThrowableTransactionCallback<R>): R =
+        useConnection(readOnly = true) { connection ->
+            connection.runTransaction { callback.execute(it) }
+        }
+
+    public suspend fun <R> readTransactionAsync(callback: suspend (tx: PowerSyncTransaction) -> R): R =
+        useConnection(readOnly = true) { connection ->
+            connection.runTransaction { callback(it) }
+        }
 
     /**
      * Obtains a connection from the read pool or an exclusive reference on the write connection.