Flyttet bekreftelse-tjeneste til intern

apps/bekreftelse-tjeneste/README.md

@@ -0,0 +1,75 @@
+```mermaid
+sequenceDiagram
+    Registeret->>PeriodeTopic: Periode startet
+    PeriodeTopic-->>MeldepliktTjeneste: Periode startet
+    MeldepliktTjeneste-->>EndringerTopic: Melding X dager før forfall
+    MeldepliktTjeneste-->>EndringerTopic: Melding OK ved innsending
+    MeldepliktTjeneste-->>EndringerTopic: Frist utløpt
+    MeldepliktTjeneste-->>EndringerTopic: Graceperiode utløpt
+    PeriodeTopic-->>MeldepliktTjeneste: Periode avsluttet
+    MeldepliktTjeneste-->>EndringerTopic: Periode avsluttet
+```
+
+```mermaid
+sequenceDiagram
+    Registeret->>PeriodeTopic: Periode startet
+    Dagpenger-->>AnsvarsTopic: Dagpenger startet
+    AnsvarsTopic-->>MeldepliktTjeneste: Dagpenger tar over
+    MeldepliktTjeneste-->>EndringerTopic: 
+```
+
+```mermaid
+graph LR
+    periodeTopic((PeriodeTopic)) -- Startet --> Ansvar[Ansvar]     
+```
+```
+ periode topic:
+    startet: lagre initiell tilstand
+    avsluttet: send ok(avsluttet)
+        :slett tilstand
+ 
+ ansvar topic:
+    tar ansvar: lagre info om ansvar
+        :sett tidspunkt for siste melding til record ts for ansvars endring
+    avslutter ansvar: slett info om ansvar
+    
+ melding topic:
+    mottatt: lagre tidspunkt for siste melding
+        :send OK(mottatt)
+        dersom ikke ønsker å fortsette:
+            :send VilAvsluttePerioden
+        
+
+ hver x time:
+    for alle perioder ingen har ansvar for:
+        dersom tid siden siste melding (eller periode start) > Y dager:
+            send melding om frist nærmer seg            
+        dersom tid siden siste melding (eller periode start) > Z dager:
+            send melding om frist utløpt
+        dersom tid siden siste melding (eller periode start) > Z+Grace dager:
+            send melding om graceperiode utløpt
+    for alle perioder andre har ansvar for:
+        dersom tid siden siste melding (eller periode start) > Z+G+1dag dager:
+            send melding om graceperiode utløpt
+                :slett ansvar, vi overtar
+```
+
+Modul: endringer til frontend
+```
+ meldings topic:
+   - OK(mottatt) -> sett oppgave fullført
+   - OK(avsluttet) -> slett oppgave kansellert
+   - OK(ansvar) -> sett oppgave fullført
+   - frist nærmer seg -> opprett oppgave
+   - frist utløpt -> opprett oppgave
+   - graceperiode utløpt -> ingenting   
+```
+
+Modul: endringer til eventlogg
+```
+ meldings topic:
+   - VilAvsluttePerioden -> send avslutt hendelse til eventlogg 
+   
+```
+* OK (grunn: RapportMottatt, AnsvarFlyttet, PeriodeAvsluttet)
+* 

apps/bekreftelse-tjeneste/build.gradle.kts

@@ -0,0 +1,35 @@
+import org.jetbrains.kotlin.gradle.tasks.KotlinCompile
+
+plugins {
+    kotlin("jvm")
+    id("com.google.cloud.tools.jib")
+}
+
+dependencies {
+    implementation(project(":lib:hoplite-config"))
+    implementation(project(":lib:kafka-streams"))
+    implementation(project(":lib:kafka-key-generator-client"))
+    implementation(project(":domain:main-avro-schema"))
+    implementation(project(":domain:rapportering-interne-hendelser"))
+    implementation(project(":domain:rapporteringsansvar-schema"))
+    implementation(project(":domain:rapporteringsmelding-schema"))
+    implementation(orgApacheKafka.kafkaStreams)
+    implementation(jackson.datatypeJsr310)
+    implementation(jackson.kotlin)
+    implementation(apacheAvro.kafkaStreamsAvroSerde)
+
+    testImplementation(orgApacheKafka.streamsTest)
+    testImplementation(testLibs.runnerJunit5)
+    testImplementation(testLibs.assertionsCore)
+}
+
+//enable context receiver
+tasks.withType<KotlinCompile>().configureEach {
+    compilerOptions {
+        freeCompilerArgs.add("-Xcontext-receivers")
+    }
+}
+
+tasks.withType<Test>().configureEach {
+    useJUnitPlatform()
+}

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/AnsvarTopology.kt

@@ -0,0 +1,15 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.config.kafka.streams.genericProcess
+import no.nav.paw.rapportering.ansvar.v1.AnsvarEndret
+import no.nav.paw.rapportering.internehendelser.RapporteringsHendelse
+import org.apache.kafka.streams.StreamsBuilder
+
+context(ApplicationConfiguration, ApplicationContext)
+fun StreamsBuilder.processAnsvarTopic() {
+    stream<Long, AnsvarEndret>(ansvarsTopic)
+        .genericProcess<Long, AnsvarEndret, Long, RapporteringsHendelse>("ansvarEndret", statStoreName) { record ->
+
+        }
+}
+

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/ApplicationConfiguration.kt

@@ -0,0 +1,12 @@
+package no.nav.paw.meldeplikttjeneste
+
+import java.time.Duration
+
+data class ApplicationConfiguration(
+    val periodeTopic: String,
+    val ansvarsTopic: String,
+    val rapporteringsTopic: String,
+    val rapporteringsHendelsesloggTopic: String,
+    val statStoreName: String,
+    val punctuateInterval: Duration
+)

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/ApplicationContext.kt

@@ -0,0 +1,9 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstandSerde
+import no.nav.paw.rapportering.internehendelser.RapporteringsHendelseSerde
+
+class ApplicationContext(
+    val internTilstandSerde: InternTilstandSerde,
+    val rapporteringsHendelseSerde: RapporteringsHendelseSerde
+)

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/KontrolerFrister.kt

@@ -0,0 +1,22 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.config.kafka.streams.Punctuation
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstand
+import no.nav.paw.meldeplikttjeneste.tilstand.RapporteringsKonfigurasjon
+import org.apache.kafka.streams.KeyValue
+import org.apache.kafka.streams.processor.PunctuationType
+import org.apache.kafka.streams.state.KeyValueStore
+import java.time.Instant
+import java.util.*
+
+
+context(ApplicationConfiguration, RapporteringsKonfigurasjon)
+fun kontrollerFrister(): Punctuation<Long, Action> = TODO()
+
+operator fun <K, V> KeyValue<K, V>.component1(): K = key
+operator fun <K, V> KeyValue<K, V>.component2(): V = value
+
+context(Instant, RapporteringsKonfigurasjon)
+fun rapporteringSkalTilgjengeliggjoeres(tilstand: InternTilstand): Boolean {
+    TODO()
+}

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/PeriodeTopology.kt

@@ -0,0 +1,67 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.arbeidssokerregisteret.api.v1.Periode
+import no.nav.paw.config.kafka.streams.genericProcess
+import no.nav.paw.config.kafka.streams.mapWithContext
+import no.nav.paw.kafkakeygenerator.client.KafkaKeysResponse
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstand
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstandSerde
+import no.nav.paw.meldeplikttjeneste.tilstand.initTilstand
+import no.nav.paw.rapportering.internehendelser.PeriodeAvsluttet
+import no.nav.paw.rapportering.internehendelser.RapporteringsHendelse
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.kstream.Produced
+import org.apache.kafka.streams.state.KeyValueStore
+import java.util.*
+
+
+context(ApplicationConfiguration, ApplicationContext)
+fun StreamsBuilder.processPeriodeTopic(kafkaKeyFunction: (String) -> KafkaKeysResponse) {
+    stream<Long, Periode>(periodeTopic)
+        .mapWithContext("lagreEllerSlettPeriode", statStoreName) { periode ->
+            val keyValueStore: KeyValueStore<UUID, InternTilstand> = getStateStore(statStoreName)
+            val currentState = keyValueStore[periode.id]
+            val (id, key) = currentState?.let { it.periode.kafkaKeysId to it.periode.recordKey } ?:
+                    kafkaKeyFunction(periode.identitetsnummer).let { it.id to it.key}
+            when {
+                currentState == null && periode.avsluttet() -> Action.DoNothing
+                periode.avsluttet() -> Action.DeleteStateAndEmit(id, periode)
+                currentState == null -> Action.UpdateState(initTilstand(id = id, key = key, periode = periode))
+                else -> Action.DoNothing
+            }
+        }
+        .genericProcess<Long, Action, Long, RapporteringsHendelse>(
+            name = "executeAction",
+            punctuation = null,
+            stateStoreNames = arrayOf(statStoreName)
+        ) { record ->
+            val keyValueStore: KeyValueStore<UUID, InternTilstand> = getStateStore(statStoreName)
+            when (val action = record.value()) {
+                is Action.DeleteStateAndEmit -> {
+                    keyValueStore.delete(action.periode.id)
+                    forward(
+                        record.withValue(
+                            PeriodeAvsluttet(
+                                UUID.randomUUID(),
+                                action.periode.id,
+                                action.periode.identitetsnummer,
+                                action.arbeidsoekerId
+                            ) as RapporteringsHendelse
+                        )
+                    )
+                }
+
+                Action.DoNothing -> {}
+                is Action.UpdateState -> keyValueStore.put(action.state.periode.periodeId, action.state)
+            }
+        }.to(rapporteringsHendelsesloggTopic, Produced.with(Serdes.Long(), rapporteringsHendelseSerde))
+}
+
+fun Periode.avsluttet(): Boolean = avsluttet != null
+
+sealed interface Action {
+    data object DoNothing : Action
+    data class DeleteStateAndEmit(val arbeidsoekerId: Long, val periode: Periode) : Action
+    data class UpdateState(val state: InternTilstand) : Action
+}

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/RapportingsMeldingTopology.kt

@@ -0,0 +1,34 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.config.kafka.streams.genericProcess
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstand
+import no.nav.paw.rapportering.internehendelser.BaOmAaAvsluttePeriode
+import no.nav.paw.rapportering.internehendelser.RapporteringsHendelse
+import no.nav.paw.rapportering.internehendelser.RapporteringsMeldingMottatt
+import no.nav.paw.rapportering.melding.v1.Melding
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.state.KeyValueStore
+import org.slf4j.LoggerFactory
+import java.util.*
+
+context(ApplicationConfiguration, ApplicationContext)
+fun StreamsBuilder.processRapporteringsMeldingTopic() {
+    stream<Long, Melding>(rapporteringsTopic)
+        .genericProcess<Long, Melding, Long, RapporteringsHendelse>(
+            name = "meldingMottatt",
+            statStoreName
+        ) { record ->
+            val gjeldeneTilstand: InternTilstand? = getStateStore<StateStore>(statStoreName)[record.value().periodeId]
+            if (gjeldeneTilstand == null) {
+                meldingsLogger.warn("Melding mottatt for periode som ikke er aktiv/eksisterer")
+            } else {
+                TODO()
+            }
+
+        }
+
+}
+
+private val meldingsLogger = LoggerFactory.getLogger("meldingsLogger")
+
+

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/Startup.kt

@@ -0,0 +1,30 @@
+package no.nav.paw.meldeplikttjeneste
+
+import io.confluent.kafka.streams.serdes.avro.SpecificAvroSerde
+import no.nav.paw.config.hoplite.loadNaisOrLocalConfiguration
+import no.nav.paw.config.kafka.KAFKA_STREAMS_CONFIG_WITH_SCHEME_REG
+import no.nav.paw.config.kafka.KafkaConfig
+import no.nav.paw.config.kafka.streams.KafkaStreamsFactory
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstandSerde
+import org.apache.kafka.common.serialization.Serdes
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.state.Stores
+
+const val STATE_STORE = "state-store"
+const val APPLICATION_ID_SUFFIX = "beta"
+
+fun main() {
+    val kafkaConfig = loadNaisOrLocalConfiguration<KafkaConfig>(KAFKA_STREAMS_CONFIG_WITH_SCHEME_REG)
+    val streamsFactory = KafkaStreamsFactory(APPLICATION_ID_SUFFIX, kafkaConfig)
+        .withDefaultKeySerde(Serdes.Long()::class)
+        .withDefaultValueSerde(SpecificAvroSerde::class)
+    val steamsBuilder = StreamsBuilder()
+        .addStateStore(
+            Stores.keyValueStoreBuilder(
+                Stores.persistentKeyValueStore(STATE_STORE),
+                Serdes.UUID(),
+                InternTilstandSerde()
+            )
+        )
+
+}

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/Topology.kt

@@ -0,0 +1,22 @@
+package no.nav.paw.meldeplikttjeneste
+
+import no.nav.paw.kafkakeygenerator.client.KafkaKeysResponse
+import no.nav.paw.meldeplikttjeneste.tilstand.InternTilstand
+import no.nav.paw.rapportering.ansvar.v1.AnsvarEndret
+import org.apache.kafka.streams.StreamsBuilder
+import org.apache.kafka.streams.Topology
+import org.apache.kafka.streams.state.KeyValueStore
+import java.util.*
+
+typealias StateStore = KeyValueStore<UUID, InternTilstand>
+
+context(ApplicationConfiguration, ApplicationContext)
+fun StreamsBuilder.appTopology(
+    kafkaKeyFunction: (String) -> KafkaKeysResponse
+): Topology {
+    processPeriodeTopic(kafkaKeyFunction)
+    processAnsvarTopic()
+    processRapporteringsMeldingTopic()
+
+    return build()
+}

apps/bekreftelse-tjeneste/src/main/kotlin/no/nav/paw/meldeplikttjeneste/tilstand/InternTilstand.kt

@@ -0,0 +1,48 @@
+package no.nav.paw.meldeplikttjeneste.tilstand
+
+import no.nav.paw.arbeidssokerregisteret.api.v1.Periode
+import no.nav.paw.rapportering.internehendelser.RapporteringsHendelse
+import java.time.Duration
+import java.time.Instant
+import java.util.UUID
+import kotlin.reflect.KClass
+
+@JvmRecord
+data class InternTilstand(
+    val periode: PeriodeInfo,
+    val utestaaende: List<Rapportering>
+)
+
+@JvmRecord
+data class Rapportering(
+    val sisteHandling: KClass<RapporteringsHendelse>,
+    val rapporteringsId: UUID,
+    val gjelderFra: Instant,
+    val gjelderTil: Instant
+)
+
+
+@JvmRecord
+data class PeriodeInfo(
+    val periodeId: UUID,
+    val identitetsnummer: String,
+    val kafkaKeysId: Long,
+    val recordKey: Long,
+    val avsluttet: Boolean
+)
+
+fun initTilstand(
+    id: Long,
+    key: Long,
+    periode: Periode
+): InternTilstand =
+    InternTilstand(
+        periode = PeriodeInfo(
+            periodeId = periode.id,
+            identitetsnummer = periode.identitetsnummer,
+            kafkaKeysId = id,
+            recordKey = key,
+            avsluttet = periode.avsluttet != null
+        ),
+        utestaaende = emptyList()
+    )