delete sync hendelser functionalities with tsm (#423)

Author: luantr

nais/topics/dinesykmeldte-lest-topic.yaml

@@ -25,5 +25,5 @@ fun Application.module() {
     configureNaisResources(get())
     configureLifecycleHooks(get())
     configureRouting()
-    configureRunningTasks(get(), get(), get())
+    configureRunningTasks(get(), get())
 }

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

@@ -22,7 +22,6 @@ data class Environment(
     val syketilfelleScope: String = getEnvVar("SYKETILLFELLE_SCOPE"),
     val electorPath: String = getEnvVar("ELECTOR_PATH"),
     val dbUrl: String = getEnvVar("NAIS_DATABASE_JDBC_URL"),
-    val syncTopic: String = "team-esyfo.dinesykmeldte-lest-topic",
 )
 
 fun getEnvVar(varName: String, defaultValue: String? = null) =

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

@@ -36,7 +36,9 @@ fun Route.registerDineSykmeldteApi(dineSykmeldteService: DineSykmeldteService) {
                 call.respond(HttpStatusCode.NotFound)
             }
             else -> {
-                securelog.info("found sykmeldt for narmestelederId: $narmestelederId, sykmeldt: $sykmeldt")
+                securelog.info(
+                    "found sykmeldt for narmestelederId: $narmestelederId, sykmeldt: $sykmeldt"
+                )
                 call.respond(sykmeldt)
             }
         }

src/main/kotlin/no/nav/syfo/dinesykmeldte/api/DineSykmeldteApi.kt

@@ -40,18 +40,12 @@ import no.nav.syfo.sykmelding.db.SykmeldtDbModel
 import no.nav.syfo.sykmelding.model.sykmelding.arbeidsgiver.BehandlerAGDTO
 import no.nav.syfo.sykmelding.model.sykmelding.arbeidsgiver.SykmeldingsperiodeAGDTO
 import no.nav.syfo.sykmelding.model.sykmelding.model.PeriodetypeDTO
-import no.nav.syfo.synchendelse.SyncHendelse
-import no.nav.syfo.synchendelse.SyncHendelseType
 import no.nav.syfo.util.logger
 import no.nav.syfo.util.securelog
 import no.nav.syfo.util.toFormattedNameString
-import org.apache.kafka.clients.producer.KafkaProducer
-import org.apache.kafka.clients.producer.ProducerRecord
 
 class MineSykmeldteService(
     private val mineSykmeldteDb: MineSykmeldteDb,
-    private val kafkaProducer: KafkaProducer<String, SyncHendelse>,
-    private val syncTopic: String,
 ) {
     private val log = logger()
 
@@ -178,53 +172,21 @@ class MineSykmeldteService(
 
     suspend fun markSykmeldingRead(sykmeldingId: String, lederFnr: String): Boolean {
         val ids = mineSykmeldteDb.markSykmeldingRead(sykmeldingId, lederFnr)
-        kafkaProducer
-            .send(ProducerRecord(syncTopic, SyncHendelse(ids, type = SyncHendelseType.SYKMELDING)))
-            .get()
         return ids.isNotEmpty()
     }
 
     suspend fun markSoknadRead(soknadId: String, lederFnr: String): Boolean {
         val ids = mineSykmeldteDb.markSoknadRead(soknadId, lederFnr)
-        kafkaProducer
-            .send(ProducerRecord(syncTopic, SyncHendelse(ids, type = SyncHendelseType.SOKNAD)))
-            .get()
         return ids.isNotEmpty()
     }
 
     suspend fun markHendelseRead(hendelseId: UUID, lederFnr: String): Boolean {
         val ids = mineSykmeldteDb.markHendelseRead(hendelseId, lederFnr)
-        kafkaProducer
-            .send(ProducerRecord(syncTopic, SyncHendelse(ids, type = SyncHendelseType.HENDELSE)))
-            .get()
         return ids.isNotEmpty()
     }
 
     suspend fun markAllSykmeldingerAndSoknaderRead(lederFnr: String) {
-        val sykmeldingIdsAndSoknadIds = mineSykmeldteDb.markAllSykmeldingAndSoknadAsRead(lederFnr)
-        val sykmeldingJob =
-            kafkaProducer.send(
-                ProducerRecord(
-                    syncTopic,
-                    SyncHendelse(
-                        sykmeldingIdsAndSoknadIds.sykmeldingIds,
-                        type = SyncHendelseType.SYKMELDING,
-                    ),
-                ),
-            )
-        val soknadJob =
-            kafkaProducer.send(
-                ProducerRecord(
-                    syncTopic,
-                    SyncHendelse(
-                        sykmeldingIdsAndSoknadIds.soknadIds,
-                        type = SyncHendelseType.SOKNAD,
-                    ),
-                ),
-            )
-
-        sykmeldingJob.get()
-        soknadJob.get()
+        mineSykmeldteDb.markAllSykmeldingAndSoknadAsRead(lederFnr)
     }
 }
 

src/main/kotlin/no/nav/syfo/minesykmeldte/MineSykmeldteService.kt

@@ -43,9 +43,6 @@ import no.nav.syfo.soknad.db.SoknadDb
 import no.nav.syfo.syketilfelle.client.SyfoSyketilfelleClient
 import no.nav.syfo.sykmelding.SykmeldingService
 import no.nav.syfo.sykmelding.db.SykmeldingDb
-import no.nav.syfo.synchendelse.SyncHendelse
-import no.nav.syfo.synchendelse.SyncHendelseConsumer
-import no.nav.syfo.synchendelse.SyncHendelseDb
 import no.nav.syfo.util.AuthConfiguration
 import no.nav.syfo.util.getWellKnownTokenX
 import no.nav.syfo.virksomhet.api.VirksomhetService
@@ -70,38 +67,10 @@ fun Application.configureDependencies() {
             databaseModule(),
             servicesModule(),
             commonKafkaConsumer(),
-            hendelseKafkaProducer(),
-            syncHendelseConsumer(),
         )
     }
 }
 
-fun hendelseKafkaProducer() = module {
-    single { createKafkaProducer<SyncHendelse>("dinesykmeldte-sync-hendelse-producer") }
-}
-
-fun syncHendelseConsumer() = module {
-    single {
-        val kafkaConsumer =
-            KafkaConsumer(
-                KafkaUtils.getKafkaConfig("dinesykmeldte-sync-hendelse-consumer")
-                    .also {
-                        it[ConsumerConfig.AUTO_OFFSET_RESET_CONFIG] = "earliest"
-                        it[ConsumerConfig.MAX_POLL_RECORDS_CONFIG] = 100
-                    }
-                    .toConsumerConfig(
-                        "esyfo-dinesykmeldte-sync-hendelse",
-                        StringDeserializer::class
-                    ),
-                StringDeserializer(),
-                StringDeserializer(),
-            )
-        val syncHendelseDb = SyncHendelseDb(get())
-        val environment = env()
-        SyncHendelseConsumer(syncHendelseDb, kafkaConsumer, environment.syncTopic)
-    }
-}
-
 private fun servicesModule() = module {
     single { AccessTokenClient(env().aadAccessTokenUrl, env().clientId, env().clientSecret, get()) }
     single { PdlClient(get(), env().pdlGraphqlPath) }
@@ -122,7 +91,7 @@ private fun servicesModule() = module {
         NarmestelederService(NarmestelederDb(get()), nlResponseProducer)
     }
     single { MineSykmeldteDb(get()) }
-    single { MineSykmeldteService(get(), get(), env().syncTopic) }
+    single { MineSykmeldteService(get()) }
     single { LeaderElection(get(), env().electorPath) }
     single { DeleteDataService(DeleteDataDb(get()), get()) }
     single { DineSykmeldteService(get()) }

src/main/kotlin/no/nav/syfo/plugins/DependencyInjection.kt

@@ -3,23 +3,18 @@ package no.nav.syfo.plugins
 import io.ktor.server.application.Application
 import io.ktor.server.application.ApplicationStopping
 import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.cancel
 import kotlinx.coroutines.launch
 import no.nav.syfo.common.delete.DeleteDataService
 import no.nav.syfo.common.kafka.CommonKafkaService
-import no.nav.syfo.synchendelse.SyncHendelseConsumer
 
 fun Application.configureRunningTasks(
     commonKafkaService: CommonKafkaService,
     deleteDataService: DeleteDataService,
-    syncHendelseConsumer: SyncHendelseConsumer
 ) {
     val commonKafkaServiceJob = launch(Dispatchers.IO) { commonKafkaService.startConsumer() }
     val deleteServiceJob = launch(Dispatchers.IO) { deleteDataService.start() }
-    val syncHendelseJob = launch(Dispatchers.IO) { syncHendelseConsumer.start() }
     monitor.subscribe(ApplicationStopping) {
         commonKafkaServiceJob.cancel()
         deleteServiceJob.cancel()
-        syncHendelseJob.cancel()
     }
 }

src/main/kotlin/no/nav/syfo/plugins/KafkaConsumers.kt

@@ -1,25 +1,21 @@
 package no.nav.syfo.minesykmeldte
 
 import io.kotest.core.spec.style.FunSpec
-import io.mockk.mockk
 import java.time.OffsetDateTime
 import java.time.ZoneOffset
 import java.util.UUID
 import no.nav.syfo.minesykmeldte.db.MineSykmeldteDb
 import no.nav.syfo.minesykmeldte.db.getSoknad
-import no.nav.syfo.synchendelse.SyncHendelse
 import no.nav.syfo.util.TestDb
 import no.nav.syfo.util.createSykmeldingDbModel
 import no.nav.syfo.util.createSykmeldtDbModel
 import no.nav.syfo.util.insertOrUpdate
 import org.amshove.kluent.shouldBeEqualTo
-import org.apache.kafka.clients.producer.KafkaProducer
 
 class MineSykmeldteServiceDbTest :
     FunSpec({
         val mineSykmeldteDb = MineSykmeldteDb(TestDb.database)
-        val kafkaProducer = mockk<KafkaProducer<String, SyncHendelse>>(relaxed = true)
-        val mineSykmeldteService = MineSykmeldteService(mineSykmeldteDb, kafkaProducer, "topic")
+        val mineSykmeldteService = MineSykmeldteService(mineSykmeldteDb)
 
         afterEach { TestDb.clearAllData() }
 

src/main/kotlin/no/nav/syfo/synchendelse/SyncHendelse.kt

@@ -50,7 +50,6 @@ import no.nav.syfo.sykmelding.model.sykmelding.model.ArbeidsrelatertArsakDTO
 import no.nav.syfo.sykmelding.model.sykmelding.model.ArbeidsrelatertArsakTypeDTO
 import no.nav.syfo.sykmelding.model.sykmelding.model.GradertDTO
 import no.nav.syfo.sykmelding.model.sykmelding.model.PeriodetypeDTO
-import no.nav.syfo.synchendelse.SyncHendelse
 import no.nav.syfo.util.createArbeidsgiverSykmelding
 import no.nav.syfo.util.createSoknad
 import no.nav.syfo.util.createSykmeldingsperiode
@@ -63,15 +62,13 @@ import org.amshove.kluent.shouldBeEqualTo
 import org.amshove.kluent.shouldHaveSize
 import org.amshove.kluent.shouldNotBe
 import org.amshove.kluent.shouldNotBeNull
-import org.apache.kafka.clients.producer.KafkaProducer
 
 @ExperimentalContracts
 class MineSykmeldteServiceTest :
     FunSpec(
         {
             val mineSykmeldteDb = mockk<MineSykmeldteDb>(relaxed = true)
-            val kafkaProducer = mockk<KafkaProducer<String, SyncHendelse>>(relaxed = true)
-            val mineSykmeldtService = MineSykmeldteService(mineSykmeldteDb, kafkaProducer, "topic")
+            val mineSykmeldtService = MineSykmeldteService(mineSykmeldteDb)
 
             afterEach { clearMocks(mineSykmeldteDb) }