Publiser meldinger med Kafka-key i API-app

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/App.kt

@@ -20,6 +20,7 @@ import no.nav.helse.rapids_rivers.RapidApplication
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.redis.RedisConnection
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.registerShutdownLifecycle
 import no.nav.helsearbeidsgiver.inntektsmelding.api.aktiveorgnr.aktiveOrgnrRoute
+import no.nav.helsearbeidsgiver.inntektsmelding.api.auth.TilgangProducer
 import no.nav.helsearbeidsgiver.inntektsmelding.api.auth.Tilgangskontroll
 import no.nav.helsearbeidsgiver.inntektsmelding.api.hentforespoersel.hentForespoersel
 import no.nav.helsearbeidsgiver.inntektsmelding.api.hentforespoerselIdListe.hentForespoerselIdListe
@@ -29,7 +30,6 @@ import no.nav.helsearbeidsgiver.inntektsmelding.api.inntekt.inntektRoute
 import no.nav.helsearbeidsgiver.inntektsmelding.api.inntektselvbestemt.inntektSelvbestemtRoute
 import no.nav.helsearbeidsgiver.inntektsmelding.api.kvittering.kvittering
 import no.nav.helsearbeidsgiver.inntektsmelding.api.lagreselvbestemtim.lagreSelvbestemtImRoute
-import no.nav.helsearbeidsgiver.inntektsmelding.api.tilgang.TilgangProducer
 import no.nav.helsearbeidsgiver.inntektsmelding.api.tilgangorgnr.tilgangOrgnrRoute
 import no.nav.helsearbeidsgiver.utils.cache.LocalCache
 import no.nav.helsearbeidsgiver.utils.json.jsonConfig

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/aktiveorgnr/AktiveOrgnrProducer.kt

@@ -33,6 +33,7 @@ class AktiveOrgnrProducer(
         ) {
             rapid
                 .publish(
+                    key = arbeidstagerFnr,
                     Key.EVENT_NAME to EventName.AKTIVE_ORGNR_REQUESTED.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/tilgang/TilgangProducer.kt → api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/auth/TilgangProducer.kt

@@ -1,4 +1,4 @@
-package no.nav.helsearbeidsgiver.inntektsmelding.api.tilgang
+package no.nav.helsearbeidsgiver.inntektsmelding.api.auth
 
 import com.github.navikt.tbd_libs.rapids_and_rivers_api.RapidsConnection
 import kotlinx.serialization.json.JsonElement
@@ -15,6 +15,7 @@ import no.nav.helsearbeidsgiver.utils.log.MdcUtils
 import no.nav.helsearbeidsgiver.utils.wrapper.Fnr
 import java.util.UUID
 
+// TODO test
 class TilgangProducer(
     private val rapid: RapidsConnection,
 ) {
@@ -29,7 +30,7 @@ class TilgangProducer(
     ) = publish(
         EventName.TILGANG_FORESPOERSEL_REQUESTED,
         transaksjonId,
-        Key.FNR to fnr.toJson(),
+        fnr,
         Key.FORESPOERSEL_ID to forespoerselId.toJson(),
     )
 
@@ -40,14 +41,15 @@ class TilgangProducer(
     ) = publish(
         EventName.TILGANG_ORG_REQUESTED,
         transaksjonId,
-        Key.FNR to fnr.toJson(),
+        fnr,
         Key.ORGNRUNDERENHET to orgnr.toJson(),
     )
 
     private fun publish(
         eventName: EventName,
         transaksjonId: UUID,
-        vararg messageFields: Pair<Key, JsonElement>,
+        fnr: Fnr,
+        dataField: Pair<Key, JsonElement>,
     ) {
         MdcUtils.withLogFields(
             Log.klasse(this),
@@ -56,9 +58,14 @@ class TilgangProducer(
         ) {
             rapid
                 .publish(
+                    key = fnr,
                     Key.EVENT_NAME to eventName.toJson(),
                     Key.UUID to transaksjonId.toJson(),
-                    Key.DATA to messageFields.toMap().toJson(),
+                    Key.DATA to
+                        mapOf(
+                            Key.FNR to fnr.toJson(),
+                            dataField,
+                        ).toJson(),
                 ).also { json ->
                     "Publiserte request om tilgang.".let {
                         logger.info(it)

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/auth/Tilgangskontroll.kt

@@ -9,7 +9,6 @@ import no.nav.helsearbeidsgiver.felles.rapidsrivers.redis.RedisPrefix
 import no.nav.helsearbeidsgiver.felles.rapidsrivers.redis.RedisStore
 import no.nav.helsearbeidsgiver.inntektsmelding.api.RedisPoller
 import no.nav.helsearbeidsgiver.inntektsmelding.api.logger
-import no.nav.helsearbeidsgiver.inntektsmelding.api.tilgang.TilgangProducer
 import no.nav.helsearbeidsgiver.utils.cache.LocalCache
 import no.nav.helsearbeidsgiver.utils.json.fromJson
 import no.nav.helsearbeidsgiver.utils.wrapper.Fnr

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/hentforespoersel/HentForespoerselProducer.kt

@@ -26,6 +26,7 @@ class HentForespoerselProducer(
     ) {
         rapid
             .publish(
+                key = request.uuid,
                 Key.EVENT_NAME to EventName.TRENGER_REQUESTED.toJson(EventName.serializer()),
                 Key.UUID to transaksjonId.toString().toJson(),
                 Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/hentforespoerselIdListe/HentForespoerslerProducer.kt

@@ -25,6 +25,7 @@ class HentForespoerslerProducer(
     ) {
         rapid
             .publish(
+                key = UUID.randomUUID(),
                 Key.EVENT_NAME to EventName.FORESPOERSLER_REQUESTED.toJson(EventName.serializer()),
                 Key.UUID to transaksjonId.toString().toJson(),
                 Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/hentselvbestemtim/HentSelvbestemtImProducer.kt

@@ -31,6 +31,7 @@ class HentSelvbestemtImProducer(
         ) {
             rapid
                 .publish(
+                    key = selvbestemtId,
                     Key.EVENT_NAME to EventName.SELVBESTEMT_IM_REQUESTED.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/innsending/InnsendingProducer.kt

@@ -27,6 +27,7 @@ class InnsendingProducer(
     ) {
         rapid
             .publish(
+                key = skjemaInntektsmelding.forespoerselId,
                 Key.EVENT_NAME to EventName.INSENDING_STARTED.toJson(),
                 Key.UUID to transaksjonId.toJson(),
                 Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/inntekt/InntektProducer.kt

@@ -32,6 +32,7 @@ class InntektProducer(
         ) {
             rapid
                 .publish(
+                    key = request.forespoerselId,
                     Key.EVENT_NAME to EventName.INNTEKT_REQUESTED.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/inntektselvbestemt/InntektSelvbestemtProducer.kt

@@ -31,6 +31,7 @@ class InntektSelvbestemtProducer(
         ) {
             rapid
                 .publish(
+                    key = request.sykmeldtFnr,
                     Key.EVENT_NAME to EventName.INNTEKT_SELVBESTEMT_REQUESTED.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/kvittering/KvitteringProducer.kt

@@ -32,6 +32,7 @@ class KvitteringProducer(
         ) {
             rapid
                 .publish(
+                    key = forespoerselId,
                     Key.EVENT_NAME to EventName.KVITTERING_REQUESTED.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

api/src/main/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/api/lagreselvbestemtim/LagreSelvbestemtImProducer.kt

@@ -33,6 +33,7 @@ class LagreSelvbestemtImProducer(
         ) {
             rapid
                 .publish(
+                    key = skjema.sykmeldtFnr,
                     Key.EVENT_NAME to EventName.SELVBESTEMT_IM_MOTTATT.toJson(),
                     Key.UUID to transaksjonId.toJson(),
                     Key.DATA to

felles/src/main/kotlin/no/nav/helsearbeidsgiver/felles/rapidsrivers/RiverUtils.kt

@@ -10,10 +10,25 @@ import kotlinx.serialization.json.JsonNull
 import no.nav.helsearbeidsgiver.felles.Key
 import no.nav.helsearbeidsgiver.utils.json.parseJson
 import no.nav.helsearbeidsgiver.utils.json.toJson
+import no.nav.helsearbeidsgiver.utils.wrapper.Fnr
+import java.util.UUID
 
-fun MessageContext.publish(vararg messageFields: Pair<Key, JsonElement>): JsonElement = publish(messageFields.toMap())
+fun MessageContext.publish(vararg messageFields: Pair<Key, JsonElement>): JsonElement = publish(null, messageFields.toMap())
 
-fun MessageContext.publish(messageFields: Map<Key, JsonElement>): JsonElement =
+fun MessageContext.publish(
+    key: Fnr,
+    vararg messageFields: Pair<Key, JsonElement>,
+): JsonElement = publish(key.verdi, messageFields.toMap())
+
+fun MessageContext.publish(
+    key: UUID,
+    vararg messageFields: Pair<Key, JsonElement>,
+): JsonElement = publish(key.toString(), messageFields.toMap())
+
+internal fun MessageContext.publish(
+    key: String?,
+    messageFields: Map<Key, JsonElement>,
+): JsonElement =
     messageFields
         .mapKeys { (key, _) -> key.toString() }
         .filterValues { it !is JsonNull }
@@ -27,5 +42,10 @@ fun MessageContext.publish(messageFields: Map<Key, JsonElement>): JsonElement =
                 randomIdGenerator = null,
             )
         }.toJson()
-        .also(::publish)
-        .parseJson()
+        .also {
+            if (key == null) {
+                publish(it)
+            } else {
+                publish(key, it)
+            }
+        }.parseJson()

felles/src/main/kotlin/no/nav/helsearbeidsgiver/felles/rapidsrivers/river/OpenRiver.kt

@@ -31,6 +31,6 @@ internal class OpenRiver(
             .toJson()
             .parseJson()
             .haandterMelding()
-            ?.also(context::publish)
+            ?.also { context.publish(null, it) }
     }
 }

felles/src/test/kotlin/no/nav/helsearbeidsgiver/felles/rapidsrivers/RiverUtilsKtTest.kt

@@ -15,6 +15,8 @@ import no.nav.helsearbeidsgiver.felles.json.toMap
 import no.nav.helsearbeidsgiver.felles.test.mock.mockInntektsmeldingV1
 import no.nav.helsearbeidsgiver.utils.json.parseJson
 import no.nav.helsearbeidsgiver.utils.json.toJson
+import no.nav.helsearbeidsgiver.utils.test.wrapper.genererGyldig
+import no.nav.helsearbeidsgiver.utils.wrapper.Fnr
 import java.util.UUID
 
 class RiverUtilsKtTest :
@@ -28,7 +30,7 @@ class RiverUtilsKtTest :
 
         context("publish") {
 
-            test("vararg pairs") {
+            test("vararg pairs (uten key)") {
                 val melding =
                     arrayOf(
                         Key.FORESPOERSEL_ID to UUID.randomUUID().toJson(),
@@ -47,18 +49,81 @@ class RiverUtilsKtTest :
                 }
             }
 
+            test("vararg pairs (fnr-key)") {
+                val key = Fnr.genererGyldig()
+                val melding =
+                    arrayOf(
+                        Key.FORESPOERSEL_ID to UUID.randomUUID().toJson(),
+                        Key.INNTEKTSMELDING to mockInntektsmeldingV1().toJson(Inntektsmelding.serializer()),
+                        Key.FNR_LISTE to setOf("111", "333", "555").toJson(String.serializer()),
+                    )
+
+                testRapid.publish(key, *melding)
+
+                verifySequence {
+                    testRapid.publish(
+                        key.toString(),
+                        withArg<String> {
+                            it.parseJson().toMap() shouldContainExactly melding.toMap()
+                        },
+                    )
+                }
+            }
+
+            test("vararg pairs (UUID-key)") {
+                val key = UUID.randomUUID()
+                val melding =
+                    arrayOf(
+                        Key.FORESPOERSEL_ID to UUID.randomUUID().toJson(),
+                        Key.INNTEKTSMELDING to mockInntektsmeldingV1().toJson(Inntektsmelding.serializer()),
+                        Key.FNR_LISTE to setOf("555", "333", "111").toJson(String.serializer()),
+                    )
+
+                testRapid.publish(key, *melding)
+
+                verifySequence {
+                    testRapid.publish(
+                        key.toString(),
+                        withArg<String> {
+                            it.parseJson().toMap() shouldContainExactly melding.toMap()
+                        },
+                    )
+                }
+            }
+
+            test("map (uten key)") {
+                val melding =
+                    mapOf(
+                        Key.FORESPOERSEL_ID to UUID.randomUUID().toJson(),
+                        Key.INNTEKTSMELDING to mockInntektsmeldingV1().toJson(Inntektsmelding.serializer()),
+                        Key.ORGNR_UNDERENHETER to setOf("666", "444", "222").toJson(String.serializer()),
+                    )
+
+                testRapid.publish(null, melding)
+
+                verifySequence {
+                    testRapid.publish(
+                        withArg<String> {
+                            it.parseJson().toMap() shouldContainExactly melding
+                        },
+                    )
+                }
+            }
+
             test("map") {
+                val key = UUID.randomUUID()
                 val melding =
                     mapOf(
                         Key.FORESPOERSEL_ID to UUID.randomUUID().toJson(),
                         Key.INNTEKTSMELDING to mockInntektsmeldingV1().toJson(Inntektsmelding.serializer()),
                         Key.ORGNR_UNDERENHETER to setOf("222", "444", "666").toJson(String.serializer()),
                     )
 
-                testRapid.publish(melding)
+                testRapid.publish(key.toString(), melding)
 
                 verifySequence {
                     testRapid.publish(
+                        key.toString(),
                         withArg<String> {
                             it.parseJson().toMap() shouldContainExactly melding
                         },
@@ -67,9 +132,11 @@ class RiverUtilsKtTest :
             }
 
             test("filtrerer ut JsonNull") {
+                val key = UUID.randomUUID()
                 val selvbestemtId = UUID.randomUUID()
 
                 testRapid.publish(
+                    key.toString(),
                     mapOf(
                         Key.SELVBESTEMT_ID to selvbestemtId.toJson(),
                         Key.FORESPOERSEL_SVAR to JsonNull,
@@ -78,6 +145,7 @@ class RiverUtilsKtTest :
 
                 verifySequence {
                     testRapid.publish(
+                        key.toString(),
                         withArg<String> {
                             it.parseJson().toMap() shouldContainExactly mapOf(Key.SELVBESTEMT_ID to selvbestemtId.toJson())
                         },

integrasjonstest/src/test/kotlin/no/nav/helsearbeidsgiver/inntektsmelding/integrasjonstest/utils/EndToEndTest.kt

@@ -32,7 +32,7 @@ import no.nav.helsearbeidsgiver.inntekt.InntektKlient
 import no.nav.helsearbeidsgiver.inntektsmelding.aareg.createAaregRiver
 import no.nav.helsearbeidsgiver.inntektsmelding.aktiveorgnrservice.createAktiveOrgnrService
 import no.nav.helsearbeidsgiver.inntektsmelding.altinn.createAltinn
-import no.nav.helsearbeidsgiver.inntektsmelding.api.tilgang.TilgangProducer
+import no.nav.helsearbeidsgiver.inntektsmelding.api.auth.TilgangProducer
 import no.nav.helsearbeidsgiver.inntektsmelding.berikinntektsmeldingservice.createBerikInntektsmeldingService
 import no.nav.helsearbeidsgiver.inntektsmelding.brospinn.SpinnKlient
 import no.nav.helsearbeidsgiver.inntektsmelding.brospinn.createHentEksternImRiver
@@ -253,7 +253,7 @@ abstract class EndToEndTest : ContainerTest() {
 
     fun publish(vararg messageFields: Pair<Key, JsonElement>) {
         println("Publiserer melding med felt: ${messageFields.toMap()}")
-        imTestRapid.publish(messageFields.toMap())
+        imTestRapid.publish(UUID.randomUUID(), *messageFields)
     }
 
     fun publish(vararg messageFields: Pair<Pri.Key, JsonElement>): JsonElement {