#190 Registrere hendelser i cpa-repo

.nais/cpa-repo-dev.yaml

@@ -57,6 +57,8 @@ spec:
         mountPath: /secrets/oracle/creds
       - kvPath: /oracle/data/dev/config/emottak_q1
         mountPath: /secrets/oracle/config
+  kafka:
+    pool: nav-dev
   webproxy: true
   accessPolicy:
     inbound:

cpa-repo/build.gradle.kts

@@ -35,6 +35,12 @@ tasks {
     }
 }
 
+tasks.withType<org.jetbrains.kotlin.gradle.tasks.KotlinCompile>().all {
+    compilerOptions {
+        freeCompilerArgs = listOf("-opt-in=kotlin.uuid.ExperimentalUuidApi,com.sksamuel.hoplite.ExperimentalHoplite")
+    }
+}
+
 dependencies {
     api(project(":felles"))
     api(project(":ebxml-processing-model"))
@@ -53,6 +59,8 @@ dependencies {
     implementation(libs.bundles.exposed)
     implementation(libs.bundles.logging)
     implementation(libs.bundles.prometheus)
+    implementation(libs.hoplite.core)
+    implementation(libs.hoplite.hocon)
     implementation(libs.ktor.serialization.kotlinx.json)
     implementation(libs.ktor.client.core)
     implementation(libs.ktor.client.cio)

cpa-repo/src/main/kotlin/no/nav/emottak/cpa/App.kt

@@ -22,13 +22,18 @@ import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
 import kotlinx.coroutines.runBlocking
 import no.nav.emottak.cpa.auth.AZURE_AD_AUTH
 import no.nav.emottak.cpa.auth.AuthConfig
+import no.nav.emottak.cpa.configuration.config
 import no.nav.emottak.cpa.persistence.CPARepository
 import no.nav.emottak.cpa.persistence.Database
 import no.nav.emottak.cpa.persistence.cpaDbConfig
 import no.nav.emottak.cpa.persistence.cpaMigrationConfig
 import no.nav.emottak.cpa.persistence.gammel.PartnerRepository
 import no.nav.emottak.cpa.persistence.oracleConfig
+import no.nav.emottak.cpa.util.EventRegistrationService
+import no.nav.emottak.cpa.util.EventRegistrationServiceImpl
 import no.nav.emottak.utils.environment.getEnvVar
+import no.nav.emottak.utils.kafka.client.EventPublisherClient
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import no.nav.security.token.support.v3.tokenValidationSupport
 import org.oasis_open.committees.ebxml_cppa.schema.cpp_cpa_2_0.CollaborationProtocolAgreement
 import org.slf4j.LoggerFactory
@@ -38,21 +43,28 @@ fun main() {
     if (getEnvVar("NAIS_CLUSTER_NAME", "local") != "prod-fss") {
         DecoroutinatorRuntime.load()
     }
+
+    val kafkaPublisherClient = EventPublisherClient(config().kafka)
+    val eventLoggingService = EventLoggingService(config().eventLogging, kafkaPublisherClient)
+    val eventRegistrationService = EventRegistrationServiceImpl(eventLoggingService)
+
     embeddedServer(
         Netty,
         port = 8080,
         module = cpaApplicationModule(
             cpaDbConfig.value,
             cpaMigrationConfig.value,
-            oracleConfig.value
+            oracleConfig.value,
+            eventRegistrationService
         )
     ).start(wait = true)
 }
 
 fun cpaApplicationModule(
     cpaDbConfig: HikariConfig,
     cpaMigrationConfig: HikariConfig,
-    emottakDbConfig: HikariConfig? = null
+    emottakDbConfig: HikariConfig? = null,
+    eventRegistrationService: EventRegistrationService
 ): Application.() -> Unit {
     return {
         val database = Database(cpaDbConfig)
@@ -77,7 +89,7 @@ fun cpaApplicationModule(
         routing {
             if (oracleDb != null) {
                 partnerId(PartnerRepository(oracleDb), cpaRepository)
-                validateCpa(cpaRepository, PartnerRepository(oracleDb))
+                validateCpa(cpaRepository, PartnerRepository(oracleDb), eventRegistrationService)
             }
             getCPA(cpaRepository)
             getTimeStamps(cpaRepository)

cpa-repo/src/main/kotlin/no/nav/emottak/cpa/Routes.kt

@@ -15,12 +15,15 @@ import io.ktor.server.routing.post
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
 import kotlinx.coroutines.Dispatchers
 import kotlinx.coroutines.withContext
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
 import no.nav.emottak.cpa.auth.AZURE_AD_AUTH
 import no.nav.emottak.cpa.feil.CpaValidationException
 import no.nav.emottak.cpa.feil.MultiplePartnerException
 import no.nav.emottak.cpa.feil.PartnerNotFoundException
 import no.nav.emottak.cpa.persistence.CPARepository
 import no.nav.emottak.cpa.persistence.gammel.PartnerRepository
+import no.nav.emottak.cpa.util.EventRegistrationService
 import no.nav.emottak.cpa.validation.MessageDirection
 import no.nav.emottak.cpa.validation.partyInfoHasRoleServiceActionCombo
 import no.nav.emottak.cpa.validation.validate
@@ -38,6 +41,8 @@ import no.nav.emottak.message.model.ValidationResult
 import no.nav.emottak.util.createX509Certificate
 import no.nav.emottak.util.marker
 import no.nav.emottak.utils.environment.getEnvVar
+import no.nav.emottak.utils.kafka.model.EventType
+import no.nav.emottak.utils.serialization.toEventDataJson
 import no.nav.security.token.support.v3.TokenValidationContextPrincipal
 import org.oasis_open.committees.ebxml_cppa.schema.cpp_cpa_2_0.CollaborationProtocolAgreement
 import java.util.Date
@@ -174,11 +179,14 @@ fun Route.postCpa(cpaRepository: CPARepository) = post("/cpa") {
         }
 }
 
-fun Route.validateCpa(cpaRepository: CPARepository, partnerRepository: PartnerRepository) = post("/cpa/validate/{$REQUEST_ID}") {
+fun Route.validateCpa(
+    cpaRepository: CPARepository,
+    partnerRepository: PartnerRepository,
+    eventRegistrationService: EventRegistrationService
+) = post("/cpa/validate/{$REQUEST_ID}") {
     val validateRequest = call.receive(ValidationRequest::class)
 
-    // TODO: Skal brukes i kall mot Event-logging:
-    // val requestId = call.parameters[REQUEST_ID] ?: throw BadRequestException("Mangler $REQUEST_ID")
+    val requestId = call.parameters[REQUEST_ID] ?: throw BadRequestException("Mangler $REQUEST_ID")
 
     try {
         log.info(validateRequest.marker(), "Validerer ebms mot CPA")
@@ -213,7 +221,6 @@ fun Route.validateCpa(cpaRepository: CPARepository, partnerRepository: PartnerRe
 
         val partnerId = runCatching { partnerRepository.findPartnerId(cpa.cpaid) }.getOrNull()
 
-        // TODO: Event-logging OK
         call.respond(
             HttpStatusCode.OK,
             ValidationResult(
@@ -232,22 +239,48 @@ fun Route.validateCpa(cpaRepository: CPARepository, partnerRepository: PartnerRe
                 partnerId
             )
         )
+
+        val eventData = Json.encodeToString(
+            mapOf("sender" to fromParty.partyName)
+        )
+
+        eventRegistrationService.registerEvent(
+            EventType.MESSAGE_VALIDATED_AGAINST_CPA,
+            validateRequest,
+            requestId,
+            eventData
+        )
     } catch (ebmsEx: EbmsException) {
-        // TODO: Event-logging feil?
+        eventRegistrationService.registerEvent(
+            EventType.VALIDATION_AGAINST_CPA_FAILED,
+            validateRequest,
+            requestId,
+            ebmsEx.toEventDataJson()
+        )
         log.error(validateRequest.marker(), ebmsEx.message, ebmsEx)
         call.respond(
             HttpStatusCode.OK,
             ValidationResult(error = ebmsEx.feil)
         )
     } catch (ex: NotFoundException) {
-        // TODO: Event-logging feil?
+        eventRegistrationService.registerEvent(
+            EventType.VALIDATION_AGAINST_CPA_FAILED,
+            validateRequest,
+            requestId,
+            ex.toEventDataJson()
+        )
         log.error(validateRequest.marker(), "${ex.message}")
         call.respond(
             HttpStatusCode.OK,
             ValidationResult(error = listOf(Feil(ErrorCode.DELIVERY_FAILURE, "${ex.message}")))
         )
     } catch (ex: Exception) {
-        // TODO: Event-logging feil
+        eventRegistrationService.registerEvent(
+            EventType.VALIDATION_AGAINST_CPA_FAILED,
+            validateRequest,
+            requestId,
+            ex.toEventDataJson()
+        )
         log.error(validateRequest.marker(), ex.message, ex)
         call.respond(
             HttpStatusCode.OK,

cpa-repo/src/main/kotlin/no/nav/emottak/cpa/configuration/Config.kt

@@ -0,0 +1,9 @@
+package no.nav.emottak.cpa.configuration
+
+import no.nav.emottak.utils.config.EventLogging
+import no.nav.emottak.utils.config.Kafka
+
+data class Config(
+    val kafka: Kafka,
+    val eventLogging: EventLogging
+)

cpa-repo/src/main/kotlin/no/nav/emottak/cpa/configuration/Configurator.kt

@@ -0,0 +1,14 @@
+package no.nav.emottak.cpa.configuration
+
+import com.sksamuel.hoplite.ConfigLoader
+import com.sksamuel.hoplite.addEnvironmentSource
+import com.sksamuel.hoplite.addResourceSource
+
+fun config() = ConfigLoader.builder()
+    .addEnvironmentSource()
+    .addResourceSource("/application-personal.conf", optional = true)
+    .addResourceSource("/kafka_common.conf")
+    .addResourceSource("/application.conf")
+    .withExplicitSealedTypes()
+    .build()
+    .loadConfigOrThrow<Config>()

cpa-repo/src/main/kotlin/no/nav/emottak/cpa/util/EventRegistration.kt

@@ -0,0 +1,57 @@
+package no.nav.emottak.cpa.util
+
+import no.nav.emottak.cpa.log
+import no.nav.emottak.message.model.ValidationRequest
+import no.nav.emottak.utils.common.parseOrGenerateUuid
+import no.nav.emottak.utils.kafka.model.Event
+import no.nav.emottak.utils.kafka.model.EventType
+import no.nav.emottak.utils.kafka.service.EventLoggingService
+
+interface EventRegistrationService {
+    suspend fun registerEvent(
+        eventType: EventType,
+        validationRequest: ValidationRequest,
+        requestId: String,
+        eventData: String = "{}"
+    )
+}
+
+class EventRegistrationServiceImpl(
+    private val eventLoggingService: EventLoggingService
+) : EventRegistrationService {
+    override suspend fun registerEvent(
+        eventType: EventType,
+        validationRequest: ValidationRequest,
+        requestId: String,
+        eventData: String
+    ) {
+        log.debug("Registering event for requestId: $requestId")
+
+        try {
+            val event = Event(
+                eventType = eventType,
+                requestId = requestId.parseOrGenerateUuid(),
+                contentId = "",
+                messageId = validationRequest.messageId,
+                eventData = eventData
+            )
+            log.debug("Registering event: {}", event)
+
+            eventLoggingService.logEvent(event)
+            log.debug("Event is registered successfully")
+        } catch (e: Exception) {
+            log.error("Error while registering event: ${e.message}", e)
+        }
+    }
+}
+
+class EventRegistrationServiceFake : EventRegistrationService {
+    override suspend fun registerEvent(
+        eventType: EventType,
+        validationRequest: ValidationRequest,
+        requestId: String,
+        eventData: String
+    ) {
+        log.debug("Registering event $eventType for validationRequest: $validationRequest and eventData: $eventData")
+    }
+}

cpa-repo/src/main/resources/application.conf

@@ -0,0 +1,4 @@
+
+kafka {
+  groupId = "cpa-repo"
+}

cpa-repo/src/test/kotlin/no/nav/emottak/cpa/CPARepoIntegrationTest.kt

@@ -26,6 +26,7 @@ import kotlinx.serialization.json.Json
 import no.nav.emottak.cpa.auth.AZURE_AD_AUTH
 import no.nav.emottak.cpa.auth.AuthConfig
 import no.nav.emottak.cpa.databasetest.PostgresOracleTest
+import no.nav.emottak.cpa.util.EventRegistrationServiceFake
 import no.nav.emottak.message.model.Addressing
 import no.nav.emottak.message.model.Direction.IN
 import no.nav.emottak.message.model.EmailAddress
@@ -55,7 +56,16 @@ import kotlin.test.assertTrue
 class CPARepoIntegrationTest : PostgresOracleTest() {
 
     private fun <T> cpaRepoTestApp(testBlock: suspend ApplicationTestBuilder.() -> T) = testApplication {
-        application(cpaApplicationModule(postgres.dataSource, postgres.dataSource, oracle.dataSource))
+        val eventRegistrationService = EventRegistrationServiceFake()
+
+        application(
+            cpaApplicationModule(
+                postgres.dataSource,
+                postgres.dataSource,
+                oracle.dataSource,
+                eventRegistrationService
+            )
+        )
         testBlock()
     }
 

cpa-repo/src/test/kotlin/no/nav/emottak/cpa/PartnerIntegrationTest.kt

@@ -9,6 +9,7 @@ import io.ktor.server.testing.ApplicationTestBuilder
 import io.ktor.server.testing.testApplication
 import no.nav.emottak.cpa.databasetest.PostgresOracleTest
 import no.nav.emottak.cpa.persistence.gammel.PARTNER_CPA
+import no.nav.emottak.cpa.util.EventRegistrationServiceFake
 import org.jetbrains.exposed.sql.deleteAll
 import org.jetbrains.exposed.sql.insert
 import org.jetbrains.exposed.sql.transactions.transaction
@@ -35,7 +36,16 @@ class PartnerIntegrationTest : PostgresOracleTest() {
     }
 
     fun <T> cpaRepoTestApp(testBlock: suspend ApplicationTestBuilder.() -> T) = testApplication {
-        application(cpaApplicationModule(postgres.dataSource, postgres.dataSource, oracle.dataSource))
+        val eventRegistrationService = EventRegistrationServiceFake()
+
+        application(
+            cpaApplicationModule(
+                postgres.dataSource,
+                postgres.dataSource,
+                oracle.dataSource,
+                eventRegistrationService
+            )
+        )
         testBlock()
     }
 

ebms-provider/src/main/kotlin/no/nav/emottak/ebms/util/EventRegistration.kt

@@ -50,12 +50,12 @@ class EventRegistrationServiceImpl(
                 messageId = ebMSDocument.transform().messageId,
                 eventData = eventData
             )
-            log.debug("Event reg. test: Publishing event: $event")
+            log.debug("Publishing event: $event")
 
             eventLoggingService.logEvent(event)
-            log.debug("Event reg. test: Event published successfully")
+            log.debug("Event published successfully")
         } catch (e: Exception) {
-            log.error("Event reg. test: Error while registering event: ${e.message}", e)
+            log.error("Error while registering event: ${e.message}", e)
         }
     }