#187 Registrere hendelser i ebms-provider

.nais/ebms-provider-dev.yaml

@@ -53,6 +53,8 @@ spec:
     - "https://ebms-provider-fss.intern.dev.nav.no"
   vault:
     enabled: true
+  kafka:
+    pool: nav-dev
   accessPolicy:
     outbound:
       rules:

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

@@ -14,9 +14,12 @@ import io.ktor.utils.io.CancellationException
 import io.micrometer.prometheusmetrics.PrometheusConfig
 import io.micrometer.prometheusmetrics.PrometheusMeterRegistry
 import kotlinx.coroutines.awaitCancellation
+import no.nav.emottak.ebms.configuration.config
 import no.nav.emottak.ebms.processing.ProcessingService
 import no.nav.emottak.ebms.sendin.SendInService
 import no.nav.emottak.ebms.validation.DokumentValidator
+import no.nav.emottak.utils.kafka.client.EventPublisherClient
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import org.slf4j.LoggerFactory
 
 val log = LoggerFactory.getLogger("no.nav.emottak.ebms.App")
@@ -34,6 +37,9 @@ fun main() = SuspendApp {
     val sendInClient = SendInClient(scopedAuthHttpClient(EBMS_SEND_IN_SCOPE))
     val sendInService = SendInService(sendInClient)
 
+    val kafkaPublisherClient = EventPublisherClient(config().kafka)
+    val eventLoggingService = EventLoggingService(kafkaPublisherClient)
+
     result {
         resourceScope {
             server(
@@ -43,7 +49,8 @@ fun main() = SuspendApp {
                     ebmsProviderModule(
                         dokumentValidator,
                         processingService,
-                        sendInService
+                        sendInService,
+                        eventLoggingService
                     )
                 }
             ).also { it.engineConfig.maxChunkSize = 100000 }
@@ -62,7 +69,8 @@ fun main() = SuspendApp {
 fun Application.ebmsProviderModule(
     validator: DokumentValidator,
     processing: ProcessingService,
-    sendInService: SendInService
+    sendInService: SendInService,
+    eventLoggingService: EventLoggingService
 ) {
     val appMicrometerRegistry = PrometheusMeterRegistry(PrometheusConfig.DEFAULT)
 
@@ -76,6 +84,6 @@ fun Application.ebmsProviderModule(
         registerPrometheusEndpoint(appMicrometerRegistry)
         registerNavCheckStatus()
 
-        postEbmsSync(validator, processing, sendInService)
+        postEbmsSync(validator, processing, sendInService, eventLoggingService)
     }
 }

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

@@ -6,6 +6,8 @@ import io.ktor.server.request.header
 import io.ktor.server.response.respond
 import io.ktor.server.routing.Route
 import io.ktor.server.routing.post
+import kotlinx.serialization.encodeToString
+import kotlinx.serialization.json.Json
 import no.nav.emottak.constants.SMTPHeaders
 import no.nav.emottak.ebms.model.signer
 import no.nav.emottak.ebms.processing.ProcessingService
@@ -24,14 +26,18 @@ import no.nav.emottak.message.model.PayloadProcessing
 import no.nav.emottak.message.model.SignatureDetails
 import no.nav.emottak.util.marker
 import no.nav.emottak.util.retrieveLoggableHeaderPairs
+import no.nav.emottak.utils.kafka.model.Event
+import no.nav.emottak.utils.kafka.model.EventType
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import kotlin.uuid.ExperimentalUuidApi
 import kotlin.uuid.Uuid
 
 @OptIn(ExperimentalUuidApi::class)
 fun Route.postEbmsSync(
     validator: DokumentValidator,
     processingService: ProcessingService,
-    sendInService: SendInService
+    sendInService: SendInService,
+    eventLoggingService: EventLoggingService
 ): Route = post("/ebms/sync") {
     log.info("Receiving synchronous request")
 
@@ -41,6 +47,10 @@ fun Route.postEbmsSync(
         call.request.validateMime()
         ebMSDocument = call.receiveEbmsDokument()
         log.info(ebMSDocument.messageHeader().marker(loggableHeaders), "Melding mottatt")
+        eventLoggingService.registerEvent(
+            EventType.MESSAGE_RECEIVED_VIA_HTTP,
+            ebMSDocument
+        )
     } catch (ex: MimeValidationException) {
         logger().error(
             call.request.headers.marker(),
@@ -102,13 +112,17 @@ fun Route.postEbmsSync(
                     processingService.proccessSyncOut(messageProcessing.first, messageProcessing.second)
                 Pair<PayloadMessage, PayloadProcessing?>(processedMessage, messageProcessing.second)
             }.let {
+                val ebMSDocument = it.first.toEbmsDokument()
                 call.respondEbmsDokument(
-                    it.first.toEbmsDokument().also { ebmsDocument ->
+                    ebMSDocument.also { ebmsDocument ->
                         ebmsDocument.signer(it.second!!.signingCertificate)
                     }
                 )
                 log.info(it.first.marker(), "Melding ferdig behandlet og svar returnert")
-                // TODO: Event-logging OK
+                eventLoggingService.registerEvent(
+                    EventType.MESSAGE_SENT_VIA_HTTP,
+                    it.first.toEbmsDokument()
+                )
                 return@post
             }
     } catch (ebmsException: EbmsException) {
@@ -118,16 +132,61 @@ fun Route.postEbmsSync(
                 it.signer(signatureDetails)
             }
             log.info(ebmsMessage.marker(), "Created MessageError response")
-            // TODO: Event-logging Feil
+
+            eventLoggingService.registerEvent(
+                EventType.ERROR_WHILE_SENDING_MESSAGE_VIA_HTTP,
+                ebMSDocument,
+                mapOf("error_message" to (ebmsException.message ?: "Unknown error"))
+            )
+
             call.respondEbmsDokument(it)
             return@post
         }
     } catch (ex: Exception) {
         log.error(ebmsMessage.marker(), "Unknown error during message processing: ${ex.message}", ex)
-        // TODO: Event-logging Feil
+
+        eventLoggingService.registerEvent(
+            EventType.ERROR_WHILE_SENDING_MESSAGE_VIA_HTTP,
+            ebMSDocument,
+            mapOf("error_message" to (ex.message ?: "Unknown error"))
+        )
+
         call.respond(
             HttpStatusCode.InternalServerError,
             ex.parseAsSoapFault()
         )
     }
 }
+
+@OptIn(ExperimentalUuidApi::class)
+suspend fun EventLoggingService.registerEvent(
+    eventType: EventType,
+    ebMSDocument: EbMSDocument,
+    eventData: Map<String, String> = emptyMap()
+) {
+    log.debug("Event reg. test: Registering event for requestId: ${ebMSDocument.requestId}")
+
+    try {
+        val requestId = try {
+            Uuid.parse(ebMSDocument.requestId)
+        } catch (e: Exception) {
+            Uuid.random()
+        }
+
+        log.debug("Event reg. test: RequestId: $requestId")
+
+        val event = Event(
+            eventType = eventType,
+            requestId = requestId,
+            contentId = "",
+            messageId = ebMSDocument.transform().messageId,
+            eventData = Json.encodeToString(eventData)
+        )
+
+        log.debug("Event reg. test: Publishing event: $event")
+        this.logEvent(event)
+        log.debug("Event reg. test: Event published successfully")
+    } catch (e: Exception) {
+        log.error("Event reg. test: Error while registering event: ${e.message}", e)
+    }
+}

ebms-provider/src/test/kotlin/no/nav/emottak/ebms/EbmsRoutFellesIT.kt

@@ -27,6 +27,7 @@ import no.nav.emottak.message.model.ValidationResult
 import no.nav.emottak.message.xml.xmlMarshaller
 import no.nav.emottak.util.decodeBase64
 import no.nav.emottak.utils.environment.getEnvVar
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import org.apache.xml.security.algorithms.MessageDigestAlgorithm
 import org.apache.xml.security.signature.XMLSignature
 import org.junit.jupiter.api.Assertions
@@ -38,6 +39,7 @@ abstract class EbmsRoutFellesIT(val endpoint: String) {
 
     val validMultipartRequest = validMultipartRequest()
     val processingService = mockk<ProcessingService>()
+    val eventLoggingService = mockk<EventLoggingService>()
     val mockProcessConfig = ProcessConfig(
         true,
         true,
@@ -67,7 +69,7 @@ abstract class EbmsRoutFellesIT(val endpoint: String) {
             }
 
             routing {
-                postEbmsSync(dokumentValidator, processingService, SendInService(sendInClient))
+                postEbmsSync(dokumentValidator, processingService, SendInService(sendInClient), eventLoggingService)
             }
         }
         externalServices {

ebms-provider/src/test/kotlin/no/nav/emottak/ebms/EbmsRouteSyncIT.kt

@@ -69,7 +69,7 @@ class EbmsRouteSyncIT : EbmsRoutFellesIT(SYNC_PATH) {
                 incomingMessage
             }
             routing {
-                postEbmsSync(dokumentValidator, processingService, SendInService(sendInClient))
+                postEbmsSync(dokumentValidator, processingService, SendInService(sendInClient), eventLoggingService)
             }
         }
         externalServices {

ebms-provider/src/test/kotlin/no/nav/emottak/ebms/test/IntegrasjonsTest.kt

@@ -12,6 +12,7 @@ import io.ktor.server.engine.embeddedServer
 import io.ktor.server.netty.Netty
 import io.ktor.server.testing.testApplication
 import io.mockk.clearAllMocks
+import io.mockk.mockk
 import kotlinx.coroutines.runBlocking
 import no.nav.emottak.constants.SMTPHeaders
 import no.nav.emottak.cpa.cpaApplicationModule
@@ -29,6 +30,7 @@ import no.nav.emottak.ebms.sendin.SendInService
 import no.nav.emottak.ebms.testConfiguration
 import no.nav.emottak.ebms.validation.DokumentValidator
 import no.nav.emottak.ebms.validation.MimeHeaders
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import no.nav.security.mock.oauth2.MockOAuth2Server
 import org.junit.jupiter.api.AfterAll
 import org.junit.jupiter.api.Assertions
@@ -45,6 +47,7 @@ open class EndToEndTest {
         val mockOAuth2Server = MockOAuth2Server().also { it.start(port = 3344) }
         val ebmsProviderUrl = "http://localhost:$portnoEbmsProvider"
         val cpaRepoUrl = "http://localhost:$portnoCpaRepo"
+        val eventLoggingService = mockk<EventLoggingService>()
 
         // TODO Start mailserver og payload processor
         val cpaRepoDbContainer: PostgreSQLContainer<Nothing>
@@ -83,7 +86,7 @@ open class EndToEndTest {
             ebmsProviderServer = embeddedServer(
                 Netty,
                 port = portnoEbmsProvider,
-                module = { ebmsProviderModule(dokumentValidator, processingService, sendInService) }
+                module = { ebmsProviderModule(dokumentValidator, processingService, sendInService, eventLoggingService) }
             ).also {
                 it.start()
             }.engine
@@ -102,7 +105,7 @@ class IntegrasjonsTest : EndToEndTest() {
 
     @Test
     fun basicEndpointTest() = testApplication {
-        application { ebmsProviderModule(dokumentValidator, processingService, sendInService) }
+        application { ebmsProviderModule(dokumentValidator, processingService, sendInService, eventLoggingService) }
         val response = client.get("/")
         Assertions.assertEquals(HttpStatusCode.OK, response.status)
         Assertions.assertEquals("{\"status\":\"Hello\"}", response.bodyAsText())

ebms-provider/src/test/kotlin/no/nav/emottak/ebms/validation/MimeValidationIT.kt

@@ -22,6 +22,7 @@ import no.nav.emottak.message.model.ErrorCode
 import no.nav.emottak.message.model.Feil
 import no.nav.emottak.message.model.ValidationResult
 import no.nav.emottak.message.xml.xmlMarshaller
+import no.nav.emottak.utils.kafka.service.EventLoggingService
 import org.junit.jupiter.api.Assertions.assertEquals
 import org.junit.jupiter.api.Assertions.assertNotNull
 import org.junit.jupiter.api.Assertions.assertTrue
@@ -35,14 +36,15 @@ class MimeValidationIT {
 
     val validMultipartRequest = validMultipartRequest()
     val cpaRepoClient = mockk<CpaRepoClient>()
+    val eventLoggingService = mockk<EventLoggingService>()
 
     fun <T> mimeTestApp(testBlock: suspend ApplicationTestBuilder.() -> T) = testApplication {
         application {
             val dokumentValidator = DokumentValidator(cpaRepoClient)
             val processingService = mockk<ProcessingService>()
             val sendInService = mockk<SendInService>()
             routing {
-                postEbmsSync(dokumentValidator, processingService, sendInService)
+                postEbmsSync(dokumentValidator, processingService, sendInService, eventLoggingService)
             }
         }
         externalServices {