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

@@ -30,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
@@ -41,7 +42,9 @@ import gov.cdc.prime.router.cli.ProcessFhirCommands
 import gov.cdc.prime.router.common.AzureHttpUtils.getSenderIP
 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.history.DetailedSubmissionHistory
 import gov.cdc.prime.router.history.azure.SubmissionsFacade
 import gov.cdc.prime.router.tokens.AuthenticatedClaims
 import gov.cdc.prime.router.tokens.Scope
@@ -68,13 +71,14 @@ class ReportFunction(
         workflowEngine.azureEventService,
         workflowEngine.reportService
     ),
+    private val hl7ACKUtils: HL7ACKUtils = HL7ACKUtils(),
 ) : RequestFunction(workflowEngine),
     Logging {
 
-        enum class IngestionMethod {
-            SFTP,
-            REST,
-        }
+    enum class IngestionMethod {
+        SFTP,
+        REST,
+    }
 
     /**
      * POST a report to the router
@@ -601,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)
@@ -633,4 +625,68 @@ class ReportFunction(
             }
         }
     }
+
+    /**
+     * 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 (
+            sender.hl7AcknowledgementEnabled &&
+            responseStatus.isSuccessful() &&
+            contentType == HttpUtilities.hl7V2MediaType &&
+            requestBody != null
+        ) {
+            try {
+                hl7ACKUtils.generateOutgoingACKMessageIfRequired(requestBody)?.let { responseBody ->
+                    logger.info("Creating HL7 ACK response")
+                    request.createResponseBuilder(responseStatus)
+                        .header(HttpHeaders.CONTENT_TYPE, HttpUtilities.hl7V2MediaType)
+                        .body(responseBody)
+                        .build()
+                }
+            } catch (ex: Exception) {
+                logger.warn("Error checking for HL7 ACK.", ex)
+                null
+            }
+        } else {
+            null
+        }
+    }
 }

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

@@ -103,12 +103,22 @@ 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
+        }
+
         /**
          * Time zone to use for ReportStream. Note that Azure runs on UTC, so this forces our local runs to also be UTC.
          */

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

@@ -0,0 +1,79 @@
+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
+import java.util.Calendar
+import java.util.Date
+import java.util.TimeZone
+import java.util.UUID
+
+/**
+ * Helper class to generate HL7 ACK response
+ */
+class HL7ACKUtils(
+    private val clock: Clock = Clock.systemUTC(),
+) {
+
+    /**
+     * 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
+        ackMsh.msh1_FieldSeparator.value = "|"
+        ackMsh.msh2_EncodingCharacters.value = "^~\\&"
+        ackMsh.msh3_SendingApplication.parse("ReportStream")
+        ackMsh.msh4_SendingFacility.parse("CDC")
+        ackMsh.msh5_ReceivingApplication.parse(HL7Reader.getSendingApplication(incomingACKMessage))
+        ackMsh.msh6_ReceivingFacility.parse(HL7Reader.getSendingFacility(incomingACKMessage))
+        ackMsh.msh7_DateTimeOfMessage.time.setValue(getTimestamp())
+        ackMsh.msh9_MessageType.parse("ACK")
+        ackMsh.msh10_MessageControlID.parse(UUID.randomUUID().toString())
+        ackMsh.msh11_ProcessingID.parse(if (Environment.isProd()) "P" else "T")
+        ackMsh.msh12_VersionID.versionID.parse("2.5.1")
+        ackMsh.msh15_AcceptAcknowledgmentType.parse("NE")
+        ackMsh.msh16_ApplicationAcknowledgmentType.parse("NE")
+
+        val ackMsa = outgoingAck.msa
+        ackMsa.msa1_AcknowledgmentCode.parse("CA")
+        ackMsa.msa2_MessageControlID.parse(HL7Reader.getMessageControlId(incomingACKMessage))
+
+        return outgoingAck.toString()
+    }
+
+    /**
+     * HL7 library requires old Java date libraries, so we do the conversion here.
+     *
+     * We must directly specify the UTC timezone or else the HL7 library will use
+     * your machines local timezone.
+     */
+    private fun getTimestamp(): Calendar {
+        val instant = clock.instant()
+        val date = Date.from(instant)
+
+        val calendar = Calendar.getInstance()
+        calendar.time = date
+        calendar.timeZone = TimeZone.getTimeZone("UTC")
+
+        return calendar
+    }
+}

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

@@ -432,5 +432,43 @@ class HL7Reader(private val actionLogger: ActionLogger) : Logging {
                 else -> null
             }
         }
+
+        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"
+        }
+
+        fun getSendingApplication(message: Message): String? {
+            return when (val structure = message[MSH_SEGMENT_NAME]) {
+                is NIST_MSH -> structure.msh3_SendingApplication.encode()
+                is v27_MSH -> structure.msh3_SendingApplication.encode()
+                is v251_MSH -> structure.msh3_SendingApplication.encode()
+                else -> null
+            }
+        }
+
+        fun getSendingFacility(message: Message): String? {
+            return when (val structure = message[MSH_SEGMENT_NAME]) {
+                is NIST_MSH -> structure.msh4_SendingFacility.encode()
+                is v27_MSH -> structure.msh4_SendingFacility.encode()
+                is v251_MSH -> structure.msh4_SendingFacility.encode()
+                else -> null
+            }
+        }
+
+        fun getMessageControlId(message: Message): String? {
+            return when (val structure = message[MSH_SEGMENT_NAME]) {
+                is NIST_MSH -> structure.msh10_MessageControlID.encode()
+                is v27_MSH -> structure.msh10_MessageControlID.encode()
+                is v251_MSH -> structure.msh10_MessageControlID.encode()
+                else -> null
+            }
+        }
     }
 }