diff --git a/CHANGELOG.md b/CHANGELOG.md new file mode 100644 index 0000000..a927bf2 --- /dev/null +++ b/CHANGELOG.md @@ -0,0 +1,85 @@ +# Changelog + +All notable changes to this project are documented in this file. + +The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/), +and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +Until `1.0.0`, breaking changes may appear in any release and are flagged with **BREAKING** below. + +## [Unreleased] + +### Changed (BREAKING) + +- **Outbox domain table renamed `outbox` → `okapi_outbox`.** Indexes follow the rename + (`idx_outbox_*` → `idx_okapi_outbox_*`). Host applications with a pre-existing `outbox` + table are no longer affected — okapi creates its own table under the `okapi_` prefix. + The new name is fixed; it is not configurable. ([#37](https://github.com/softwaremill/okapi/issues/37)) +- **Liquibase tracking tables default to `okapi_databasechangelog` / + `okapi_databasechangeloglock`.** Previously okapi shared the application's + default `databasechangelog` / `databasechangeloglock`. Override the new defaults + via configuration to keep the shared-table layout (see Added below). + ([#37](https://github.com/softwaremill/okapi/issues/37)) + +### Added + +- `okapi.liquibase.changelog-table` — Spring Boot property that configures the + `databaseChangeLogTable` of okapi's autoconfigured `SpringLiquibase` beans + (`okapiPostgresLiquibase` / `okapiMysqlLiquibase`). Default: `okapi_databasechangelog`. +- `okapi.liquibase.changelog-lock-table` — likewise for `databaseChangeLogLockTable`. + Default: `okapi_databasechangeloglock`. + +### Migration from 0.2.x + +These are breaking changes; existing deployments must take action before the first +`0.3.0` startup. The README has the full SQL: see +[Database migrations § Upgrading from 0.2.x](README.md#upgrading-from-02x). +Two paths are documented: rename in place (recommended) or stay on the legacy +changelog table names by overriding `okapi.liquibase.changelog-table` / +`changelog-lock-table`. The domain-table rename has no opt-out — run the +provided `ALTER TABLE ... RENAME TO okapi_outbox` script. + +## [0.2.0] — 2026-04-29 + +### Added + +- Observability: `OutboxProcessorListener` API and the `okapi-micrometer` module + (counters, timers, gauges; Spring Boot Actuator integration). ([#27](https://github.com/softwaremill/okapi/pull/27)) +- Multi-datasource transaction validation in `okapi-spring-boot` + (`SpringTransactionContextValidator`, `okapi.datasource-qualifier` property). ([#17](https://github.com/softwaremill/okapi/pull/17)) +- `@JvmOverloads` / `@JvmStatic` annotations across the public API for Java interop. ([#24](https://github.com/softwaremill/okapi/pull/24)) +- Maven Central release pipeline. ([#18](https://github.com/softwaremill/okapi/pull/18)) + +### Changed + +- `OutboxStore` migrated from JetBrains Exposed to plain JDBC in + `okapi-postgres` and `okapi-mysql`. The Exposed-based path remains + available via the optional `okapi-exposed` module. ([#26](https://github.com/softwaremill/okapi/pull/26)) +- Configuration unification: `Duration` types throughout, dedicated + `OutboxPurgerConfig` and `OutboxSchedulerConfig`. ([#16](https://github.com/softwaremill/okapi/pull/16)) +- `OutboxProcessorScheduler` and `OutboxPurger` v2 — configurable interval, + batch size, retention; reliable shutdown via `SmartLifecycle`. ([#11](https://github.com/softwaremill/okapi/pull/11), [#14](https://github.com/softwaremill/okapi/pull/14)) + +### Fixed + +- Actionable error message in `ExposedConnectionProvider` when no transaction is + bound to the current thread. ([#32](https://github.com/softwaremill/okapi/pull/32)) +- `okapi-micrometer` artifact published to Maven Central; the + `okapi.metrics.refresh-interval` property documented. ([#29](https://github.com/softwaremill/okapi/pull/29)) + +## [0.1.0] — 2026-04-07 + +Initial public release. + +### Added + +- Transactional outbox pattern for Kotlin/JVM with PostgreSQL and MySQL stores. +- `okapi-http` and `okapi-kafka` deliverers; pluggable `MessageDeliverer` API. +- `OutboxProcessor` with configurable `RetryPolicy` and delivery-result + classification (`Success` / `RetriableFailure` / `PermanentFailure`). +- `okapi-spring-boot` autoconfiguration for stores, transports, scheduler, and purger. +- `okapi-exposed` integration (transaction runner, connection provider, validator). +- Concurrent processing via `FOR UPDATE SKIP LOCKED`. + +[Unreleased]: https://github.com/softwaremill/okapi/compare/v0.2.0...HEAD +[0.2.0]: https://github.com/softwaremill/okapi/compare/v0.1.0...v0.2.0 +[0.1.0]: https://github.com/softwaremill/okapi/releases/tag/v0.1.0 diff --git a/README.md b/README.md index 6cb6c3a..bb417a5 100644 --- a/README.md +++ b/README.md @@ -95,6 +95,70 @@ Okapi implements the [transactional outbox pattern](https://softwaremill.com/mic - **Concurrent processing** — multiple processors can run in parallel using `FOR UPDATE SKIP LOCKED`, so messages are never processed twice simultaneously. - **Delivery result classification** — each transport classifies errors as `Success`, `RetriableFailure`, or `PermanentFailure`. For example, HTTP 429 is retriable while HTTP 400 is permanent. +## Database migrations + +Okapi ships Liquibase changelogs that create the outbox table and its indexes: + +- `classpath:com/softwaremill/okapi/db/changelog.xml` — PostgreSQL (from `okapi-postgres`) +- `classpath:com/softwaremill/okapi/db/mysql/changelog.xml` — MySQL (from `okapi-mysql`) + +When `okapi-spring-boot` is on the classpath, these run automatically against the configured `DataSource` on application startup. Without Spring Boot, point your own Liquibase setup at the paths above and pass an `outboxTable` change-log parameter (see below). + +### Configuration + +Okapi's table names are fixed under the `okapi_` prefix so its schema stays out of the way of any pre-existing tables in the host application (`outbox`, `databasechangelog`, etc.): + +| Table | Purpose | +|-------|---------| +| `okapi_outbox` | Domain table holding outbox entries (created by the bundled Liquibase changesets, queried by `PostgresOutboxStore` / `MysqlOutboxStore`). | +| `okapi_databasechangelog` | Liquibase changeset history for okapi (configurable). | +| `okapi_databasechangeloglock` | Liquibase concurrency lock for okapi (configurable). | + +The Liquibase tracking-table names are configurable in case the host application wants to share them with its own Liquibase setup: + +| Property | Default | Description | +|----------|---------|-------------| +| `okapi.liquibase.changelog-table` | `okapi_databasechangelog` | Liquibase changeset history for okapi | +| `okapi.liquibase.changelog-lock-table` | `okapi_databasechangeloglock` | Liquibase concurrency lock for okapi | + +These properties affect the autoconfigured `okapiPostgresLiquibase` / `okapiMysqlLiquibase` beans only. If you run Liquibase yourself, configure the table names there directly. The domain table name (`okapi_outbox`) is fixed. + +### Upgrading from 0.2.x + +Releases up to 0.2.x wrote to shared tables `databasechangelog` / `databasechangeloglock` and the domain table `outbox`. From 0.3.0 these are renamed to `okapi_*`. Two upgrade paths: + +**Stay on the existing changelog tables** (simplest for the Liquibase tracking pair, zero-downtime) — opt out of the new defaults: + +```yaml +okapi: + liquibase: + changelog-table: databasechangelog + changelog-lock-table: databasechangeloglock +``` + +The domain table `outbox` cannot be opted out via configuration — see the migration steps below. + +**Migrate to dedicated tables** — run before the first 0.3.0 startup (PostgreSQL syntax shown): + +```sql +-- Outbox domain table: rename in place. Indexes follow the table. +ALTER TABLE outbox RENAME TO okapi_outbox; +ALTER INDEX idx_outbox_status_last_attempt RENAME TO idx_okapi_outbox_status_last_attempt; +ALTER INDEX idx_outbox_status_created_at RENAME TO idx_okapi_outbox_status_created_at; + +-- Liquibase tracking: split okapi rows into the new tables. +CREATE TABLE okapi_databasechangelog (LIKE databasechangelog INCLUDING ALL); +CREATE TABLE okapi_databasechangeloglock (LIKE databasechangeloglock INCLUDING ALL); +INSERT INTO okapi_databasechangelog + SELECT * FROM databasechangelog WHERE filename LIKE '%com/softwaremill/okapi/%'; +INSERT INTO okapi_databasechangeloglock SELECT * FROM databasechangeloglock; +DELETE FROM databasechangelog WHERE filename LIKE '%com/softwaremill/okapi/%'; +``` + +Without one of these steps, Liquibase will see an empty changelog table on the first 0.3.0 startup and try to re-run okapi's migrations — which fails if rows already exist under the legacy `outbox` table while okapi now writes to `okapi_outbox`. + +Full release history: [CHANGELOG.md](CHANGELOG.md). + ## Observability Add `okapi-micrometer` alongside `okapi-spring-boot` (from the Quick Start above) to get Micrometer metrics: diff --git a/gradle.properties b/gradle.properties index 20e2ceb..64f5318 100644 --- a/gradle.properties +++ b/gradle.properties @@ -7,4 +7,4 @@ org.gradle.caching=true org.gradle.configuration-cache=true GROUP=com.softwaremill.okapi -VERSION_NAME=0.2.0 +VERSION_NAME=0.3.0 diff --git a/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/support/PostgresBenchmarkSupport.kt b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/support/PostgresBenchmarkSupport.kt index 98a502a..a97e5a7 100644 --- a/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/support/PostgresBenchmarkSupport.kt +++ b/okapi-benchmarks/src/jmh/kotlin/com/softwaremill/okapi/benchmarks/support/PostgresBenchmarkSupport.kt @@ -39,7 +39,7 @@ class PostgresBenchmarkSupport { fun truncate() { jdbc.withTransaction { jdbc.withConnection { conn -> - conn.createStatement().use { it.execute("TRUNCATE TABLE outbox") } + conn.createStatement().use { it.execute("TRUNCATE TABLE okapi_outbox") } } } } diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/MysqlTestSupport.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/MysqlTestSupport.kt index 4d16282..723a012 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/MysqlTestSupport.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/MysqlTestSupport.kt @@ -32,7 +32,7 @@ class MysqlTestSupport { fun truncate() { jdbc.withTransaction { jdbc.withConnection { conn -> - conn.createStatement().use { it.execute("DELETE FROM outbox") } + conn.createStatement().use { it.execute("DELETE FROM okapi_outbox") } } } } diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/PostgresTestSupport.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/PostgresTestSupport.kt index 04d3cf6..1d9bc45 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/PostgresTestSupport.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/support/PostgresTestSupport.kt @@ -32,7 +32,7 @@ class PostgresTestSupport { fun truncate() { jdbc.withTransaction { jdbc.withConnection { conn -> - conn.createStatement().use { it.execute("TRUNCATE TABLE outbox") } + conn.createStatement().use { it.execute("TRUNCATE TABLE okapi_outbox") } } } } diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/ConnectionLeakProofTest.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/ConnectionLeakProofTest.kt index 2c76dcb..cba7c23 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/ConnectionLeakProofTest.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/ConnectionLeakProofTest.kt @@ -49,7 +49,7 @@ class ConnectionLeakProofTest : FunSpec({ beforeEach { counter.delegate.connection.use { conn -> - conn.createStatement().use { it.execute("TRUNCATE TABLE outbox") } + conn.createStatement().use { it.execute("TRUNCATE TABLE okapi_outbox") } } counter.opened.set(0) counter.closed.set(0) diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MultiDataSourceTransactionTest.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MultiDataSourceTransactionTest.kt index aa75755..0f77226 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MultiDataSourceTransactionTest.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MultiDataSourceTransactionTest.kt @@ -92,7 +92,7 @@ class MultiDataSourceTransactionTest : FunSpec({ beforeEach { outboxDataSource.connection.use { conn -> - conn.createStatement().use { it.execute("TRUNCATE TABLE outbox") } + conn.createStatement().use { it.execute("TRUNCATE TABLE okapi_outbox") } } } diff --git a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MysqlConnectionLeakProofTest.kt b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MysqlConnectionLeakProofTest.kt index bc1eb40..b91fd55 100644 --- a/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MysqlConnectionLeakProofTest.kt +++ b/okapi-integration-tests/src/test/kotlin/com/softwaremill/okapi/test/transaction/MysqlConnectionLeakProofTest.kt @@ -48,7 +48,7 @@ class MysqlConnectionLeakProofTest : FunSpec({ beforeEach { counter.delegate.connection.use { conn -> - conn.createStatement().use { it.execute("DELETE FROM outbox") } + conn.createStatement().use { it.execute("DELETE FROM okapi_outbox") } } counter.opened.set(0) counter.closed.set(0) diff --git a/okapi-mysql/src/main/kotlin/com/softwaremill/okapi/mysql/MysqlOutboxStore.kt b/okapi-mysql/src/main/kotlin/com/softwaremill/okapi/mysql/MysqlOutboxStore.kt index 14ab76c..39eeb48 100644 --- a/okapi-mysql/src/main/kotlin/com/softwaremill/okapi/mysql/MysqlOutboxStore.kt +++ b/okapi-mysql/src/main/kotlin/com/softwaremill/okapi/mysql/MysqlOutboxStore.kt @@ -19,7 +19,7 @@ class MysqlOutboxStore( override fun persist(entry: OutboxEntry): OutboxEntry { val sql = """ - INSERT INTO outbox (id, message_type, payload, delivery_type, status, created_at, updated_at, retries, last_attempt, last_error, delivery_metadata) + INSERT INTO okapi_outbox (id, message_type, payload, delivery_type, status, created_at, updated_at, retries, last_attempt, last_error, delivery_metadata) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE status = VALUES(status), @@ -61,8 +61,8 @@ class MysqlOutboxStore( // that FOR UPDATE SKIP LOCKED only row-locks the rows actually returned // by LIMIT, rather than every row matching the WHERE clause. val sql = """ - SELECT * FROM outbox - FORCE INDEX (idx_outbox_status_created_at) + SELECT * FROM okapi_outbox + FORCE INDEX (idx_okapi_outbox_status_created_at) WHERE status = ? ORDER BY created_at ASC LIMIT ? @@ -84,9 +84,9 @@ class MysqlOutboxStore( override fun removeDeliveredBefore(time: Instant, limit: Int): Int { val sql = """ - DELETE FROM outbox WHERE id IN ( + DELETE FROM okapi_outbox WHERE id IN ( SELECT id FROM ( - SELECT id FROM outbox + SELECT id FROM okapi_outbox WHERE status = ? AND last_attempt < ? ORDER BY id @@ -109,7 +109,7 @@ class MysqlOutboxStore( override fun findOldestCreatedAt(statuses: Set): Map { val result = statuses.associateWith { clock.instant() }.toMutableMap() val placeholders = statuses.joinToString(",") { "?" } - val sql = "SELECT status, MIN(created_at) AS min_created_at FROM outbox WHERE status IN ($placeholders) GROUP BY status" + val sql = "SELECT status, MIN(created_at) AS min_created_at FROM okapi_outbox WHERE status IN ($placeholders) GROUP BY status" connectionProvider.withConnection { conn -> conn.prepareStatement(sql).use { stmt -> @@ -126,7 +126,7 @@ class MysqlOutboxStore( } override fun countByStatuses(): Map { - val sql = "SELECT status, COUNT(*) AS count FROM outbox GROUP BY status" + val sql = "SELECT status, COUNT(*) AS count FROM okapi_outbox GROUP BY status" val counts = mutableMapOf() connectionProvider.withConnection { conn -> diff --git a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/001__create_outbox_table.sql b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/001__create_outbox_table.sql index 89aed15..b54bf2d 100644 --- a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/001__create_outbox_table.sql +++ b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/001__create_outbox_table.sql @@ -1,7 +1,7 @@ --liquibase formatted sql --changeset outbox:001 -CREATE TABLE IF NOT EXISTS outbox +CREATE TABLE IF NOT EXISTS okapi_outbox ( id CHAR(36) NOT NULL PRIMARY KEY, message_type VARCHAR(255) NOT NULL, diff --git a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/002__add_purger_index.sql b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/002__add_purger_index.sql index b299cb0..b53bf16 100644 --- a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/002__add_purger_index.sql +++ b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/002__add_purger_index.sql @@ -1,4 +1,4 @@ --liquibase formatted sql --changeset outbox:002 -CREATE INDEX idx_outbox_status_last_attempt ON outbox(status, last_attempt); +CREATE INDEX idx_okapi_outbox_status_last_attempt ON okapi_outbox(status, last_attempt); diff --git a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/003__add_claim_index.sql b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/003__add_claim_index.sql index 70ca899..4cffaea 100644 --- a/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/003__add_claim_index.sql +++ b/okapi-mysql/src/main/resources/com/softwaremill/okapi/db/mysql/003__add_claim_index.sql @@ -1,4 +1,4 @@ --liquibase formatted sql --changeset outbox:003 -CREATE INDEX idx_outbox_status_created_at ON outbox (status, created_at); +CREATE INDEX idx_okapi_outbox_status_created_at ON okapi_outbox (status, created_at); diff --git a/okapi-postgres/src/main/kotlin/com/softwaremill/okapi/postgres/PostgresOutboxStore.kt b/okapi-postgres/src/main/kotlin/com/softwaremill/okapi/postgres/PostgresOutboxStore.kt index 5890cd5..2c4036d 100644 --- a/okapi-postgres/src/main/kotlin/com/softwaremill/okapi/postgres/PostgresOutboxStore.kt +++ b/okapi-postgres/src/main/kotlin/com/softwaremill/okapi/postgres/PostgresOutboxStore.kt @@ -19,7 +19,7 @@ class PostgresOutboxStore( override fun persist(entry: OutboxEntry): OutboxEntry { val sql = """ - INSERT INTO outbox (id, message_type, payload, delivery_type, status, created_at, updated_at, retries, last_attempt, last_error, delivery_metadata) + INSERT INTO okapi_outbox (id, message_type, payload, delivery_type, status, created_at, updated_at, retries, last_attempt, last_error, delivery_metadata) VALUES (?::uuid, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?::jsonb) ON CONFLICT (id) DO UPDATE SET status = EXCLUDED.status, @@ -58,7 +58,7 @@ class PostgresOutboxStore( override fun claimPending(limit: Int): List { val sql = """ - SELECT * FROM outbox + SELECT * FROM okapi_outbox WHERE status = ? ORDER BY created_at ASC LIMIT ? @@ -80,8 +80,8 @@ class PostgresOutboxStore( override fun removeDeliveredBefore(time: Instant, limit: Int): Int { val sql = """ - DELETE FROM outbox WHERE id IN ( - SELECT id FROM outbox + DELETE FROM okapi_outbox WHERE id IN ( + SELECT id FROM okapi_outbox WHERE status = ? AND last_attempt < ? ORDER BY id @@ -103,7 +103,7 @@ class PostgresOutboxStore( override fun findOldestCreatedAt(statuses: Set): Map { val result = statuses.associateWith { clock.instant() }.toMutableMap() val placeholders = statuses.joinToString(",") { "?" } - val sql = "SELECT status, MIN(created_at) AS min_created_at FROM outbox WHERE status IN ($placeholders) GROUP BY status" + val sql = "SELECT status, MIN(created_at) AS min_created_at FROM okapi_outbox WHERE status IN ($placeholders) GROUP BY status" connectionProvider.withConnection { conn -> conn.prepareStatement(sql).use { stmt -> @@ -120,7 +120,7 @@ class PostgresOutboxStore( } override fun countByStatuses(): Map { - val sql = "SELECT status, COUNT(*) AS count FROM outbox GROUP BY status" + val sql = "SELECT status, COUNT(*) AS count FROM okapi_outbox GROUP BY status" val counts = mutableMapOf() connectionProvider.withConnection { conn -> diff --git a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/001__create_outbox_table.sql b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/001__create_outbox_table.sql index b26d93f..466e065 100644 --- a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/001__create_outbox_table.sql +++ b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/001__create_outbox_table.sql @@ -1,7 +1,7 @@ --liquibase formatted sql --changeset outbox:001 -CREATE TABLE IF NOT EXISTS outbox +CREATE TABLE IF NOT EXISTS okapi_outbox ( id UUID NOT NULL PRIMARY KEY, message_type VARCHAR(255) NOT NULL, diff --git a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/002__add_delivery_type_column.sql b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/002__add_delivery_type_column.sql index 94f3f43..e75c2e1 100644 --- a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/002__add_delivery_type_column.sql +++ b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/002__add_delivery_type_column.sql @@ -1,8 +1,8 @@ --liquibase formatted sql --changeset outbox:002 -ALTER TABLE outbox ADD COLUMN IF NOT EXISTS delivery_type VARCHAR(50); +ALTER TABLE okapi_outbox ADD COLUMN IF NOT EXISTS delivery_type VARCHAR(50); -UPDATE outbox SET delivery_type = delivery_metadata->>'type' WHERE delivery_type IS NULL; +UPDATE okapi_outbox SET delivery_type = delivery_metadata->>'type' WHERE delivery_type IS NULL; -ALTER TABLE outbox ALTER COLUMN delivery_type SET NOT NULL; +ALTER TABLE okapi_outbox ALTER COLUMN delivery_type SET NOT NULL; diff --git a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/003__add_purger_index.sql b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/003__add_purger_index.sql index ef3e169..2fe114a 100644 --- a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/003__add_purger_index.sql +++ b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/003__add_purger_index.sql @@ -1,4 +1,4 @@ --liquibase formatted sql --changeset outbox:003 -CREATE INDEX idx_outbox_status_last_attempt ON outbox(status, last_attempt); +CREATE INDEX idx_okapi_outbox_status_last_attempt ON okapi_outbox(status, last_attempt); diff --git a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/004__add_claim_index.sql b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/004__add_claim_index.sql index 2e4bafc..154e846 100644 --- a/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/004__add_claim_index.sql +++ b/okapi-postgres/src/main/resources/com/softwaremill/okapi/db/004__add_claim_index.sql @@ -1,4 +1,4 @@ --liquibase formatted sql --changeset outbox:004 -CREATE INDEX IF NOT EXISTS idx_outbox_status_created_at ON outbox (status, created_at); +CREATE INDEX IF NOT EXISTS idx_okapi_outbox_status_created_at ON okapi_outbox (status, created_at); diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OkapiProperties.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OkapiProperties.kt index 863a904..099130e 100644 --- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OkapiProperties.kt +++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OkapiProperties.kt @@ -7,10 +7,33 @@ import org.springframework.validation.annotation.Validated @Validated data class OkapiProperties( val datasourceQualifier: String? = null, + val liquibase: Liquibase = Liquibase(), ) { 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." } } + + /** + * Liquibase tracking-table names used by okapi's bundled migrations. + * + * Defaults to dedicated tables (`okapi_databasechangelog` / `okapi_databasechangeloglock`) + * so okapi's migration history is isolated from the host application's. Override via + * `okapi.liquibase.changelog-table` / `okapi.liquibase.changelog-lock-table` to point at + * existing tables (e.g. `databasechangelog`) when migrating from a setup that shared them. + */ + data class Liquibase( + val changelogTable: String = "okapi_databasechangelog", + val changelogLockTable: String = "okapi_databasechangeloglock", + ) { + init { + require(changelogTable.isNotBlank()) { + "okapi.liquibase.changelog-table must not be blank." + } + require(changelogLockTable.isNotBlank()) { + "okapi.liquibase.changelog-lock-table must not be blank." + } + } + } } diff --git a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt index b5a9d22..efd3e16 100644 --- a/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt +++ b/okapi-spring-boot/src/main/kotlin/com/softwaremill/okapi/springboot/OutboxAutoConfiguration.kt @@ -163,6 +163,13 @@ class OutboxAutoConfiguration( clock = clock.getIfAvailable { Clock.systemUTC() }, ) + /** + * Runs okapi's bundled PostgreSQL changelog (creates `okapi_outbox` and its indexes) + * on application startup. Tracks its history in dedicated tables to keep okapi's + * migrations isolated from the host application's. Override the tracking-table names + * via `okapi.liquibase.changelog-table` / `okapi.liquibase.changelog-lock-table` + * (see [OkapiProperties.Liquibase]). + */ @Bean("okapiPostgresLiquibase") @ConditionalOnClass(SpringLiquibase::class) @ConditionalOnBean(value = [DataSource::class, PostgresOutboxStore::class]) @@ -170,6 +177,8 @@ class OutboxAutoConfiguration( fun okapiPostgresLiquibase(): SpringLiquibase = SpringLiquibase().apply { dataSource = resolveDataSource(dataSources, primaryDataSource, okapiProperties) changeLog = "classpath:com/softwaremill/okapi/db/changelog.xml" + databaseChangeLogTable = okapiProperties.liquibase.changelogTable + databaseChangeLogLockTable = okapiProperties.liquibase.changelogLockTable } } @@ -188,6 +197,13 @@ class OutboxAutoConfiguration( clock = clock.getIfAvailable { Clock.systemUTC() }, ) + /** + * Runs okapi's bundled MySQL changelog (creates `okapi_outbox` and its indexes) + * on application startup. Tracks its history in dedicated tables to keep okapi's + * migrations isolated from the host application's. Override the tracking-table names + * via `okapi.liquibase.changelog-table` / `okapi.liquibase.changelog-lock-table` + * (see [OkapiProperties.Liquibase]). + */ @Bean("okapiMysqlLiquibase") @ConditionalOnClass(SpringLiquibase::class) @ConditionalOnBean(value = [DataSource::class, MysqlOutboxStore::class]) @@ -195,6 +211,8 @@ class OutboxAutoConfiguration( fun okapiMysqlLiquibase(): SpringLiquibase = SpringLiquibase().apply { dataSource = resolveDataSource(dataSources, primaryDataSource, okapiProperties) changeLog = "classpath:com/softwaremill/okapi/db/mysql/changelog.xml" + databaseChangeLogTable = okapiProperties.liquibase.changelogTable + databaseChangeLogLockTable = okapiProperties.liquibase.changelogLockTable } } diff --git a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseAutoConfigurationTest.kt b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseAutoConfigurationTest.kt new file mode 100644 index 0000000..d802d74 --- /dev/null +++ b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseAutoConfigurationTest.kt @@ -0,0 +1,153 @@ +package com.softwaremill.okapi.springboot + +import com.softwaremill.okapi.core.DeliveryResult +import com.softwaremill.okapi.core.MessageDeliverer +import com.softwaremill.okapi.core.OutboxEntry +import com.softwaremill.okapi.core.OutboxStatus +import com.softwaremill.okapi.core.OutboxStore +import io.kotest.assertions.throwables.shouldThrow +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.shouldBe +import org.springframework.boot.autoconfigure.AutoConfigurations +import org.springframework.boot.test.context.runner.ApplicationContextRunner +import org.springframework.jdbc.datasource.SimpleDriverDataSource +import java.time.Instant +import javax.sql.DataSource + +/** + * Verifies that okapi configures dedicated Liquibase tracking tables (issue #37) so its migration + * history stays isolated from the host application's `databasechangelog`. + * + * The bean-wiring contexts instantiate the inner @Configuration classes directly to avoid + * `afterPropertiesSet()` (which would try to run Liquibase against a fake DataSource). + * + * The standalone property-binding test pins down the YAML contract — the keys + * `okapi.liquibase.changelog-table` / `okapi.liquibase.changelog-lock-table` and their mapping + * to the nested [OkapiProperties.Liquibase] data class. Without it, a refactor renaming the + * Kotlin fields would silently break user configuration. + */ +class LiquibaseAutoConfigurationTest : FunSpec({ + + val dataSource: DataSource = SimpleDriverDataSource() + val dataSources = mapOf("primary" to dataSource) + + fun postgresConfig(props: OkapiProperties = OkapiProperties()) = + OutboxAutoConfiguration.PostgresStoreConfiguration(dataSources, dataSource, props) + + fun mysqlConfig(props: OkapiProperties = OkapiProperties()) = + OutboxAutoConfiguration.MysqlStoreConfiguration(dataSources, dataSource, props) + + val contextRunner = ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(OutboxAutoConfiguration::class.java)) + .withBean(OutboxStore::class.java, { stubStore() }) + .withBean(MessageDeliverer::class.java, { stubDeliverer() }) + .withBean(DataSource::class.java, { SimpleDriverDataSource() }) + + context("postgres liquibase") { + test("uses dedicated changelog tables by default") { + val liquibase = postgresConfig().okapiPostgresLiquibase() + + liquibase.databaseChangeLogTable shouldBe "okapi_databasechangelog" + liquibase.databaseChangeLogLockTable shouldBe "okapi_databasechangeloglock" + } + + test("honours custom changelog table names") { + val props = OkapiProperties( + liquibase = OkapiProperties.Liquibase( + changelogTable = "custom_changelog", + changelogLockTable = "custom_changelog_lock", + ), + ) + + val liquibase = postgresConfig(props).okapiPostgresLiquibase() + + liquibase.databaseChangeLogTable shouldBe "custom_changelog" + liquibase.databaseChangeLogLockTable shouldBe "custom_changelog_lock" + } + } + + context("mysql liquibase") { + test("uses dedicated changelog tables by default") { + val liquibase = mysqlConfig().okapiMysqlLiquibase() + + liquibase.databaseChangeLogTable shouldBe "okapi_databasechangelog" + liquibase.databaseChangeLogLockTable shouldBe "okapi_databasechangeloglock" + } + + test("honours custom changelog table names") { + val props = OkapiProperties( + liquibase = OkapiProperties.Liquibase( + changelogTable = "shared_changelog", + changelogLockTable = "shared_changelog_lock", + ), + ) + + val liquibase = mysqlConfig(props).okapiMysqlLiquibase() + + liquibase.databaseChangeLogTable shouldBe "shared_changelog" + liquibase.databaseChangeLogLockTable shouldBe "shared_changelog_lock" + } + } + + context("validation rejects blank table names") { + data class BlankCase( + val label: String, + val build: () -> OkapiProperties.Liquibase, + val expectedMessage: String, + ) + + val tableMsg = "okapi.liquibase.changelog-table must not be blank." + val lockMsg = "okapi.liquibase.changelog-lock-table must not be blank." + + listOf( + BlankCase("changelog-table — empty", { OkapiProperties.Liquibase(changelogTable = "") }, tableMsg), + BlankCase("changelog-table — whitespace", { OkapiProperties.Liquibase(changelogTable = " ") }, tableMsg), + BlankCase("changelog-lock-table — empty", { OkapiProperties.Liquibase(changelogLockTable = "") }, lockMsg), + BlankCase("changelog-lock-table — whitespace", { OkapiProperties.Liquibase(changelogLockTable = " ") }, lockMsg), + ).forEach { case -> + test(case.label) { + val ex = shouldThrow { case.build() } + ex.message shouldBe case.expectedMessage + } + } + } + + test("okapi.liquibase.* properties bind to nested config") { + contextRunner + .withPropertyValues( + "okapi.liquibase.changelog-table=app_changelog", + "okapi.liquibase.changelog-lock-table=app_changelog_lock", + ) + .run { ctx -> + val props = ctx.getBean(OkapiProperties::class.java) + props.liquibase.changelogTable shouldBe "app_changelog" + props.liquibase.changelogLockTable shouldBe "app_changelog_lock" + } + } + + test("blank changelog-table property triggers startup failure") { + // Pins that init { require(isNotBlank()) } actually propagates through Spring's + // Binder — without this, a future refactor of OkapiProperties.Liquibase that bypasses + // the constructor could silently let blank table names through. + contextRunner + .withPropertyValues("okapi.liquibase.changelog-table= ") + .run { ctx -> + val rootCause = generateSequence(ctx.startupFailure) { it.cause }.last() + rootCause.message shouldBe "okapi.liquibase.changelog-table must not be blank." + } + } +}) + +private fun stubStore() = object : OutboxStore { + override fun persist(entry: OutboxEntry) = entry + override fun claimPending(limit: Int) = emptyList() + override fun updateAfterProcessing(entry: OutboxEntry) = entry + override fun removeDeliveredBefore(time: Instant, limit: Int) = 0 + override fun findOldestCreatedAt(statuses: Set) = emptyMap() + override fun countByStatuses() = emptyMap() +} + +private fun stubDeliverer() = object : MessageDeliverer { + override val type = "stub" + override fun deliver(entry: OutboxEntry) = DeliveryResult.Success +} diff --git a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseE2ETest.kt b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseE2ETest.kt new file mode 100644 index 0000000..f53bdd6 --- /dev/null +++ b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/LiquibaseE2ETest.kt @@ -0,0 +1,195 @@ +package com.softwaremill.okapi.springboot + +import com.mysql.cj.jdbc.MysqlDataSource +import com.softwaremill.okapi.core.DeliveryResult +import com.softwaremill.okapi.core.MessageDeliverer +import com.softwaremill.okapi.core.OutboxEntry +import com.softwaremill.okapi.postgres.PostgresOutboxStore +import io.kotest.core.spec.style.FunSpec +import io.kotest.matchers.collections.shouldContain +import io.kotest.matchers.collections.shouldNotContain +import io.kotest.matchers.nulls.shouldBeNull +import org.postgresql.ds.PGSimpleDataSource +import org.springframework.boot.autoconfigure.AutoConfigurations +import org.springframework.boot.test.context.FilteredClassLoader +import org.springframework.boot.test.context.runner.ApplicationContextRunner +import org.testcontainers.containers.MySQLContainer +import org.testcontainers.containers.PostgreSQLContainer +import javax.sql.DataSource + +/** + * End-to-end check that [OutboxAutoConfiguration] runs okapi's Liquibase migrations against real + * databases and writes its history into the dedicated tracking tables (issue #37). + * + * Unit tests in [LiquibaseAutoConfigurationTest] verify bean wiring and YAML property binding; + * this test proves the setters actually flow through SpringLiquibase to real DDL — i.e. that + * `okapi_databasechangelog` exists after startup and the host application's `databasechangelog` + * stays untouched. Postgres and MySQL get equal coverage because they use different Liquibase + * adapters and different DDL semantics. + */ +class LiquibaseE2ETest : FunSpec({ + + val postgres = PostgreSQLContainer("postgres:16") + val mysql = MySQLContainer("mysql:8.0") + + beforeSpec { + postgres.start() + mysql.start() + } + afterSpec { + postgres.stop() + mysql.stop() + } + + context("postgres") { + fun dataSource(): DataSource = PGSimpleDataSource().apply { + setURL(postgres.jdbcUrl) + user = postgres.username + password = postgres.password + } + + fun resetSchema() { + dataSource().connection.use { conn -> + conn.createStatement().use { stmt -> + stmt.execute("DROP SCHEMA public CASCADE") + stmt.execute("CREATE SCHEMA public") + } + } + } + + fun listTables(ds: DataSource): Set = ds.connection.use { conn -> + conn.metaData.getTables(null, "public", "%", arrayOf("TABLE")).use { rs -> + buildSet { while (rs.next()) add(rs.getString("TABLE_NAME").lowercase()) } + } + } + + fun runner(ds: DataSource) = ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of(OutboxAutoConfiguration::class.java)) + .withBean(MessageDeliverer::class.java, { stubDeliverer() }) + .withBean(DataSource::class.java, { ds }) + .withPropertyValues( + "okapi.processor.enabled=false", + "okapi.purger.enabled=false", + ) + + beforeEach { resetSchema() } + + test("autoconfig creates okapi_databasechangelog and runs okapi migrations") { + val ds = dataSource() + + runner(ds).run { ctx -> + ctx.startupFailure.shouldBeNull() + + val tables = listTables(ds) + tables shouldContain "okapi_databasechangelog" + tables shouldContain "okapi_databasechangeloglock" + tables shouldContain "okapi_outbox" + tables shouldNotContain "outbox" + tables shouldNotContain "databasechangelog" + tables shouldNotContain "databasechangeloglock" + } + } + + test("custom changelog-table property creates the named table instead") { + val ds = dataSource() + + runner(ds) + .withPropertyValues( + "okapi.liquibase.changelog-table=my_outbox_changelog", + "okapi.liquibase.changelog-lock-table=my_outbox_changelog_lock", + ) + .run { ctx -> + ctx.startupFailure.shouldBeNull() + + val tables = listTables(ds) + tables shouldContain "my_outbox_changelog" + tables shouldContain "my_outbox_changelog_lock" + tables shouldContain "okapi_outbox" + tables shouldNotContain "okapi_databasechangelog" + } + } + } + + context("mysql") { + fun dataSource(): DataSource = MysqlDataSource().apply { + setURL(mysql.jdbcUrl) + user = mysql.username + setPassword(mysql.password) + } + + fun resetSchema() { + dataSource().connection.use { conn -> + conn.createStatement().use { stmt -> + stmt.execute("SET FOREIGN_KEY_CHECKS = 0") + val tables = mutableListOf() + conn.metaData.getTables(mysql.databaseName, null, "%", arrayOf("TABLE")).use { rs -> + while (rs.next()) tables.add(rs.getString("TABLE_NAME")) + } + tables.forEach { stmt.execute("DROP TABLE IF EXISTS `$it`") } + stmt.execute("SET FOREIGN_KEY_CHECKS = 1") + } + } + } + + fun listTables(ds: DataSource): Set = ds.connection.use { conn -> + conn.metaData.getTables(mysql.databaseName, null, "%", arrayOf("TABLE")).use { rs -> + buildSet { while (rs.next()) add(rs.getString("TABLE_NAME").lowercase()) } + } + } + + // Hide PostgresOutboxStore from the classpath: both `okapi-postgres` and `okapi-mysql` are + // on the test classpath, and PostgresStoreConfiguration would otherwise activate first and + // try to run Postgres-specific Liquibase changesets against this MySQL container. + fun runner(ds: DataSource) = ApplicationContextRunner() + .withClassLoader(FilteredClassLoader(PostgresOutboxStore::class.java)) + .withConfiguration(AutoConfigurations.of(OutboxAutoConfiguration::class.java)) + .withBean(MessageDeliverer::class.java, { stubDeliverer() }) + .withBean(DataSource::class.java, { ds }) + .withPropertyValues( + "okapi.processor.enabled=false", + "okapi.purger.enabled=false", + ) + + beforeEach { resetSchema() } + + test("autoconfig creates okapi_databasechangelog and runs okapi migrations") { + val ds = dataSource() + + runner(ds).run { ctx -> + ctx.startupFailure.shouldBeNull() + + val tables = listTables(ds) + tables shouldContain "okapi_databasechangelog" + tables shouldContain "okapi_databasechangeloglock" + tables shouldContain "okapi_outbox" + tables shouldNotContain "outbox" + tables shouldNotContain "databasechangelog" + tables shouldNotContain "databasechangeloglock" + } + } + + test("custom changelog-table property creates the named table instead") { + val ds = dataSource() + + runner(ds) + .withPropertyValues( + "okapi.liquibase.changelog-table=my_outbox_changelog", + "okapi.liquibase.changelog-lock-table=my_outbox_changelog_lock", + ) + .run { ctx -> + ctx.startupFailure.shouldBeNull() + + val tables = listTables(ds) + tables shouldContain "my_outbox_changelog" + tables shouldContain "my_outbox_changelog_lock" + tables shouldContain "okapi_outbox" + tables shouldNotContain "okapi_databasechangelog" + } + } + } +}) + +private fun stubDeliverer() = object : MessageDeliverer { + override val type = "stub" + override fun deliver(entry: OutboxEntry) = DeliveryResult.Success +} diff --git a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxMysqlEndToEndTest.kt b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxMysqlEndToEndTest.kt index 93b61fc..6362096 100644 --- a/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxMysqlEndToEndTest.kt +++ b/okapi-spring-boot/src/test/kotlin/com/softwaremill/okapi/springboot/OutboxMysqlEndToEndTest.kt @@ -73,7 +73,7 @@ class OutboxMysqlEndToEndTest : beforeEach { wiremock.resetAll() jdbc.withTransaction { - jdbc.withConnection { conn -> conn.createStatement().use { it.execute("DELETE FROM outbox") } } + jdbc.withConnection { conn -> conn.createStatement().use { it.execute("DELETE FROM okapi_outbox") } } } }