Update event status (#284)

* update event status update

* update

* auto patch increment

* Removing intermediate status in progress

Co-authored-by: ras-rm-pr-bot <[email protected]>

_infra/helm/collection-exercise/Chart.yaml

@@ -14,8 +14,8 @@ type: application
 
 # This is the chart version. This version number should be incremented each time you make changes
 # to the chart and its templates, including the app version.
-version: 12.1.7
+version: 12.1.8
 
 # This is the version number of the application being deployed. This version number should be
 # incremented each time you make changes to the application.
-appVersion: 12.1.7
+appVersion: 12.1.8

_infra/helm/collection-exercise/templates/deployment.yaml

@@ -279,6 +279,8 @@ spec:
             value: "{{ .Values.gcp.caseNotificationTopic }}"
           - name: GCP_SAMPLESUMMARYACTIVATIONSTATUSSUBSCRIPTION
             value: "{{ .Values.gcp.sampleSummaryActivationStatusSubscription }}"
+          - name: GCP_COLLECTIONEXERCISEEVENTSTATUSUPDATESUBSCRIPTION
+            value: "{{ .Values.gcp.collectionExerciseEventStatusUpdateSubscription }}"
           - name: GCP_SAMPLESUMMARYACTIVATIONTOPIC
             value: "{{ .Values.gcp.sampleSummaryActivationTopic }}"
           - name: GCP_COLLECTIONEXERCISEENDTOPIC

_infra/helm/collection-exercise/values.yaml

@@ -71,4 +71,5 @@ gcp:
   caseNotificationTopic: "case-notification-topic"
   sampleSummaryActivationStatusSubscription: sample-summary-activation-status
   sampleSummaryActivationTopic: sample-summary-activation
-  collectionExerciseEndTopic: collection-exercise-end
+  collectionExerciseEndTopic: collection-exercise-end
+  collectionExerciseEventStatusUpdateSubscription: collection-exercise-event-update

docker-compose.yml

@@ -16,7 +16,7 @@ services:
    ports:
      - "18681:8681"
    environment:
-     - PUBSUB_PROJECT1=test,sample_unit_topic:sample_unit_subscription
+     - PUBSUB_PROJECT1=test,sample_unit_topic:sample_unit_subscription,event_status_topic:event_status_subscription
  survey:
    container_name: survey-it
    image: eu.gcr.io/ons-rasrmbs-management/survey

src/main/java/uk/gov/ons/ctp/response/collection/exercise/CollectionExerciseApplication.java

@@ -2,6 +2,7 @@
 
 import com.godaddy.logging.LoggingConfigs;
 import javax.annotation.PostConstruct;
+import lombok.extern.slf4j.Slf4j;
 import net.sourceforge.cobertura.CoverageIgnore;
 import org.redisson.Redisson;
 import org.redisson.api.RedissonClient;
@@ -54,6 +55,7 @@
 @EntityScan("uk.gov.ons.ctp.response")
 @EnableScheduling
 @EnableCaching
+@Slf4j
 public class CollectionExerciseApplication {
 
   private static final String VALIDATION_LIST = "collectionexercisesvc.sample.validation";
@@ -271,4 +273,25 @@ public void initJsonLogging() {
       LoggingConfigs.setCurrent(LoggingConfigs.getCurrent().useJson());
     }
   }
+
+  @Bean
+  public PubSubInboundChannelAdapter eventStatusUpdateChannelAdapter(
+      @Qualifier("collectionExerciseEventStatusUpdateChannel") MessageChannel inputChannel,
+      PubSubTemplate pubSubTemplate) {
+    String subscriptionName =
+        appConfig.getGcp().getCollectionExerciseEventStatusUpdateSubscription();
+    log.info(
+        "Application is listening for case event status update on subscription id {}",
+        subscriptionName);
+    PubSubInboundChannelAdapter adapter =
+        new PubSubInboundChannelAdapter(pubSubTemplate, subscriptionName);
+    adapter.setOutputChannel(inputChannel);
+    adapter.setAckMode(AckMode.MANUAL);
+    return adapter;
+  }
+
+  @Bean
+  public MessageChannel collectionExerciseEventStatusUpdateChannel() {
+    return new PublishSubscribeChannel();
+  }
 }

src/main/java/uk/gov/ons/ctp/response/collection/exercise/config/GCP.java

@@ -10,4 +10,5 @@ public class GCP {
   String sampleSummaryActivationStatusSubscription;
   String sampleSummaryActivationTopic;
   String collectionExerciseEndTopic;
+  String collectionExerciseEventStatusUpdateSubscription;
 }

src/main/java/uk/gov/ons/ctp/response/collection/exercise/message/CaseActionEventStatusReceiver.java

@@ -0,0 +1,49 @@
+package uk.gov.ons.ctp.response.collection.exercise.message;
+
+import static net.logstash.logback.argument.StructuredArguments.kv;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.godaddy.logging.Logger;
+import com.godaddy.logging.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.cloud.gcp.pubsub.support.BasicAcknowledgeablePubsubMessage;
+import org.springframework.cloud.gcp.pubsub.support.GcpPubSubHeaders;
+import org.springframework.integration.annotation.ServiceActivator;
+import org.springframework.messaging.Message;
+import org.springframework.messaging.handler.annotation.Header;
+import org.springframework.stereotype.Component;
+import uk.gov.ons.ctp.response.collection.exercise.message.dto.CaseActionEventStatusDTO;
+import uk.gov.ons.ctp.response.collection.exercise.service.EventService;
+
+@Component
+public class CaseActionEventStatusReceiver {
+  private static final Logger log = LoggerFactory.getLogger(CaseActionEventStatusReceiver.class);
+  @Autowired private ObjectMapper objectMapper;
+  @Autowired private EventService eventService;
+
+  @ServiceActivator(inputChannel = "collectionExerciseEventStatusUpdateChannel")
+  public void messageReceiver(
+      Message message,
+      @Header(GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage pubSubMsg) {
+    log.info(
+        "Receiving message ID from PubSub",
+        kv("messageId", pubSubMsg.getPubsubMessage().getMessageId()));
+    String payload = new String((byte[]) message.getPayload());
+    log.with("payload", payload).info("New request to update case action event status");
+    try {
+      log.info("Mapping payload to CaseActionEventStatus object");
+      CaseActionEventStatusDTO eventStatus =
+          objectMapper.readValue(payload, CaseActionEventStatusDTO.class);
+      log.info("Mapping successful, accepting action case notification");
+      eventService.updateEventStatus(eventStatus);
+      pubSubMsg.ack();
+    } catch (JsonProcessingException e) {
+      log.with(e)
+          .error(
+              "Something went wrong while processing message received from PubSub for updating action event status",
+              e);
+      pubSubMsg.nack();
+    }
+  }
+}

src/main/java/uk/gov/ons/ctp/response/collection/exercise/message/dto/CaseActionEventStatusDTO.java

@@ -0,0 +1,33 @@
+package uk.gov.ons.ctp.response.collection.exercise.message.dto;
+
+import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import java.util.UUID;
+import javax.validation.constraints.NotNull;
+import lombok.AccessLevel;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+import uk.gov.ons.ctp.response.collection.exercise.representation.EventDTO;
+
+@Data
+@AllArgsConstructor
+@NoArgsConstructor(access = AccessLevel.PUBLIC)
+@JsonIgnoreProperties(ignoreUnknown = true)
+public class CaseActionEventStatusDTO {
+  public enum EventTag {
+    mps,
+    go_live,
+    reminder,
+    reminder2,
+    reminder3,
+    nudge_email_0,
+    nudge_email_1,
+    nudge_email_2,
+    nudge_email_3,
+    nudge_email_4
+  }
+
+  @NotNull private UUID collectionExerciseID;
+  @NotNull private EventTag tag;
+  @NotNull private EventDTO.Status status;
+}

src/main/java/uk/gov/ons/ctp/response/collection/exercise/repository/EventRepository.java

@@ -15,6 +15,8 @@ public interface EventRepository extends JpaRepository<Event, Integer> {
 
   Event findOneByCollectionExerciseAndTag(CollectionExercise collex, String tag);
 
+  Event findOneByCollectionExerciseIdAndTag(UUID collexId, String tag);
+
   List<Event> findByCollectionExerciseId(UUID collexId);
 
   List<Event> findByCollectionExerciseIdIn(List<UUID> collexId);

src/main/java/uk/gov/ons/ctp/response/collection/exercise/representation/EventDTO.java

@@ -27,7 +27,9 @@ public class EventDTO {
   @Nullable
   public enum Status {
     SCHEDULED,
-    PROCESSED,
-    FAILED
+    COMPLETED,
+    FAILED,
+    RETRY,
+    PROCESSING
   }
 }

src/main/java/uk/gov/ons/ctp/response/collection/exercise/service/EventService.java

@@ -21,6 +21,7 @@
 import uk.gov.ons.ctp.response.collection.exercise.lib.common.error.CTPException;
 import uk.gov.ons.ctp.response.collection.exercise.lib.common.error.CTPException.Fault;
 import uk.gov.ons.ctp.response.collection.exercise.message.CollectionExerciseEndPublisher;
+import uk.gov.ons.ctp.response.collection.exercise.message.dto.CaseActionEventStatusDTO;
 import uk.gov.ons.ctp.response.collection.exercise.repository.EventRepository;
 import uk.gov.ons.ctp.response.collection.exercise.representation.CollectionExerciseDTO;
 import uk.gov.ons.ctp.response.collection.exercise.representation.EventDTO;
@@ -337,6 +338,21 @@ public boolean isScheduled(UUID collexUuid) throws CTPException {
         >= numberOfMandatoryEvents;
   }
 
+  @Transactional
+  public void updateEventStatus(CaseActionEventStatusDTO eventStatus) {
+    Event eventToBeUpdated =
+        eventRepository.findOneByCollectionExerciseIdAndTag(
+            eventStatus.getCollectionExerciseID(), eventStatus.getTag().toString());
+    if (null == eventToBeUpdated) {
+      log.with("collectionExerciseId", eventStatus.getCollectionExerciseID().toString())
+          .with("eventTag", eventStatus.getTag().toString())
+          .error("Unable to find an event for the matching combination.");
+      return;
+    }
+    eventToBeUpdated.setStatus(eventStatus.getStatus());
+    eventRepository.saveAndFlush(eventToBeUpdated);
+  }
+
   /** Get all the scheduled events and send them to action to be acted on. */
   @Transactional
   public void processEvents() {
@@ -416,17 +432,18 @@ public void processEvents() {
                 }
 
                 if (success) {
-                  log.info("Event processing succeeded, setting to PROCESSED state");
-                  event.setStatus(EventDTO.Status.PROCESSED);
+                  log.info("Event processing succeeded, setting to PROCESSING state");
+                  event.setStatus(EventDTO.Status.PROCESSING);
                   event.setMessageSent(Timestamp.from(Instant.now()));
                 } else {
-                  log.error("Event processing failed, setting to FAILED state");
-                  event.setStatus(EventDTO.Status.FAILED);
+                  log.error(
+                      "Event processing failed, due to service call hence keeping SCHEDULED state");
+                  event.setStatus(EventDTO.Status.SCHEDULED);
                 }
               } else {
                 log.with("tag", event.getTag())
-                    .debug("Event is not actionable, setting to PROCESSED state");
-                event.setStatus(EventDTO.Status.PROCESSED);
+                    .debug("Event is not actionable, setting to COMPLETED state");
+                event.setStatus(EventDTO.Status.COMPLETED);
               }
               eventRepository.saveAndFlush(event);
             } else {

src/main/resources/application.yml

@@ -190,3 +190,4 @@ gcp:
   sampleSummaryActivationStatusSubscription: "sample-summary-activation-status"
   sampleSummaryActivationTopic: "sample-summary-activation"
   collectionExerciseEndTopic: "collect-exercise-end"
+  collectionExerciseEventStatusUpdateSubscription: "event_status_subscription"

src/test/java/uk/gov/ons/ctp/response/collection/exercise/endpoint/CollectionExerciseClient.java

@@ -27,7 +27,7 @@
 import uk.gov.ons.ctp.response.collection.exercise.representation.SampleLinkDTO;
 
 /** A class to wrap the collection exercise REST API in Java using Unirest */
-class CollectionExerciseClient {
+public class CollectionExerciseClient {
 
   private ObjectMapper jacksonMapper;
   private int port;
@@ -42,7 +42,7 @@ class CollectionExerciseClient {
    * @param aPassword collection exercise API password
    * @param aMapper an object mapper
    */
-  CollectionExerciseClient(
+  public CollectionExerciseClient(
       final int aPort, final String aUsername, final String aPassword, final ObjectMapper aMapper) {
     this.port = aPort;
     this.jacksonMapper = aMapper;
@@ -52,7 +52,7 @@ class CollectionExerciseClient {
     UnirestInitialiser.initialise(jacksonMapper);
   }
 
-  List<EventDTO> getEvents(final UUID collexId) throws CTPException {
+  public List<EventDTO> getEvents(final UUID collexId) throws CTPException {
     try {
       return new ArrayList<>(
           Arrays.asList(
@@ -70,7 +70,7 @@ List<EventDTO> getEvents(final UUID collexId) throws CTPException {
     }
   }
 
-  HttpResponse updateEvent(final EventDTO event) {
+  public HttpResponse updateEvent(final EventDTO event) {
     final OffsetDateTime offsetDateTime =
         OffsetDateTime.ofInstant(event.getTimestamp().toInstant(), ZoneOffset.systemDefault());
     final String date = DateTimeFormatter.ISO_DATE_TIME.format(offsetDateTime);
@@ -112,7 +112,7 @@ HttpResponse updateEvent(final EventDTO event) {
    *     has been created
    * @throws CTPException thrown if an error occurred creating the collection exercise
    */
-  Pair<Integer, String> createCollectionExercise(
+  public Pair<Integer, String> createCollectionExercise(
       final UUID surveyId, final String exerciseRef, final String userDescription)
       throws CTPException {
     CollectionExerciseDTO inputDto = new CollectionExerciseDTO();
@@ -170,7 +170,7 @@ CollectionExerciseDTO getCollectionExercise(final UUID collexId) throws CTPExcep
    * @return a representation of the collection exercise
    * @throws CTPException thrown if there was an error retrieving the collection exercise
    */
-  CollectionExerciseDTO getCollectionExercise(final String uriStr) throws CTPException {
+  public CollectionExerciseDTO getCollectionExercise(final String uriStr) throws CTPException {
     try {
       return Unirest.get(uriStr)
           .basicAuth(this.username, this.password)
@@ -254,7 +254,7 @@ List<SampleLinkDTO> getSampleLinks(final UUID collexId) throws CTPException {
     }
   }
 
-  void createCollectionExerciseEvent(EventDTO event) {
+  public void createCollectionExerciseEvent(EventDTO event) {
     HttpResponse<String> response = null;
     try {
       response =