repliker bestilling- og fakturastatus fra tiltaksøkonomi

Author: sondrele

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/ApplicationConfig.kt

@@ -95,6 +95,8 @@ data class KafkaProducers(
 
 data class KafkaConsumers(
     val gjennomforingerV1: KafkaTopicConsumer.Config,
+    val replicateBestillingStatus: KafkaTopicConsumer.Config,
+    val replicateFakturaStatus: KafkaTopicConsumer.Config,
     val amtDeltakerV1: KafkaTopicConsumer.Config,
     val amtVirksomheterV1: KafkaTopicConsumer.Config,
     val amtArrangorMeldingV1: KafkaTopicConsumer.Config,

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/ApplicationConfigDev.kt

@@ -80,6 +80,14 @@ val ApplicationConfigDev = AppConfig(
                 id = "amt-tiltakskoordinators-deltakerliste",
                 topic = "amt.tiltakskoordinators-deltakerliste-v1",
             ),
+            replicateBestillingStatus = KafkaTopicConsumer.Config(
+                id = "replicate-bestilling-status",
+                topic = "team-mulighetsrommet.tiltaksokonomi.bestilling-status-v1",
+            ),
+            replicateFakturaStatus = KafkaTopicConsumer.Config(
+                id = "replicate-faktura-status",
+                topic = "team-mulighetsrommet.tiltaksokonomi.faktura-status-v1",
+            ),
         ),
     ),
     auth = AuthConfig(

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/ApplicationConfigLocal.kt

@@ -91,6 +91,14 @@ val ApplicationConfigLocal = AppConfig(
                 id = "amt-tiltakskoordinators-deltakerliste",
                 topic = "amt.tiltakskoordinators-deltakerliste-v1",
             ),
+            replicateBestillingStatus = KafkaTopicConsumer.Config(
+                id = "replicate-bestilling-status",
+                topic = "tiltaksokonomi.bestilling-status-v1",
+            ),
+            replicateFakturaStatus = KafkaTopicConsumer.Config(
+                id = "replicate-faktura-status",
+                topic = "tiltaksokonomi.faktura-status-v1",
+            ),
         ),
     ),
     auth = AuthConfig(

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/ApplicationConfigProd.kt

@@ -80,6 +80,14 @@ val ApplicationConfigProd = AppConfig(
                 id = "amt-tiltakskoordinators-deltakerliste",
                 topic = "amt.tiltakskoordinators-deltakerliste-v1",
             ),
+            replicateBestillingStatus = KafkaTopicConsumer.Config(
+                id = "replicate-bestilling-status",
+                topic = "team-mulighetsrommet.tiltaksokonomi.bestilling-status-v1",
+            ),
+            replicateFakturaStatus = KafkaTopicConsumer.Config(
+                id = "replicate-faktura-status",
+                topic = "team-mulighetsrommet.tiltaksokonomi.faktura-status-v1",
+            ),
         ),
     ),
     auth = AuthConfig(

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/plugins/DependencyInjection.kt

@@ -59,13 +59,15 @@ import no.nav.mulighetsrommet.api.tasks.GenerateValidationReport
 import no.nav.mulighetsrommet.api.tasks.NotifyFailedKafkaEvents
 import no.nav.mulighetsrommet.api.tilsagn.OkonomiBestillingService
 import no.nav.mulighetsrommet.api.tilsagn.TilsagnService
+import no.nav.mulighetsrommet.api.tilsagn.kafka.ReplicateOkonomiBestillingStatus
 import no.nav.mulighetsrommet.api.tiltakstype.TiltakstypeService
 import no.nav.mulighetsrommet.api.tiltakstype.kafka.SisteTiltakstyperV2KafkaProducer
 import no.nav.mulighetsrommet.api.tiltakstype.task.InitialLoadTiltakstyper
 import no.nav.mulighetsrommet.api.utbetaling.HentAdressebeskyttetPersonBolkPdlQuery
 import no.nav.mulighetsrommet.api.utbetaling.UtbetalingService
 import no.nav.mulighetsrommet.api.utbetaling.kafka.AmtArrangorMeldingV1KafkaConsumer
 import no.nav.mulighetsrommet.api.utbetaling.kafka.AmtDeltakerV1KafkaConsumer
+import no.nav.mulighetsrommet.api.utbetaling.kafka.ReplicateOkonomiFakturaStatus
 import no.nav.mulighetsrommet.api.utbetaling.task.GenerateUtbetaling
 import no.nav.mulighetsrommet.api.utbetaling.task.JournalforUtbetaling
 import no.nav.mulighetsrommet.api.veilederflate.services.BrukerService
@@ -165,18 +167,11 @@ private fun kafka(appConfig: AppConfig) = module {
                 db = get(),
                 utbetalingService = get(),
             ),
-            AmtVirksomheterV1KafkaConsumer(
-                config = config.consumers.amtVirksomheterV1,
-                get(),
-            ),
-            AmtArrangorMeldingV1KafkaConsumer(
-                config = config.consumers.amtArrangorMeldingV1,
-                db = get(),
-            ),
-            AmtKoordinatorGjennomforingV1KafkaConsumer(
-                config = config.consumers.amtKoordinatorMeldingV1,
-                db = get(),
-            ),
+            AmtVirksomheterV1KafkaConsumer(config.consumers.amtVirksomheterV1, get()),
+            AmtArrangorMeldingV1KafkaConsumer(config.consumers.amtArrangorMeldingV1, get()),
+            AmtKoordinatorGjennomforingV1KafkaConsumer(config.consumers.amtKoordinatorMeldingV1, get()),
+            ReplicateOkonomiBestillingStatus(config.consumers.replicateBestillingStatus, get()),
+            ReplicateOkonomiFakturaStatus(config.consumers.replicateFakturaStatus, get()),
         )
         KafkaConsumerOrchestrator(
             consumerPreset = config.consumerPreset,

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/tilsagn/kafka/ReplicateOkonomiBestillingStatus.kt

@@ -0,0 +1,30 @@
+package no.nav.mulighetsrommet.api.tilsagn.kafka
+
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.decodeFromJsonElement
+import no.nav.common.kafka.consumer.util.deserializer.Deserializers.stringDeserializer
+import no.nav.mulighetsrommet.api.ApiDatabase
+import no.nav.mulighetsrommet.kafka.KafkaTopicConsumer
+import no.nav.mulighetsrommet.kafka.serialization.JsonElementDeserializer
+import no.nav.mulighetsrommet.serialization.json.JsonIgnoreUnknownKeys
+import no.nav.tiltak.okonomi.BestillingStatus
+import org.slf4j.LoggerFactory
+
+class ReplicateOkonomiBestillingStatus(
+    config: Config,
+    private val db: ApiDatabase,
+) : KafkaTopicConsumer<String, JsonElement>(
+    config,
+    stringDeserializer(),
+    JsonElementDeserializer(),
+) {
+    private val logger = LoggerFactory.getLogger(javaClass)
+
+    override suspend fun consume(key: String, message: JsonElement): Unit = db.session {
+        logger.info("Konsumerer statusmelding bestillingsnummer=$key")
+
+        val (bestillingsnummer, status) = JsonIgnoreUnknownKeys.decodeFromJsonElement<BestillingStatus>(message)
+
+        queries.tilsagn.setBestillingStatus(bestillingsnummer, status)
+    }
+}

mulighetsrommet-api/src/main/kotlin/no/nav/mulighetsrommet/api/utbetaling/kafka/ReplicateOkonomiFakturaStatus.kt

@@ -0,0 +1,30 @@
+package no.nav.mulighetsrommet.api.utbetaling.kafka
+
+import kotlinx.serialization.json.JsonElement
+import kotlinx.serialization.json.decodeFromJsonElement
+import no.nav.common.kafka.consumer.util.deserializer.Deserializers.stringDeserializer
+import no.nav.mulighetsrommet.api.ApiDatabase
+import no.nav.mulighetsrommet.kafka.KafkaTopicConsumer
+import no.nav.mulighetsrommet.kafka.serialization.JsonElementDeserializer
+import no.nav.mulighetsrommet.serialization.json.JsonIgnoreUnknownKeys
+import no.nav.tiltak.okonomi.FakturaStatus
+import org.slf4j.LoggerFactory
+
+class ReplicateOkonomiFakturaStatus(
+    config: Config,
+    private val db: ApiDatabase,
+) : KafkaTopicConsumer<String, JsonElement>(
+    config,
+    stringDeserializer(),
+    JsonElementDeserializer(),
+) {
+    private val logger = LoggerFactory.getLogger(javaClass)
+
+    override suspend fun consume(key: String, message: JsonElement): Unit = db.session {
+        logger.info("Konsumerer statusmelding fakturanummer=$key")
+
+        val (fakturanummer, status) = JsonIgnoreUnknownKeys.decodeFromJsonElement<FakturaStatus>(message)
+
+        queries.delutbetaling.setFakturaStatus(fakturanummer, status)
+    }
+}