implementer KafkaProducerRepositoryImpl

Author: sondrele

common/kafka/src/main/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRepositoryImpl.kt

@@ -0,0 +1,62 @@
+package no.nav.mulighetsrommet.kafka
+
+import kotliquery.Row
+import kotliquery.queryOf
+import no.nav.common.kafka.producer.feilhandtering.KafkaProducerRepository
+import no.nav.common.kafka.producer.feilhandtering.StoredProducerRecord
+import no.nav.mulighetsrommet.database.Database
+import no.nav.mulighetsrommet.database.createTextArray
+import org.intellij.lang.annotations.Language
+
+class KafkaProducerRepositoryImpl(private val db: Database) : KafkaProducerRepository {
+    override fun storeRecord(record: StoredProducerRecord): Long = db.session { session ->
+        @Language("PostgreSQL")
+        val sql = """
+            insert into kafka_producer_record (topic, key, value, headers_json) values (?, ?, ?, ?)
+            returning id
+        """.trimIndent()
+
+        val query = queryOf(sql, record.topic, record.key, record.value, record.headersJson)
+
+        session.single(query) { it.long("id") }!!
+    }
+
+    override fun deleteRecords(ids: List<Long>): Unit = db.session { session ->
+        @Language("PostgreSQL")
+        val sql = """
+            delete from kafka_producer_record where id = any(?::bigint[])
+        """.trimIndent()
+
+        session.update(queryOf(sql, session.createArrayOf("bigint", ids)))
+    }
+
+    override fun getRecords(maxMessages: Int): List<StoredProducerRecord> = db.session { session ->
+        @Language("PostgreSQL")
+        val sql = """
+            select * from kafka_producer_record order by id limit ?
+        """.trimIndent()
+
+        val query = queryOf(sql, maxMessages)
+
+        session.list(query) { toStoredProducerRecord(it) }
+    }
+
+    override fun getRecords(maxMessages: Int, topics: List<String>): List<StoredProducerRecord> = db.session { session ->
+        @Language("PostgreSQL")
+        val sql = """
+            select * from kafka_producer_record where topic = any(?::text[]) order by id limit ?
+        """.trimIndent()
+
+        val query = queryOf(sql, session.createTextArray(topics), maxMessages)
+
+        session.list(query) { toStoredProducerRecord(it) }
+    }
+}
+
+private fun toStoredProducerRecord(row: Row) = StoredProducerRecord(
+    row.long("id"),
+    row.string("topic"),
+    row.bytes("key"),
+    row.bytes("value"),
+    row.string("headers_json"),
+)

common/kafka/src/test/kotlin/no/nav/mulighetsrommet/kafka/KafkaProducerRepositoryTest.kt

@@ -0,0 +1,56 @@
+package no.nav.mulighetsrommet.kafka
+
+import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
+import no.nav.common.kafka.producer.feilhandtering.StoredProducerRecord
+import no.nav.mulighetsrommet.database.kotest.extensions.FlywayDatabaseTestListener
+
+class KafkaProducerRepositoryTest : FunSpec({
+    val database = extension(FlywayDatabaseTestListener(testDatabaseConfig))
+
+    lateinit var kafkaProducerRepository: KafkaProducerRepositoryImpl
+
+    beforeSpec {
+        kafkaProducerRepository = KafkaProducerRepositoryImpl(database.db)
+    }
+
+    test("should store records") {
+        val records = (0..2).map {
+            StoredProducerRecord(
+                "topic$it",
+                "key$it".toByteArray(),
+                "value$it".toByteArray(),
+                "{}",
+            )
+        }
+
+        records.forEach { kafkaProducerRepository.storeRecord(it) }
+
+        database.assertTable("kafka_producer_record").hasNumberOfRows(3)
+    }
+
+    test("should delete records") {
+        kafkaProducerRepository.deleteRecords(listOf(1, 2))
+        database.assertTable("kafka_producer_record").hasNumberOfRows(1)
+    }
+
+    test("should get all records") {
+        val result = kafkaProducerRepository.getRecords(10)
+        result.size shouldBe 1
+        result[0].id shouldBe 3
+    }
+
+    test("should get records by topics") {
+        kafkaProducerRepository.storeRecord(
+            StoredProducerRecord(
+                "topic1",
+                "key1".toByteArray(),
+                "value1".toByteArray(),
+                "{}",
+            ),
+        )
+
+        val result = kafkaProducerRepository.getRecords(10, listOf("topic1"))
+        result.size shouldBe 1
+    }
+})

common/kafka/src/test/resources/db/migration/V1__test-database.sql

@@ -24,6 +24,16 @@ create table failed_events
     unique (topic, partition, record_offset)
 );
 
+create table kafka_producer_record
+(
+    id           bigint generated always as identity,
+    topic        text                                not null,
+    key          bytea,
+    value        bytea,
+    headers_json text,
+    created_at   timestamp default current_timestamp not null
+);
+
 create type topic_type as enum ('CONSUMER', 'PRODUCER');
 
 create table topics