621: Implement check for linked orders/results

etor/src/main/java/gov/hhs/cdc/trustedintermediary/etor/metadata/partner/PartnerMetadataOrchestrator.java

@@ -1,6 +1,7 @@
 package gov.hhs.cdc.trustedintermediary.etor.metadata.partner;
 
 import gov.hhs.cdc.trustedintermediary.etor.RSEndpointClient;
+import gov.hhs.cdc.trustedintermediary.external.database.DatabaseLinkedMessageStorage;
 import gov.hhs.cdc.trustedintermediary.external.reportstream.ReportStreamEndpointClientException;
 import gov.hhs.cdc.trustedintermediary.wrappers.Logger;
 import gov.hhs.cdc.trustedintermediary.wrappers.formatter.Formatter;
@@ -11,6 +12,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Optional;
+import java.util.Set;
 import java.util.stream.Collectors;
 import javax.inject.Inject;
 
@@ -24,6 +26,7 @@ public class PartnerMetadataOrchestrator {
     private static final PartnerMetadataOrchestrator INSTANCE = new PartnerMetadataOrchestrator();
 
     @Inject PartnerMetadataStorage partnerMetadataStorage;
+    @Inject DatabaseLinkedMessageStorage linkedMessageStorage;
     @Inject RSEndpointClient rsclient;
     @Inject Formatter formatter;
     @Inject Logger logger;
@@ -143,43 +146,45 @@ public Optional<PartnerMetadata> getMetadata(String receivedSubmissionId)
 
         PartnerMetadata partnerMetadata = optionalPartnerMetadata.get();
         var sentSubmissionId = partnerMetadata.sentSubmissionId();
-        if (metadataIsStale(partnerMetadata) && sentSubmissionId != null) {
-            logger.logInfo(
-                    "Receiver name not found in metadata or delivery status still pending, looking up {} from RS history API",
-                    sentSubmissionId);
-
-            String receiver;
-            String rsStatus;
-            String rsMessage = "";
-            String timeDelivered;
-            try {
-                String bearerToken = rsclient.getRsToken();
-                String responseBody =
-                        rsclient.requestHistoryEndpoint(sentSubmissionId, bearerToken);
-                var parsedResponseBody = getDataFromReportStream(responseBody);
-                receiver = parsedResponseBody[0];
-                rsStatus = parsedResponseBody[1];
-                rsMessage = parsedResponseBody[2];
-                timeDelivered = parsedResponseBody[3];
-            } catch (ReportStreamEndpointClientException | FormatterProcessingException e) {
-                throw new PartnerMetadataException(
-                        "Unable to retrieve metadata from RS history API", e);
-            }
 
-            var ourStatus = ourStatusFromReportStreamStatus(rsStatus);
+        if (!metadataIsStale(partnerMetadata) || sentSubmissionId == null) {
+            return Optional.of(partnerMetadata);
+        }
 
-            logger.logInfo("Updating metadata with receiver {} and status {}", receiver, ourStatus);
-            partnerMetadata = partnerMetadata.withReceiver(receiver).withDeliveryStatus(ourStatus);
+        logger.logInfo(
+                "Receiver name not found in metadata or delivery status still pending, looking up {} from RS history API",
+                sentSubmissionId);
 
-            if (ourStatus == PartnerMetadataStatus.FAILED) {
-                partnerMetadata = partnerMetadata.withFailureMessage(rsMessage);
-            } else if (ourStatus == PartnerMetadataStatus.DELIVERED && timeDelivered != null) {
-                partnerMetadata = partnerMetadata.withTimeDelivered(Instant.parse(timeDelivered));
-            }
+        String receiver;
+        String rsStatus;
+        String rsMessage;
+        String timeDelivered;
+        try {
+            String bearerToken = rsclient.getRsToken();
+            String responseBody = rsclient.requestHistoryEndpoint(sentSubmissionId, bearerToken);
+            var parsedResponseBody = getDataFromReportStream(responseBody);
+            receiver = parsedResponseBody[0];
+            rsStatus = parsedResponseBody[1];
+            rsMessage = parsedResponseBody[2];
+            timeDelivered = parsedResponseBody[3];
+        } catch (ReportStreamEndpointClientException | FormatterProcessingException e) {
+            throw new PartnerMetadataException(
+                    "Unable to retrieve metadata from RS history API", e);
+        }
 
-            partnerMetadataStorage.saveMetadata(partnerMetadata);
+        var ourStatus = ourStatusFromReportStreamStatus(rsStatus);
+
+        logger.logInfo("Updating metadata with receiver {} and status {}", receiver, ourStatus);
+        partnerMetadata = partnerMetadata.withReceiver(receiver).withDeliveryStatus(ourStatus);
+
+        if (ourStatus == PartnerMetadataStatus.FAILED) {
+            partnerMetadata = partnerMetadata.withFailureMessage(rsMessage);
+        } else if (ourStatus == PartnerMetadataStatus.DELIVERED && timeDelivered != null) {
+            partnerMetadata = partnerMetadata.withTimeDelivered(Instant.parse(timeDelivered));
         }
 
+        partnerMetadataStorage.saveMetadata(partnerMetadata);
+
         return Optional.of(partnerMetadata);
     }
 
@@ -319,6 +324,23 @@ PartnerMetadataStatus ourStatusFromReportStreamStatus(String rsStatus) {
         };
     }
 
+    boolean isMessageLinked(String receivedSubmissionId) throws Exception {
+        var linkedMessageSet = linkedMessageStorage.readLinkedMessages(receivedSubmissionId);
+        return !linkedMessageSet.isEmpty();
+    }
+
+    Set<String> findMessagesToLink(String receivedSubmissionId) throws PartnerMetadataException {
+        var metadataSet =
+                partnerMetadataStorage.readMetadataForLinkingMessages(receivedSubmissionId);
+        return metadataSet.stream()
+                .map(PartnerMetadata::receivedSubmissionId)
+                .collect(Collectors.toSet());
+    }
+
+    void linkMessages(Set<String> messagesToLink) {
+        // create entries in message_link table using receivedSubmissionIds in messagesToLink
+    }
+
     private boolean metadataIsStale(PartnerMetadata partnerMetadata) {
         return partnerMetadata.receiver() == null
                 || partnerMetadata.deliveryStatus() == PartnerMetadataStatus.PENDING;

etor/src/main/java/gov/hhs/cdc/trustedintermediary/etor/metadata/partner/PartnerMetadataStorage.java

@@ -30,4 +30,7 @@ Optional<PartnerMetadata> readMetadata(String receivedSubmissionId)
      * @return a set of {@link PartnerMetadata}s.
      */
     Set<PartnerMetadata> readMetadataForSender(String sender) throws PartnerMetadataException;
+
+    Set<PartnerMetadata> readMetadataForLinkingMessages(String submissionId)
+            throws PartnerMetadataException;
 }

etor/src/main/java/gov/hhs/cdc/trustedintermediary/external/database/DatabaseLinkedMessageStorage.java

@@ -0,0 +1,31 @@
+package gov.hhs.cdc.trustedintermediary.external.database;
+
+import gov.hhs.cdc.trustedintermediary.wrappers.Logger;
+import java.sql.SQLException;
+import java.util.Set;
+import javax.inject.Inject;
+
+public class DatabaseLinkedMessageStorage {
+
+    @Inject DbDao dao;
+
+    @Inject Logger logger;
+
+    private static final DatabaseLinkedMessageStorage INSTANCE = new DatabaseLinkedMessageStorage();
+
+    private DatabaseLinkedMessageStorage() {}
+
+    public static DatabaseLinkedMessageStorage getInstance() {
+        return INSTANCE;
+    }
+
+    public Set<?> readLinkedMessages(String submissionId) throws Exception {
+        Set<?> linkedMessageSet;
+        try {
+            linkedMessageSet = dao.fetchLinkedMessages(submissionId);
+        } catch (SQLException e) {
+            throw new Exception("Error retrieving metadata", e);
+        }
+        return linkedMessageSet;
+    }
+}

etor/src/main/java/gov/hhs/cdc/trustedintermediary/external/database/DatabasePartnerMetadataStorage.java

@@ -66,4 +66,16 @@ public Set<PartnerMetadata> readMetadataForSender(String sender)
         }
         return consolidatedMetadata;
     }
+
+    @Override
+    public Set<PartnerMetadata> readMetadataForLinkingMessages(String submissionId)
+            throws PartnerMetadataException {
+        Set<PartnerMetadata> metadataSet;
+        try {
+            metadataSet = dao.fetchMetadataForMessageLinking(submissionId);
+        } catch (SQLException e) {
+            throw new PartnerMetadataException("Error retrieving metadata", e);
+        }
+        return metadataSet;
+    }
 }

etor/src/main/java/gov/hhs/cdc/trustedintermediary/external/database/DbDao.java

@@ -22,7 +22,11 @@ void upsertMetadata(
             PartnerMetadataMessageType messageType)
             throws SQLException;
 
+    Object fetchMetadata(String uniqueId) throws SQLException;
+
     Set<PartnerMetadata> fetchMetadataForSender(String sender) throws SQLException;
 
-    Object fetchMetadata(String uniqueId) throws SQLException;
+    Set<PartnerMetadata> fetchMetadataForMessageLinking(String submissionId) throws SQLException;
+
+    Set<PartnerMetadata> fetchLinkedMessages(String messageId) throws SQLException;
 }

etor/src/main/java/gov/hhs/cdc/trustedintermediary/external/database/PostgresDao.java

@@ -128,6 +128,62 @@ public PartnerMetadata fetchMetadata(String submissionId) throws SQLException {
         }
     }
 
+    @Override
+    public Set<PartnerMetadata> fetchMetadataForMessageLinking(String submissionId)
+            throws SQLException {
+        var sql =
+                """
+                SELECT m2.*
+                FROM metadata m1
+                JOIN metadata m2
+                    ON m1.placer_order_number = m2.placer_order_number
+                        AND m1.sending_application_id = m2.sending_application_id
+                        AND m1.sending_facility_id = m2.sending_facility_id
+                WHERE m1.sent_message_id = ?;
+                """;
+
+        try (Connection conn = connectionPool.getConnection();
+                PreparedStatement statement = conn.prepareStatement(sql)) {
+            statement.setString(1, submissionId);
+            ResultSet resultSet = statement.executeQuery();
+
+            Set<PartnerMetadata> metadataSet = new HashSet<>();
+
+            while (resultSet.next()) {
+                metadataSet.add(partnerMetadataFromResultSet(resultSet));
+            }
+
+            return metadataSet;
+        }
+    }
+
+    @Override
+    public Set<PartnerMetadata> fetchLinkedMessages(String messageId) throws SQLException {
+        var sql =
+                """
+                SELECT *
+                FROM message_link
+                WHERE link_id = (
+                    SELECT link_id
+                    FROM message_link
+                    WHERE message_id = ?);
+                """;
+
+        try (Connection conn = connectionPool.getConnection();
+                PreparedStatement statement = conn.prepareStatement(sql)) {
+            statement.setString(1, messageId);
+            ResultSet resultSet = statement.executeQuery();
+
+            Set<PartnerMetadata> metadataSet = new HashSet<>();
+
+            while (resultSet.next()) {
+                metadataSet.add(partnerMetadataFromResultSet(resultSet));
+            }
+
+            return metadataSet;
+        }
+    }
+
     private PartnerMetadata partnerMetadataFromResultSet(ResultSet resultSet) throws SQLException {
         Instant timeReceived = null;
         Instant timeDelivered = null;