Rapid

build.gradle.kts

@@ -1,6 +1,6 @@
 plugins {
     application
-    kotlin("jvm") version "1.9.21"
+    kotlin("jvm") version "2.2.21"
 }
 
 group = "no.nav"
@@ -42,6 +42,8 @@ dependencies {
     implementation("org.postgresql:postgresql:42.7.3")
     implementation("org.flywaydb:flyway-core:10.10.0")
     implementation("org.flywaydb:flyway-database-postgresql:10.10.0")
+    implementation("com.github.navikt:rapids-and-rivers:2025110410541762250064.d7e58c3fad81")
+
 
     testImplementation(platform("org.junit:junit-bom:5.10.2"))
     testImplementation("org.junit.jupiter:junit-jupiter")
@@ -52,6 +54,7 @@ dependencies {
     testImplementation("org.wiremock:wiremock:3.3.1")
     testImplementation("org.skyscreamer:jsonassert:1.5.1")
     testImplementation("org.testcontainers:postgresql:1.19.7")
+    testImplementation("io.mockk:mockk:1.13.8")
 }
 
 tasks.test {

src/main/kotlin/no/nav/toi/kandidatvarsel/Javalin.kt

@@ -4,6 +4,7 @@ import auth.obo.KandidatsokApiKlient
 import com.fasterxml.jackson.databind.SerializationFeature
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule
 import com.fasterxml.jackson.module.kotlin.jacksonObjectMapper
+import com.github.navikt.tbd_libs.rapids_and_rivers.KafkaRapid
 import com.zaxxer.hikari.HikariDataSource
 import io.javalin.Javalin
 import io.javalin.http.HttpStatus
@@ -18,6 +19,7 @@ fun startJavalin(
     dataSource: HikariDataSource,
     migrateResult: AtomicReference<MigrateResult>,
     kandidatsokApiKlient: KandidatsokApiKlient,
+    kafkaRapid: KafkaRapid,
     port: Int = 8080,
 ): Javalin = Javalin
     .create {
@@ -37,7 +39,7 @@ fun startJavalin(
     }
     .apply {
         azureAdAuthentication(azureAdConfig)
-        handleHealth(dataSource, migrateResult)
+        handleHealth(dataSource, migrateResult, kafkaRapid)
         handleVarsler(dataSource, kandidatsokApiKlient)
         handleMeldingsmal()
 

src/main/kotlin/no/nav/toi/kandidatvarsel/KubernetesHealthApi.kt

@@ -1,5 +1,6 @@
 package no.nav.toi.kandidatvarsel
 
+import com.github.navikt.tbd_libs.rapids_and_rivers.KafkaRapid
 import io.javalin.Javalin
 import io.javalin.http.Handler
 import io.javalin.http.HttpStatus
@@ -8,21 +9,27 @@ import org.flywaydb.core.api.output.MigrateResult
 import java.util.concurrent.atomic.AtomicReference
 import javax.sql.DataSource
 
-fun Javalin.handleHealth(dataSource: DataSource, migrationResult: AtomicReference<MigrateResult>) {
+fun Javalin.handleHealth(
+    dataSource: DataSource,
+    migrationResult: AtomicReference<MigrateResult>,
+    kafkaRapid: KafkaRapid
+) {
     fun checks(checks: Map<String, () -> Boolean>) = Handler { ctx ->
         val checkOutcomes = checks.mapValues { it.value() }
         val httpStatus = if (checkOutcomes.all { it.value }) HttpStatus.OK else HttpStatus.SERVICE_UNAVAILABLE
         ctx.status(httpStatus).json(checkOutcomes)
     }
 
-    val isReadyChecks = mapOf(
-        "database" to { dataSource.isReady() },
-        "migration" to { migrationResult.get()?.success == true }
-    )
+    val isReadyChecks = buildMap {
+        put("database") { dataSource.isReady() }
+        put("migration") { migrationResult.get()?.success == true }
+        put("rapid") { kafkaRapid.isRunning() }
+    }
 
-    val isAliveChecks = mapOf(
-        "migration" to { migrationResult.get()?.success != false }
-    )
+    val isAliveChecks = buildMap {
+        put("migration") { migrationResult.get()?.success != false }
+        put("rapid") { kafkaRapid.isRunning() }
+    }
 
     get("/internal/ready", checks(isReadyChecks), UNPROTECTED)
     get("/internal/alive", checks(isAliveChecks), UNPROTECTED)

src/main/kotlin/no/nav/toi/kandidatvarsel/Main.kt

@@ -2,103 +2,180 @@ package no.nav.toi.kandidatvarsel
 
 import auth.obo.KandidatsokApiKlient
 import auth.obo.OnBehalfOfTokenClient
+import com.github.navikt.tbd_libs.rapids_and_rivers.KafkaRapid
+import com.github.navikt.tbd_libs.rapids_and_rivers_api.RapidsConnection
+import com.zaxxer.hikari.HikariDataSource
+import no.nav.helse.rapids_rivers.RapidApplication
 import no.nav.toi.kandidatvarsel.minside.bestillVarsel
 import no.nav.toi.kandidatvarsel.minside.sjekkVarselOppdateringer
+import no.nav.toi.kandidatvarsel.rapids.lyttere.InvitertTreffKandidatEndretLytter
+import no.nav.toi.kandidatvarsel.rapids.lyttere.KandidatInvitertLytter
 import org.flywaydb.core.api.output.MigrateResult
 import org.slf4j.Logger
 import org.slf4j.LoggerFactory
 import java.lang.System.getenv
 import java.util.concurrent.atomic.AtomicBoolean
 import java.util.concurrent.atomic.AtomicReference
 import kotlin.concurrent.thread
+import kotlin.system.exitProcess
 import kotlin.time.Duration.Companion.milliseconds
 import kotlin.time.Duration.Companion.seconds
 
 private val log = LoggerFactory.getLogger("no.nav.toi.kandidatvarsel.Main")!!
 
 fun main() {
-    val dataSource = DatabaseConfig.nais().createDataSource()
-
-    /* Status på migrering, så ready-endepunktet kan fortelle om vi er klare for å motta api-kall. */
-    val migrateResult = AtomicReference<MigrateResult>()
-
-
-
-    while (!dataSource.isReady())  {
-        log.info("Database not ready. Sleeping")
-        Thread.sleep(100.milliseconds.inWholeMilliseconds)
+    log.info("Starter applikasjon")
+
+    try {
+        lateinit var kafkaRapid: KafkaRapid
+        val rapidsConnection = RapidApplication.create(
+            System.getenv(),
+            builder = { withHttpPort(9000) },
+            configure = { _, rapid ->
+                kafkaRapid = rapid
+            }
+        )
+
+        val dataSource = DatabaseConfig.nais().createDataSource()
+
+        startOppApplikasjon(
+            rapidsConnection = rapidsConnection,
+            dataSource = dataSource,
+            kafkaRapid = kafkaRapid
+        )
+    } catch (e: Exception) {
+        secureLog.error("Uhåndtert exception, stanser applikasjonen", e)
+        log.error("Uhåndtert exception, stanser applikasjonen (se securelog)")
+        exitProcess(1)
     }
-    migrateResult.set(dataSource.migrate())
+}
 
-    val shutdown = AtomicBoolean(false)
+fun startOppApplikasjon(
+    rapidsConnection: RapidsConnection,
+    dataSource: HikariDataSource,
+    kafkaRapid: KafkaRapid
+) {
+    val migreringsResultat = AtomicReference<MigrateResult>()
+    val avsluttSignal = AtomicBoolean(false)
 
-    val kafkaConfig = KafkaConfig.nais()
+    ventPåDatabase(dataSource)
+    migreringsResultat.set(dataSource.migrate())
 
+    val kafkaConfig = KafkaConfig.nais()
     val minsideBestillingProducer = kafkaConfig.minsideBestillingsProducer()
     val minsideOppdateringConsumer = kafkaConfig.minsideOppdateringsConsumer()
 
-    val azureTokenClient = AzureTokenClient(
-        tokenEndpoint = getenvOrThrow("AZURE_OPENID_CONFIG_TOKEN_ENDPOINT"),
-        clientId = getenvOrThrow("AZURE_APP_CLIENT_ID"),
-        clientSecret = getenvOrThrow("AZURE_APP_CLIENT_SECRET"),
-        scope = "api://${getenv("NAIS_CLUSTER_NAME")}.toi.rekrutteringsbistand-stilling-api/.default"
-    )
-
-    val onBehalfOfTokenClient = OnBehalfOfTokenClient(
-        tokenEndpoint = getenvOrThrow("AZURE_OPENID_CONFIG_TOKEN_ENDPOINT"),
-        clientId = getenvOrThrow("AZURE_APP_CLIENT_ID"),
-        clientSecret = getenvOrThrow("AZURE_APP_CLIENT_SECRET"),
-        scope = "api://${getenv("NAIS_CLUSTER_NAME")}.toi.rekrutteringsbistand-kandidatsok-api/.default",
-        issuernavn = getenvOrThrow("AZURE_OPENID_CONFIG_ISSUER")
-    )
-
-
-
+    val azureTokenClient = opprettAzureTokenClient()
+    val onBehalfOfTokenClient = opprettOnBehalfOfTokenClient()
     val stillingClient = StillingClientImpl(azureTokenClient)
     val kandidatsokApiKlient = KandidatsokApiKlient(onBehalfOfTokenClient, getenvOrThrow("KANDIDATSOK_API_URL"))
 
-    val minsideBestillingThread = backgroundThread("minside-utsending", shutdown) {
+    val minsideBestillingThread = backgroundThread("minside-utsending", avsluttSignal) {
         if (!bestillVarsel(dataSource, stillingClient, minsideBestillingProducer)) {
             Thread.sleep(1.seconds.inWholeMilliseconds)
         }
     }
 
-    val minsideOppdateringThread = backgroundThread(name = "minside-oppdatering", shutdown) {
+    val minsideOppdateringThread = backgroundThread("minside-oppdatering", avsluttSignal) {
         sjekkVarselOppdateringer(dataSource, minsideOppdateringConsumer)
     }
 
     val javalin = startJavalin(
         azureAdConfig = AzureAdConfig.nais(),
         dataSource = dataSource,
-        migrateResult = migrateResult,
-        kandidatsokApiKlient = kandidatsokApiKlient
+        migrateResult = migreringsResultat,
+        kandidatsokApiKlient = kandidatsokApiKlient,
+        kafkaRapid = kafkaRapid
+    )
+
+    registrerRapidsLyttere(rapidsConnection, dataSource)
+    rapidsConnection.start()
+    log.info("RapidApplication startet")
+
+    registrerShutdownHook(
+        avsluttSignal = avsluttSignal,
+        rapidsConnection = rapidsConnection,
+        minsideBestillingThread = minsideBestillingThread,
+        minsideBestillingProducer = minsideBestillingProducer,
+        minsideOppdateringThread = minsideOppdateringThread,
+        minsideOppdateringConsumer = minsideOppdateringConsumer,
+        javalin = javalin,
+        dataSource = dataSource
     )
+}
 
+private fun ventPåDatabase(dataSource: HikariDataSource) {
+    while (!dataSource.isReady()) {
+        log.info("Database er ikke klar. Venter...")
+        Thread.sleep(100.milliseconds.inWholeMilliseconds)
+    }
+}
+
+private fun opprettAzureTokenClient() = AzureTokenClient(
+    tokenEndpoint = getenvOrThrow("AZURE_OPENID_CONFIG_TOKEN_ENDPOINT"),
+    clientId = getenvOrThrow("AZURE_APP_CLIENT_ID"),
+    clientSecret = getenvOrThrow("AZURE_APP_CLIENT_SECRET"),
+    scope = "api://${getenv("NAIS_CLUSTER_NAME")}.toi.rekrutteringsbistand-stilling-api/.default"
+)
+
+private fun opprettOnBehalfOfTokenClient() = OnBehalfOfTokenClient(
+    tokenEndpoint = getenvOrThrow("AZURE_OPENID_CONFIG_TOKEN_ENDPOINT"),
+    clientId = getenvOrThrow("AZURE_APP_CLIENT_ID"),
+    clientSecret = getenvOrThrow("AZURE_APP_CLIENT_SECRET"),
+    scope = "api://${getenv("NAIS_CLUSTER_NAME")}.toi.rekrutteringsbistand-kandidatsok-api/.default",
+    issuernavn = getenvOrThrow("AZURE_OPENID_CONFIG_ISSUER")
+)
+
+private fun registrerRapidsLyttere(rapidsConnection: RapidsConnection, dataSource: HikariDataSource) {
+    try {
+        KandidatInvitertLytter(rapidsConnection, dataSource)
+        InvitertTreffKandidatEndretLytter(rapidsConnection, dataSource)
+    } catch (e: Exception) {
+        log.error("Feil ved oppstart av RapidApplication (se securelog)")
+        secureLog.error("Feil ved oppstart av RapidApplication", e)
+        throw e
+    }
+}
+
+private fun registrerShutdownHook(
+    avsluttSignal: AtomicBoolean,
+    rapidsConnection: RapidsConnection,
+    minsideBestillingThread: Thread,
+    minsideBestillingProducer: org.apache.kafka.clients.producer.KafkaProducer<String, String>,
+    minsideOppdateringThread: Thread,
+    minsideOppdateringConsumer: org.apache.kafka.clients.consumer.KafkaConsumer<String, String>,
+    javalin: io.javalin.Javalin,
+    dataSource: HikariDataSource
+) {
     Runtime.getRuntime().addShutdownHook(Thread {
-        shutdown.set(true)
+        log.info("Shutdownhook kjører")
+        avsluttSignal.set(true)
+        rapidsConnection.stop()
         minsideBestillingThread.join()
         minsideBestillingProducer.close()
         minsideOppdateringThread.join()
         minsideOppdateringConsumer.close()
         javalin.stop()
         dataSource.close()
-        log.info("Shutdownhook ran successfully to completion")
+        log.info("Shutdownhook fullført")
     })
 }
 
-private fun backgroundThread(name: String, shutdown: AtomicBoolean, body: () -> Unit): Thread = thread(name = name) {
-    while (!shutdown.get()) {
-        try {
-            body()
-        } catch (e: Exception) {
-            log.error("Exception in background thread $name (Se secure log)")
-            secureLog.error("Exception in background thread $name", e)
-            Thread.sleep(1.seconds.inWholeMilliseconds)
+private fun backgroundThread(navn: String, avsluttSignal: AtomicBoolean, oppgave: () -> Unit): Thread =
+    thread(name = navn) {
+        while (!avsluttSignal.get()) {
+            try {
+                oppgave()
+            } catch (e: Exception) {
+                log.error("Exception i bakgrunnsThread $navn (se secure log)")
+                secureLog.error("Exception i bakgrunnsThread $navn", e)
+                Thread.sleep(1.seconds.inWholeMilliseconds)
+            }
         }
     }
-}
 
 val Any.log: Logger
     get() = LoggerFactory.getLogger(this::class.java)
 
-fun getenvOrThrow(name: String): String = getenv(name) ?: throw IllegalStateException("Mangler miljøvariabel '$name'")
+fun getenvOrThrow(name: String): String = 
+    getenv(name) ?: throw IllegalStateException("Mangler miljøvariabel '$name'")