lagre statusmeldinger som utgående meldinger

utgående meldinger blir prosessert og sendt på topic via
KafkaProducerRecordProcessor

Author: sondrele

iac/kafka-topics/dev/tiltaksokonomi.bestilling-status-v1.yaml

@@ -0,0 +1,26 @@
+apiVersion: kafka.nais.io/v1
+kind: Topic
+metadata:
+  name: tiltaksokonomi.bestilling-status-v1
+  namespace: team-mulighetsrommet
+  labels:
+    team: team-mulighetsrommet
+spec:
+  pool: nav-dev
+  config:
+    cleanupPolicy: compact
+    minimumInSyncReplicas: 2
+    partitions: 2
+    replication: 3
+    retentionHours: 1
+    segmentHours: 1
+  acl:
+    - team: team-mulighetsrommet
+      application: tiltaksokonomi
+      access: readwrite
+    - team: team-mulighetsrommet
+      application: mulighetsrommet-api
+      access: read
+    - team: team-mulighetsrommet
+      application: mulighetsrommet-kafka-manager
+      access: read

iac/kafka-topics/dev/tiltaksokonomi.faktura-status-v1.yaml

@@ -0,0 +1,26 @@
+apiVersion: kafka.nais.io/v1
+kind: Topic
+metadata:
+  name: tiltaksokonomi.faktura-status-v1
+  namespace: team-mulighetsrommet
+  labels:
+    team: team-mulighetsrommet
+spec:
+  pool: nav-dev
+  config:
+    cleanupPolicy: compact
+    minimumInSyncReplicas: 2
+    partitions: 2
+    replication: 3
+    retentionHours: 1
+    segmentHours: 1
+  acl:
+    - team: team-mulighetsrommet
+      application: tiltaksokonomi
+      access: readwrite
+    - team: team-mulighetsrommet
+      application: mulighetsrommet-api
+      access: read
+    - team: team-mulighetsrommet
+      application: mulighetsrommet-kafka-manager
+      access: read

mulighetsrommet-tiltaksokonomi/build.gradle.kts

@@ -29,6 +29,9 @@ dependencies {
     implementation(projects.common.ktorClients)
     implementation(projects.common.metrics)
     implementation(projects.common.tiltaksokonomiClient)
+    implementation(projects.common.tasks)
+
+    implementation(libs.shedlock.jdbc)
 
     // Cache
     implementation(libs.caffeine)

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/Application.kt

@@ -3,11 +3,16 @@ package no.nav.tiltak.okonomi
 import io.ktor.server.application.*
 import io.ktor.server.engine.*
 import io.ktor.server.netty.*
+import net.javacrumbs.shedlock.provider.jdbc.JdbcLockProvider
+import no.nav.common.job.leader_election.ShedLockLeaderElectionClient
+import no.nav.common.kafka.producer.feilhandtering.KafkaProducerRecordProcessor
+import no.nav.common.kafka.producer.util.KafkaProducerClientBuilder
 import no.nav.mulighetsrommet.brreg.BrregClient
 import no.nav.mulighetsrommet.database.Database
 import no.nav.mulighetsrommet.database.FlywayMigrationManager
 import no.nav.mulighetsrommet.env.NaisEnv
 import no.nav.mulighetsrommet.kafka.KafkaConsumerOrchestrator
+import no.nav.mulighetsrommet.kafka.KafkaProducerRepositoryImpl
 import no.nav.mulighetsrommet.ktor.plugins.configureMonitoring
 import no.nav.mulighetsrommet.tokenprovider.CachedTokenProvider
 import no.nav.tiltak.okonomi.api.configureApi
@@ -44,34 +49,39 @@ fun Application.configure(config: AppConfig) {
     configureMonitoring({ db.isHealthy() })
     configureHTTP()
 
-    val okonomiDb = OkonomiDatabase(db)
-
-    val cachedTokenProvider = CachedTokenProvider.init(config.auth.azure.audience, config.auth.azure.tokenEndpointUrl, config.auth.azure.privateJwk)
+    val cachedTokenProvider = CachedTokenProvider.init(
+        config.auth.azure.audience,
+        config.auth.azure.tokenEndpointUrl,
+        config.auth.azure.privateJwk,
+    )
 
     val oebs = OebsPoApClient(
         engine = config.httpClientEngine,
         baseUrl = config.clients.oebsPoAp.url,
         tokenProvider = cachedTokenProvider.withScope(config.clients.oebsPoAp.scope),
     )
+
     val brreg = BrregClient(config.httpClientEngine)
-    val okonomi = OkonomiService(okonomiDb, oebs, brreg)
-    val kafka = configureKafka(config.kafka, db, okonomi)
 
-    configureApi(kafka, okonomiDb, okonomi)
+    val okonomiDb = OkonomiDatabase(db)
 
-    monitor.subscribe(ApplicationStarted) {
-        kafka.enableFailedRecordProcessor()
-    }
+    val okonomi = OkonomiService(
+        topics = config.kafka.topics,
+        db = okonomiDb,
+        oebs = oebs,
+        brreg = brreg,
+    )
 
-    monitor.subscribe(ApplicationStopPreparing) {
-        kafka.disableFailedRecordProcessor()
-        kafka.stopPollingTopicChanges()
+    val kafka = configureKafka(config.kafka, db, okonomi)
 
+    configureApi(kafka, okonomiDb, okonomi)
+
+    monitor.subscribe(ApplicationStopped) {
         db.close()
     }
 }
 
-fun configureKafka(
+private fun Application.configureKafka(
     config: KafkaConfig,
     db: Database,
     okonomi: OkonomiService,
@@ -81,9 +91,33 @@ fun configureKafka(
         okonomi = okonomi,
     )
 
-    return KafkaConsumerOrchestrator(
+    val producerClient = KafkaProducerClientBuilder.builder<ByteArray, ByteArray?>()
+        .withProperties(config.producerPropertiesPreset)
+        .build()
+    val shedLockLeaderElectionClient = ShedLockLeaderElectionClient(JdbcLockProvider(db.getDatasource()))
+    val producerRecordProcessor = KafkaProducerRecordProcessor(
+        KafkaProducerRepositoryImpl(db),
+        producerClient,
+        shedLockLeaderElectionClient,
+    )
+
+    val kafkaConsumerOrchestrator = KafkaConsumerOrchestrator(
         consumerPreset = config.consumerPropertiesPreset,
         db = db,
         consumers = listOf(bestilling),
     )
+
+    monitor.subscribe(ApplicationStarted) {
+        kafkaConsumerOrchestrator.enableFailedRecordProcessor()
+        producerRecordProcessor.start()
+    }
+
+    monitor.subscribe(ApplicationStopPreparing) {
+        kafkaConsumerOrchestrator.disableFailedRecordProcessor()
+        kafkaConsumerOrchestrator.stopPollingTopicChanges()
+        producerRecordProcessor.close()
+        shedLockLeaderElectionClient.close()
+    }
+
+    return kafkaConsumerOrchestrator
 }

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/ApplicationConfig.kt

@@ -20,9 +20,16 @@ data class AppConfig(
 
 data class KafkaConfig(
     val consumerPropertiesPreset: Properties,
+    val producerPropertiesPreset: Properties,
+    val topics: KafkaTopics,
     val clients: KafkaClients,
 )
 
+data class KafkaTopics(
+    val bestillingStatus: String,
+    val fakturaStatus: String,
+)
+
 data class KafkaClients(
     val okonomiBestillingConsumer: KafkaTopicConsumer.Config,
 )

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/ApplicationConfigDev.kt

@@ -30,6 +30,11 @@ val ApplicationConfigDev = AppConfig(
     ),
     kafka = KafkaConfig(
         consumerPropertiesPreset = KafkaPropertiesPreset.aivenDefaultConsumerProperties("team-mulighetsrommet.tiltaksokonomi.v1"),
+        producerPropertiesPreset = KafkaPropertiesPreset.aivenByteProducerProperties("team-mulighetsrommet.tiltaksokonomi.v1"),
+        topics = KafkaTopics(
+            bestillingStatus = "team-mulighetsrommet.tiltaksokonomi-bestilling-status-v1",
+            fakturaStatus = "team-mulighetsrommet.tiltaksokonomi-faktura-status-v1",
+        ),
         clients = KafkaClients(
             okonomiBestillingConsumer = KafkaTopicConsumer.Config(
                 id = "bestilling",

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/ApplicationConfigLocal.kt

@@ -11,6 +11,7 @@ import no.nav.mulighetsrommet.ktor.respondJson
 import no.nav.mulighetsrommet.tokenprovider.createMockRSAKey
 import no.nav.tiltak.okonomi.oebs.OebsPoApClient
 import org.apache.kafka.common.serialization.ByteArrayDeserializer
+import org.apache.kafka.common.serialization.ByteArraySerializer
 import org.intellij.lang.annotations.Language
 
 val mockClientEngine = createMockEngine {
@@ -70,6 +71,16 @@ val ApplicationConfigLocal = AppConfig(
             .withBrokerUrl("localhost:29092")
             .withDeserializers(ByteArrayDeserializer::class.java, ByteArrayDeserializer::class.java)
             .build(),
+        producerPropertiesPreset = KafkaPropertiesBuilder.producerBuilder()
+            .withBaseProperties()
+            .withProducerId("tiltaksokonomi.v1")
+            .withBrokerUrl("localhost:29092")
+            .withSerializers(ByteArraySerializer::class.java, ByteArraySerializer::class.java)
+            .build(),
+        topics = KafkaTopics(
+            bestillingStatus = "tiltaksokonomi-bestilling-status-v1",
+            fakturaStatus = "tiltaksokonomi-faktura-status-v1",
+        ),
         clients = KafkaClients(
             okonomiBestillingConsumer = KafkaTopicConsumer.Config(
                 id = "bestilling",

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/db/OkonomiDatabase.kt

@@ -4,6 +4,7 @@ import kotliquery.Session
 import no.nav.mulighetsrommet.database.Database
 import no.nav.tiltak.okonomi.db.queries.BestillingQueries
 import no.nav.tiltak.okonomi.db.queries.FakturaQueries
+import no.nav.tiltak.okonomi.db.queries.KafkaProducerRecordQueries
 import no.nav.tiltak.okonomi.db.queries.TiltakKonteringQueries
 import javax.sql.DataSource
 
@@ -38,5 +39,6 @@ class QueryContext(val session: Session) {
         val bestilling = BestillingQueries(session)
         val faktura = FakturaQueries(session)
         val kontering = TiltakKonteringQueries(session)
+        val kafkaProducerRecord = KafkaProducerRecordQueries(session)
     }
 }

mulighetsrommet-tiltaksokonomi/src/main/kotlin/no/nav/tiltak/okonomi/db/queries/KafkaProducerRecordQueries.kt

@@ -0,0 +1,46 @@
+package no.nav.tiltak.okonomi.db.queries
+
+import kotlinx.serialization.json.Json
+import kotliquery.Session
+import kotliquery.queryOf
+import no.nav.common.kafka.producer.util.ProducerUtils
+import no.nav.tiltak.okonomi.api.BestillingStatus
+import no.nav.tiltak.okonomi.api.FakturaStatus
+import org.apache.kafka.clients.producer.ProducerRecord
+import org.intellij.lang.annotations.Language
+
+class KafkaProducerRecordQueries(private val session: Session) {
+
+    fun insertBestillingStatus(topic: String, status: BestillingStatus) {
+        val record = ProducerRecord(
+            topic,
+            status.bestillingsnummer.toByteArray(),
+            Json.encodeToString(status).toByteArray(),
+        )
+
+        insert(record)
+    }
+
+    fun insertFakturaStatus(topic: String, status: FakturaStatus) {
+        val record = ProducerRecord(
+            topic,
+            status.fakturanummer.toByteArray(),
+            Json.encodeToString(status).toByteArray(),
+        )
+
+        insert(record)
+    }
+
+    private fun insert(record: ProducerRecord<ByteArray?, ByteArray?>) {
+        val storedRecord = ProducerUtils.mapToStoredRecord(record)
+
+        @Language("PostgreSQL")
+        val sql = """
+            insert into kafka_producer_record (topic, key, value, headers_json) values (?, ?, ?, ?)
+        """.trimIndent()
+
+        val query = queryOf(sql, storedRecord.topic, storedRecord.key, storedRecord.value, storedRecord.headersJson)
+
+        session.execute(query)
+    }
+}