fixup! 16148: send reports from the submissions directly to FHIR Convert

CDCgov · Oct 31, 2024 · 5b10fe7 · 5b10fe7
1 parent 09a0d07
commit 5b10fe7
Show file tree

Hide file tree

Showing 7 changed files with 368 additions and 106 deletions.
diff --git a/prime-router/src/main/kotlin/azure/observability/event/ReportStreamEventBuilder.kt b/prime-router/src/main/kotlin/azure/observability/event/ReportStreamEventBuilder.kt
@@ -167,6 +167,7 @@ open class ReportStreamItemEventBuilder(
     private var theParentItemIndex = 1
     private var theChildIndex = 1
     private var theTrackingId: String? = null
+    private var theSender: String? = null
 
     fun trackingId(bundle: Bundle) {
         theTrackingId = AzureEventUtils.getIdentifier(bundle).value
@@ -180,6 +181,10 @@ open class ReportStreamItemEventBuilder(
         theChildIndex = childItemIndex
     }
 
+    fun sender(sender: String) {
+        theSender = sender
+    }
+
     protected fun getItemEventData(): ItemEventData {
         if (theParentReportId == null) {
             throw IllegalStateException("Parent Report ID must be set to generate an ItemEvent")
@@ -188,7 +193,8 @@ open class ReportStreamItemEventBuilder(
             theChildIndex,
             theParentReportId!!,
             theParentItemIndex,
-            theTrackingId
+            theTrackingId,
+            theSender
         )
     }
 

diff --git a/prime-router/src/main/kotlin/azure/observability/event/ReportStreamEventService.kt b/prime-router/src/main/kotlin/azure/observability/event/ReportStreamEventService.kt
@@ -202,6 +202,7 @@ interface IReportStreamEventService {
         parentReportId: UUID,
         parentItemIndex: Int,
         trackingId: String?,
+        senderString: String? = null,
     ): ItemEventData
 }
 
@@ -400,11 +401,15 @@ class ReportStreamEventService(
         val submittedReportIds = if (parentReportId != null) {
             val rootReports = reportService.getRootReports(parentReportId)
             rootReports.ifEmpty {
-                listOf(dbAccess.fetchReportFile(parentReportId))
+                try {
+                    listOf(dbAccess.fetchReportFile(parentReportId))
+                } catch (ex: IllegalStateException) {
+                    emptyList()
+                }
             }
         } else {
             emptyList()
-        }.map { it.reportId }
+        }.map { it.reportId }.ifEmpty { if (parentReportId != null) listOf(parentReportId) else emptyList() }
 
         return ReportEventData(
             childReportId,
@@ -423,16 +428,23 @@ class ReportStreamEventService(
         parentReportId: UUID,
         parentItemIndex: Int,
         trackingId: String?,
+        senderString: String?,
     ): ItemEventData {
         val submittedIndex = reportService.getRootItemIndex(parentReportId, parentItemIndex) ?: parentItemIndex
-        val rootReport =
-            reportService.getRootReports(parentReportId).firstOrNull() ?: dbAccess.fetchReportFile(parentReportId)
+        val sender = if (senderString != null) {
+            senderString
+        } else {
+            val rootReport =
+                reportService.getRootReports(parentReportId).firstOrNull() ?: dbAccess.fetchReportFile(parentReportId)
+            "${rootReport.sendingOrg}.${rootReport.sendingOrgClient}"
+        }
+
         return ItemEventData(
             childItemIndex,
             parentItemIndex,
             submittedIndex,
             trackingId,
-            "${rootReport.sendingOrg}.${rootReport.sendingOrgClient}"
+            sender
         )
     }
 }
diff --git a/prime-router/src/main/kotlin/fhirengine/azure/FHIRFunctions.kt b/prime-router/src/main/kotlin/fhirengine/azure/FHIRFunctions.kt
@@ -25,6 +25,7 @@ import gov.cdc.prime.router.fhirengine.engine.FHIRTranslator
 import gov.cdc.prime.router.fhirengine.engine.FhirReceiveQueueMessage
 import gov.cdc.prime.router.fhirengine.engine.PrimeRouterQueueMessage
 import gov.cdc.prime.router.fhirengine.engine.ReportPipelineMessage
+import gov.cdc.prime.router.fhirengine.engine.SubmissionSenderNotFound
 import org.apache.commons.lang3.StringUtils
 import org.apache.logging.log4j.kotlin.Logging
 import org.jooq.exception.DataAccessException
@@ -166,6 +167,17 @@ class FHIRFunctions(
             // DB connectivity issues that are resolved without intervention
             logger.error(ex)
             throw ex
+        } catch (ex: SubmissionSenderNotFound) {
+            logger.error(ex)
+            val tableEntity = Submission(
+                ex.reportId.toString(),
+                "Rejected",
+                ex.blobURL,
+                actionLogger.errors.takeIf { it.isNotEmpty() }?.map { it.detail.message }?.toString()
+            )
+            submissionTableService.insertSubmission(tableEntity)
+            queueAccess.sendMessage("${messageContent.messageQueueName}-poison", message)
+            return emptyList()
         } catch (ex: Exception) {
             // We're catching anything else that occurs because the most likely cause is a code or configuration error
             // that will not be resolved if the message is automatically retried