PI-2605 Send domain events for address changes (#4399)

* PI-2605 Send domain events for address changes

* Formatting changes

---------

Co-authored-by: probation-integration-bot[bot] <177347787+probation-integration-bot[bot]@users.noreply.github.com>

projects/approved-premises-and-delius/deploy/values.yaml

@@ -25,6 +25,8 @@ generic-service:
       SENTRY_DSN: SENTRY_DSN
     approved-premises-and-delius-queue:
       MESSAGING_CONSUMER_QUEUE: QUEUE_NAME
+    hmpps-domain-events-topic:
+      MESSAGING_PRODUCER_TOPIC: topic_arn
 
 generic-prometheus-alerts:
   targetApplication: approved-premises-and-delius

projects/approved-premises-and-delius/src/integrationTest/kotlin/uk/gov/justice/digital/hmpps/MessagingIntegrationTest.kt

@@ -5,6 +5,7 @@ import org.hamcrest.MatcherAssert.assertThat
 import org.hamcrest.Matchers.containsString
 import org.hamcrest.Matchers.equalTo
 import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.BeforeEach
 import org.junit.jupiter.api.MethodOrderer.OrderAnnotation
 import org.junit.jupiter.api.Order
 import org.junit.jupiter.api.Test
@@ -36,6 +37,7 @@ import uk.gov.justice.digital.hmpps.integrations.delius.person.address.PersonAdd
 import uk.gov.justice.digital.hmpps.integrations.delius.referencedata.ApprovedPremisesCategoryCode
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.StaffRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.getByCode
+import uk.gov.justice.digital.hmpps.message.HmppsDomainEvent
 import uk.gov.justice.digital.hmpps.messaging.HmppsChannelManager
 import uk.gov.justice.digital.hmpps.messaging.crn
 import uk.gov.justice.digital.hmpps.messaging.telemetryProperties
@@ -51,6 +53,9 @@ internal class MessagingIntegrationTest {
     @Value("\${messaging.consumer.queue}")
     lateinit var queueName: String
 
+    @Value("\${messaging.producer.topic}")
+    lateinit var topicName: String
+
     @Autowired
     lateinit var channelManager: HmppsChannelManager
 
@@ -81,6 +86,14 @@ internal class MessagingIntegrationTest {
     @Autowired
     private lateinit var staffRepository: StaffRepository
 
+    @BeforeEach
+    fun clearTopic() {
+        val topic = channelManager.getChannel(topicName)
+        do {
+            val message = topic.receive()?.also { topic.done(it.id) }
+        } while (message != null)
+    }
+
     @Test
     fun `application submission creates an alert contact`() {
         // Given an application-submitted event
@@ -299,6 +312,13 @@ internal class MessagingIntegrationTest {
         assertThat(main.postcode, equalTo(ap.postcode))
         assertThat(main.telephoneNumber, equalTo(ap.telephoneNumber))
 
+        // And a domain event is published for the new address
+        val domainEvent = channelManager.getChannel(topicName).receive()?.message as HmppsDomainEvent
+        assertThat(domainEvent.eventType, equalTo("probation-case.address.created"))
+        assertThat(domainEvent.crn(), equalTo(event.message.crn()))
+        assertThat(domainEvent.additionalInformation["addressId"], equalTo(main.id))
+        assertThat(domainEvent.additionalInformation["addressStatus"], equalTo("Main Address"))
+
         val keyWorker = staffRepository.getByCode("N54A001")
         val residences = residenceRepository.findAll().filter { it.personId == contact.person.id }
         assertThat(residences.size, equalTo(1))
@@ -350,6 +370,11 @@ internal class MessagingIntegrationTest {
 
         assertNull(personAddressRepository.findMainAddress(PersonGenerator.DEFAULT.id))
 
+        val domainEvent = channelManager.getChannel(topicName).receive()?.message as HmppsDomainEvent
+        assertThat(domainEvent.eventType, equalTo("probation-case.address.updated"))
+        assertThat(domainEvent.crn(), equalTo(event.message.crn()))
+        assertThat(domainEvent.additionalInformation["addressStatus"], equalTo("Previous Address"))
+
         val residence = residenceRepository.findAll().first { it.personId == contact.person.id }
         assertThat(residence.departureDate, equalTo(nsi.actualEndDate))
         assertThat(residence.departureReasonId, equalTo(ReferenceDataGenerator.ORDER_EXPIRED.id))

projects/approved-premises-and-delius/src/main/kotlin/uk/gov/justice/digital/hmpps/messaging/Notifier.kt

@@ -0,0 +1,64 @@
+package uk.gov.justice.digital.hmpps.messaging
+
+import org.openfolder.kotlinasyncapi.annotation.Schema
+import org.openfolder.kotlinasyncapi.annotation.channel.Channel
+import org.openfolder.kotlinasyncapi.annotation.channel.Message
+import org.openfolder.kotlinasyncapi.annotation.channel.Subscribe
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.stereotype.Service
+import uk.gov.justice.digital.hmpps.message.*
+import uk.gov.justice.digital.hmpps.publisher.NotificationPublisher
+
+@Service
+@Channel("hmpps-domain-events-topic")
+class Notifier(@Qualifier("topicPublisher") private val topicPublisher: NotificationPublisher) {
+    @Subscribe(
+        messages = [
+            Message(title = "probation-case.address.created", payload = Schema(HmppsDomainEvent::class)),
+            Message(title = "probation-case.address.updated", payload = Schema(HmppsDomainEvent::class))
+        ]
+    )
+    fun addressCreated(crn: String, addressId: Long, addressStatus: String) {
+        topicPublisher.publish(
+            Notification(
+                message = HmppsDomainEvent(
+                    version = 1,
+                    eventType = "probation-case.address.created",
+                    description = "A new address has been created on the probation case",
+                    personReference = PersonReference(
+                        identifiers = listOf(
+                            PersonIdentifier("CRN", crn),
+                        ),
+                    ),
+                    additionalInformation = mapOf(
+                        "addressStatus" to addressStatus,
+                        "addressId" to addressId
+                    )
+                ),
+                attributes = MessageAttributes("probation-case.address.created")
+            )
+        )
+    }
+
+    fun addressUpdated(crn: String, addressId: Long, addressStatus: String) {
+        topicPublisher.publish(
+            Notification(
+                message = HmppsDomainEvent(
+                    version = 1,
+                    eventType = "probation-case.address.updated",
+                    description = "An address has been updated on the probation case",
+                    personReference = PersonReference(
+                        identifiers = listOf(
+                            PersonIdentifier("CRN", crn),
+                        ),
+                    ),
+                    additionalInformation = mapOf(
+                        "addressStatus" to addressStatus,
+                        "addressId" to addressId
+                    )
+                ),
+                attributes = MessageAttributes("probation-case.address.updated")
+            )
+        )
+    }
+}

projects/approved-premises-and-delius/src/main/kotlin/uk/gov/justice/digital/hmpps/service/AddressService.kt

@@ -17,14 +17,19 @@ class AddressService(
     private val personAddressRepository: PersonAddressRepository,
     private val referenceDataRepository: ReferenceDataRepository
 ) {
-    fun updateMainAddress(person: Person, details: PersonArrived, ap: ApprovedPremises) {
-        endMainAddress(person, details.arrivedAt.toLocalDate())
-        ap.arrival(person, details).apply(personAddressRepository::save)
+    fun updateMainAddress(
+        person: Person,
+        details: PersonArrived,
+        ap: ApprovedPremises
+    ): Pair<PersonAddress?, PersonAddress> {
+        val previous = endMainAddress(person, details.arrivedAt.toLocalDate())
+        val current = ap.arrival(person, details).let(personAddressRepository::save)
+        return previous to current
     }
 
-    fun endMainAddress(person: Person, endDate: LocalDate) {
+    fun endMainAddress(person: Person, endDate: LocalDate): PersonAddress? {
         val currentMain = personAddressRepository.findMainAddress(person.id)
-        currentMain?.apply {
+        return currentMain?.also {
             val previousStatus = referenceDataRepository.previousAddressStatus()
             currentMain.status = previousStatus
             currentMain.endDate = endDate

projects/approved-premises-and-delius/src/main/kotlin/uk/gov/justice/digital/hmpps/service/ApprovedPremisesService.kt

@@ -12,6 +12,7 @@ import uk.gov.justice.digital.hmpps.integrations.delius.person.getByCrn
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.StaffRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.getByCode
 import uk.gov.justice.digital.hmpps.message.HmppsDomainEvent
+import uk.gov.justice.digital.hmpps.messaging.Notifier
 import uk.gov.justice.digital.hmpps.messaging.crn
 import uk.gov.justice.digital.hmpps.messaging.url
 
@@ -24,7 +25,8 @@ class ApprovedPremisesService(
     private val eventRepository: EventRepository,
     private val contactService: ContactService,
     private val nsiService: NsiService,
-    private val referralService: ReferralService
+    private val referralService: ReferralService,
+    private val notifier: Notifier,
 ) {
     fun applicationSubmitted(event: HmppsDomainEvent) {
         val details = approvedPremisesApiClient.getApplicationSubmittedDetails(event.url()).eventDetails
@@ -115,13 +117,18 @@ class ApprovedPremisesService(
         val details = approvedPremisesApiClient.getPersonArrivedDetails(event.url()).eventDetails
         val person = personRepository.getByCrn(event.crn())
         val ap = approvedPremisesRepository.getApprovedPremises(details.premises.legacyApCode)
-        nsiService.personArrived(person, details, ap)
+        nsiService.personArrived(person, details, ap)?.let { (previousAddress, newAddress) ->
+            notifier.addressCreated(person.crn, newAddress.id, newAddress.status.description)
+            previousAddress?.let { notifier.addressUpdated(person.crn, it.id, it.status.description) }
+        }
     }
 
     fun personDeparted(event: HmppsDomainEvent) {
         val details = approvedPremisesApiClient.getPersonDepartedDetails(event.url()).eventDetails
         val person = personRepository.getByCrn(event.crn())
         val ap = approvedPremisesRepository.getApprovedPremises(details.premises.legacyApCode)
-        nsiService.personDeparted(person, details, ap)
+        nsiService.personDeparted(person, details, ap)?.let { updatedAddress ->
+            notifier.addressUpdated(person.crn, updatedAddress.id, updatedAddress.status.description)
+        }
     }
 }

projects/approved-premises-and-delius/src/main/kotlin/uk/gov/justice/digital/hmpps/service/NsiService.kt

@@ -11,6 +11,7 @@ import uk.gov.justice.digital.hmpps.integrations.delius.contact.outcome.ContactO
 import uk.gov.justice.digital.hmpps.integrations.delius.contact.type.ContactTypeCode
 import uk.gov.justice.digital.hmpps.integrations.delius.nonstatutoryintervention.entity.*
 import uk.gov.justice.digital.hmpps.integrations.delius.person.Person
+import uk.gov.justice.digital.hmpps.integrations.delius.person.address.PersonAddress
 import uk.gov.justice.digital.hmpps.integrations.delius.referencedata.ReferenceDataRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.referencedata.referralCompleted
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.StaffRepository
@@ -38,7 +39,7 @@ class NsiService(
         person: Person,
         details: PersonArrived,
         ap: ApprovedPremises
-    ) {
+    ): Pair<PersonAddress?, PersonAddress>? {
         val externalReference = Nsi.EXT_REF_BOOKING_PREFIX + details.bookingId
         nsiRepository.findByPersonIdAndExternalReference(person.id, externalReference) ?: run {
             val staff = staffRepository.getByCode(details.keyWorker.staffCode)
@@ -69,7 +70,6 @@ class NsiService(
                     transferReason = transferReasonRepository.getNsiTransferReason()
                 )
             )
-            addressService.updateMainAddress(person, details, ap)
             contactService.createContact(
                 ContactDetails(
                     date = details.arrivedAt,
@@ -88,15 +88,16 @@ class NsiService(
                 probationAreaCode = ap.probationArea.code
             )
             referralService.personArrived(person, ap, details)
+            return addressService.updateMainAddress(person, details, ap)
         }
+        return null
     }
 
-    fun personDeparted(person: Person, details: PersonDeparted, ap: ApprovedPremises) {
+    fun personDeparted(person: Person, details: PersonDeparted, ap: ApprovedPremises): PersonAddress? {
         val nsi =
             nsiRepository.findByPersonIdAndExternalReference(person.id, Nsi.EXT_REF_BOOKING_PREFIX + details.bookingId)
         nsi?.actualEndDate = details.departedAt
         nsi?.outcome = referenceDataRepository.referralCompleted()
-        addressService.endMainAddress(person, details.departedAt.toLocalDate())
         contactService.createContact(
             ContactDetails(
                 date = details.departedAt,
@@ -114,5 +115,6 @@ class NsiService(
             probationAreaCode = ap.probationArea.code
         )
         referralService.personDeparted(person, details)
+        return addressService.endMainAddress(person, details.departedAt.toLocalDate())
     }
 }

projects/approved-premises-and-delius/src/main/resources/application.yml

@@ -65,6 +65,7 @@ wiremock.enabled: true
 context.initializer.classes: uk.gov.justice.digital.hmpps.wiremock.WireMockInitialiser
 
 messaging.consumer.queue: message-queue
+messaging.producer.topic: domain-events
 
 integrations:
   approved-premises-api:

projects/approved-premises-and-delius/src/test/kotlin/uk/gov/justice/digital/hmpps/service/ApprovedPremisesServiceTest.kt

@@ -40,6 +40,7 @@ import uk.gov.justice.digital.hmpps.integrations.delius.staff.Staff
 import uk.gov.justice.digital.hmpps.integrations.delius.staff.StaffRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.team.Team
 import uk.gov.justice.digital.hmpps.integrations.delius.team.TeamRepository
+import uk.gov.justice.digital.hmpps.messaging.Notifier
 import uk.gov.justice.digital.hmpps.messaging.crn
 import uk.gov.justice.digital.hmpps.messaging.url
 import uk.gov.justice.digital.hmpps.prepEvent
@@ -131,6 +132,9 @@ internal class ApprovedPremisesServiceTest {
     @Mock
     lateinit var applicationStartedEvent: ApplicationStartedEvent
 
+    @Mock
+    lateinit var notifier: Notifier
+
     lateinit var addressService: AddressService
     lateinit var contactService: ContactService
     lateinit var nsiService: NsiService
@@ -191,7 +195,8 @@ internal class ApprovedPremisesServiceTest {
             eventRepository,
             contactService,
             nsiService,
-            referralService
+            referralService,
+            notifier,
         )
     }
 
@@ -304,6 +309,7 @@ internal class ApprovedPremisesServiceTest {
         givenAddressTypes(listOf(ReferenceDataGenerator.AP_ADDRESS_TYPE))
         givenAuditUser()
         givenReferral(person, details.eventDetails.bookingId)
+        whenever(personAddressRepository.save(any())).thenAnswer { it.arguments[0] }
 
         approvedPremisesService.personArrived(personArrivedEvent)
 

projects/cas3-and-delius/deploy/values.yaml

@@ -22,6 +22,8 @@ generic-service:
       SENTRY_DSN: SENTRY_DSN
     cas3-and-delius-queue:
       MESSAGING_CONSUMER_QUEUE: QUEUE_NAME
+    hmpps-domain-events-topic:
+      MESSAGING_PRODUCER_TOPIC: topic_arn
 
 generic-prometheus-alerts:
   targetApplication: cas3-and-delius

projects/cas3-and-delius/src/integrationTest/kotlin/uk/gov/justice/digital/hmpps/CASIntegrationTest.kt

@@ -23,6 +23,7 @@ import uk.gov.justice.digital.hmpps.integrations.approvedpremises.*
 import uk.gov.justice.digital.hmpps.integrations.delius.entity.ContactRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.entity.PersonAddressRepository
 import uk.gov.justice.digital.hmpps.integrations.delius.entity.PersonRepository
+import uk.gov.justice.digital.hmpps.message.HmppsDomainEvent
 import uk.gov.justice.digital.hmpps.messaging.HmppsChannelManager
 import uk.gov.justice.digital.hmpps.messaging.crn
 import uk.gov.justice.digital.hmpps.resourceloader.ResourceLoader
@@ -37,6 +38,9 @@ internal class CASIntegrationTest {
     @Value("\${messaging.consumer.queue}")
     lateinit var queueName: String
 
+    @Value("\${messaging.producer.topic}")
+    lateinit var topicName: String
+
     @Autowired
     lateinit var channelManager: HmppsChannelManager
 
@@ -182,6 +186,12 @@ internal class CASIntegrationTest {
         assertThat(address.streetName, equalTo("12 Church Street"))
         assertThat(address.county, equalTo("Bibbinghammcshireshire"))
         assertThat(address.postcode, equalTo("BB1 1BB"))
+
+        val domainEvent = channelManager.getChannel(topicName).receive()?.message as HmppsDomainEvent
+        assertThat(domainEvent.eventType, equalTo("probation-case.address.created"))
+        assertThat(domainEvent.crn(), equalTo(event.message.crn()))
+        assertThat(domainEvent.additionalInformation["addressId"], equalTo(address.id))
+        assertThat(domainEvent.additionalInformation["addressStatus"], equalTo("Main Address"))
     }
 
     @Test
@@ -215,6 +225,12 @@ internal class CASIntegrationTest {
         assertThat(address!!.status.code, equalTo("P"))
         assertThat(contact.teamId, equalTo(ProviderGenerator.DEFAULT_TEAM.id))
         assertThat(contact.staffId, equalTo(ProviderGenerator.DEFAULT_STAFF.id))
+
+        val domainEvent = channelManager.getChannel(topicName).receive()?.message as HmppsDomainEvent
+        assertThat(domainEvent.eventType, equalTo("probation-case.address.updated"))
+        assertThat(domainEvent.crn(), equalTo(event.message.crn()))
+        assertThat(domainEvent.additionalInformation["addressId"], equalTo(address.id))
+        assertThat(domainEvent.additionalInformation["addressStatus"], equalTo("Main Address"))
     }
 
     @Test