Merge pull request #25 from navikt/ekstern-varsling-status

Ekstern varsling status

Author: chris-santa

build.gradle.kts

@@ -26,6 +26,8 @@ dependencies {
     implementation(Flyway.core)
     implementation(Flyway.postgres)
     implementation(Hikari.cp)
+    implementation(JacksonDatatype.datatypeJsr310)
+    implementation(JacksonDatatype.moduleKotlin)
     implementation(KotlinLogging.logging)
     implementation(Ktor.Server.core)
     implementation(Ktor.Server.netty)
@@ -35,11 +37,12 @@ dependencies {
     implementation(Ktor.Server.statusPages)
     implementation(Ktor.Serialization.jackson)
     implementation(KtorHtml.htmlBuilder)
+    implementation(Logstash.logbackEncoder)
     implementation(TmsCommonLib.utils)
     implementation(TmsKtorTokenSupport.azureValidation)
     implementation(Postgresql.postgresql)
-    implementation(RapidsAndRivers.rapidsAndRivers)
     implementation(KotliQuery.kotliquery)
+    implementation(TmsKafkaTools.kafkaApplication)
 
     testImplementation(Junit.api)
     testImplementation(Junit.engine)

buildSrc/src/main/kotlin/default/dependencies.kt

@@ -228,7 +228,7 @@ interface TmsCommonLibDefaults: DependencyGroup {
 
 interface TmsKafkaToolsDefaults: DependencyGroup {
     override val groupId get() = "no.nav.tms.kafka"
-    override val version get() = "1.4.4"
+    override val version get() = "1.5.0-onReady-5"
 
     val kafkaApplication get() = dependency("kafka-application")
 }

src/main/kotlin/no/nav/tms/statistikk/api/StatistikkContent.kt

@@ -25,10 +25,10 @@ fun HTML.buildStats() {
     }
 }
 
-internal fun OutputStream.writeInnloggingCSV(innlogingPerDag: Int) =
+fun OutputStream.writeInnloggingCSV(innlogingPerDag: Int) =
     bufferedWriter().apply {
         write("""Innlogging etter ekstern varsling,$innlogingPerDag""")
         flush()
     }
 
-class StatistikkContentException(message: String) : Exception(message)
+class StatistikkContentException(message: String) : Exception(message)

src/main/kotlin/no/nav/tms/statistikk/api/routes.kt

@@ -8,7 +8,7 @@ import io.ktor.server.routing.*
 import no.nav.tms.statistikk.login.LoginRepository
 
 
-internal fun Routing.statistikk(loginRepository: LoginRepository) {
+fun Routing.statistikk(loginRepository: LoginRepository) {
     route("/hent") {
 
         get {

src/main/kotlin/no/nav/tms/statistikk/application.kt

@@ -1,18 +1,14 @@
 package no.nav.tms.statistikk
 
-import no.nav.helse.rapids_rivers.RapidApplication
-import no.nav.helse.rapids_rivers.RapidApplication.RapidApplicationConfig.Companion.fromEnv
-import no.nav.helse.rapids_rivers.RapidsConnection
+import no.nav.tms.kafka.application.KafkaApplication
 import no.nav.tms.statistikk.database.Flyway
 import no.nav.tms.statistikk.database.PostgresDatabase
 import no.nav.tms.statistikk.eksternVarsling.EksternVarslingRepository
-import no.nav.tms.statistikk.eksternVarsling.EksternVarslingSink
+import no.nav.tms.statistikk.eksternVarsling.EksternVarslingSubscriber
 import no.nav.tms.statistikk.login.LoginRepository
 import no.nav.tms.statistikk.microfrontends.MicrofrontendRepository
-import no.nav.tms.statistikk.microfrontends.MicrofrontendSink
-import no.nav.tms.statistikk.utkast.UtkastRespository
-import no.nav.tms.statistikk.utkast.UtkastCreatedSink
-import no.nav.tms.statistikk.utkast.UtkastDeletedSink
+import no.nav.tms.statistikk.microfrontends.MicrofrontendSubscriber
+import no.nav.tms.statistikk.utkast.*
 import no.nav.tms.statistikk.varsel.*
 
 fun main() {
@@ -32,21 +28,33 @@ private fun startRapid(
     val eksternVarslingRepository = EksternVarslingRepository(database)
     val utkastRespository = UtkastRespository(database)
 
-    RapidApplication.Builder(fromEnv(environment.rapidConfig())).withKtorModule {
-        statistikkApi(loginRepository)
-    }.build().apply {
-        VarselAktivertSink(this, varselRepository)
-        VarselInaktivertSink(this, varselRepository)
-        VarselPerDagSink(this, varselRepository)
-        EksternVarslingSink(this, eksternVarslingRepository)
-        UtkastCreatedSink(this, utkastRespository)
-        UtkastDeletedSink(this,utkastRespository)
-        MicrofrontendSink(this, MicrofrontendRepository(database))
-    }.apply {
-        register(object : RapidsConnection.StatusListener {
-            override fun onStartup(rapidsConnection: RapidsConnection) {
-                Flyway.runFlywayMigrations(environment)
-            }
-        })
+    KafkaApplication.build {
+        kafkaConfig {
+            groupId = environment.groupId
+            readTopics(
+                "min-side.brukervarsel-v1",
+                "min-side.aapen-utkast-v1",
+                "min-side.aapen-microfrontend-v1"
+            )
+            eventNameFields("@event_name", "@action")
+        }
+
+        ktorModule {
+            statistikkApi(loginRepository)
+        }
+
+        subscribers(
+            VarselAktivertSubscriber(varselRepository),
+            VarselInaktivertSubscriber(varselRepository),
+            VarselPerDagSubscriber(varselRepository),
+            EksternVarslingSubscriber(eksternVarslingRepository),
+            UtkastCreatedSubscriber(utkastRespository),
+            UtkastDeletedSubscriber(utkastRespository),
+            MicrofrontendSubscriber(MicrofrontendRepository(database))
+        )
+
+        onStartup {
+            Flyway.runFlywayMigrations(environment)
+        }
     }.start()
 }

src/main/kotlin/no/nav/tms/statistikk/eksternVarsling/EksternVarslingSink.kt

@@ -0,0 +1,45 @@
+package no.nav.tms.statistikk.eksternVarsling
+
+import no.nav.tms.kafka.application.JsonMessage
+import no.nav.tms.kafka.application.Subscriber
+import no.nav.tms.kafka.application.Subscription
+import no.nav.tms.statistikk.asUtcDateTime
+import java.time.LocalDateTime
+
+
+class EksternVarslingSubscriber(
+    private val eksternVarslingRepository: EksternVarslingRepository
+) : Subscriber() {
+
+    override fun subscribe() = Subscription.forEvents("eksternStatusOppdatert", "eksternVarslingStatusOppdatert")
+        .withFields(
+            "kanal",
+            "varselId",
+            "ident",
+            "tidspunkt"
+        )
+        .withValue("status", "sendt")
+
+    override suspend fun receive(jsonMessage: JsonMessage) {
+        eksternVarslingRepository.insertEksternVarsling(
+            eventId = jsonMessage.varselId,
+            kanal = jsonMessage.kanal,
+            ident = jsonMessage.ident,
+            tidspunkt = jsonMessage.tidspunkt
+        )
+        eksternVarslingRepository.updateVarsel(jsonMessage.varselId, jsonMessage.kanal)
+    }
+}
+
+val JsonMessage.ident: String
+    get() = get("ident").asText()
+val JsonMessage.kanal: String
+    get() = get("kanal").asText()
+val JsonMessage.varselId: String
+    get() = get("varselId").asText()
+val JsonMessage.tidspunkt: LocalDateTime
+    get() = get("tidspunkt").asUtcDateTime()
+
+
+
+

src/main/kotlin/no/nav/tms/statistikk/eksternVarsling/EksternVarslingSubscriber.kt

@@ -0,0 +1,28 @@
+package no.nav.tms.statistikk.microfrontends
+
+import no.nav.tms.kafka.application.JsonMessage
+import no.nav.tms.kafka.application.Subscriber
+import no.nav.tms.kafka.application.Subscription
+
+class MicrofrontendSubscriber(
+    private val microfrontendRepository: MicrofrontendRepository
+) : Subscriber() {
+
+    override fun subscribe() = Subscription.forAllEvents()
+        .withFields(
+            "@action",
+            "ident",
+            "microfrontend_id"
+        )
+        .withOptionalFields("@initiated_by")
+
+
+    override suspend fun receive(jsonMessage: JsonMessage) {
+        microfrontendRepository.insertMicrofrontend(
+            action = jsonMessage["@action"].asText(),
+            microfrontendId = jsonMessage["microfrontend_id"].asText(),
+            ident = jsonMessage["ident"].asText(),
+            initiatedBy = jsonMessage.getOrNull("@initiated_by")?.textValue()
+        )
+    }
+}