La inn støtte for å oppdage id merges

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/Applikasjon.kt

@@ -4,19 +4,36 @@ import io.opentelemetry.instrumentation.annotations.WithSpan
 import no.nav.paw.kafkakeygenerator.FailureCode.CONFLICT
 import no.nav.paw.kafkakeygenerator.FailureCode.DB_NOT_FOUND
 import no.nav.paw.kafkakeygenerator.api.v2.*
+import no.nav.paw.kafkakeygenerator.mergedetector.findMerge
+import no.nav.paw.kafkakeygenerator.mergedetector.hentLagretData
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.MergeDetected
 import no.nav.paw.kafkakeygenerator.pdl.PdlIdentitesTjeneste
-import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
-import no.nav.paw.kafkakeygenerator.vo.CallId
-import no.nav.paw.kafkakeygenerator.vo.Identitetsnummer
-import kotlin.math.absoluteValue
+import no.nav.paw.kafkakeygenerator.vo.*
 
 class Applikasjon(
     private val kafkaKeys: KafkaKeys,
     private val identitetsTjeneste: PdlIdentitesTjeneste
 ) {
+    @WithSpan
+    suspend fun validerLagretData(callId: CallId, identitet: Identitetsnummer): Either<Failure, InfoResponse> {
+        return hentInfo(callId, identitet)
+            .flatMap { info ->
+                hentLagretData(
+                    hentArbeidssoekerId = kafkaKeys::hent,
+                    info = info
+                ).map { info to it }
+            }
+            .map { (info, lagretDatra ) -> info to findMerge(lagretDatra) }
+            .map { (info, merge) ->
+                InfoResponse(
+                    info = info,
+                    mergeDetected = merge as? MergeDetected
+                )
+            }
+    }
 
     @WithSpan
-    suspend fun hentInfo(callId: CallId, identitet: Identitetsnummer): Either<Failure, InfoResponse> {
+    suspend fun hentInfo(callId: CallId, identitet: Identitetsnummer): Either<Failure, Info> {
         val pdlIdInfo = identitetsTjeneste.hentIdentInformasjon(
             callId = callId,
             identitet = identitet,
@@ -29,11 +46,16 @@ class Applikasjon(
                     recordKey = publicTopicKeyFunction(arbeidssoekerId).value
                 )
             }.map { lokalIdData ->
-                InfoResponse(
+                Info(
+                    identitetsnummer = identitet.value,
                     lagretData = lokalIdData,
                     pdlData = pdlIdInfo.fold(
                         { PdlData(error = it.code.name, id = null) },
-                        { PdlData(error = null, id = it.map { identInfo ->  PdlId(identInfo.gruppe.name, identInfo.ident) })}
+                        {
+                            PdlData(
+                                error = null,
+                                id = it.map { identInfo -> PdlId(identInfo.gruppe.name, identInfo.ident) })
+                        }
                     )
                 )
             }
@@ -42,15 +64,15 @@ class Applikasjon(
     @WithSpan
     suspend fun hent(callId: CallId, identitet: Identitetsnummer): Either<Failure, ArbeidssoekerId> {
         return kafkaKeys.hent(identitet)
-            .recover(DB_NOT_FOUND) {
+            .suspendingRecover(DB_NOT_FOUND) {
                 sjekkMotAliaser(callId, identitet)
             }
     }
 
     @WithSpan
     suspend fun hentEllerOpprett(callId: CallId, identitet: Identitetsnummer): Either<Failure, ArbeidssoekerId> {
         return hent(callId, identitet)
-            .recover(DB_NOT_FOUND) {
+            .suspendingRecover(DB_NOT_FOUND) {
                 kafkaKeys.opprett(identitet)
             }.recover(CONFLICT) {
                 kafkaKeys.hent(identitet)

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/Either.kt

@@ -15,6 +15,7 @@ fun <L1, R1, R2> Either<L1, R1>.flatMap(f: (R1) -> Either<L1, R2>): Either<L1, R
     is Right -> f(right)
     is Left -> this
 }
+
 data class Right<R>(val right: R) : Either<Nothing, R> {
     override fun <R2> map(f: (R) -> R2): Right<R2> = Right(f(right))
     override fun <L2> mapLeft(f: (Nothing) -> L2): Right<R> = this
@@ -24,10 +25,34 @@ data class Right<R>(val right: R) : Either<Nothing, R> {
     override fun <V> fold(onLeft: (Nothing) -> V, onRight: (R) -> V): V = onRight(right)
 }
 
-data class Left<L>(val left: L): Either<L, Nothing> {
+data class Left<L>(val left: L) : Either<L, Nothing> {
     override fun <R2> map(f: (Nothing) -> R2): Left<L> = this
     override fun <L2> mapLeft(f: (L) -> L2): Left<L2> = Left(f(left))
     override fun onLeft(f: (L) -> Unit) = this.also { f(left) }
     override fun onRight(f: (Nothing) -> Unit) = this
     override fun <V> fold(onLeft: (L) -> V, onRight: (Nothing) -> V): V = onLeft(left)
-}
+}
+
+fun <L, R, C : Collection<Either<L, R>>> Either<L, C>.flatten(): Either<L, List<R>> {
+    return flatMap { list ->
+        val lefts = list.filterIsInstance<Left<L>>()
+        if (lefts.isNotEmpty()) {
+            lefts.first()
+        } else {
+            list.filterIsInstance<Right<R>>()
+                .map { it.right }
+                .let(::right)
+        }
+    }
+}
+
+fun <L, R, C : Collection<Either<L, R>>> C.flatten(): Either<L, List<R>> {
+    val lefts = filterIsInstance<Left<L>>()
+    return if (lefts.isNotEmpty()) {
+        lefts.first()
+    } else {
+        filterIsInstance<Right<R>>()
+            .map { it.right }
+            .let(::right)
+    }
+}

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/Result.kt

@@ -40,7 +40,18 @@ fun <R> Either<Exception, R>.mapToFailure(
     }
 }
 
-suspend fun <R> Either<Failure, R>.recover(failureCode: FailureCode, f: suspend () -> Either<Failure, R>): Either<Failure, R> {
+suspend fun <R> Either<Failure, R>.suspendingRecover(failureCode: FailureCode, f: suspend () -> Either<Failure, R>): Either<Failure, R> {
+    return when (this) {
+        is Right -> this
+        is Left -> if (left.code == failureCode) {
+            f()
+        } else {
+            this
+        }
+    }
+}
+
+fun <R> Either<Failure, R>.recover(failureCode: FailureCode, f: () -> Either<Failure, R>): Either<Failure, R> {
     return when (this) {
         is Right -> this
         is Left -> if (left.code == failureCode) {

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/api/v2/ApiEndepunktV2.kt

@@ -45,7 +45,7 @@ suspend fun PipelineContext<Unit, ApplicationCall>.hentInfo(
         ?.let { CallId(it) }
         ?: CallId(UUID.randomUUID().toString())
     val request = call.receive<RequestV2>()
-    when (val resultat = applikasjon.hentInfo(callId, Identitetsnummer(request.ident))) {
+    when (val resultat = applikasjon.validerLagretData(callId, Identitetsnummer(request.ident))) {
         is Left -> call.respond(
             status = InternalServerError,
             message = resultat.left.code.name

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/api/v2/InfoResponse.kt

@@ -1,21 +1,9 @@
 package no.nav.paw.kafkakeygenerator.api.v2
 
-data class InfoResponse(
-    val lagretData: LokalIdData?,
-    val pdlData: PdlData
-)
-
-data class LokalIdData(
-    val arbeidsoekerId: Long,
-    val recordKey: Long
-)
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.MergeDetected
+import no.nav.paw.kafkakeygenerator.vo.Info
 
-data class PdlData(
-    val error: String?,
-    val id: List<PdlId>?
-)
-
-data class PdlId(
-    val gruppe: String,
-    val id: String
+data class InfoResponse(
+    val info: Info,
+    val mergeDetected: MergeDetected?
 )

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/mergedetector/FailureBuilder.kt

@@ -0,0 +1,14 @@
+package no.nav.paw.kafkakeygenerator.mergedetector
+
+import no.nav.paw.kafkakeygenerator.Failure
+import no.nav.paw.kafkakeygenerator.FailureCode
+
+fun failure(code: FailureCode) : Failure =
+    Failure(
+        system = when (code) {
+            FailureCode.PDL_NOT_FOUND -> "pdl"
+            FailureCode.DB_NOT_FOUND -> "database"
+            else -> "other"
+        },
+        code = code
+    )

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/mergedetector/FindMerge.kt

@@ -0,0 +1,20 @@
+package no.nav.paw.kafkakeygenerator.mergedetector
+
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.LagretData
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.MergeDetected
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.MergeDetectorResult
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.NoMergeDetected
+
+fun findMerge(lagretData: LagretData): MergeDetectorResult {
+    val etterArbeidssoekerId = lagretData.lagretData
+        .toList()
+        .mapNotNull { (id, arbeidssoekerId) -> arbeidssoekerId?.let { it to id } }
+        .groupBy { it.first }
+        .toMap()
+        .mapValues { (_, value) -> value.map { it.second } }
+    return if (etterArbeidssoekerId.size > 1) {
+        MergeDetected(lagretData.identitetsnummer, etterArbeidssoekerId)
+    } else {
+        NoMergeDetected(lagretData.identitetsnummer)
+    }
+}

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/mergedetector/HentLagretData.kt

@@ -0,0 +1,43 @@
+package no.nav.paw.kafkakeygenerator.mergedetector
+
+import io.opentelemetry.instrumentation.annotations.WithSpan
+import kotlinx.coroutines.yield
+import no.nav.paw.kafkakeygenerator.*
+import no.nav.paw.kafkakeygenerator.FailureCode.DB_NOT_FOUND
+import no.nav.paw.kafkakeygenerator.FailureCode.PDL_NOT_FOUND
+import no.nav.paw.kafkakeygenerator.vo.Info
+import no.nav.paw.kafkakeygenerator.vo.PdlId
+import no.nav.paw.kafkakeygenerator.mergedetector.vo.LagretData
+import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
+import no.nav.paw.kafkakeygenerator.vo.Identitetsnummer
+import kotlin.coroutines.Continuation
+import kotlin.coroutines.CoroutineContext
+import kotlin.coroutines.EmptyCoroutineContext
+import kotlin.coroutines.createCoroutine
+
+@WithSpan
+fun hentLagretData(
+    hentArbeidssoekerId: (Identitetsnummer) -> Either<Failure, ArbeidssoekerId>,
+    info: Info
+): Either<Failure, LagretData> {
+    if (info.lagretData == null) return left(failure(DB_NOT_FOUND)) // Er ikke noe lagret kan vi ikke ha en merge
+    val pdlData = info.pdlData.id
+    if (pdlData.isNullOrEmpty()) return left(failure(PDL_NOT_FOUND)) // Er ikke noe pdl data kan vi ikke ha en merge
+    return pdlData
+        .asSequence()
+        .filter { it.gruppe == "FOLKEREGISTERIDENT" }
+        .map(PdlId::id)
+        .map(::Identitetsnummer)
+        .map { identitetsnummer ->
+            hentArbeidssoekerId(identitetsnummer)
+                .recover(DB_NOT_FOUND) { right(null) }
+                .map { identitetsnummer to it }
+        }
+        .toList()
+        .flatten()
+        .map { it.toMap() }
+        .map { LagretData(
+            identitetsnummer = Identitetsnummer(info.identitetsnummer),
+            lagretData = it
+        ) }
+}

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/mergedetector/vo/LagretData.kt

@@ -0,0 +1,9 @@
+package no.nav.paw.kafkakeygenerator.mergedetector.vo
+
+import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
+import no.nav.paw.kafkakeygenerator.vo.Identitetsnummer
+
+data class LagretData(
+    val identitetsnummer: Identitetsnummer,
+    val lagretData: Map<Identitetsnummer, ArbeidssoekerId?>
+)

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/mergedetector/vo/MergeDetected.kt

@@ -0,0 +1,17 @@
+package no.nav.paw.kafkakeygenerator.mergedetector.vo
+
+import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
+import no.nav.paw.kafkakeygenerator.vo.Identitetsnummer
+
+interface MergeDetectorResult {
+    val id: Identitetsnummer
+}
+
+data class NoMergeDetected(
+    override val id: Identitetsnummer
+) : MergeDetectorResult
+
+data class MergeDetected(
+    override val id: Identitetsnummer,
+    val map: Map<ArbeidssoekerId, List<Identitetsnummer>>
+) : MergeDetectorResult

apps/kafka-key-generator/src/main/kotlin/no/nav/paw/kafkakeygenerator/vo/Info.kt

@@ -0,0 +1,22 @@
+package no.nav.paw.kafkakeygenerator.vo
+
+data class Info(
+    val identitetsnummer: String,
+    val lagretData: LokalIdData?,
+    val pdlData: PdlData
+)
+
+data class LokalIdData(
+    val arbeidsoekerId: Long,
+    val recordKey: Long
+)
+
+data class PdlData(
+    val error: String?,
+    val id: List<PdlId>?
+)
+
+data class PdlId(
+    val gruppe: String,
+    val id: String
+)

apps/kafka-key-generator/src/main/resources/openapi/documentation.yaml

@@ -103,6 +103,24 @@ components:
     InfoResponse:
       type: "object"
       properties:
+        info:
+          $ref: "#/components/schemas/Info"
+        mergeDetected:
+          type: "object"
+          properties:
+            id:
+              type: "object"
+              properties:
+                value:
+                  type: "string"
+            map:
+              type: "object"
+              additionalProperties: true
+    Info:
+      type: "object"
+      properties:
+        identitetsnummer:
+          type: "string"
         lagretData:
           type: "object"
           properties: