feat(okapi-exposed): ExposedConnectionProvider support for multiple databases

CHANGELOG.md

@@ -8,6 +8,17 @@ Until `1.0.0`, breaking changes may appear in any release and are flagged with *
 
 ## [Unreleased]
 
+### Changed (BREAKING)
+
+- **`ExposedConnectionProvider`** now requires a `database: Database` constructor argument and
+  reads the active transaction from `database.transactionManager.currentOrNull()` instead of the
+  global `TransactionManager.currentOrNull()`. Previously, in a multi-database Exposed app, the
+  provider would silently return whichever transaction happened to be innermost-active on the
+  calling thread — regardless of which `Database` it belonged to — since every
+  `OutboxStore`/`OutboxPublisher` operation routes through `withConnection`. Construct one
+  instance per `Database`, matching `ExposedTransactionRunner` and
+  `ExposedTransactionContextValidator`. ([#96](https://github.com/softwaremill/okapi/issues/96))
+
 ## [1.0.0] — 2026-07-28
 
 First stable release. The public API now follows semantic versioning — breaking changes will

okapi-exposed/src/main/kotlin/com/softwaremill/okapi/exposed/ExposedConnectionProvider.kt

@@ -1,27 +1,42 @@
 package com.softwaremill.okapi.exposed
 
 import com.softwaremill.okapi.core.ConnectionProvider
-import org.jetbrains.exposed.v1.jdbc.transactions.TransactionManager
+import org.jetbrains.exposed.v1.jdbc.Database
+import org.jetbrains.exposed.v1.jdbc.transactions.currentOrNull
+import org.jetbrains.exposed.v1.jdbc.transactions.transactionManager
 import java.sql.Connection
 
 /**
  * Exposed implementation of [ConnectionProvider].
  *
- * Reads the JDBC [Connection] from Exposed's active transaction (via `TransactionManager.currentOrNull()`) and
- * passes it to the caller's block. Exposed owns the connection's lifecycle — it commits
- * or rolls back, and returns the connection to the pool when the enclosing
- * `transaction(database) { }` block completes — so this provider performs no cleanup.
+ * Reads the JDBC [Connection] from the active Exposed transaction **on [database]** (via the
+ * per-database `database.transactionManager.currentOrNull()`) and passes it to the caller's
+ * block. Exposed owns the connection's lifecycle — it commits or rolls back, and returns the
+ * connection to the pool when the enclosing `transaction(database) { }` block completes — so
+ * this provider performs no cleanup.
  *
- * Use when your application manages transactions via Exposed (e.g. Ktor + Exposed apps).
- * Must be called from within an active Exposed transaction; otherwise [withConnection]
- * throws an [IllegalStateException] pointing the caller at the missing `transaction { }`
- * block, instead of letting Exposed's own less specific error surface.
+ * Use when your application manages transactions via Exposed (e.g. Ktor + Exposed apps). In a
+ * multi-database app, construct one instance per [Database] — same as [ExposedTransactionRunner]
+ * and [ExposedTransactionContextValidator]. Scoping to a specific database (issue #96) matters
+ * because every [com.softwaremill.okapi.core.OutboxStore] operation (`persist`, `claimPending`,
+ * `updateAfterProcessing`, ...) goes through [withConnection]: without this, a global
+ * `TransactionManager.currentOrNull()` would silently return whichever transaction happens to be
+ * innermost-active on the calling thread — potentially the *wrong* database's connection in a
+ * nested-transaction, multi-database app, with no error.
+ *
+ * Must be called from within an active Exposed transaction on [database]; otherwise
+ * [withConnection] throws an [IllegalStateException] pointing the caller at the missing
+ * `transaction(database) { }` block, instead of letting Exposed's own less specific error surface.
+ *
+ * @param database The [Database] instance where the outbox table resides.
  */
-class ExposedConnectionProvider : ConnectionProvider {
+class ExposedConnectionProvider(private val database: Database) : ConnectionProvider {
     override fun <T> withConnection(block: (Connection) -> T): T {
-        val transaction = TransactionManager.currentOrNull()
+        val transaction = database.transactionManager.currentOrNull()
             ?: throw IllegalStateException(
-                "ExposedConnectionProvider.withConnection must be called within an Exposed transaction { } block",
+                "ExposedConnectionProvider.withConnection must be called within an Exposed " +
+                    "transaction(database) { } block using this specific Database instance " +
+                    "($database) -- the one where the outbox table lives.",
             )
         return block(transaction.connection.connection as Connection)
     }

okapi-exposed/src/test/kotlin/com/softwaremill/okapi/exposed/ExposedConnectionProviderTest.kt

@@ -9,26 +9,46 @@ import org.jetbrains.exposed.v1.jdbc.transactions.transaction
 
 class ExposedConnectionProviderTest : FunSpec({
 
-    val provider = ExposedConnectionProvider()
+    val outboxDbName = "exposed_provider_outbox_test_${System.nanoTime()}"
+    val otherDbName = "exposed_provider_other_test_${System.nanoTime()}"
+    val outboxDb = Database.connect("jdbc:h2:mem:$outboxDbName;DB_CLOSE_DELAY=-1", driver = "org.h2.Driver")
+    val otherDb = Database.connect("jdbc:h2:mem:$otherDbName;DB_CLOSE_DELAY=-1", driver = "org.h2.Driver")
+    val provider = ExposedConnectionProvider(outboxDb)
 
     test("throws IllegalStateException with actionable message when called outside an Exposed transaction") {
         val ex = shouldThrow<IllegalStateException> {
             provider.withConnection { /* unreachable */ }
         }
         ex.message shouldContain "ExposedConnectionProvider.withConnection"
-        ex.message shouldContain "Exposed transaction { } block"
+        ex.message shouldContain "transaction(database) { }"
+        ex.message shouldContain "specific Database instance"
     }
 
     test("supplies the active Exposed transaction's connection to the block") {
-        val db = Database.connect(
-            "jdbc:h2:mem:exposed_provider_test_${System.nanoTime()};DB_CLOSE_DELAY=-1",
-            driver = "org.h2.Driver",
-        )
-
-        val connectionWasOpen: Boolean = transaction(db) {
+        val connectionWasOpen: Boolean = transaction(outboxDb) {
             provider.withConnection { conn -> !conn.isClosed }
         }
 
         connectionWasOpen shouldBe true
     }
+
+    test("throws when only a transaction on a DIFFERENT Database is active") {
+        val ex = shouldThrow<IllegalStateException> {
+            transaction(otherDb) {
+                provider.withConnection { /* unreachable */ }
+            }
+        }
+        ex.message shouldContain "ExposedConnectionProvider.withConnection"
+        ex.message shouldContain "specific Database instance"
+    }
+
+    test("resolves the outbox Database's connection even when a transaction on another Database is nested inside it") {
+        val connectionUrl: String = transaction(outboxDb) {
+            transaction(otherDb) {
+                provider.withConnection { conn -> conn.metaData.url }
+            }
+        }
+
+        connectionUrl shouldContain outboxDbName
+    }
 })