16394 add ack functionality

prime-router/metadata/json_schema/organizations/organizations.json

@@ -213,6 +213,9 @@
             "HL7_BATCH"
           ]
         },
+        "hl7AcknowledgementEnabled": {
+          "type": "boolean"
+        },
         "name": {
           "type": "string"
         },

prime-router/src/main/kotlin/Sender.kt

@@ -31,6 +31,7 @@ import java.time.OffsetDateTime
  * @property allowDuplicates if false a duplicate submission will be rejected
  * @property senderType one of four broad sender categories
  * @property primarySubmissionMethod Sender preference for submission - manual or automatic
+ * @property hl7AcknowledgementEnabled should we return an HL7 ACK response if MSH.15 == "AL"?
  */
 @JsonIgnoreProperties(ignoreUnknown = true)
 @JsonTypeInfo(
@@ -59,6 +60,7 @@ abstract class Sender(
     val allowDuplicates: Boolean = true,
     val senderType: SenderType? = null,
     val primarySubmissionMethod: PrimarySubmissionMethod? = null,
+    val hl7AcknowledgementEnabled: Boolean = false,
     override var version: Int? = null,
     override var createdBy: String? = null,
     override var createdAt: OffsetDateTime? = null,
@@ -194,6 +196,7 @@ class UniversalPipelineSender : Sender {
         allowDuplicates: Boolean = true,
         senderType: SenderType? = null,
         primarySubmissionMethod: PrimarySubmissionMethod? = null,
+        hl7AcknowledgementEnabled: Boolean = false,
         topic: Topic,
     ) : super(
         topic,
@@ -205,7 +208,8 @@ class UniversalPipelineSender : Sender {
         processingType,
         allowDuplicates,
         senderType,
-        primarySubmissionMethod
+        primarySubmissionMethod,
+        hl7AcknowledgementEnabled
     )
 
     constructor(copy: UniversalPipelineSender) : this(
@@ -214,6 +218,7 @@ class UniversalPipelineSender : Sender {
         copy.format,
         copy.customerStatus,
         copy.schemaName,
+        hl7AcknowledgementEnabled = copy.hl7AcknowledgementEnabled,
         topic = copy.topic,
     )
 

prime-router/src/main/kotlin/azure/HttpUtilities.kt

@@ -24,6 +24,7 @@ class HttpUtilities {
     companion object : Logging {
         const val jsonMediaType = "application/json"
         const val fhirMediaType = "application/fhir+json"
+        const val hl7V2MediaType = "application/hl7-v2"
         const val oldApi = "/api/reports"
         const val watersApi = "/api/waters"
         const val tokenApi = "/api/token"
@@ -434,6 +435,14 @@ class HttpUtilities {
                 return responseCode to response
             }
         }
+
+        /**
+         * Is status code 2xx?
+         */
+        fun HttpStatus.isSuccessful(): Boolean {
+            val status = this.value()
+            return status in 200..299
+        }
     }
 }
 

prime-router/src/main/kotlin/azure/ReportFunction.kt

@@ -19,7 +19,6 @@ import com.microsoft.azure.functions.annotation.StorageAccount
 import gov.cdc.prime.router.ActionError
 import gov.cdc.prime.router.ActionLog
 import gov.cdc.prime.router.ActionLogLevel
-import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.CustomerStatus
 import gov.cdc.prime.router.InvalidParamMessage
 import gov.cdc.prime.router.InvalidReportMessage
@@ -31,6 +30,7 @@ import gov.cdc.prime.router.Sender.ProcessingType
 import gov.cdc.prime.router.SubmissionReceiver
 import gov.cdc.prime.router.UniversalPipelineReceiver
 import gov.cdc.prime.router.azure.BlobAccess.Companion.getBlobContainer
+import gov.cdc.prime.router.azure.HttpUtilities.Companion.isSuccessful
 import gov.cdc.prime.router.azure.db.enums.TaskAction
 import gov.cdc.prime.router.azure.db.tables.pojos.ReportFile
 import gov.cdc.prime.router.azure.observability.event.IReportStreamEventService
@@ -44,7 +44,7 @@ import gov.cdc.prime.router.common.Environment
 import gov.cdc.prime.router.common.JacksonMapperUtilities
 import gov.cdc.prime.router.fhirengine.translation.hl7.utils.HL7ACKUtils
 import gov.cdc.prime.router.fhirengine.utils.FhirTranscoder
-import gov.cdc.prime.router.fhirengine.utils.HL7Reader
+import gov.cdc.prime.router.history.DetailedSubmissionHistory
 import gov.cdc.prime.router.history.azure.SubmissionsFacade
 import gov.cdc.prime.router.tokens.AuthenticatedClaims
 import gov.cdc.prime.router.tokens.Scope
@@ -75,10 +75,10 @@ class ReportFunction(
 ) : RequestFunction(workflowEngine),
     Logging {
 
-        enum class IngestionMethod {
-            SFTP,
-            REST,
-        }
+    enum class IngestionMethod {
+        SFTP,
+        REST,
+    }
 
     /**
      * POST a report to the router
@@ -499,12 +499,6 @@ class ReportFunction(
         request: HttpRequestMessage<String?>,
         sender: Sender,
     ): HttpResponseMessage {
-        // check for ACK request
-        val maybeACKResponse = handleAckRequest(request)
-        if (maybeACKResponse != null) {
-            return maybeACKResponse
-        }
-
         // determine if we should be following the sync or async workflow
         val isAsync = processingType(request, sender) == ProcessingType.async
         // allow duplicates 'override' param
@@ -611,19 +605,7 @@ class ReportFunction(
             SubmissionsFacade.instance.findDetailedSubmissionHistory(txn, null, actionHistory.action)
         }
 
-        val response = request.createResponseBuilder(httpStatus)
-            .header(HttpHeaders.CONTENT_TYPE, "application/json")
-            .body(
-                JacksonMapperUtilities.allowUnknownsMapper
-                    .writeValueAsString(submission)
-            )
-            .header(
-                HttpHeaders.LOCATION,
-                request.uri.resolve(
-                    "/api/waters/report/${submission?.reportId}/history"
-                ).toString()
-            )
-            .build()
+        val response = buildResponse(request, httpStatus, submission, sender)
 
         // queue messages here after all task / action records are in
         actionHistory.queueMessages(workflowEngine)
@@ -644,34 +626,66 @@ class ReportFunction(
         }
     }
 
-    private fun handleAckRequest(request: HttpRequestMessage<String?>): HttpResponseMessage? {
-        // why does Azure handle Headers case-sensitive???
+    /**
+     * Returns an ACK response if required. Otherwise returns a JSON response.
+     */
+    private fun buildResponse(
+        request: HttpRequestMessage<String?>,
+        responseStatus: HttpStatus,
+        submission: DetailedSubmissionHistory?,
+        sender: Sender,
+    ): HttpResponseMessage {
+        return handleAckRequest(request, responseStatus, sender) ?: request
+            .createResponseBuilder(responseStatus)
+            .header(HttpHeaders.CONTENT_TYPE, "application/json")
+            .body(
+                JacksonMapperUtilities.allowUnknownsMapper
+                    .writeValueAsString(submission)
+            )
+            .header(
+                HttpHeaders.LOCATION,
+                request.uri.resolve(
+                    "/api/waters/report/${submission?.reportId}/history"
+                ).toString()
+            )
+            .build()
+    }
+
+    /**
+     * This function will return an HL7 ACK response if the following conditions are met:
+     * - The sender has the "hl7AcknowledgementEnabled" field set to true
+     * - The HL7 message has been processed successfully
+     * - The submitted HL7 contains MSH.15 == "AL"
+     *
+     * @return ACK response or null
+     */
+    private fun handleAckRequest(
+        request: HttpRequestMessage<String?>,
+        responseStatus: HttpStatus,
+        sender: Sender,
+    ): HttpResponseMessage? {
+        // Azure handles all headers as lowercase
         val contentType = request.headers[HttpHeaders.CONTENT_TYPE.lowercase()]
         val requestBody = request.body
-        return if (contentType == "application/hl7-v2" && requestBody != null) {
+        return if (
+            sender.hl7AcknowledgementEnabled &&
+            responseStatus.isSuccessful() &&
+            contentType == HttpUtilities.hl7V2MediaType &&
+            requestBody != null
+        ) {
             try {
-                // Parse HL7 message
-                val maybeMessage = HL7Reader(ActionLogger())
-                    .getMessages(requestBody)
-                    .firstOrNull()
-
-                // Is the message an ACK?
-                if (maybeMessage != null && HL7Reader.isAckMessage(maybeMessage)) {
+                hl7ACKUtils.generateOutgoingACKMessageIfRequired(requestBody)?.let { responseBody ->
                     logger.info("Creating HL7 ACK response")
-                    request.createResponseBuilder(HttpStatus.OK)
-                        .header(HttpHeaders.CONTENT_TYPE, "application/hl7-v2")
-                        .body(hl7ACKUtils.generateOutgoingACKMessage(maybeMessage))
+                    request.createResponseBuilder(responseStatus)
+                        .header(HttpHeaders.CONTENT_TYPE, HttpUtilities.hl7V2MediaType)
+                        .body(responseBody)
                         .build()
-                } else {
-                    logger.trace("Not an HL7 ACK message. Continuing.")
-                    null
                 }
             } catch (ex: Exception) {
-                logger.warn("Error checking for HL7 ACK. Continuing normal pipeline execution.", ex)
+                logger.warn("Error checking for HL7 ACK.", ex)
                 null
             }
         } else {
-            logger.trace("Not an HL7 ACK message. Continuing.")
             null
         }
     }

prime-router/src/main/kotlin/common/Environment.kt

@@ -103,12 +103,18 @@ enum class Environment(
 
         /**
          * Checks if the current environment is the local environment.
+         *
          * @return true if local environment, false otherwise
          */
         fun isLocal(): Boolean {
             return get() == LOCAL
         }
 
+        /**
+         * Checks if the current environment is the production environment.
+         *
+         * @return true if production environment, false otherwise
+         */
         fun isProd(): Boolean {
             return get() == PROD
         }

prime-router/src/main/kotlin/fhirengine/translation/hl7/utils/HL7ACKUtils.kt

@@ -2,6 +2,7 @@ package gov.cdc.prime.router.fhirengine.translation.hl7.utils
 
 import ca.uhn.hl7v2.model.Message
 import ca.uhn.hl7v2.model.v251.message.ACK
+import gov.cdc.prime.router.ActionLogger
 import gov.cdc.prime.router.common.Environment
 import gov.cdc.prime.router.fhirengine.utils.HL7Reader
 import java.time.Clock
@@ -17,7 +18,24 @@ class HL7ACKUtils(
     private val clock: Clock = Clock.systemUTC(),
 ) {
 
-    fun generateOutgoingACKMessage(incomingACKMessage: Message): String {
+    /**
+     * Parses out raw HL7 message and then checks if MSH.15 == "AL".
+     *
+     * @return HL7 ACK message body or null
+     */
+    fun generateOutgoingACKMessageIfRequired(rawHL7: String): String? {
+        val maybeMessage = HL7Reader(ActionLogger())
+            .getMessages(rawHL7)
+            .firstOrNull()
+
+        return if (maybeMessage != null && HL7Reader.requiresAckMessageResponse(maybeMessage)) {
+            generateOutgoingACKMessage(maybeMessage)
+        } else {
+            null
+        }
+    }
+
+    private fun generateOutgoingACKMessage(incomingACKMessage: Message): String {
         val outgoingAck = ACK()
 
         val ackMsh = outgoingAck.msh

prime-router/src/main/kotlin/fhirengine/utils/HL7Reader.kt

@@ -433,13 +433,14 @@ class HL7Reader(private val actionLogger: ActionLogger) : Logging {
             }
         }
 
-        fun isAckMessage(message: Message): Boolean {
+        fun requiresAckMessageResponse(message: Message): Boolean {
             val acceptAcknowledgmentType = when (val structure = message[MSH_SEGMENT_NAME]) {
                 is NIST_MSH -> structure.msh15_AcceptAcknowledgmentType.value
                 is v27_MSH -> structure.msh15_AcceptAcknowledgmentType.value
                 is v251_MSH -> structure.msh15_AcceptAcknowledgmentType.value
                 else -> null
             }
+            // AL means Always
             return acceptAcknowledgmentType == "AL"
         }