feat: multi-datasource transaction validation (KOJAK-63) (#17)

## Summary

- `SpringTransactionContextValidator` now checks the specific DataSource
via `TransactionSynchronizationManager.getResource(dataSource)` —
rejects publishes in transactions on the wrong database
- `ExposedTransactionContextValidator` uses
`database.transactionManager.currentOrNull()` for per-database
validation in Ktor/Exposed setups
- New `okapi.datasource-qualifier` property — one line of YAML to
configure multi-datasource (e.g. `okapi.datasource-qualifier:
storeDataSource`)
- Full autocomplete support via `spring-configuration-metadata.json`

## Motivation

In multi-datasource setups the previous validator only checked
`isActualTransactionActive()` — it accepted ANY active transaction, even
from a different DataSource. This silently broke the outbox pattern's
atomicity guarantee.

## Changes

- `SpringTransactionContextValidator(dataSource)` — adds third check:
`getResource(dataSource) != null`
- `SpringOutboxPublisher(delegate, dataSource)` — passes DataSource to
validator
- `OutboxAutoConfiguration` — resolves DataSource by
`okapi.datasource-qualifier` or falls back to primary
- `OkapiProperties` — new `@ConfigurationProperties(prefix = "okapi")`
with `datasourceQualifier`
- `ExposedTransactionContextValidator(database)` — Exposed counterpart
using per-database `currentOrNull()`
- `spring-configuration-metadata.json` — added
`okapi.datasource-qualifier` for IDE autocomplete

## Test plan

- [x] `SpringTransactionContextValidatorTest` — 5 scenarios (no tx,
correct DS, wrong DS, read-only, both DS)
- [x] `MultiDataSourceTransactionTest` — integration test with 2
Postgres Testcontainers + 2 TransactionManagers
- [x] `ExposedTransactionContextValidatorTest` — 6 scenarios with H2 (no
tx, correct DB, wrong DB, nested both ways, read-only)
- [x] `DataSourceQualifierAutoConfigurationTest` — 4 scenarios (single
DS, qualifier hit, qualifier miss, multiple DS)
- [x] All existing tests pass (backward compatible)

Author: endrju19

gradle/libs.versions.toml

@@ -15,6 +15,7 @@ vanniktechPublish = "0.36.0"
 wiremock = "3.13.2"
 slf4j = "2.0.17"
 assertj = "3.27.7"
+h2 = "2.3.232"
 
 [libraries]
 kotlinGradlePlugin = { module = "org.jetbrains.kotlin:kotlin-gradle-plugin", version.ref = "kotlin" }
@@ -26,6 +27,7 @@ exposedCore = { module = "org.jetbrains.exposed:exposed-core", version.ref = "ex
 exposedJdbc = { module = "org.jetbrains.exposed:exposed-jdbc", version.ref = "exposed" }
 exposedJson = { module = "org.jetbrains.exposed:exposed-json", version.ref = "exposed" }
 exposedJavaTime = { module = "org.jetbrains.exposed:exposed-java-time", version.ref = "exposed" }
+exposedSpringTransaction = { module = "org.jetbrains.exposed:spring7-transaction", version.ref = "exposed" }
 jacksonModuleKotlin = { module = "com.fasterxml.jackson.module:jackson-module-kotlin", version.ref = "jackson" }
 jacksonDatatypeJsr310 = { module = "com.fasterxml.jackson.datatype:jackson-datatype-jsr310", version.ref = "jackson" }
 liquibaseCore = { module = "org.liquibase:liquibase-core", version.ref = "liquibase" }
@@ -37,13 +39,15 @@ mysql = { module = "com.mysql:mysql-connector-j", version.ref = "mysql" }
 kafkaClients = { module = "org.apache.kafka:kafka-clients", version.ref = "kafkaClients" }
 springContext = { module = "org.springframework:spring-context", version.ref = "spring" }
 springTx = { module = "org.springframework:spring-tx", version.ref = "spring" }
+springJdbc = { module = "org.springframework:spring-jdbc", version.ref = "spring" }
 springBootAutoconfigure = { module = "org.springframework.boot:spring-boot-autoconfigure", version.ref = "springBoot" }
 springBootStarterValidation = { module = "org.springframework.boot:spring-boot-starter-validation", version.ref = "springBoot" }
 springBootTest = { module = "org.springframework.boot:spring-boot-test", version.ref = "springBoot" }
 assertjCore = { module = "org.assertj:assertj-core", version.ref = "assertj" }
 wiremock = { module = "org.wiremock:wiremock", version.ref = "wiremock" }
 slf4jApi = { module = "org.slf4j:slf4j-api", version.ref = "slf4j" }
 slf4jSimple = { module = "org.slf4j:slf4j-simple", version.ref = "slf4j" }
+h2 = { module = "com.h2database:h2", version.ref = "h2" }
 
 [plugins]
 ktlint = { id = "org.jlleitschuh.gradle.ktlint", version.ref = "ktlint" }

okapi-core/build.gradle.kts

@@ -7,7 +7,10 @@ description = "Core outbox abstractions and processing engine"
 
 dependencies {
     implementation(libs.slf4jApi)
+    compileOnly(libs.exposedJdbc)
     testImplementation(libs.kotestRunnerJunit5)
     testImplementation(libs.kotestAssertionsCore)
+    testImplementation(libs.exposedJdbc)
+    testImplementation(libs.h2)
     testRuntimeOnly(libs.slf4jSimple)
 }

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/ExposedTransactionContextValidator.kt

@@ -0,0 +1,31 @@
+package com.softwaremill.okapi.core
+
+import org.jetbrains.exposed.v1.jdbc.Database
+import org.jetbrains.exposed.v1.jdbc.transactions.currentOrNull
+import org.jetbrains.exposed.v1.jdbc.transactions.transactionManager
+
+/**
+ * Exposed implementation of [TransactionContextValidator].
+ *
+ * Validates that the current thread is inside an active, non-read-only
+ * Exposed transaction on the specified [database] instance (the one
+ * where the outbox table lives).
+ *
+ * Uses the per-database variant of `currentOrNull()` which searches the
+ * thread-local transaction stack filtered by [Database] instance.
+ * This correctly handles multiple databases and nested transactions.
+ *
+ * @param database The [Database] instance where the outbox table resides.
+ */
+class ExposedTransactionContextValidator(
+    private val database: Database,
+) : TransactionContextValidator {
+    override fun isInActiveReadWriteTransaction(): Boolean {
+        val transaction = database.transactionManager.currentOrNull() ?: return false
+        return !transaction.readOnly
+    }
+
+    override val failureMessage: String
+        get() = "No active read-write Exposed transaction on the outbox Database. " +
+            "Ensure publish() is called within a transaction(database) { } block using the outbox Database instance."
+}

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/TransactionContextValidator.kt

@@ -4,10 +4,14 @@ package com.softwaremill.okapi.core
  * Checks whether the current execution context is inside an active read-write transaction.
  *
  * Framework-specific modules provide implementations:
- * - `okapi-spring`: checks via `TransactionSynchronizationManager`
- * - `okapi-ktor`: checks via Exposed's `TransactionManager.currentOrNull()`
+ * - `okapi-spring`: [SpringTransactionContextValidator][com.softwaremill.okapi.springboot.SpringTransactionContextValidator]
+ *   — checks via `TransactionSynchronizationManager`
+ * - `okapi-core`: [ExposedTransactionContextValidator] — checks via Exposed's `TransactionManager.currentOrNull()`
  * - Standalone: no-op (always returns true)
  */
 interface TransactionContextValidator {
     fun isInActiveReadWriteTransaction(): Boolean
+
+    val failureMessage: String
+        get() = "No active read-write transaction. Ensure that publish() is called within a transactional context."
 }

okapi-core/src/main/kotlin/com/softwaremill/okapi/core/TransactionalOutboxPublisher.kt

@@ -14,9 +14,7 @@ class TransactionalOutboxPublisher(
     private val validator: TransactionContextValidator,
 ) {
     fun publish(outboxMessage: OutboxMessage, deliveryInfo: DeliveryInfo): OutboxId {
-        check(validator.isInActiveReadWriteTransaction()) {
-            "No active read-write transaction. Ensure that publish() is called within a transactional context."
-        }
+        check(validator.isInActiveReadWriteTransaction()) { validator.failureMessage }
         return delegate.publish(outboxMessage, deliveryInfo)
     }
 }

okapi-core/src/test/kotlin/com/softwaremill/okapi/core/ExposedTransactionContextValidatorTest.kt

@@ -0,0 +1,82 @@
+package com.softwaremill.okapi.core
+
+import io.kotest.core.spec.style.BehaviorSpec
+import io.kotest.matchers.shouldBe
+import org.jetbrains.exposed.v1.jdbc.Database
+import org.jetbrains.exposed.v1.jdbc.transactions.transaction
+
+class ExposedTransactionContextValidatorTest : BehaviorSpec({
+
+    val outboxDb = Database.connect(
+        "jdbc:h2:mem:outbox_test_${System.nanoTime()};DB_CLOSE_DELAY=-1",
+        driver = "org.h2.Driver",
+    )
+    val otherDb = Database.connect(
+        "jdbc:h2:mem:other_test_${System.nanoTime()};DB_CLOSE_DELAY=-1",
+        driver = "org.h2.Driver",
+    )
+    val validator = ExposedTransactionContextValidator(outboxDb)
+
+    given("no active Exposed transaction") {
+        then("returns false") {
+            validator.isInActiveReadWriteTransaction() shouldBe false
+        }
+    }
+
+    given("active transaction on outbox Database") {
+        then("returns true") {
+            transaction(outboxDb) {
+                validator.isInActiveReadWriteTransaction() shouldBe true
+            }
+        }
+    }
+
+    given("active transaction on OTHER Database") {
+        then("returns false") {
+            transaction(otherDb) {
+                validator.isInActiveReadWriteTransaction() shouldBe false
+            }
+        }
+    }
+
+    given("nested transaction: outboxDb outer, otherDb inner") {
+        then("returns true — outboxDb transaction is still active on the stack") {
+            transaction(outboxDb) {
+                transaction(otherDb) {
+                    validator.isInActiveReadWriteTransaction() shouldBe true
+                }
+            }
+        }
+    }
+
+    given("nested transaction: otherDb outer, outboxDb inner") {
+        then("returns true inside inner transaction") {
+            transaction(otherDb) {
+                transaction(outboxDb) {
+                    validator.isInActiveReadWriteTransaction() shouldBe true
+                }
+            }
+        }
+    }
+
+    given("nested: RW outer on outboxDb, read-only inner on outboxDb") {
+        then("returns true — Exposed currentOrNull() finds the outer RW transaction for same-db nesting") {
+            transaction(outboxDb) {
+                // Exposed does not stack same-db transactions independently;
+                // currentOrNull() finds the outer (RW) transaction, so the validator sees RW.
+                val innerResult = transaction(db = outboxDb, readOnly = true) {
+                    validator.isInActiveReadWriteTransaction()
+                }
+                innerResult shouldBe true
+            }
+        }
+    }
+
+    given("read-only transaction on outbox Database") {
+        then("returns false") {
+            transaction(db = outboxDb, readOnly = true) {
+                validator.isInActiveReadWriteTransaction() shouldBe false
+            }
+        }
+    }
+})

okapi-integration-tests/build.gradle.kts

@@ -29,6 +29,7 @@ dependencies {
     testImplementation(libs.exposedJdbc)
     testImplementation(libs.exposedJson)
     testImplementation(libs.exposedJavaTime)
+    testImplementation(libs.exposedSpringTransaction)
 
     // Liquibase (schema migrations in tests)
     testImplementation(libs.liquibaseCore)
@@ -46,4 +47,5 @@ dependencies {
     testImplementation(libs.springContext)
     testImplementation(libs.springTx)
     testImplementation(libs.springBootAutoconfigure)
+    testImplementation(libs.springJdbc)
 }

okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MultiDataSourceTransactionTest.kt

@@ -0,0 +1,147 @@
+package com.softwaremill.okapi.test.transaction
+
+import com.softwaremill.okapi.core.DeliveryInfo
+import com.softwaremill.okapi.core.OutboxMessage
+import com.softwaremill.okapi.core.OutboxPublisher
+import com.softwaremill.okapi.core.OutboxStatus
+import com.softwaremill.okapi.postgres.PostgresOutboxStore
+import com.softwaremill.okapi.springboot.SpringOutboxPublisher
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.maps.shouldContain
+import io.kotest.matchers.shouldNotBe
+import liquibase.Liquibase
+import liquibase.database.DatabaseFactory
+import liquibase.database.jvm.JdbcConnection
+import liquibase.resource.ClassLoaderResourceAccessor
+import org.jetbrains.exposed.v1.spring7.transaction.SpringTransactionManager
+import org.postgresql.ds.PGSimpleDataSource
+import org.springframework.jdbc.datasource.DataSourceTransactionManager
+import org.springframework.transaction.support.TransactionTemplate
+import org.testcontainers.containers.PostgreSQLContainer
+import java.sql.DriverManager
+import java.time.Clock
+import javax.sql.DataSource
+
+/**
+ * Integration test verifying that [SpringOutboxPublisher] correctly validates
+ * the DataSource-aware transactional context in a multi-datasource setup.
+ *
+ * Uses two separate PostgreSQL containers:
+ * - **outboxContainer**: hosts the outbox table (Liquibase migration applied)
+ * - **otherContainer**: a second database with no outbox table
+ *
+ * The outbox DataSource uses [SpringTransactionManager] (Exposed-compatible),
+ * while the other DataSource uses plain [DataSourceTransactionManager].
+ */
+class MultiDataSourceTransactionTest : FunSpec({
+
+    val outboxContainer = PostgreSQLContainer<Nothing>("postgres:16")
+    val otherContainer = PostgreSQLContainer<Nothing>("postgres:16")
+
+    lateinit var outboxDataSource: DataSource
+    lateinit var otherDataSource: DataSource
+    lateinit var outboxTxTemplate: TransactionTemplate
+    lateinit var otherTxTemplate: TransactionTemplate
+    lateinit var publisher: SpringOutboxPublisher
+    lateinit var store: PostgresOutboxStore
+
+    val clock: Clock = Clock.systemUTC()
+
+    val stubDeliveryInfo = object : DeliveryInfo {
+        override val type = "stub"
+        override fun serialize(): String = """{"type":"stub"}"""
+    }
+
+    val testMessage = OutboxMessage(messageType = "test.event", payload = """{"key":"value"}""")
+
+    beforeSpec {
+        outboxContainer.start()
+        otherContainer.start()
+
+        outboxDataSource = PGSimpleDataSource().apply {
+            setURL(outboxContainer.jdbcUrl)
+            user = outboxContainer.username
+            password = outboxContainer.password
+        }
+
+        otherDataSource = PGSimpleDataSource().apply {
+            setURL(otherContainer.jdbcUrl)
+            user = otherContainer.username
+            password = otherContainer.password
+        }
+
+        // Run Liquibase migration only on the outbox database
+        runLiquibase(outboxContainer)
+
+        // SpringTransactionManager (Exposed) for the outbox DataSource —
+        // PostgresOutboxStore uses Exposed internally, so the transaction must be Exposed-compatible
+        val outboxTxManager = SpringTransactionManager(outboxDataSource)
+        outboxTxTemplate = TransactionTemplate(outboxTxManager)
+
+        // Plain DataSourceTransactionManager for the other DataSource
+        val otherTxManager = DataSourceTransactionManager(otherDataSource)
+        otherTxTemplate = TransactionTemplate(otherTxManager)
+
+        store = PostgresOutboxStore(clock)
+        val corePublisher = OutboxPublisher(store, clock)
+        publisher = SpringOutboxPublisher(delegate = corePublisher, dataSource = outboxDataSource)
+    }
+
+    afterSpec {
+        outboxContainer.stop()
+        otherContainer.stop()
+    }
+
+    beforeEach {
+        outboxDataSource.connection.use { conn ->
+            conn.createStatement().use { it.execute("TRUNCATE TABLE outbox") }
+        }
+    }
+
+    test("publish succeeds when in transaction on outbox DataSource") {
+        val outboxId = outboxTxTemplate.execute {
+            publisher.publish(testMessage, stubDeliveryInfo)
+        }
+
+        outboxId shouldNotBe null
+
+        val counts = outboxTxTemplate.execute { store.countByStatuses() }
+        counts shouldContain (OutboxStatus.PENDING to 1L)
+    }
+
+    test("publish fails with IllegalStateException when in transaction on OTHER DataSource") {
+        shouldThrow<IllegalStateException> {
+            otherTxTemplate.execute {
+                publisher.publish(testMessage, stubDeliveryInfo)
+            }
+        }
+    }
+
+    test("publish fails with IllegalStateException when called outside any transaction") {
+        shouldThrow<IllegalStateException> {
+            publisher.publish(testMessage, stubDeliveryInfo)
+        }
+    }
+
+    test("publish succeeds with nested transaction (savepoint) on outbox DataSource") {
+        val outboxId = outboxTxTemplate.execute {
+            // Nested call creates a savepoint (PROPAGATION_REQUIRED by default nests via savepoints)
+            outboxTxTemplate.execute {
+                publisher.publish(testMessage, stubDeliveryInfo)
+            }
+        }
+
+        outboxId shouldNotBe null
+
+        val counts = outboxTxTemplate.execute { store.countByStatuses() }
+        counts shouldContain (OutboxStatus.PENDING to 1L)
+    }
+})
+
+private fun runLiquibase(container: PostgreSQLContainer<Nothing>) {
+    val connection = DriverManager.getConnection(container.jdbcUrl, container.username, container.password)
+    val db = DatabaseFactory.getInstance().findCorrectDatabaseImplementation(JdbcConnection(connection))
+    Liquibase("com/softwaremill/okapi/db/changelog.xml", ClassLoaderResourceAccessor(), db).use { it.update("") }
+    connection.close()
+}

okapi-spring-boot/build.gradle.kts

@@ -24,6 +24,7 @@ dependencies {
     testImplementation(libs.kotestAssertionsCore)
     testImplementation(libs.springContext)
     testImplementation(libs.springTx)
+    testImplementation(libs.springJdbc)
     testImplementation(libs.springBootAutoconfigure)
     testImplementation(libs.springBootTest)
     testImplementation(libs.assertjCore)

okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OkapiProperties.kt

@@ -0,0 +1,16 @@
+package com.softwaremill.okapi.springboot
+
+import org.springframework.boot.context.properties.ConfigurationProperties
+import org.springframework.validation.annotation.Validated
+
+@ConfigurationProperties(prefix = "okapi")
+@Validated
+data class OkapiProperties(
+    val datasourceQualifier: String? = null,
+) {
+    init {
+        require(datasourceQualifier == null || datasourceQualifier.isNotBlank()) {
+            "okapi.datasource-qualifier must not be blank. Set it to the bean name of the outbox DataSource, or remove the property."
+        }
+    }
+}