Endret bekreftelse-api til å benytte database

apps/api-start-stopp-perioder/build.gradle.kts

@@ -39,7 +39,7 @@ dependencies {
     implementation(libs.paw.pdl.client)
     implementation(libs.logbackClassic)
     implementation(libs.logstashLogbackEncoder)
-    implementation(libs.paw.kafkaClients)
+    implementation(libs.kafka.clients)
     implementation(libs.ktor.client.contentNegotiation)
     implementation(libs.ktor.client.core)
     implementation(libs.ktor.client.cio)

apps/bekreftelse-api/build.gradle.kts

@@ -59,15 +59,24 @@ dependencies {
     implementation(libs.opentelemetry.api)
     implementation(libs.opentelemetry.annotations)
 
+    // Database
+    implementation(libs.exposed.jdbc)
+    implementation(libs.exposed.json)
+    implementation(libs.exposed.javaTime)
+    implementation(libs.database.hikari.connectionPool)
+    implementation(libs.database.postgres.driver)
+    implementation(libs.database.flyway.postgres)
+
     // Kafka
-    implementation(libs.kafka.streams.core)
+    implementation(libs.kafka.clients)
     implementation(libs.avro.kafkaStreamsSerde)
 
     // Test
     testImplementation(libs.ktor.server.testJvm)
     testImplementation(libs.ktor.client.mock)
     testImplementation(libs.bundles.testLibsWithUnitTesting)
     testImplementation(libs.test.mockOauth2Server)
+    testImplementation(libs.test.testContainers.postgresql)
 }
 
 java {

apps/bekreftelse-api/nais/nais-dev.yaml

@@ -21,8 +21,6 @@ spec:
       value: "paw.arbeidssoker-bekreftelse-hendelseslogg-beta-v2"
     - name: CORS_ALLOW_ORIGINS
       value: "www.intern.dev.nav.no"
-  ingresses:
-    - https://bekreftelse-arbeidssoekerregisteret.intern.dev.nav.no
   replicas:
     min: 2
     max: 2
@@ -41,10 +39,6 @@ spec:
       claims:
         extra:
           - NAVident
-  idporten:
-    enabled: true
-    sidecar:
-      enabled: true
   liveness:
     path: /internal/isAlive
     initialDelay: 10
@@ -61,6 +55,12 @@ spec:
   kafka:
     pool: nav-dev
     streams: true
+  gcp:
+    sqlInstances:
+      - type: POSTGRES_16
+        tier: db-f1-micro
+        databases:
+          - name: bekreftelser
   accessPolicy:
     inbound:
       rules:

apps/bekreftelse-api/nais/nais-prod.yaml

@@ -39,10 +39,6 @@ spec:
       claims:
         extra:
           - NAVident
-  idporten:
-    enabled: true
-    sidecar:
-      enabled: true
   liveness:
     path: /internal/isAlive
     initialDelay: 10
@@ -59,6 +55,12 @@ spec:
   kafka:
     pool: nav-prod
     streams: true
+  gcp:
+    sqlInstances:
+      - type: POSTGRES_16
+        tier: db-custom-1-6144
+        databases:
+          - name: bekreftelser
   accessPolicy:
     inbound:
       rules:

apps/bekreftelse-api/src/main/kotlin/no/nav/paw/bekreftelse/api/Application.kt

@@ -7,6 +7,7 @@ import io.ktor.server.netty.Netty
 import io.ktor.server.routing.routing
 import no.nav.paw.bekreftelse.api.context.ApplicationContext
 import no.nav.paw.bekreftelse.api.plugins.configureAuthentication
+import no.nav.paw.bekreftelse.api.plugins.configureDatabase
 import no.nav.paw.bekreftelse.api.plugins.configureHTTP
 import no.nav.paw.bekreftelse.api.plugins.configureKafka
 import no.nav.paw.bekreftelse.api.plugins.configureLogging
@@ -49,6 +50,7 @@ fun Application.module(applicationContext: ApplicationContext) {
     configureLogging()
     configureSerialization()
     configureTracing()
+    configureDatabase(applicationContext)
     configureKafka(applicationContext)
 
     routing {

apps/bekreftelse-api/src/main/kotlin/no/nav/paw/bekreftelse/api/config/ApplicationConfig.kt

@@ -3,31 +3,25 @@ package no.nav.paw.bekreftelse.api.config
 import no.nav.paw.config.kafka.KafkaConfig
 import no.nav.paw.kafkakeygenerator.auth.AzureM2MConfig
 import no.nav.paw.kafkakeygenerator.client.KafkaKeyConfig
+import java.time.Duration
 
 const val APPLICATION_CONFIG_FILE_NAME = "application_config.toml"
 
 data class ApplicationConfig(
     val autorisasjon: AutorisasjonConfig,
-    val kafkaTopology: KafkaTopologyConfig,
-    val authProviders: AuthProviders,
+    val authProviders: List<AuthProvider>,
     val azureM2M: AzureM2MConfig,
     val poaoClientConfig: ServiceClientConfig,
     val kafkaKeysClient: KafkaKeyConfig,
-    val kafkaClients: KafkaConfig
+    val kafkaClients: KafkaConfig,
+    val kafkaTopology: KafkaTopologyConfig,
+    val database: DatabaseConfig
 )
 
 data class AutorisasjonConfig(
     val corsAllowOrigins: String? = null
 )
 
-data class KafkaTopologyConfig(
-    val applicationIdSuffix: String,
-    val producerId: String,
-    val bekreftelseTopic: String,
-    val bekreftelseHendelsesloggTopic: String,
-    val internStateStoreName: String
-)
-
 data class ServiceClientConfig(
     val url: String,
     val scope: String
@@ -40,9 +34,27 @@ data class AuthProvider(
     val claims: AuthProviderClaims
 )
 
-typealias AuthProviders = List<AuthProvider>
-
 data class AuthProviderClaims(
     val map: List<String>,
     val combineWithOr: Boolean = false
+)
+
+data class KafkaTopologyConfig(
+    val version: Int,
+    val antallPartitioner: Int,
+    val producerId: String,
+    val consumerId: String,
+    val consumerGroupId: String,
+    val bekreftelseTopic: String,
+    val bekreftelseHendelsesloggTopic: String
+)
+
+data class DatabaseConfig(
+    val jdbcUrl: String,
+    val driverClassName: String,
+    val autoCommit: Boolean = false,
+    val maxPoolSize: Int = 10,
+    val connectionTimeout: Duration = Duration.ofSeconds(30),
+    val idleTimeout: Duration = Duration.ofMinutes(10),
+    val maxLifetime: Duration = Duration.ofMinutes(30)
 )

apps/bekreftelse-api/src/main/kotlin/no/nav/paw/bekreftelse/api/context/ApplicationContext.kt

@@ -1,39 +1,48 @@
 package no.nav.paw.bekreftelse.api.context
 
-import io.ktor.client.HttpClient
-import io.ktor.client.plugins.HttpResponseValidator
-import io.ktor.client.plugins.contentnegotiation.ContentNegotiation
-import io.ktor.serialization.jackson.jackson
 import io.micrometer.prometheusmetrics.PrometheusConfig
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
 import no.nav.paw.bekreftelse.api.config.APPLICATION_CONFIG_FILE_NAME
 import no.nav.paw.bekreftelse.api.config.ApplicationConfig
 import no.nav.paw.bekreftelse.api.config.SERVER_CONFIG_FILE_NAME
 import no.nav.paw.bekreftelse.api.config.ServerConfig
-import no.nav.paw.bekreftelse.api.consumer.BekreftelseHttpConsumer
-import no.nav.paw.bekreftelse.api.plugins.buildKafkaStreams
+import no.nav.paw.bekreftelse.api.handler.KafkaConsumerExceptionHandler
 import no.nav.paw.bekreftelse.api.producer.BekreftelseKafkaProducer
+import no.nav.paw.bekreftelse.api.repository.BekreftelseRepository
 import no.nav.paw.bekreftelse.api.services.AuthorizationService
 import no.nav.paw.bekreftelse.api.services.BekreftelseService
-import no.nav.paw.bekreftelse.api.topology.buildBekreftelseTopology
-import no.nav.paw.bekreftelse.api.utils.configureJackson
-import no.nav.paw.bekreftelse.api.utils.handleError
+import no.nav.paw.bekreftelse.api.utils.BekreftelseAvroSerializer
+import no.nav.paw.bekreftelse.api.utils.createDataSource
+import no.nav.paw.bekreftelse.internehendelser.BekreftelseHendelse
+import no.nav.paw.bekreftelse.internehendelser.BekreftelseHendelseDeserializer
+import no.nav.paw.bekreftelse.melding.v1.Bekreftelse
 import no.nav.paw.config.hoplite.loadNaisOrLocalConfiguration
+import no.nav.paw.config.kafka.KafkaFactory
+import no.nav.paw.health.model.HealthStatus
+import no.nav.paw.health.model.LivenessHealthIndicator
+import no.nav.paw.health.model.ReadinessHealthIndicator
 import no.nav.paw.health.repository.HealthIndicatorRepository
 import no.nav.paw.kafkakeygenerator.auth.azureAdM2MTokenClient
 import no.nav.paw.kafkakeygenerator.client.KafkaKeysClient
 import no.nav.paw.kafkakeygenerator.client.kafkaKeysClient
 import no.nav.poao_tilgang.client.PoaoTilgangCachedClient
 import no.nav.poao_tilgang.client.PoaoTilgangHttpClient
-import org.apache.kafka.streams.KafkaStreams
+import org.apache.kafka.clients.consumer.KafkaConsumer
+import org.apache.kafka.clients.producer.Producer
+import org.apache.kafka.common.serialization.LongDeserializer
+import org.apache.kafka.common.serialization.LongSerializer
+import javax.sql.DataSource
 
 data class ApplicationContext(
     val serverConfig: ServerConfig,
     val applicationConfig: ApplicationConfig,
+    val dataSource: DataSource,
     val kafkaKeysClient: KafkaKeysClient,
     val prometheusMeterRegistry: PrometheusMeterRegistry,
     val healthIndicatorRepository: HealthIndicatorRepository,
-    val bekreftelseKafkaStreams: KafkaStreams,
+    val bekreftelseKafkaProducer: Producer<Long, Bekreftelse>,
+    val bekreftelseKafkaConsumer: KafkaConsumer<Long, BekreftelseHendelse>,
+    val kafkaConsumerExceptionHandler: KafkaConsumerExceptionHandler,
     val authorizationService: AuthorizationService,
     val bekreftelseService: BekreftelseService
 ) {
@@ -42,6 +51,8 @@ data class ApplicationContext(
             val serverConfig = loadNaisOrLocalConfiguration<ServerConfig>(SERVER_CONFIG_FILE_NAME)
             val applicationConfig = loadNaisOrLocalConfiguration<ApplicationConfig>(APPLICATION_CONFIG_FILE_NAME)
 
+            val dataSource = createDataSource(applicationConfig.database)
+
             val azureM2MTokenClient = azureAdM2MTokenClient(
                 serverConfig.runtimeEnvironment, applicationConfig.azureM2M
             )
@@ -54,17 +65,6 @@ data class ApplicationContext(
 
             val healthIndicatorRepository = HealthIndicatorRepository()
 
-            val httpClient = HttpClient {
-                install(ContentNegotiation) {
-                    jackson {
-                        configureJackson()
-                    }
-                }
-                HttpResponseValidator {
-                    validateResponse(::handleError)
-                }
-            }
-
             val poaoTilgangClient = PoaoTilgangCachedClient(
                 PoaoTilgangHttpClient(
                     baseUrl = applicationConfig.poaoClientConfig.url,
@@ -79,33 +79,48 @@ data class ApplicationContext(
                 poaoTilgangClient
             )
 
-            val bekreftelseTopology = buildBekreftelseTopology(applicationConfig, prometheusMeterRegistry)
-            val bekreftelseKafkaStreams = buildKafkaStreams(
-                serverConfig,
-                applicationConfig,
-                healthIndicatorRepository,
-                bekreftelseTopology
+            val kafkaConsumerExceptionHandler = KafkaConsumerExceptionHandler(
+                healthIndicatorRepository.addLivenessIndicator(LivenessHealthIndicator(HealthStatus.HEALTHY)),
+                healthIndicatorRepository.addReadinessIndicator(ReadinessHealthIndicator(HealthStatus.HEALTHY))
             )
 
-            val bekreftelseKafkaProducer = BekreftelseKafkaProducer(applicationConfig, prometheusMeterRegistry)
+            val kafkaFactory = KafkaFactory(applicationConfig.kafkaClients)
 
-            val bekreftelseHttpConsumer = BekreftelseHttpConsumer(httpClient)
+            val kafkaProducer = kafkaFactory.createProducer<Long, Bekreftelse>(
+                clientId = applicationConfig.kafkaTopology.producerId,
+                keySerializer = LongSerializer::class,
+                valueSerializer = BekreftelseAvroSerializer::class
+            )
+
+            val kafkaConsumer = kafkaFactory.createConsumer(
+                clientId = applicationConfig.kafkaTopology.consumerId,
+                groupId = applicationConfig.kafkaTopology.consumerGroupId,
+                keyDeserializer = LongDeserializer::class,
+                valueDeserializer = BekreftelseHendelseDeserializer::class,
+                autoCommit = false
+            )
+
+            val bekreftelseKafkaProducer = BekreftelseKafkaProducer(applicationConfig, kafkaProducer)
+            val bekreftelseRepository = BekreftelseRepository()
 
             val bekreftelseService = BekreftelseService(
                 serverConfig,
                 applicationConfig,
-                bekreftelseHttpConsumer,
-                bekreftelseKafkaStreams,
-                bekreftelseKafkaProducer
+                prometheusMeterRegistry,
+                bekreftelseKafkaProducer,
+                bekreftelseRepository
             )
 
             return ApplicationContext(
                 serverConfig,
                 applicationConfig,
+                dataSource,
                 kafkaKeysClient,
                 prometheusMeterRegistry,
                 healthIndicatorRepository,
-                bekreftelseKafkaStreams,
+                kafkaProducer,
+                kafkaConsumer,
+                kafkaConsumerExceptionHandler,
                 authorizationService,
                 bekreftelseService
             )