-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
IS-2271: Add consuming of arbeidsuforhet vurdering and storing of res…
…ult in db (#364) * IS-2271: WIP * IS-2271: WIP * IS-2271: Fix tests * IS-2271: Updated tests
- Loading branch information
1 parent
9e4c60e
commit 0dfcab7
Showing
38 changed files
with
560 additions
and
49 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
11 changes: 11 additions & 0 deletions
11
src/main/kotlin/no/nav/syfo/personstatus/application/IPersonOversiktStatusRepository.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| package no.nav.syfo.personstatus.application | ||
|
|
||
| import no.nav.syfo.domain.PersonIdent | ||
| import no.nav.syfo.personstatus.domain.PersonOversiktStatus | ||
|
|
||
| interface IPersonOversiktStatusRepository { | ||
|
|
||
| fun updateArbeidsuforhetvurderingStatus(personident: PersonIdent, isAktivVurdering: Boolean): Result<Int> | ||
|
|
||
| fun getPersonOversiktStatus(personident: PersonIdent): PersonOversiktStatus? | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
128 changes: 128 additions & 0 deletions
128
...kotlin/no/nav/syfo/personstatus/infrastructure/database/PersonOversiktStatusRepository.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| package no.nav.syfo.personstatus.infrastructure.database | ||
|
|
||
| import no.nav.syfo.application.database.DatabaseInterface | ||
| import no.nav.syfo.application.database.toList | ||
| import no.nav.syfo.domain.PersonIdent | ||
| import no.nav.syfo.personstatus.application.IPersonOversiktStatusRepository | ||
| import no.nav.syfo.personstatus.domain.PPersonOversiktStatus | ||
| import no.nav.syfo.personstatus.domain.PersonOversiktStatus | ||
| import no.nav.syfo.personstatus.domain.toPersonOversiktStatus | ||
| import java.sql.ResultSet | ||
| import java.sql.Timestamp | ||
| import java.time.Instant | ||
| import java.time.LocalDate | ||
| import java.time.OffsetDateTime | ||
| import java.util.* | ||
|
|
||
| class PersonOversiktStatusRepository(private val database: DatabaseInterface) : IPersonOversiktStatusRepository { | ||
|
|
||
| override fun updateArbeidsuforhetvurderingStatus( | ||
| personIdent: PersonIdent, | ||
| isAktivVurdering: Boolean | ||
| ): Result<Int> { | ||
| return try { | ||
| database.connection.use { connection -> | ||
| val tidspunkt = Timestamp.from(Instant.now()) | ||
| val rowsUpdated = connection.prepareStatement(UPDATE_OR_INSERT_PERSON_OVERSIKT_STATUS).use { | ||
| it.setString(1, UUID.randomUUID().toString()) | ||
| it.setString(2, personIdent.value) | ||
| it.setBoolean(3, isAktivVurdering) | ||
| it.setTimestamp(4, tidspunkt) | ||
| it.setTimestamp(5, tidspunkt) | ||
| it.executeUpdate() | ||
| } | ||
| val isSuccess = rowsUpdated == 1 | ||
| if (isSuccess) { | ||
| connection.commit() | ||
| Result.success(rowsUpdated) | ||
| } else { | ||
| connection.rollback() | ||
| Result.failure(RuntimeException("Failed to update arbeidsuforhet vurdering status for person with fnr: ${personIdent.value}")) | ||
| } | ||
| } | ||
| } catch (e: Exception) { | ||
| Result.failure(e) | ||
| } | ||
| } | ||
|
|
||
| override fun getPersonOversiktStatus(personIdent: PersonIdent): PersonOversiktStatus? { | ||
| database.connection.use { connection -> | ||
| val personoversiktStatus = connection.prepareStatement(GET_PERSON_OVERSIKT_STATUS).use { | ||
| it.setString(1, personIdent.value) | ||
| it.executeQuery().toList { toPPersonOversiktStatus() } | ||
| } | ||
| return personoversiktStatus.firstOrNull()?.toPersonOversiktStatus() | ||
| } | ||
| } | ||
|
|
||
| companion object { | ||
| private const val GET_PERSON_OVERSIKT_STATUS = | ||
| """ | ||
| SELECT * | ||
| FROM PERSON_OVERSIKT_STATUS | ||
| WHERE fnr = ? | ||
| """ | ||
|
|
||
| private const val UPDATE_OR_INSERT_PERSON_OVERSIKT_STATUS = | ||
| """ | ||
| INSERT INTO person_oversikt_status ( | ||
| id, | ||
| uuid, | ||
| fnr, | ||
| arbeidsuforhet_aktiv_vurdering, | ||
| opprettet, | ||
| sist_endret | ||
| ) VALUES (DEFAULT, ?, ?, ?, ?, ?) | ||
| ON CONFLICT (fnr) | ||
| DO UPDATE SET | ||
| arbeidsuforhet_aktiv_vurdering = EXCLUDED.arbeidsuforhet_aktiv_vurdering, | ||
| sist_endret = EXCLUDED.sist_endret | ||
| """ | ||
| } | ||
| } | ||
|
|
||
| private fun ResultSet.toPPersonOversiktStatus(): PPersonOversiktStatus = | ||
| PPersonOversiktStatus( | ||
| id = getInt("id"), | ||
| uuid = getString("uuid").let { UUID.fromString(it) }, | ||
| veilederIdent = getString("tildelt_veileder"), | ||
| fnr = getString("fnr"), | ||
| navn = getString("name"), | ||
| enhet = getString("tildelt_enhet"), | ||
| tildeltEnhetUpdatedAt = getObject("tildelt_enhet_updated_at", OffsetDateTime::class.java), | ||
| motebehovUbehandlet = getObject("motebehov_ubehandlet") as Boolean?, | ||
| oppfolgingsplanLPSBistandUbehandlet = getObject("oppfolgingsplan_lps_bistand_ubehandlet") as Boolean?, | ||
| dialogmotesvarUbehandlet = getObject("dialogmotesvar_ubehandlet") as Boolean, | ||
| dialogmotekandidat = getObject("dialogmotekandidat") as Boolean?, | ||
| dialogmotekandidatGeneratedAt = getObject("dialogmotekandidat_generated_at", OffsetDateTime::class.java), | ||
| motestatus = getString("motestatus"), | ||
| motestatusGeneratedAt = getObject("motestatus_generated_at", OffsetDateTime::class.java), | ||
| oppfolgingstilfelleUpdatedAt = getObject("oppfolgingstilfelle_updated_at", OffsetDateTime::class.java), | ||
| oppfolgingstilfelleGeneratedAt = getObject("oppfolgingstilfelle_generated_at", OffsetDateTime::class.java), | ||
| oppfolgingstilfelleStart = getObject("oppfolgingstilfelle_start", LocalDate::class.java), | ||
| oppfolgingstilfelleEnd = getObject("oppfolgingstilfelle_end", LocalDate::class.java), | ||
| oppfolgingstilfelleBitReferanseUuid = getString("oppfolgingstilfelle_bit_referanse_uuid")?.let { | ||
| UUID.fromString( | ||
| it | ||
| ) | ||
| }, | ||
| oppfolgingstilfelleBitReferanseInntruffet = getObject( | ||
| "oppfolgingstilfelle_bit_referanse_inntruffet", | ||
| OffsetDateTime::class.java | ||
| ), | ||
| aktivitetskrav = getString("aktivitetskrav"), | ||
| aktivitetskravStoppunkt = getObject("aktivitetskrav_stoppunkt", LocalDate::class.java), | ||
| aktivitetskravUpdatedAt = getObject("aktivitetskrav_sist_vurdert", OffsetDateTime::class.java), | ||
| aktivitetskravVurderingFrist = getObject("aktivitetskrav_vurdering_frist", LocalDate::class.java), | ||
| behandlerdialogSvarUbehandlet = getBoolean("behandlerdialog_svar_ubehandlet"), | ||
| behandlerdialogUbesvartUbehandlet = getBoolean("behandlerdialog_ubesvart_ubehandlet"), | ||
| behandlerdialogAvvistUbehandlet = getBoolean("behandlerdialog_avvist_ubehandlet"), | ||
| aktivitetskravVurderStansUbehandlet = getBoolean("aktivitetskrav_vurder_stans_ubehandlet"), | ||
| trengerOppfolging = getBoolean("trenger_oppfolging") as Boolean, | ||
| trengerOppfolgingFrist = getObject("trenger_oppfolging_frist", LocalDate::class.java), | ||
| behandlerBerOmBistandUbehandlet = getBoolean("behandler_bistand_ubehandlet"), | ||
| antallSykedager = getObject("antall_sykedager") as Int?, | ||
| arbeidsuforhetVurderAvslagUbehandlet = getBoolean("arbeidsuforhet_vurder_avslag_ubehandlet"), | ||
| isAktivArbeidsuforhetvurdering = getBoolean("arbeidsuforhet_aktiv_vurdering"), | ||
| friskmeldingTilArbeidsformidlingFom = getObject("friskmelding_til_arbeidsformidling_fom", LocalDate::class.java), | ||
| ) |
70 changes: 70 additions & 0 deletions
70
...n/kotlin/no/nav/syfo/personstatus/infrastructure/kafka/ArbeidsuforhetvurderingConsumer.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,70 @@ | ||
| package no.nav.syfo.personstatus.infrastructure.kafka | ||
|
|
||
| import no.nav.syfo.application.ApplicationState | ||
| import no.nav.syfo.application.kafka.KafkaEnvironment | ||
| import no.nav.syfo.application.kafka.kafkaAivenConsumerConfig | ||
| import no.nav.syfo.domain.PersonIdent | ||
| import no.nav.syfo.kafka.KafkaConsumerService | ||
| import no.nav.syfo.kafka.launchKafkaTask | ||
| import no.nav.syfo.personstatus.PersonoversiktStatusService | ||
| import no.nav.syfo.util.configuredJacksonMapper | ||
| import org.apache.kafka.clients.consumer.ConsumerConfig | ||
| import org.apache.kafka.clients.consumer.ConsumerRecords | ||
| import org.apache.kafka.clients.consumer.KafkaConsumer | ||
| import org.apache.kafka.common.serialization.Deserializer | ||
| import org.slf4j.Logger | ||
| import org.slf4j.LoggerFactory | ||
| import java.time.Duration | ||
| import java.util.* | ||
|
|
||
| class ArbeidsuforhetvurderingConsumer( | ||
| private val personoversiktStatusService: PersonoversiktStatusService, | ||
| ) : KafkaConsumerService<ArbeidsuforhetvurderingRecord> { | ||
|
|
||
| override val pollDurationInMillis: Long = 1000 | ||
|
|
||
| override fun pollAndProcessRecords(kafkaConsumer: KafkaConsumer<String, ArbeidsuforhetvurderingRecord>) { | ||
| val records = kafkaConsumer.poll(Duration.ofMillis(pollDurationInMillis)) | ||
| if (records.count() > 0) { | ||
| log.info("ArbeidsuforhetvurderingConsumer trace: Received ${records.count()} records") | ||
| processRecords(records = records) | ||
| kafkaConsumer.commitSync() | ||
| } | ||
| } | ||
|
|
||
| private fun processRecords(records: ConsumerRecords<String, ArbeidsuforhetvurderingRecord>): List<Result<Int>> { | ||
| val validRecords = records.requireNoNulls() | ||
| return validRecords.map { record -> | ||
| val recordValue = record.value() | ||
| personoversiktStatusService.updateArbeidsuforhetvurderingStatus( | ||
| personident = PersonIdent(recordValue.personident), | ||
| isAktivVurdering = !recordValue.isFinal, | ||
| ) | ||
| } | ||
| } | ||
|
|
||
| fun start(applicationState: ApplicationState, kafkaEnvironment: KafkaEnvironment) { | ||
| val consumerProperties = Properties().apply { | ||
| putAll(kafkaAivenConsumerConfig(kafkaEnvironment = kafkaEnvironment)) | ||
| this[ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG] = | ||
| ArbeidsuforhetvurderingRecordDeserializer::class.java.canonicalName | ||
| } | ||
| launchKafkaTask( | ||
| applicationState = applicationState, | ||
| kafkaConsumerService = this, | ||
| consumerProperties = consumerProperties, | ||
| topic = ARBEIDSUFORHET_VURDERING_TOPIC, | ||
| ) | ||
| } | ||
|
|
||
| companion object { | ||
| private const val ARBEIDSUFORHET_VURDERING_TOPIC = "teamsykefravr.arbeidsuforhet-vurdering" | ||
| private val log: Logger = LoggerFactory.getLogger(this::class.java) | ||
| } | ||
| } | ||
|
|
||
| class ArbeidsuforhetvurderingRecordDeserializer : Deserializer<ArbeidsuforhetvurderingRecord> { | ||
| private val mapper = configuredJacksonMapper() | ||
| override fun deserialize(topic: String, data: ByteArray): ArbeidsuforhetvurderingRecord = | ||
| mapper.readValue(data, ArbeidsuforhetvurderingRecord::class.java) | ||
| } |
Oops, something went wrong.