Working on API implementations

Author: deusaquilus

controller-android/src/androidMain/kotlin/io/exoquery/controller/android/AndroidDatabaseController.kt

@@ -151,7 +151,7 @@ class AndroidDatabaseController internal constructor(
   }
 
   // Is there an open writer?
-  override fun CoroutineContext.hasOpenConnection(): Boolean {
+  override suspend fun CoroutineContext.hasOpenConnection(): Boolean {
     val session = get(sessionKey)?.session
     return session != null && session.isWriter && !isClosedSession(session)
   }
@@ -401,7 +401,7 @@ interface WithReadOnlyVerbs: RequiresSession<Connection, SupportSQLiteStatement,
     prepareSession(pool.borrowReader())
 
   // Check if there is at least a reader on th context, if it has a writer that's fine too
-  fun CoroutineContext.hasOpenReadOrWriteConnection(): Boolean {
+  suspend fun CoroutineContext.hasOpenReadOrWriteConnection(): Boolean {
     val session = get(sessionKey)?.session
     return session != null && !isClosedSession(session)
   }

controller-core/src/commonMain/kotlin/io/exoquery/controller/Controller.kt

@@ -86,12 +86,12 @@ interface RequiresSession<Session, Stmt, ExecutionOpts> {
   // Methods that implementors need to provide
   val sessionKey: CoroutineContext.Key<CoroutineSession<Session>>
   abstract suspend fun newSession(executionOptions: ExecutionOpts): Session
-  abstract fun closeSession(session: Session): Unit
-  abstract fun isClosedSession(session: Session): Boolean
+  abstract suspend fun closeSession(session: Session): Unit
+  abstract suspend fun isClosedSession(session: Session): Boolean
   suspend fun <R> accessStmt(sql: String, conn: Session, block: suspend (Stmt) -> R): R
   suspend fun <R> accessStmtReturning(sql: String, conn: Session, options: ExecutionOpts, returningColumns: List<String>, block: suspend (Stmt) -> R): R
 
-  fun CoroutineContext.hasOpenConnection(): Boolean {
+  suspend fun CoroutineContext.hasOpenConnection(): Boolean {
     val session = get(sessionKey)?.session
     return session != null && !isClosedSession(session)
   }

…n/io/exoquery/controller/JavaEncoding.kt …re/src/jvmMain/kotlin/JavaSqlEncoding.ktcontroller-jdbc/src/main/kotlin/io/exoquery/controller/JavaEncoding.kt renamed to controller-core/src/jvmMain/kotlin/JavaSqlEncoding.kt controller-jdbc/src/main/kotlin/io/exoquery/controller/JavaEncoding.kt renamed to controller-core/src/jvmMain/kotlin/JavaSqlEncoding.kt

@@ -1,9 +1,5 @@
 package io.exoquery.controller
 
-import io.exoquery.controller.SqlDecoder
-import io.exoquery.controller.SqlEncoder
-import io.exoquery.controller.SqlEncoding
-import io.exoquery.controller.TimeEncoding
 import java.time.*
 import java.math.BigDecimal
 import java.util.Date

controller-jdbc/src/main/kotlin/io/exoquery/controller/jdbc/JdbcContextMixins.kt

@@ -63,8 +63,8 @@ interface HasSessionJdbc: RequiresSession<Connection, PreparedStatement, JdbcExe
     conn
   }
 
-  override open fun closeSession(session: Connection): Unit = session.close()
-  override open fun isClosedSession(session: Connection): Boolean = session.isClosed
+  override open suspend fun closeSession(session: Connection): Unit = session.close()
+  override open suspend fun isClosedSession(session: Connection): Boolean = session.isClosed
 
   override open suspend fun <R> accessStmtReturning(sql: String, conn: Connection, options: JdbcExecutionOptions, returningColumns: List<String>, block: suspend (PreparedStatement) -> R): R {
     val stmt =

controller-r2dbc/src/main/kotlin/io/exoquery/controller/r2dbc/R2dbcController.kt

@@ -3,96 +3,131 @@ package io.exoquery.controller.r2dbc
 import io.exoquery.controller.*
 import io.r2dbc.spi.Connection
 import io.r2dbc.spi.ConnectionFactory
-import io.r2dbc.spi.Result
+import io.r2dbc.spi.Row
+import io.r2dbc.spi.Statement
 import kotlinx.coroutines.flow.Flow
-import kotlinx.coroutines.flow.collect
 import kotlinx.coroutines.flow.emptyFlow
-import kotlinx.coroutines.flow.flatMapConcat
+import kotlinx.coroutines.flow.map
 import kotlinx.coroutines.flow.toList
 import kotlinx.coroutines.reactive.asFlow
 import kotlinx.coroutines.reactive.awaitFirstOrNull
-import kotlinx.coroutines.reactive.awaitSingle
-import org.reactivestreams.Publisher
+import kotlinx.coroutines.reactive.collect
 
 // Minimal execution options placeholder for R2DBC
 data class R2dbcExecutionOptions(
-  val debug: Boolean = false
+  val sessionTimeout: Int? = null,
+  val fetchSize: Int? = null,
+  val queryTimeout: Int? = null
 ) {
   companion object {
     fun Default() = R2dbcExecutionOptions()
   }
 }
 
-class R2dbcController(private val connectionFactory: ConnectionFactory): ControllerVerbs<R2dbcExecutionOptions> {
+class R2dbcController(
+  override val encodingConfig: R2dbcEncodingConfig = R2dbcEncodingConfig.Default(),
+  override val connectionFactory: ConnectionFactory
+):
+  WithEncoding<Connection, Statement, Row>,
+  ControllerVerbs<R2dbcExecutionOptions>,
+  HasTransactionalityR2dbc
+{
   override fun DefaultOpts(): R2dbcExecutionOptions = R2dbcExecutionOptions.Default()
 
+  override val encodingApi: R2dbcSqlEncoding =
+    object: JavaSqlEncoding<Connection, Statement, Row>,
+      BasicEncoding<Connection, Statement, Row> by R2dbcBasicEncoding,
+      JavaTimeEncoding<Connection, Statement, Row> by R2dbcTimeEncoding,
+      JavaUuidEncoding<Connection, Statement, Row> by R2dbcUuidEncoding {}
+
+
   // Helper to create a connection and ensure closure
-  private suspend fun <T> withConnection(block: suspend (Connection) -> T): T {
-    val conn = connectionFactory.create().awaitSingle()
-    try {
-      return block(conn)
-    } finally {
-      conn.close().awaitFirstOrNull()
+  //private suspend fun <T> withConnection(block: suspend (Connection) -> T): T {
+  //  val conn = connectionFactory.create().awaitSingle()
+  //  try {
+  //    return block(conn)
+  //  } finally {
+  //    conn.close().awaitFirstOrNull()
+  //  }
+  //}
+
+  override fun extractColumnInfo(row: Row): List<ColumnInfo>? {
+    val meta = row.metadata
+    val cols = meta.columnMetadatas
+    return cols.map { cmd ->
+      ColumnInfo(cmd.name, cmd.type.name)
     }
   }
 
-  private fun bindParams(stmt: io.r2dbc.spi.Statement, params: List<StatementParam<*>>) {
-    var idx = 0
-    for (p in params) {
-      // Bind by index; most R2DBC drivers accept basic Kotlin/Java types directly
-      @Suppress("UNCHECKED_CAST")
-      val v: Any? = (p.value as Any?)
-      stmt.bind(idx, v)
-      idx += 1
+  override suspend fun <T> stream(act: ControllerQuery<T>, options: R2dbcExecutionOptions): Flow<T> =
+    withConnection(options) {
+      val conn = localConnection()
+      accessStmt(act.sql, conn) { stmt ->
+        prepare(stmt, conn, act.params)
+        val pub = stmt.execute() // TODO try-catch here?
+        pub.awaitFirstOrNull()?.map { row, meta ->
+          val resultMaker = act.resultMaker.makeExtractor(QueryDebugInfo(act.sql))
+          PubResult(resultMaker(conn, row))
+        }?.asFlow()?.map { it.value } ?: emptyFlow()
+      }
     }
-  }
-
-  override suspend fun <T> stream(query: ControllerQuery<T>, options: R2dbcExecutionOptions): Flow<T> {
-    // Decoding using resultMaker requires a full encoding implementation which is out of scope here.
-    // Provided for API completeness.
-    throw UnsupportedOperationException("R2dbcController.stream(query) decoding not yet implemented")
-  }
 
   override suspend fun <T> stream(query: ControllerBatchActionReturning<T>, options: R2dbcExecutionOptions): Flow<T> {
     throw UnsupportedOperationException("R2dbcController.stream(batchReturning) not yet implemented")
   }
 
-  override suspend fun <T> stream(query: ControllerActionReturning<T>, options: R2dbcExecutionOptions): Flow<T> {
-    throw UnsupportedOperationException("R2dbcController.stream(actionReturning) not yet implemented")
-  }
+  override suspend fun <T> stream(act: ControllerActionReturning<T>, options: R2dbcExecutionOptions): Flow<T> =
+    withConnection(options) {
+      val conn = localConnection()
+      accessStmt(act.sql, conn) { stmt ->
+        prepare(stmt, conn, act.params)
+        val results = mutableListOf<List<Pair<String, String?>>>()
+        val pub = stmt.execute() // TODO try-catch here?
+        // Each Result may contain rows; map them to name->string pairs for all columns
+
+        // convert the publisher into a suspeding function
+        pub.awaitFirstOrNull()?.map { row, meta ->
+          val resultMaker = act.resultMaker.makeExtractor(QueryDebugInfo(act.sql))
+          PubResult(resultMaker(conn, row))
+        }?.asFlow()?.map { it.value } ?: emptyFlow()
+      }
+    }
+
+  /** Need a temporary wrapper to work around limitation of pub-result being not-nullable */
+  @JvmInline
+  private value class PubResult<T>(val value: T)
 
   override suspend fun <T> run(query: ControllerQuery<T>, options: R2dbcExecutionOptions): List<T> = stream(query, options).toList()
 
-  override suspend fun run(query: ControllerAction, options: R2dbcExecutionOptions): Long =
-    withConnection { conn ->
-      val stmt = conn.createStatement(query.sql)
-      bindParams(stmt, query.params)
-      // Execute and sum rowsUpdated across possibly multiple results
-      val pub = stmt.execute()
-      var total = 0L
-      pub.asFlow().collect { result ->
-        val updated = result.rowsUpdated.awaitFirstOrNull() ?: 0
-        total += updated.toLong()
+  override suspend fun run(act: ControllerAction, options: R2dbcExecutionOptions): Long =
+    withConnection(options) {
+      val conn = localConnection()
+      accessStmt(act.sql, conn) { stmt ->
+        prepare(stmt, conn, act.params)
+        // Execute and sum rowsUpdated across possibly multiple results
+        val pub = stmt.execute()
+        pub.awaitFirstOrNull()?.rowsUpdated?.awaitFirstOrNull() ?: 0
       }
-      total
     }
 
   override suspend fun run(query: ControllerBatchAction, options: R2dbcExecutionOptions): List<Long> =
-    withConnection { conn ->
+    withConnection(options) {
+      val conn = localConnection()
       // TODO this statement works very well with caching, should look into reusing statements across calls
-      val stmt = conn.createStatement(query.sql)
-      // Add batches
-      query.params.forEach { batch ->
-        bindParams(stmt, batch)
-        stmt.add()
-      }
-      val results = mutableListOf<Long>()
-      val pub = stmt.execute()
-      pub.asFlow().collect { result ->
-        val updated = result.rowsUpdated.awaitFirstOrNull() ?: 0
-        results.add(updated.toLong())
+      accessStmtReturning(query.sql, conn, options, emptyList()) { stmt ->
+        query.params.forEach { batch ->
+          prepare(stmt, conn, batch)
+          stmt.add()
+        }
+        val results = mutableListOf<Long>()
+        val pub = stmt.execute()
+        // Here using the asFlow and connect actually makes sense because multiple results are expected
+        pub.asFlow().collect { result ->
+          val updated = result.rowsUpdated.awaitFirstOrNull() ?: 0
+          results.add(updated.toLong())
+        }
+        results
       }
-      results
     }
 
   override suspend fun <T> run(query: ControllerActionReturning<T>, options: R2dbcExecutionOptions): T =
@@ -101,40 +136,25 @@ class R2dbcController(private val connectionFactory: ConnectionFactory): Control
   override suspend fun <T> run(query: ControllerBatchActionReturning<T>, options: R2dbcExecutionOptions): List<T> =
     stream(query, options).toList()
 
-  override suspend fun <T> runRaw(query: ControllerQuery<T>, options: R2dbcExecutionOptions): List<List<Pair<String, String?>>> =
-    withConnection { conn ->
+  override suspend fun <T> runRaw(query: ControllerQuery<T>, options: R2dbcExecutionOptions) =
+    withConnection(options) {
+      val conn = localConnection()
       val stmt = conn.createStatement(query.sql)
-      bindParams(stmt, query.params)
+      prepare(stmt, conn, query.params)
       val results = mutableListOf<List<Pair<String, String?>>>()
       val pub = stmt.execute()
       // Each Result may contain rows; map them to name->string pairs for all columns
 
       // convert the publisher into a suspeding function
-      //pub.awaitFirstOrNull()?.map { row, meta ->
-      //  val cols = meta.columnMetadatas
-      //  val pairs = cols.mapIndexed { i, md ->
-      //    val name = md.name
-      //    val value = row.get(i, Any::class.java)
-      //    name to value?.toString()
-      //  }
-      //  pairs
-      //}?.asFlow() ?: emptyFlow()
-
-      pub.asFlow()
-        .flatMapConcat { r ->
-          r.map { row, meta ->
-            val cols = meta.columnMetadatas
-            val pairs = cols.mapIndexed { i, md ->
-              val name = md.name
-              val value = row.get(i, Any::class.java)
-              name to value?.toString()
-            }
-            pairs
-          }.asFlow()
-        }
-        .collect { rowPairs ->
-          results.add(rowPairs)
+      pub.awaitFirstOrNull()?.map { row, meta ->
+        val cols = meta.columnMetadatas
+        val pairs = cols.mapIndexed { i, md ->
+          val name = md.name
+          val value = row.get(i, Any::class.java)
+          name to value?.toString()
         }
+        pairs
+      }?.collect { rowPairs -> results.add(rowPairs) }
       results
     }
 }

controller-r2dbc/src/main/kotlin/io/exoquery/controller/r2dbc/R2dbcControllerMixins.kt

@@ -0,0 +1,84 @@
+package io.exoquery.controller.r2dbc
+
+import io.exoquery.controller.CoroutineSession
+import io.exoquery.controller.RequiresSession
+import io.exoquery.controller.RequiresTransactionality
+import io.exoquery.controller.jdbc.CoroutineTransaction
+import io.r2dbc.spi.Connection
+import io.r2dbc.spi.Row
+import io.r2dbc.spi.Statement
+import io.r2dbc.spi.ValidationDepth
+import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.reactive.awaitFirstOrNull
+import kotlinx.coroutines.reactive.awaitSingle
+import kotlinx.coroutines.withContext
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.coroutineContext
+
+object R2dbcCoroutineContext: CoroutineContext.Key<CoroutineSession<Connection>> {}
+
+interface HasTransactionalityR2dbc: RequiresTransactionality<Connection, Statement, R2dbcExecutionOptions>, HasSessionR2dbc {
+  override val sessionKey: CoroutineContext.Key<CoroutineSession<Connection>> get() = R2dbcCoroutineContext
+
+  override suspend fun <T> runTransactionally(block: suspend CoroutineScope.() -> T): T {
+    val session = coroutineContext[sessionKey]?.session ?: error("No connection found")
+    session.runWithManualCommit {
+      val transaction = CoroutineTransaction()
+      try {
+        val result = withContext(transaction) { block() }
+        commitTransaction()
+        return result
+      } catch (ex: Throwable) {
+        rollbackTransaction()
+        throw ex
+      } finally {
+        transaction.complete()
+      }
+    }
+  }
+}
+
+internal inline fun <T> Connection.runWithManualCommit(block: Connection.() -> T): T {
+  val before = this.isAutoCommit
+
+  return try {
+    this.setAutoCommit(false)
+    this.run(block)
+  } finally {
+    this.setAutoCommit(before)
+  }
+}
+
+interface HasSessionR2dbc: RequiresSession<Connection, Statement, R2dbcExecutionOptions> {
+  override val sessionKey: CoroutineContext.Key<CoroutineSession<Connection>> get() = R2dbcCoroutineContext
+  val connectionFactory: io.r2dbc.spi.ConnectionFactory
+
+  override suspend fun newSession(executionOptions: R2dbcExecutionOptions): Connection =
+    connectionFactory.create().awaitFirstOrNull() ?: error("Failed to create R2DBC connection")
+
+  override suspend fun closeSession(session: Connection) =
+    session.close().awaitFirstOrNull().run { Unit }
+
+  override suspend fun isClosedSession(session: Connection): Boolean =
+    session.validate(ValidationDepth.REMOTE).awaitFirstOrNull() == true
+
+  override suspend fun <R> accessStmt(sql: String, conn: Connection, block: suspend (Statement) -> R): R =
+    block(conn.createStatement(sql))
+
+  override suspend fun <R> accessStmtReturning(sql: String, conn: Connection, options: R2dbcExecutionOptions, returningColumns: List<String>, block: suspend (Statement) -> R): R =
+    conn.createStatement(sql).let {
+      val preparedWithColumns =
+        if (returningColumns.isNotEmpty()) {
+          it.returnGeneratedValues(*returningColumns.toTypedArray())
+        } else {
+          it
+        }
+
+      val fetchSize = options.fetchSize
+      val preparedWithOptions =
+        (fetchSize?.let { preparedWithColumns.fetchSize(it) } ?: preparedWithColumns)
+
+      block(preparedWithOptions)
+    }
+
+}