diff --git a/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaMigrationIT.kt b/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaMigrationIT.kt index 9ca613a..77bcc49 100644 --- a/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaMigrationIT.kt +++ b/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaMigrationIT.kt @@ -277,4 +277,100 @@ class LedgerSchemaMigrationIT { (balance.compareTo(BigDecimal.ZERO) == 0) shouldBe true (version.compareTo(BigDecimal.ZERO) == 0) shouldBe true } + + @Test + fun `should create idempotency_keys table with primary key and expires index`() { + testDb.boolQuery("SELECT to_regclass('platform.idempotency_keys') IS NOT NULL") shouldBe true + testDb.intQuery( + "SELECT count(*) FROM pg_constraint " + + "WHERE conrelid = 'platform.idempotency_keys'::regclass AND conname = 'pk_idempotency_keys'", + ) shouldBe 1 + testDb.boolQuery("SELECT to_regclass('platform.idx_idempotency_keys_expires') IS NOT NULL") shouldBe true + } + + @Test + fun `should reject a duplicate idempotency key hash`() { + testDb.open().use { connection -> + testDb.insertIdempotencyKey(connection, keyHash = REQUEST_HASH) + shouldThrow { testDb.insertIdempotencyKey(connection, keyHash = REQUEST_HASH) } + } + } + + @Test + fun `should default created_at and leave optional columns null on minimal idempotency insert`() { + testDb.open().use { connection -> testDb.insertIdempotencyKey(connection, keyHash = MINIMAL_KEY_HASH) } + testDb.boolQuery( + "SELECT created_at IS NOT NULL AND status_code IS NULL AND response_body IS NULL " + + "FROM platform.idempotency_keys WHERE key_hash = '$MINIMAL_KEY_HASH'", + ) shouldBe true + } + + @Test + fun `should create outbox_events table with primary key status check and dispatcher indexes`() { + testDb.boolQuery("SELECT to_regclass('platform.outbox_events') IS NOT NULL") shouldBe true + testDb.intQuery( + "SELECT count(*) FROM pg_constraint " + + "WHERE conrelid = 'platform.outbox_events'::regclass " + + "AND conname IN ('pk_outbox_events','ck_outbox_events_status')", + ) shouldBe 2 + testDb.boolQuery("SELECT to_regclass('platform.idx_outbox_events_pending') IS NOT NULL") shouldBe true + testDb.boolQuery("SELECT to_regclass('platform.idx_outbox_events_aggregate') IS NOT NULL") shouldBe true + testDb.boolQuery( + "SELECT indexdef LIKE '%WHERE%' FROM pg_indexes " + + "WHERE schemaname = 'platform' AND indexname = 'idx_outbox_events_pending'", + ) shouldBe true + } + + @Test + fun `should accept every outbox status value defined by the enum`() { + testDb.open().use { connection -> + OUTBOX_STATUSES.forEach { status -> + shouldNotThrowAny { testDb.insertOutboxEvent(connection, aggregateId = "ac5-$status", status = status) } + } + } + } + + @Test + fun `should reject an unknown outbox status value`() { + testDb.open().use { connection -> + shouldThrow { testDb.insertOutboxEvent(connection, aggregateId = "ac6", status = "BOGUS") } + } + } + + @Test + fun `should store the longest outbox status without truncation`() { + testDb.open().use { connection -> + testDb.insertOutboxEvent(connection, aggregateId = "ac7", status = PERMANENTLY_FAILED_STATUS) + } + testDb.boolQuery( + "SELECT EXISTS(SELECT 1 FROM platform.outbox_events " + + "WHERE aggregate_id = 'ac7' AND status = 'PERMANENTLY_FAILED')", + ) shouldBe true + } + + @Test + fun `should generate an outbox id when none is supplied`() { + testDb.open().use { connection -> testDb.insertOutboxEventMinimal(connection, aggregateId = "ac8") } + testDb.intQuery( + "SELECT count(*) FROM platform.outbox_events WHERE aggregate_id = 'ac8' AND id IS NOT NULL", + ) shouldBe 1 + } + + @Test + fun `should default outbox status attempts and created_at on minimal insert`() { + testDb.open().use { connection -> testDb.insertOutboxEventMinimal(connection, aggregateId = "ac9") } + testDb.boolQuery( + "SELECT status = 'PENDING' AND attempts = 0 AND created_at IS NOT NULL " + + "AND last_error IS NULL AND published_at IS NULL " + + "FROM platform.outbox_events WHERE aggregate_id = 'ac9'", + ) shouldBe true + } + + @Test + fun `should not define a version column on outbox_events`() { + testDb.intQuery( + "SELECT count(*) FROM information_schema.columns " + + "WHERE table_schema = 'platform' AND table_name = 'outbox_events' AND column_name = 'version'", + ) shouldBe 0 + } } diff --git a/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaTestDb.kt b/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaTestDb.kt index 3d44f18..979abe2 100644 --- a/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaTestDb.kt +++ b/services/ledger/src/integrationTest/kotlin/com/fincore/ledger/db/LedgerSchemaTestDb.kt @@ -11,6 +11,7 @@ import liquibase.database.jvm.JdbcConnection import liquibase.resource.ClassLoaderResourceAccessor import org.testcontainers.containers.PostgreSQLContainer import java.math.BigDecimal +import java.security.MessageDigest import java.sql.Connection import java.sql.DriverManager import java.sql.PreparedStatement @@ -33,6 +34,23 @@ internal const val NEG_PRECISE = "-100.123456789012345678" internal const val CREATED_Q2 = "2026-06-01T12:00:00Z" internal const val CREATED_Q3 = "2026-08-15T12:00:00Z" +private fun sha256Hex(value: String): String = + MessageDigest + .getInstance("SHA-256") + .digest(value.toByteArray()) + .joinToString("") { byte -> "%02x".format(byte) } + +internal val KEY_HASH = sha256Hex("fincore-idempotency-key") +internal val REQUEST_HASH = sha256Hex("fincore-request-body") +internal val MINIMAL_KEY_HASH = sha256Hex("fincore-minimal-key") +internal const val EXPIRES_AT = "2026-06-02T12:00:00Z" +internal const val AGGREGATE_TYPE = "Account" +internal const val EVENT_TYPE = "AccountOpened" +internal const val EMPTY_JSON = "{}" +internal const val PENDING_STATUS = "PENDING" +internal const val PERMANENTLY_FAILED_STATUS = "PERMANENTLY_FAILED" +internal val OUTBOX_STATUSES = listOf("PENDING", "PUBLISHING", "PUBLISHED", "FAILED", "PERMANENTLY_FAILED") + @Suppress("MagicNumber") // positional JDBC parameter indices internal class LedgerSchemaTestDb( private val postgres: PostgreSQLContainer<*>, @@ -184,6 +202,60 @@ internal class LedgerSchemaTestDb( } } + fun insertIdempotencyKey( + connection: Connection, + keyHash: String = KEY_HASH, + requestHash: String = REQUEST_HASH, + expiresAt: String = EXPIRES_AT, + ) { + connection + .prepareStatement( + "INSERT INTO platform.idempotency_keys(key_hash,request_hash,expires_at) " + + "VALUES (?,?,?::timestamptz)", + ).use { statement -> + statement.setString(1, keyHash) + statement.setString(2, requestHash) + statement.setString(3, expiresAt) + statement.executeUpdate() + } + } + + fun insertOutboxEvent( + connection: Connection, + aggregateId: String, + status: String = PENDING_STATUS, + ) { + connection + .prepareStatement( + "INSERT INTO platform.outbox_events(aggregate_type,aggregate_id,event_type,payload,status) " + + "VALUES (?,?,?,?::jsonb,?)", + ).use { statement -> + statement.setString(1, AGGREGATE_TYPE) + statement.setString(2, aggregateId) + statement.setString(3, EVENT_TYPE) + statement.setString(4, EMPTY_JSON) + statement.setString(5, status) + statement.executeUpdate() + } + } + + fun insertOutboxEventMinimal( + connection: Connection, + aggregateId: String, + ) { + connection + .prepareStatement( + "INSERT INTO platform.outbox_events(aggregate_type,aggregate_id,event_type,payload) " + + "VALUES (?,?,?,?::jsonb)", + ).use { statement -> + statement.setString(1, AGGREGATE_TYPE) + statement.setString(2, aggregateId) + statement.setString(3, EVENT_TYPE) + statement.setString(4, EMPTY_JSON) + statement.executeUpdate() + } + } + fun entryCount(txId: UUID): Int = open().use { connection -> connection.prepareStatement("SELECT count(*) FROM ledger.entries WHERE transaction_id = ?").use { statement -> diff --git a/services/ledger/src/main/resources/db/changelog/db.changelog-master.yaml b/services/ledger/src/main/resources/db/changelog/db.changelog-master.yaml index 48883eb..789c4da 100644 --- a/services/ledger/src/main/resources/db/changelog/db.changelog-master.yaml +++ b/services/ledger/src/main/resources/db/changelog/db.changelog-master.yaml @@ -19,3 +19,9 @@ databaseChangeLog: - include: file: v0.1/014-account-balances.sql relativeToChangelogFile: true + - include: + file: v0.1/015-idempotency-keys.sql + relativeToChangelogFile: true + - include: + file: v0.1/016-outbox-events.sql + relativeToChangelogFile: true diff --git a/services/ledger/src/main/resources/db/changelog/v0.1/015-idempotency-keys.sql b/services/ledger/src/main/resources/db/changelog/v0.1/015-idempotency-keys.sql new file mode 100644 index 0000000..b1f7c99 --- /dev/null +++ b/services/ledger/src/main/resources/db/changelog/v0.1/015-idempotency-keys.sql @@ -0,0 +1,15 @@ +--liquibase formatted sql +-- SPDX-License-Identifier: BUSL-1.1 +-- SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +--changeset fincore:015-idempotency-keys dbms:postgresql +CREATE TABLE IF NOT EXISTS platform.idempotency_keys ( + key_hash VARCHAR(64) NOT NULL, + request_hash VARCHAR(64) NOT NULL, + status_code INT, + response_body JSONB, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + expires_at TIMESTAMPTZ NOT NULL, + CONSTRAINT pk_idempotency_keys PRIMARY KEY (key_hash) +); +CREATE INDEX IF NOT EXISTS idx_idempotency_keys_expires ON platform.idempotency_keys(expires_at); diff --git a/services/ledger/src/main/resources/db/changelog/v0.1/016-outbox-events.sql b/services/ledger/src/main/resources/db/changelog/v0.1/016-outbox-events.sql new file mode 100644 index 0000000..09abdc8 --- /dev/null +++ b/services/ledger/src/main/resources/db/changelog/v0.1/016-outbox-events.sql @@ -0,0 +1,25 @@ +--liquibase formatted sql +-- SPDX-License-Identifier: BUSL-1.1 +-- SPDX-FileCopyrightText: 2026 FinCore Engine Authors + +--changeset fincore:016-outbox-events dbms:postgresql +CREATE TABLE IF NOT EXISTS platform.outbox_events ( + id UUID NOT NULL DEFAULT gen_random_uuid(), + aggregate_type VARCHAR(64) NOT NULL, + aggregate_id VARCHAR(64) NOT NULL, + event_type VARCHAR(128) NOT NULL, + payload JSONB NOT NULL, + status VARCHAR(24) NOT NULL DEFAULT 'PENDING', + attempts INT NOT NULL DEFAULT 0, + last_error TEXT, + created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(), + published_at TIMESTAMPTZ, + CONSTRAINT pk_outbox_events PRIMARY KEY (id), + CONSTRAINT ck_outbox_events_status + CHECK (status IN ('PENDING', 'PUBLISHING', 'PUBLISHED', 'FAILED', 'PERMANENTLY_FAILED')) +); +CREATE INDEX IF NOT EXISTS idx_outbox_events_pending + ON platform.outbox_events(status, created_at) + WHERE status = 'PENDING'; +CREATE INDEX IF NOT EXISTS idx_outbox_events_aggregate + ON platform.outbox_events(aggregate_type, aggregate_id, created_at);