Endret til å bare telle antall utestående merges

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

@@ -27,7 +27,7 @@ fun Routing.konfigurereHelse(
         call.respondText(
             mergeDetector
                 .findMerges(900)
-                .map { "Number of pending merges: ${it.size } "}
+                .map { "Number of pending merges: $it "}
                 .onRight { mergeLogger.info(it) }
                 .onLeft { mergeLogger.error("Error: ${it.system}:${it.code}", it.exception) }
                 .fold( { "Error: ${it.system}:${it.code}" }, { it } )

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

@@ -11,7 +11,7 @@ class MergeDetector(
     private val pdlIdentitesTjeneste: PdlIdentitesTjeneste,
     private val kafkaKeys: KafkaKeys
 ) {
-    suspend fun findMerges(batchSize: Int): Either<Failure, List<MergeDetected>> {
+    suspend fun findMerges(batchSize: Int): Either<Failure, Long> {
         require(batchSize > 0) { "Batch size must be greater than 0" }
         return kafkaKeys.hentSisteArbeidssoekerId()
             .map { it.value }
@@ -20,7 +20,7 @@ class MergeDetector(
                     stopAt = max,
                     maxSize = batchSize,
                     currentPos = 0L,
-                    right(emptyList())
+                    right(0L)
                 )
             }
     }
@@ -29,8 +29,8 @@ class MergeDetector(
         stopAt: Long,
         maxSize: Int,
         currentPos: Long,
-        results: Either<Failure, List<MergeDetected>>
-    ): Either<Failure, List<MergeDetected>> {
+        results: Either<Failure, Long>
+    ): Either<Failure, Long> {
         return when (results) {
             is Left -> {
                 return results
@@ -45,9 +45,10 @@ class MergeDetector(
                         .suspendingFlatMap {
                             pdlIdentitesTjeneste.hentIdenter(it.keys.toList()).map { res -> it to res }
                         }
-                        .map { (local, pdl) ->
-                            detectMerges(local, pdl)
-                        }.map(results.right::plus)
+                        .map { (local, pdl) -> detectMerges(local, pdl) }
+                        .map(Sequence<MergeDetected>::count)
+                        .map(Int::toLong)
+                        .map(results.right::plus)
                     val newStart =
                         storedData.fold({ -1L }, { it.values.maxOfOrNull(ArbeidssoekerId::value)?.plus(1) ?: -1 })
                     processRange(stopAt, maxSize, newStart, detected)
@@ -60,7 +61,7 @@ class MergeDetector(
 fun detectMerges(
     local: Map<Identitetsnummer, ArbeidssoekerId>,
     pdl: Map<String, List<IdentInformasjon>>
-): List<MergeDetected> {
+): Sequence<MergeDetected> {
     return pdl.asSequence()
         .mapNotNull { (searchedId, resultIds) ->
             val arbIds = resultIds
@@ -78,7 +79,7 @@ fun detectMerges(
             } else {
                 null
             }
-        }.toList()
+        }
 }
 
 suspend fun <L, R, A> Either<L, R>.suspendingFlatMap(f: suspend (R) -> Either<L, A>): Either<L, A> =

apps/kafka-key-generator/src/test/kotlin/no/nav/paw/kafkakeygenerator/merge/MergeDetectorKtTest.kt

@@ -0,0 +1,70 @@
+package no.nav.paw.kafkakeygenerator.merge
+
+import io.kotest.core.spec.style.FreeSpec
+import io.kotest.matchers.collections.shouldBeEmpty
+import io.kotest.matchers.sequences.shouldBeEmpty
+import io.kotest.matchers.should
+import io.kotest.matchers.shouldBe
+import no.nav.paw.kafkakeygenerator.vo.ArbeidssoekerId
+import no.nav.paw.kafkakeygenerator.vo.Identitetsnummer
+import no.nav.paw.pdl.graphql.generated.enums.IdentGruppe
+import no.nav.paw.pdl.graphql.generated.hentidenter.IdentInformasjon
+import org.junit.jupiter.api.Test
+
+import org.junit.jupiter.api.Assertions.*
+
+class MergeDetectorKtTest : FreeSpec({
+
+    "Når 2 identiteter er samme person i PDL, men har forskjellige arbeidssøkerId skal det slå ut som 'merge detected" {
+        val local = mapOf(
+            Identitetsnummer("1") to ArbeidssoekerId(11),
+            Identitetsnummer("2") to ArbeidssoekerId(12)
+        )
+        val pdl = mapOf(
+            "1" to listOf(
+                IdentInformasjon(
+                    ident = "1",
+                    gruppe = IdentGruppe.FOLKEREGISTERIDENT,
+                    historisk = false
+                ),
+                IdentInformasjon(
+                    ident = "2",
+                    gruppe = IdentGruppe.FOLKEREGISTERIDENT,
+                    historisk = false
+                )
+            )
+        )
+        detectMerges(
+            local = local,
+            pdl = pdl
+        ) should { merges ->
+            merges.toList().size shouldBe 1
+        }
+    }
+
+    "Når 2 identiteter er samme person i PDL, og har arbeidssøkerId skal det ikke slå ut som 'merge detected" {
+        val local = mapOf(
+            Identitetsnummer("1") to ArbeidssoekerId(11),
+            Identitetsnummer("2") to ArbeidssoekerId(11)
+        )
+        val pdl = mapOf(
+            "1" to listOf(
+                IdentInformasjon(
+                    ident = "1",
+                    gruppe = IdentGruppe.FOLKEREGISTERIDENT,
+                    historisk = false
+                ),
+                IdentInformasjon(
+                    ident = "2",
+                    gruppe = IdentGruppe.FOLKEREGISTERIDENT,
+                    historisk = false
+                )
+            )
+        )
+        detectMerges(
+            local = local,
+            pdl = pdl
+        ).shouldBeEmpty()
+    }
+
+})