CCM-12616: add messageReference to the event data

lapenna-bjss · lapenna-bjss · commit 105450591c1c · 2025-11-28T10:25:51.000Z
diff --git a/lambdas/mesh-download/src/__tests__/test_processor.py b/lambdas/mesh-download/src/__tests__/test_processor.py
@@ -45,7 +45,7 @@ def create_valid_cloud_event():
         'data': {
             'meshMessageId': 'test_message_123',
             'senderId': 'TEST_SENDER',
-            'workflowId': 'TEST_WORKFLOW'
+            'messageReference': 'ref_001'
         }
     }
 
@@ -419,3 +419,44 @@ def getenv_side_effect(key, default=''):
         assert len(published_events) == 1
         message_uri = published_events[0]['data']['messageUri']
         assert message_uri.startswith('s3://test-pii-bucket/')
+
+    @patch('src.processor.os.getenv')
+    def test_message_reference_mismatch_raises_error(self, mock_getenv):
+        """When messageReference from event doesn't match local_id from MESH, should raise ValueError"""
+        from src.processor import MeshDownloadProcessor
+
+        mesh_client, log, s3_client, event_publisher, document_store = setup_mocks()
+
+        def getenv_side_effect(key, default=None):
+            return {
+                'EVENT_PUBLISHER_EVENT_BUS_ARN': 'arn:aws:events:test',
+                'EVENT_PUBLISHER_DLQ_URL': 'https://sqs.test.com/dlq',
+                'ENVIRONMENT': 'development',
+                'DEPLOYMENT': 'dev-1',
+                'PII_BUCKET': 'test-pii-bucket',
+                'MOCK_MESH_BUCKET': 'test-mock-bucket'
+            }.get(key, default)
+
+        mock_getenv.side_effect = getenv_side_effect
+
+        processor = MeshDownloadProcessor(
+            mesh_client=mesh_client,
+            log=log,
+            s3_client=s3_client,
+            document_store=document_store,
+            event_publisher=event_publisher
+        )
+
+        # Create MESH message with different local_id than event messageReference
+        mesh_message = create_mesh_message(local_id='different_ref')
+        mesh_client.retrieve_message.return_value = mesh_message
+
+        # Event has messageReference='ref_001' (from create_valid_cloud_event)
+        sqs_record = create_sqs_record()
+
+        with pytest.raises(ValueError, match="messageReference mismatch"):
+            processor.process_sqs_message(sqs_record)
+
+        # Should not acknowledge or publish if validation fails
+        mesh_message.acknowledge.assert_not_called()
+        event_publisher.send_events.assert_not_called()
diff --git a/lambdas/mesh-download/src/processor.py b/lambdas/mesh-download/src/processor.py
@@ -104,6 +104,7 @@ def process_sqs_message(self, sqs_record):
             data = validated_event.data
             mesh_message_id = data.meshMessageId
             sender_id = data.senderId
+            message_reference = data.messageReference
 
             if not mesh_message_id:
                 self.__log.error("Missing meshMessageId in event data", sqs_record=sqs_record)
@@ -113,15 +114,15 @@ def process_sqs_message(self, sqs_record):
             logger.info("Processing MESH download request")
 
             # Download and store the MESH message
-            self.download_and_store_message(mesh_message_id, sender_id, logger)
+            self.download_and_store_message(mesh_message_id, sender_id, message_reference, logger)
 
         except Exception as exc:
             self.__log.error("Error processing SQS message",
                             error=str(exc),
                             sqs_record=sqs_record)
             raise
 
-    def download_and_store_message(self, mesh_message_id, sender_id, logger):
+    def download_and_store_message(self, mesh_message_id, sender_id, message_reference, logger):
         """
         Downloads a MESH message and stores it in S3
         """
@@ -135,7 +136,7 @@ def download_and_store_message(self, mesh_message_id, sender_id, logger):
 
             # Extract data from MESH message headers
             sender_mailbox_id = getattr(message, 'sender', '')
-            message_reference = getattr(message, 'local_id', '')
+            local_id = getattr(message, 'local_id', '')
             message_type = getattr(message, 'message_type', '')
             subject = getattr(message, 'subject', '')
             workflow_id = getattr(message, 'workflow_id', '')
@@ -146,10 +147,22 @@ def download_and_store_message(self, mesh_message_id, sender_id, logger):
                 sender=sender_mailbox_id,
                 workflow_id=workflow_id,
                 subject=subject,
-                local_id=message_reference,
-                message_type=message_type
+                local_id=local_id,
+                message_type=message_type,
+                message_reference_from_event=message_reference
             )
 
+            # Validate messageReference matches local_id from MESH
+            if message_reference != local_id:
+                logger.warning(
+                    "messageReference mismatch between event and MESH message",
+                    expected=message_reference,
+                    actual=local_id
+                )
+                raise ValueError(
+                    f"messageReference mismatch: event has '{message_reference}' but MESH message has '{local_id}'"
+                )
+
             logger.info("Processing message based on type")
 
             # Process DATA message (document reference)
diff --git a/lambdas/mesh-poll/src/processor.py b/lambdas/mesh-poll/src/processor.py
@@ -94,14 +94,14 @@ def process_message(self, message):
         sender_mailbox_id = getattr(message, "sender", "")
         workflow_id = getattr(message, "workflow_id", "")
         subject = getattr(message, "subject", "")
-        local_id = getattr(message, "local_id", "")
+        message_reference = getattr(message, "local_id", "")
 
         logger = self.__log.bind(
             message_id=message.id(),
             sender=sender_mailbox_id,
             workflow_id=workflow_id,
             subject=subject,
-            local_id=local_id,
+            local_id=message_reference,
             message_type=message_type,
         )
 
@@ -121,7 +121,8 @@ def process_message(self, message):
             event_detail = {
                 "data": {
                     "meshMessageId": message_id,
-                    "senderId": client_id
+                    "senderId": client_id,
+                    "messageReference": message_reference
                 }
             }
 
diff --git a/utils/event-publisher-py/event_publisher/__tests__/test_event_publisher.py b/utils/event-publisher-py/event_publisher/__tests__/test_event_publisher.py
@@ -46,16 +46,18 @@ def valid_cloud_event():
         'specversion': '1.0',
         'source': '/nhs/england/notify/production/primary/data-plane/digitalletters/mesh',
         'subject': 'customer/920fca11-596a-4eca-9c47-99f624614658/recipient/769acdd4-6a47-496f-999f-76a6fd2c3959',
-        'type': 'uk.nhs.notify.digital.letters.sent.v1',
+        'type': 'uk.nhs.notify.digital.letters.mesh.inbox.message.received.v1',
         'time': '2023-06-20T12:00:00Z',
         'recordedtime': '2023-06-20T12:00:00.250Z',
         'severitynumber': 2,
+        'severitytext': 'INFO',
         'traceparent': '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01',
-        'dataschema': 'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10/digital-letter-base-data.schema.json',
+        'dataschema': 'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10/digital-letters-mesh-inbox-message-received-data.schema.json',
+        'dataschemaversion': '1.0',
         'data': {
-            'digital-letter-id': '123e4567-e89b-12d3-a456-426614174000',
-            'messageReference': 'ref1',
+            'meshMessageId': 'test-123',
             'senderId': 'sender1',
+            'messageReference': 'ref_001'
         },
     }
 
@@ -67,18 +69,19 @@ def valid_cloud_event2():
         'profilepublished': '2025-10',
         'id': '550e8400-e29b-41d4-a716-446655440002',
         'specversion': '1.0',
-        'source': '/nhs/england/notify/development/primary/data-plane/digital-letters',
+        'source': '/nhs/england/notify/development/primary/data-plane/digitalletters/mesh',
         'subject': 'customer/920fca11-596a-4eca-9c47-99f624614658/recipient/769acdd4-6a47-496f-999f-76a6fd2c3959',
-        'type': 'uk.nhs.notify.digital.letters.sent.v2',
+        'type': 'uk.nhs.notify.digital.letters.mesh.inbox.message.received.v1',
         'time': '2023-06-20T12:00:00Z',
         'recordedtime': '2023-06-20T12:00:00.250Z',
         'severitynumber': 2,
         'traceparent': '00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01',
-        'dataschema': 'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10/digital-letter-base-data.schema.json',
+        'dataschema': 'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10/digital-letters-mesh-inbox-message-received-data.schema.json',
+        'dataschemaversion': '1.0',
         'data': {
-            'digital-letter-id': '123e4567-e89b-12d3-a456-426614174001',
-            'messageReference': 'ref1',
+            'meshMessageId': 'test-123',
             'senderId': 'sender1',
+            'messageReference': 'ref_001'
         },
     }
 
@@ -101,11 +104,14 @@ def test_should_return_empty_array_when_no_events_provided(self, test_config, mo
         mock_events_client.put_events.assert_not_called()
         mock_sqs_client.send_message_batch.assert_not_called()
 
-    def test_should_send_valid_events_to_eventbridge(self, test_config, mock_events_client, valid_cloud_event, valid_cloud_event2):
+    def test_should_send_valid_events_to_eventbridge(self, test_config, mock_events_client, mock_sqs_client, valid_cloud_event, valid_cloud_event2):
         mock_events_client.put_events.return_value = {
             'FailedEntryCount': 0,
             'Entries': [{'EventId': 'event-1'}]
         }
+        mock_sqs_client.send_message_batch.return_value = {
+            'Successful': []
+        }
 
         publisher = EventPublisher(**test_config)
         result = publisher.send_events([valid_cloud_event, valid_cloud_event2])
@@ -154,6 +160,7 @@ def test_should_send_failed_eventbridge_events_to_dlq(self, test_config, mock_ev
 
         assert result == []
         assert mock_events_client.put_events.call_count == 1
+        # Should call DLQ once for the failed event
         assert mock_sqs_client.send_message_batch.call_count == 1
 
         # Verify EventBridge was called with both events
@@ -180,6 +187,7 @@ def test_should_handle_eventbridge_send_error_and_send_all_events_to_dlq(self, t
 
         assert result == []
         assert mock_events_client.put_events.call_count == 1
+        # Should call DLQ once for all events after EventBridge failure
         assert mock_sqs_client.send_message_batch.call_count == 1
 
     def test_should_return_failed_events_when_dlq_also_fails(self, test_config, mock_sqs_client, invalid_cloud_event):
@@ -280,11 +288,14 @@ def test_should_throw_error_when_dlq_url_is_missing(self, test_config):
         with pytest.raises(ValueError, match='dlq_url has not been specified'):
             EventPublisher(**test_config)
 
-    def test_should_be_reusable_for_multiple_calls(self, test_config, mock_events_client, valid_cloud_event, valid_cloud_event2):
+    def test_should_be_reusable_for_multiple_calls(self, test_config, mock_events_client, mock_sqs_client, valid_cloud_event, valid_cloud_event2):
         mock_events_client.put_events.return_value = {
             'FailedEntryCount': 0,
             'Entries': [{'EventId': 'event-1'}]
         }
+        mock_sqs_client.send_message_batch.return_value = {
+            'Successful': []
+        }
 
         publisher = EventPublisher(**test_config)
 
diff --git a/utils/event-publisher-py/event_publisher/models.py b/utils/event-publisher-py/event_publisher/models.py
@@ -174,6 +174,7 @@ class MeshInboxMessageData(BaseModel):
     """Data payload for MESH inbox message received event"""
     meshMessageId: str = Field(..., min_length=1)
     senderId: str = Field(..., min_length=1)
+    messageReference: str = Field(..., min_length=1)
 
 
 class MeshInboxMessageEvent(CloudEvent):
