feat(platform): idempotency and outbox tables

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<SQLException> { 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<SQLException> { 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
+    }
 }

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 ->

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

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);

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);