Merge pull request #134 from companieshouse/bugfix/DSND-2813

initial commit

src/main/java/uk/gov/companieshouse/pscdataapi/api/ChsKafkaApiService.java

@@ -4,6 +4,7 @@
 import com.fasterxml.jackson.databind.ObjectMapper;
 import java.time.OffsetDateTime;
 import java.util.HashMap;
+
 import org.springframework.beans.factory.annotation.Value;
 import org.springframework.stereotype.Service;
 import uk.gov.companieshouse.api.InternalApiClient;
@@ -15,7 +16,8 @@
 import uk.gov.companieshouse.logging.Logger;
 import uk.gov.companieshouse.pscdataapi.exceptions.SerDesException;
 import uk.gov.companieshouse.pscdataapi.exceptions.ServiceUnavailableException;
-import uk.gov.companieshouse.pscdataapi.models.PscData;
+import uk.gov.companieshouse.pscdataapi.models.PscDocument;
+import uk.gov.companieshouse.pscdataapi.transform.CompanyPscTransformer;
 import uk.gov.companieshouse.pscdataapi.util.PscTransformationHelper;
 
 @Service
@@ -26,6 +28,7 @@ public class ChsKafkaApiService {
     private static final String CHANGED_EVENT_TYPE = "changed";
     private static final String DELETE_EVENT_TYPE = "deleted";
 
+    private final CompanyPscTransformer companyPscTransformer;
     private final InternalApiClient internalApiClient;
     private final Logger logger;
     private final ObjectMapper objectMapper;
@@ -35,10 +38,12 @@ public class ChsKafkaApiService {
     @Value("${chs.api.kafka.resource-changed.uri}")
     private String resourceChangedUri;
 
-    public ChsKafkaApiService(InternalApiClient internalApiClient, Logger logger, ObjectMapper objectMapper) {
+    public ChsKafkaApiService(InternalApiClient internalApiClient, Logger logger,
+                              ObjectMapper objectMapper, CompanyPscTransformer companyPscTransformer) {
         this.internalApiClient = internalApiClient;
         this.logger = logger;
         this.objectMapper = objectMapper;
+        this.companyPscTransformer = companyPscTransformer;
     }
 
     /**
@@ -73,34 +78,51 @@ public ApiResponse<Void> invokeChsKafkaApi(String contextId, String companyNumbe
     public ApiResponse<Void> invokeChsKafkaApiWithDeleteEvent(String contextId,
                                                               String companyNumber,
                                                               String notificationId,
-                                                              String kind, PscData pscData) {
+                                                              String kind, PscDocument pscDocument) {
         internalApiClient.setBasePath(chsKafkaApiUrl);
         PrivateChangedResourcePost changedResourcePost =
                 internalApiClient.privateChangedResourceHandler()
                         .postChangedResource(resourceChangedUri,
                                 mapChangedResource(contextId, companyNumber,
-                                        notificationId, kind, true, pscData));
+                                        notificationId, kind, true, pscDocument));
         return handleApiCall(changedResourcePost);
     }
 
     private ChangedResource mapChangedResource(String contextId, String companyNumber,
                                                String notificationId,
-                                               String kind, boolean isDelete, PscData pscData) {
+                                               String kind, boolean isDelete, PscDocument pscDocument) {
         ChangedResourceEvent event = new ChangedResourceEvent();
         ChangedResource changedResource = new ChangedResource();
         event.setPublishedAt(String.valueOf(OffsetDateTime.now()));
         if (isDelete) {
             event.setType(DELETE_EVENT_TYPE);
-            try {
-                // This write value/read value is necessary to remove null fields during the jackson conversion
-                Object pscDataAsObject = objectMapper.readValue(
-                        objectMapper.writeValueAsString(pscData),
-                        Object.class);
-                changedResource.setDeletedData(pscDataAsObject);
-            } catch (JsonProcessingException ex) {
-                throw new SerDesException("Failed to serialise/deserialise psc data", ex);
+            if (pscDocument != null) {
+                // This write-value/read-value is necessary to remove null fields during the jackson conversion
+                try {
+                    Object pscObject = switch (pscDocument.getData().getKind()) {
+                        case "individual-person-with-significant-control" ->
+                                companyPscTransformer.transformPscDocToIndividual(pscDocument, false);
+                        case "individual-beneficial-owner" ->
+                                companyPscTransformer.transformPscDocToIndividualBeneficialOwner(pscDocument, false);
+                        case "corporate-entity-person-with-significant-control" ->
+                                companyPscTransformer.transformPscDocToCorporateEntity(pscDocument);
+                        case "corporate-entity-beneficial-owner" ->
+                                companyPscTransformer.transformPscDocToCorporateEntityBeneficialOwner(pscDocument);
+                        case "legal-person-person-with-significant-control" ->
+                                companyPscTransformer.transformPscDocToLegalPerson(pscDocument);
+                        case "legal-person-beneficial-owner" ->
+                                companyPscTransformer.transformPscDocToLegalPersonBeneficialOwner(pscDocument);
+                        case "super-secure-person-with-significant-control" ->
+                                companyPscTransformer.transformPscDocToSuperSecure(pscDocument);
+                        case "super-secure-beneficial-owner" ->
+                                companyPscTransformer.transformPscDocToSuperSecureBeneficialOwner(pscDocument);
+                        default -> null;
+                    };
+                    changedResource.setDeletedData(deserializedData(pscObject));
+                } catch (JsonProcessingException e) {
+                    throw new SerDesException("Failed to serialise/deserialise psc data", e);
+                }
             }
-
         } else {
             event.setType(CHANGED_EVENT_TYPE);
         }
@@ -137,4 +159,8 @@ private ApiResponse<Void> handleApiCall(PrivateChangedResourcePost changedResour
             throw exception;
         }
     }
+
+    private Object deserializedData(Object pscDocument) throws JsonProcessingException {
+        return objectMapper.readValue(objectMapper.writeValueAsString(pscDocument), Object.class);
+    }
 }

src/main/java/uk/gov/companieshouse/pscdataapi/service/CompanyPscService.java

@@ -173,7 +173,7 @@ public void deletePsc(String companyNumber, String notificationId, String contex
         repository.delete(pscDocument);
         try {
             chsKafkaApiService.invokeChsKafkaApiWithDeleteEvent(contextId,
-                    companyNumber, notificationId, kind, pscDocument.getData());
+                    companyNumber, notificationId, kind, pscDocument);
         } catch (Exception exception) {
             throw new ServiceUnavailableException(exception.getMessage());
         }