IS-2685: Publish forhandsvarsel to esyfovarsel (#27)

src/main/kotlin/no/nav/syfo/App.kt

@@ -6,6 +6,7 @@ import io.ktor.server.config.*
 import io.ktor.server.engine.*
 import io.ktor.server.netty.*
 import no.nav.syfo.api.apiModule
+import no.nav.syfo.application.VarselService
 import no.nav.syfo.application.VurderingService
 import no.nav.syfo.infrastructure.clients.azuread.AzureAdClient
 import no.nav.syfo.infrastructure.clients.dokarkiv.DokarkivClient
@@ -17,10 +18,14 @@ import no.nav.syfo.infrastructure.clients.wellknown.getWellKnown
 import no.nav.syfo.infrastructure.cronjob.launchCronjobs
 import no.nav.syfo.infrastructure.database.applicationDatabase
 import no.nav.syfo.infrastructure.database.databaseModule
+import no.nav.syfo.infrastructure.database.repository.VarselRepository
 import no.nav.syfo.infrastructure.database.repository.VurderingRepository
 import no.nav.syfo.infrastructure.journalforing.JournalforingService
+import no.nav.syfo.infrastructure.kafka.VarselProducer
 import no.nav.syfo.infrastructure.kafka.VurderingProducer
 import no.nav.syfo.infrastructure.kafka.VurderingRecordSerializer
+import no.nav.syfo.infrastructure.kafka.esyfovarsel.EsyfoVarselHendelseSerializer
+import no.nav.syfo.infrastructure.kafka.esyfovarsel.EsyfovarselHendelseProducer
 import no.nav.syfo.infrastructure.kafka.kafkaAivenProducerConfig
 import org.apache.kafka.clients.producer.KafkaProducer
 import org.slf4j.LoggerFactory
@@ -54,7 +59,11 @@ fun main() {
     val pdfGenClient = PdfGenClient(
         pdfGenBaseUrl = environment.clients.ispdfgen.baseUrl,
     )
-
+    val arbeidstakerForhandsvarselProducer = EsyfovarselHendelseProducer(
+        producer = KafkaProducer(
+            kafkaAivenProducerConfig<EsyfoVarselHendelseSerializer>(kafkaEnvironment = environment.kafka)
+        )
+    )
     val vurderingProducer = VurderingProducer(
         producer = KafkaProducer(kafkaAivenProducerConfig<VurderingRecordSerializer>(kafkaEnvironment = environment.kafka))
     )
@@ -68,8 +77,8 @@ fun main() {
         pdlClient = pdlClient,
     )
 
-    lateinit var vurderingRepository: VurderingRepository
     lateinit var vurderingService: VurderingService
+    lateinit var varselService: VarselService
 
     val applicationEngineEnvironment =
         applicationEngineEnvironment {
@@ -83,16 +92,23 @@ fun main() {
                     databaseEnvironment = environment.database,
                 )
 
-                vurderingRepository = VurderingRepository(
+                val vurderingRepository = VurderingRepository(
                     database = applicationDatabase,
                 )
+                val varselRepository = VarselRepository(database = applicationDatabase)
 
                 vurderingService = VurderingService(
                     journalforingService = journalforingService,
                     vurderingRepository = vurderingRepository,
                     vurderingProducer = vurderingProducer,
                     vurderingPdfService = vurderingPdfService,
                 )
+                varselService = VarselService(
+                    varselRepository = varselRepository,
+                    varselProducer = VarselProducer(
+                        arbeidstakerForhandsvarselProducer = arbeidstakerForhandsvarselProducer,
+                    ),
+                )
 
                 apiModule(
                     applicationState = applicationState,
@@ -113,6 +129,7 @@ fun main() {
             applicationState = applicationState,
             environment = environment,
             vurderingService = vurderingService,
+            varselService = varselService,
         )
     }
 

src/main/kotlin/no/nav/syfo/application/IVarselProducer.kt

@@ -0,0 +1,13 @@
+package no.nav.syfo.application
+
+import no.nav.syfo.domain.JournalpostId
+import no.nav.syfo.domain.Personident
+import no.nav.syfo.domain.Varsel
+
+interface IVarselProducer {
+    fun sendArbeidstakerForhandsvarsel(
+        personIdent: Personident,
+        journalpostId: JournalpostId,
+        varsel: Varsel
+    ): Result<Varsel>
+}

src/main/kotlin/no/nav/syfo/application/IVarselRepository.kt

@@ -0,0 +1,11 @@
+package no.nav.syfo.application
+
+import no.nav.syfo.domain.JournalpostId
+import no.nav.syfo.domain.Personident
+import no.nav.syfo.domain.Varsel
+import java.util.*
+
+interface IVarselRepository {
+    fun getUnpublishedVarsler(): List<Triple<Personident, JournalpostId, Varsel>>
+    fun updatePublishedAt(varselUuid: UUID)
+}

src/main/kotlin/no/nav/syfo/application/VarselService.kt

@@ -0,0 +1,24 @@
+package no.nav.syfo.application
+
+import no.nav.syfo.domain.Varsel
+
+class VarselService(
+    private val varselRepository: IVarselRepository,
+    private val varselProducer: IVarselProducer
+) {
+
+    fun publishUnpublishedVarsler(): List<Result<Varsel>> {
+        val unpublishedVarsler = varselRepository.getUnpublishedVarsler()
+
+        return unpublishedVarsler.map { (personident, journalpostId, varsel) ->
+            varselProducer.sendArbeidstakerForhandsvarsel(
+                personIdent = personident,
+                journalpostId = journalpostId,
+                varsel = varsel,
+            ).map {
+                varselRepository.updatePublishedAt(it.uuid)
+                it
+            }
+        }
+    }
+}

src/main/kotlin/no/nav/syfo/infrastructure/cronjob/CronjobModule.kt

@@ -2,6 +2,7 @@ package no.nav.syfo.infrastructure.cronjob
 
 import no.nav.syfo.ApplicationState
 import no.nav.syfo.Environment
+import no.nav.syfo.application.VarselService
 import no.nav.syfo.application.VurderingService
 import no.nav.syfo.infrastructure.clients.leaderelection.LeaderPodClient
 import no.nav.syfo.launchBackgroundTask
@@ -10,6 +11,7 @@ fun launchCronjobs(
     applicationState: ApplicationState,
     environment: Environment,
     vurderingService: VurderingService,
+    varselService: VarselService,
 ) {
     val leaderPodClient = LeaderPodClient(
         electorPath = environment.electorPath
@@ -20,6 +22,9 @@ fun launchCronjobs(
     )
     val cronjobs = mutableListOf<Cronjob>()
 
+    val publishForhandsvarselCronjob = PublishForhandsvarselCronjob(varselService = varselService)
+    cronjobs.add(publishForhandsvarselCronjob)
+
     val journalforVurderingerCronjob = JournalforVurderingerCronjob(vurderingService)
     cronjobs.add(journalforVurderingerCronjob)
 

src/main/kotlin/no/nav/syfo/infrastructure/cronjob/PublishForhandsvarselCronjob.kt

@@ -0,0 +1,10 @@
+package no.nav.syfo.infrastructure.cronjob
+
+import no.nav.syfo.application.VarselService
+
+class PublishForhandsvarselCronjob(private val varselService: VarselService) : Cronjob {
+    override val initialDelayMinutes: Long = 4
+    override val intervalDelayMinutes: Long = 10
+
+    override suspend fun run() = varselService.publishUnpublishedVarsler()
+}

src/main/kotlin/no/nav/syfo/infrastructure/database/repository/VarselRepository.kt

@@ -0,0 +1,64 @@
+package no.nav.syfo.infrastructure.database.repository
+
+import no.nav.syfo.application.IVarselRepository
+import no.nav.syfo.domain.JournalpostId
+import no.nav.syfo.domain.Personident
+import no.nav.syfo.domain.Varsel
+import no.nav.syfo.infrastructure.database.DatabaseInterface
+import no.nav.syfo.infrastructure.database.toList
+import no.nav.syfo.util.nowUTC
+import java.sql.ResultSet
+import java.sql.SQLException
+import java.time.OffsetDateTime
+import java.util.*
+
+class VarselRepository(private val database: DatabaseInterface) : IVarselRepository {
+
+    override fun getUnpublishedVarsler(): List<Triple<Personident, JournalpostId, Varsel>> = database.connection.use { connection ->
+        connection.prepareStatement(GET_UNPUBLISHED_VARSEL).use {
+            it.executeQuery().toList { Triple(Personident(getString("personident")), JournalpostId(getString("journalpost_id")), toPVarsel()) }
+        }
+    }.map { (personident, journalpostId, pVarsel) -> Triple(personident, journalpostId, pVarsel.toVarsel()) }
+
+    override fun updatePublishedAt(varselUuid: UUID) =
+        database.connection.use { connection ->
+            connection.prepareStatement(UPDATE_PUBLISHED_AT).use {
+                it.setObject(1, nowUTC())
+                it.setObject(2, nowUTC())
+                it.setString(3, varselUuid.toString())
+                val updated = it.executeUpdate()
+                if (updated != 1) {
+                    throw SQLException("Expected a single row to be updated, got update count $updated")
+                }
+            }
+            connection.commit()
+        }
+
+    companion object {
+        private const val GET_UNPUBLISHED_VARSEL =
+            """
+                SELECT vu.personident, vu.journalpost_id, v.* FROM varsel v
+                INNER JOIN vurdering vu
+                ON v.vurdering_id = vu.id
+                WHERE vu.journalpost_id IS NOT NULL AND v.published_at IS NULL
+            """
+
+        private const val UPDATE_PUBLISHED_AT =
+            """
+                 UPDATE varsel
+                 SET published_at = ?, updated_at = ?
+                 WHERE uuid = ?
+            """
+    }
+}
+
+internal fun ResultSet.toPVarsel(): PVarsel =
+    PVarsel(
+        id = getInt("id"),
+        uuid = UUID.fromString(getString("uuid")),
+        vurderingId = getInt("vurdering_id"),
+        createdAt = getObject("created_at", OffsetDateTime::class.java),
+        updatedAt = getObject("updated_at", OffsetDateTime::class.java),
+        svarfrist = getDate("svarfrist").toLocalDate(),
+        publishedAt = getObject("published_at", OffsetDateTime::class.java),
+    )

src/main/kotlin/no/nav/syfo/infrastructure/database/repository/VurderingRepository.kt

@@ -294,17 +294,6 @@ internal fun ResultSet.toPVurdering(): PVurdering =
         publishedAt = getObject("published_at", OffsetDateTime::class.java),
     )
 
-private fun ResultSet.toPVarsel(): PVarsel =
-    PVarsel(
-        id = getInt("id"),
-        uuid = UUID.fromString(getString("uuid")),
-        vurderingId = getInt("vurdering_id"),
-        createdAt = getObject("created_at", OffsetDateTime::class.java),
-        updatedAt = getObject("updated_at", OffsetDateTime::class.java),
-        svarfrist = getDate("svarfrist").toLocalDate(),
-        publishedAt = getObject("published_at", OffsetDateTime::class.java),
-    )
-
 fun ResultSet.toPVurderingPdf(): PVurderingPdf =
     PVurderingPdf(
         id = getInt("id"),

src/main/kotlin/no/nav/syfo/infrastructure/kafka/VarselProducer.kt

@@ -0,0 +1,23 @@
+package no.nav.syfo.infrastructure.kafka
+
+import no.nav.syfo.application.IVarselProducer
+import no.nav.syfo.domain.JournalpostId
+import no.nav.syfo.domain.Personident
+import no.nav.syfo.domain.Varsel
+import no.nav.syfo.infrastructure.kafka.esyfovarsel.EsyfovarselHendelseProducer
+
+class VarselProducer(
+    private val arbeidstakerForhandsvarselProducer: EsyfovarselHendelseProducer,
+) : IVarselProducer {
+
+    override fun sendArbeidstakerForhandsvarsel(
+        personIdent: Personident,
+        journalpostId: JournalpostId,
+        varsel: Varsel
+    ): Result<Varsel> =
+        arbeidstakerForhandsvarselProducer.sendArbeidstakerForhandsvarsel(
+            personIdent = personIdent,
+            journalpostId = journalpostId,
+            varsel = varsel
+        )
+}

src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/EsyfoVarselHendelseSerializer.kt

@@ -0,0 +1,11 @@
+package no.nav.syfo.infrastructure.kafka.esyfovarsel
+
+import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.EsyfovarselHendelse
+import no.nav.syfo.util.configuredJacksonMapper
+import org.apache.kafka.common.serialization.Serializer
+
+class EsyfoVarselHendelseSerializer : Serializer<EsyfovarselHendelse> {
+    private val mapper = configuredJacksonMapper()
+    override fun serialize(topic: String?, data: EsyfovarselHendelse?): ByteArray =
+        mapper.writeValueAsBytes(data)
+}

src/main/kotlin/no/nav/syfo/infrastructure/kafka/esyfovarsel/EsyfovarselHendelseProducer.kt

@@ -1,6 +1,8 @@
 package no.nav.syfo.infrastructure.kafka.esyfovarsel
 
-import no.nav.syfo.domain.ManglendeMedvirkningVurdering
+import no.nav.syfo.domain.JournalpostId
+import no.nav.syfo.domain.Personident
+import no.nav.syfo.domain.Varsel
 import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.ArbeidstakerHendelse
 import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.EsyfovarselHendelse
 import no.nav.syfo.infrastructure.kafka.esyfovarsel.dto.HendelseType
@@ -14,18 +16,18 @@ import java.util.*
 class EsyfovarselHendelseProducer(
     private val producer: KafkaProducer<String, EsyfovarselHendelse>,
 ) {
-
-    fun sendVurderingVarsel(vurdering: ManglendeMedvirkningVurdering): Result<ManglendeMedvirkningVurdering> {
-        if (vurdering.journalpostId == null)
-            throw IllegalStateException("JournalpostId is null for vurdering ${vurdering.uuid}")
-
+    fun sendArbeidstakerForhandsvarsel(
+        personIdent: Personident,
+        journalpostId: JournalpostId,
+        varsel: Varsel
+    ): Result<Varsel> {
         val varselHendelse = ArbeidstakerHendelse(
             type = HendelseType.SM_FORHANDSVARSEL_MANGLENDE_MEDVIRKNING,
-            arbeidstakerFnr = vurdering.personident.value,
+            arbeidstakerFnr = personIdent.value,
             data = VarselData(
                 journalpost = VarselDataJournalpost(
-                    uuid = vurdering.uuid.toString(),
-                    id = vurdering.journalpostId!!.value,
+                    uuid = varsel.uuid.toString(),
+                    id = journalpostId.value,
                 ),
             ),
             orgnummer = null,
@@ -35,13 +37,13 @@ class EsyfovarselHendelseProducer(
             producer.send(
                 ProducerRecord(
                     ESYFOVARSEL_TOPIC,
-                    UUID.nameUUIDFromBytes(vurdering.personident.value.toByteArray()).toString(),
+                    UUID.randomUUID().toString(),
                     varselHendelse,
                 )
             ).get()
-            Result.success(vurdering)
+            Result.success(varsel)
         } catch (e: Exception) {
-            log.error("Exception was thrown when attempting to send hendelse varsel (uuid: ${vurdering.uuid}) to esyfovarsel: ${e.message}")
+            log.error("Exception was thrown when attempting to send hendelse varsel (uuid: ${varsel.uuid}) to esyfovarsel: ${e.message}")
             Result.failure(e)
         }
     }