feat(eventbus): add idempotent consumer dedup primitive

gradle/libs.versions.toml

@@ -38,6 +38,7 @@ spring-security-test = { module = "org.springframework.security:spring-security-
 
 # Spring Kafka (version managed by the Spring Boot BOM)
 spring-kafka = { module = "org.springframework.kafka:spring-kafka" }
+spring-jdbc = { module = "org.springframework:spring-jdbc" }
 
 # Hibernate
 hibernate-core = { module = "org.hibernate.orm:hibernate-core", version.ref = "hibernate" }

libs/fincore-eventbus/build.gradle.kts

@@ -27,6 +27,8 @@ dependencies {
     // spring-kafka is exposed: the auto-configuration's @Bean methods return KafkaTemplate/KafkaAdmin,
     // so a consuming service (the ledger dispatcher) compiles against these types.
     api(libs.spring.kafka)
+    // spring-jdbc backs JdbcProcessedEventStore (consumer dedup over a processed_events table).
+    implementation(libs.spring.jdbc)
 
     testImplementation(platform(springBootBom))
     testImplementation(libs.spring.boot.starter.test)
@@ -52,7 +54,9 @@ dependencies {
     "integrationTestImplementation"(platform(springBootBom))
     "integrationTestImplementation"(libs.testcontainers.core)
     "integrationTestImplementation"(libs.testcontainers.redpanda)
+    "integrationTestImplementation"(libs.testcontainers.postgres)
     "integrationTestImplementation"(libs.testcontainers.junit5)
+    "integrationTestImplementation"(libs.postgres.jdbc)
 }
 
 val integrationTest =

libs/fincore-eventbus/src/integrationTest/kotlin/com/fincore/eventbus/consumer/JdbcProcessedEventStoreIT.kt

@@ -0,0 +1,104 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import io.kotest.matchers.shouldBe
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.postgresql.ds.PGSimpleDataSource
+import org.springframework.core.io.ClassPathResource
+import org.springframework.jdbc.core.JdbcTemplate
+import org.springframework.jdbc.datasource.DataSourceTransactionManager
+import org.springframework.transaction.support.TransactionTemplate
+import org.testcontainers.containers.PostgreSQLContainer
+import org.testcontainers.junit.jupiter.Container
+import org.testcontainers.junit.jupiter.Testcontainers
+import java.util.UUID
+import java.util.concurrent.CopyOnWriteArrayList
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+@Testcontainers
+class JdbcProcessedEventStoreIT {
+    private val jdbcTemplate = JdbcTemplate(dataSource)
+    private val transactionTemplate = TransactionTemplate(DataSourceTransactionManager(dataSource))
+    private val store = JdbcProcessedEventStore(jdbcTemplate)
+
+    @BeforeEach
+    fun resetTable() {
+        val ddl = ClassPathResource("db/processed-events.sql").inputStream.bufferedReader().use { it.readText() }
+        jdbcTemplate.execute(ddl)
+        jdbcTemplate.execute("TRUNCATE processed_events")
+    }
+
+    @Test
+    fun `should return first-seen then duplicate and persist a single row`() {
+        val id = UUID.randomUUID()
+
+        store.markIfFirstSeen(id, "group") shouldBe true
+        store.markIfFirstSeen(id, "group") shouldBe false
+        rowCount(id) shouldBe 1
+    }
+
+    @Test
+    fun `should undo the claim when the surrounding transaction rolls back`() {
+        val id = UUID.randomUUID()
+
+        transactionTemplate.executeWithoutResult { status ->
+            store.markIfFirstSeen(id, "group")
+            status.setRollbackOnly()
+        }
+
+        rowCount(id) shouldBe 0
+        store.markIfFirstSeen(id, "group") shouldBe true
+    }
+
+    @Test
+    fun `should grant first-seen to exactly one of two concurrent transactions`() {
+        val id = UUID.randomUUID()
+        val results = CopyOnWriteArrayList<Boolean>()
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(2)
+        val pool = Executors.newFixedThreadPool(2)
+        repeat(2) {
+            pool.submit {
+                start.await()
+                transactionTemplate.executeWithoutResult { results.add(store.markIfFirstSeen(id, "group")) }
+                done.countDown()
+            }
+        }
+
+        start.countDown()
+        done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+        pool.shutdown()
+
+        results.size shouldBe 2
+        results.count { it } shouldBe 1
+        rowCount(id) shouldBe 1
+    }
+
+    private fun rowCount(id: UUID): Int =
+        jdbcTemplate.queryForObject(
+            "SELECT count(*) FROM processed_events WHERE envelope_id = ?",
+            Int::class.java,
+            id,
+        ) ?: 0
+
+    companion object {
+        private const val TIMEOUT_SECONDS = 10L
+
+        @Container
+        @JvmStatic
+        val postgres: PostgreSQLContainer<*> = PostgreSQLContainer("postgres:17-alpine")
+
+        private val dataSource: PGSimpleDataSource by lazy {
+            PGSimpleDataSource().apply {
+                setUrl(postgres.jdbcUrl)
+                user = postgres.username
+                password = postgres.password
+            }
+        }
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/consumer/IdempotentEventProcessor.kt

@@ -0,0 +1,27 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import java.util.UUID
+
+/**
+ * Runs a handler exactly once per (envelope id, consumer group). Claim-then-handle: the dedup claim
+ * is taken first, then the handler runs. Call this inside the consumer's transaction so a thrown
+ * handler rolls back the claim and the event is retried (at-least-once, exactly-once effect).
+ */
+class IdempotentEventProcessor(
+    private val store: ProcessedEventStore,
+) {
+    fun process(
+        envelopeId: UUID,
+        consumerGroup: String,
+        handler: () -> Unit,
+    ): EventProcessingOutcome =
+        if (store.markIfFirstSeen(envelopeId, consumerGroup)) {
+            handler()
+            EventProcessingOutcome.PROCESSED
+        } else {
+            EventProcessingOutcome.DUPLICATE_SKIPPED
+        }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/consumer/InMemoryProcessedEventStore.kt

@@ -0,0 +1,21 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import java.util.UUID
+import java.util.concurrent.ConcurrentHashMap
+
+/**
+ * Non-persistent dedup store for tests and local development. State is lost on restart, so it gives
+ * no exactly-once effect across restarts - a persistent store (see [JdbcProcessedEventStore]) is
+ * required in production.
+ */
+class InMemoryProcessedEventStore : ProcessedEventStore {
+    private val seen: MutableSet<Pair<UUID, String>> = ConcurrentHashMap.newKeySet()
+
+    override fun markIfFirstSeen(
+        envelopeId: UUID,
+        consumerGroup: String,
+    ): Boolean = seen.add(envelopeId to consumerGroup)
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/consumer/JdbcProcessedEventStore.kt

@@ -0,0 +1,39 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import org.springframework.jdbc.core.JdbcTemplate
+import java.sql.Timestamp
+import java.time.Instant
+import java.util.UUID
+
+/**
+ * Persistent dedup store over a `processed_events` table (see db/processed-events.sql). The claim is
+ * an INSERT ... ON CONFLICT DO NOTHING that participates in the ambient transaction, so a failing
+ * handler rolls it back and the event is retried.
+ */
+class JdbcProcessedEventStore(
+    private val jdbcTemplate: JdbcTemplate,
+    private val tableName: String = "processed_events",
+) : ProcessedEventStore {
+    init {
+        require(tableName.matches(SAFE_IDENTIFIER)) { "tableName must be a simple SQL identifier" }
+    }
+
+    override fun markIfFirstSeen(
+        envelopeId: UUID,
+        consumerGroup: String,
+    ): Boolean =
+        jdbcTemplate.update(
+            "INSERT INTO $tableName (envelope_id, consumer_group, processed_at) VALUES (?, ?, ?) " +
+                "ON CONFLICT (envelope_id, consumer_group) DO NOTHING",
+            envelopeId,
+            consumerGroup,
+            Timestamp.from(Instant.now()),
+        ) == 1
+
+    private companion object {
+        val SAFE_IDENTIFIER = Regex("^[A-Za-z_][A-Za-z0-9_]*$")
+    }
+}

libs/fincore-eventbus/src/main/kotlin/com/fincore/eventbus/consumer/ProcessedEventStore.kt

@@ -0,0 +1,26 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import java.util.UUID
+
+/**
+ * Dedup store for at-least-once consumers. Keyed by (envelope id, consumer group) so independent
+ * consumers each process an event exactly once.
+ */
+interface ProcessedEventStore {
+    /**
+     * Atomically claims an envelope for a consumer group. Returns true the first time the pair is
+     * seen (the caller should process it) and false on every subsequent call (a duplicate to skip).
+     */
+    fun markIfFirstSeen(
+        envelopeId: UUID,
+        consumerGroup: String,
+    ): Boolean
+}
+
+enum class EventProcessingOutcome {
+    PROCESSED,
+    DUPLICATE_SKIPPED,
+}

libs/fincore-eventbus/src/main/resources/db/processed-events.sql

@@ -0,0 +1,13 @@
+-- SPDX-License-Identifier: BUSL-1.1
+-- SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+--
+-- Reference DDL for the consumer dedup table used by JdbcProcessedEventStore. This is NOT a
+-- drop-in migration: a consuming service copies this statement into its own Liquibase formatted-SQL
+-- changeset (per CLAUDE.md 8.5.1) in its own schema. Idempotent and re-runnable.
+
+CREATE TABLE IF NOT EXISTS processed_events (
+    envelope_id    UUID         NOT NULL,
+    consumer_group VARCHAR(128) NOT NULL,
+    processed_at   TIMESTAMPTZ  NOT NULL DEFAULT now(),
+    CONSTRAINT pk_processed_events PRIMARY KEY (envelope_id, consumer_group)
+);

libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/consumer/IdempotentEventProcessorTest.kt

@@ -0,0 +1,61 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import io.kotest.assertions.throwables.shouldThrow
+import io.kotest.matchers.shouldBe
+import org.junit.jupiter.api.Test
+import java.util.UUID
+
+class IdempotentEventProcessorTest {
+    private val processor = IdempotentEventProcessor(InMemoryProcessedEventStore())
+
+    @Test
+    fun `should run the handler and report processed when the envelope is first seen`() {
+        var calls = 0
+
+        val outcome = processor.process(UUID.randomUUID(), "group", { calls++ })
+
+        outcome shouldBe EventProcessingOutcome.PROCESSED
+        calls shouldBe 1
+    }
+
+    @Test
+    fun `should skip the handler and report duplicate when the envelope was already processed`() {
+        val id = UUID.randomUUID()
+        var calls = 0
+        processor.process(id, "group", { calls++ })
+
+        val outcome = processor.process(id, "group", { calls++ })
+
+        outcome shouldBe EventProcessingOutcome.DUPLICATE_SKIPPED
+        calls shouldBe 1
+    }
+
+    @Test
+    fun `should process the same envelope once per consumer group`() {
+        val id = UUID.randomUUID()
+        var calls = 0
+        processor.process(id, "group-a", { calls++ })
+
+        val outcome = processor.process(id, "group-b", { calls++ })
+
+        outcome shouldBe EventProcessingOutcome.PROCESSED
+        calls shouldBe 2
+    }
+
+    @Test
+    fun `should propagate the exception when the handler throws`() {
+        val id = UUID.randomUUID()
+
+        shouldThrow<IllegalStateException> {
+            processor.process(id, "group", { error("handler failed") })
+        }
+
+        // The in-memory store has no transaction, so the claim survives a thrown handler: the
+        // re-process is skipped. Rollback of the claim on handler failure is a property of a
+        // transactional store (JdbcProcessedEventStore), proven by JdbcProcessedEventStoreIT.
+        processor.process(id, "group", { }) shouldBe EventProcessingOutcome.DUPLICATE_SKIPPED
+    }
+}

libs/fincore-eventbus/src/test/kotlin/com/fincore/eventbus/consumer/InMemoryProcessedEventStoreTest.kt

@@ -0,0 +1,51 @@
+// SPDX-License-Identifier: BUSL-1.1
+// SPDX-FileCopyrightText: 2026 FinCore Engine Authors
+
+package com.fincore.eventbus.consumer
+
+import io.kotest.matchers.shouldBe
+import org.junit.jupiter.api.Test
+import java.util.UUID
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
+
+class InMemoryProcessedEventStoreTest {
+    private val store = InMemoryProcessedEventStore()
+
+    @Test
+    fun `should return first-seen then duplicate for the same key`() {
+        val id = UUID.randomUUID()
+
+        store.markIfFirstSeen(id, "group") shouldBe true
+        store.markIfFirstSeen(id, "group") shouldBe false
+    }
+
+    @Test
+    fun `should grant first-seen to exactly one thread when racing on the same key`() {
+        val id = UUID.randomUUID()
+        val successes = AtomicInteger(0)
+        val start = CountDownLatch(1)
+        val done = CountDownLatch(THREADS)
+        val pool = Executors.newFixedThreadPool(THREADS)
+        repeat(THREADS) {
+            pool.submit {
+                start.await()
+                if (store.markIfFirstSeen(id, "group")) successes.incrementAndGet()
+                done.countDown()
+            }
+        }
+
+        start.countDown()
+        done.await(TIMEOUT_SECONDS, TimeUnit.SECONDS)
+        pool.shutdown()
+
+        successes.get() shouldBe 1
+    }
+
+    private companion object {
+        const val THREADS = 16
+        const val TIMEOUT_SECONDS = 10L
+    }
+}