CCM-14566: Add FHIR validation to mesh-download (#276)

* CCM-14566: Add FHIR validation to mesh-download

* CCM-14566: Update component tests

---------

Co-authored-by: Gareth Allan <157592212+gareth-allan@users.noreply.github.com>

Author: simonlabarere

infrastructure/terraform/components/dl/data/failure_codes.csv

@@ -2,4 +2,6 @@ code,description
 DL_PDMV_001,Letter rejected by PDM
 DL_PDMV_002,Timeout waiting for letter storage
 DL_CLIV_003,Attachment contains a virus
+DL_CLIV_004,Duplicate request
+DL_CLIV_005,Invalid FHIR resource
 DL_INTE_001,Request rejected by Core API

infrastructure/terraform/components/dl/s3_object_failure_codes.tf

@@ -1,6 +1,3 @@
-# Auto-generated CSV containing failure code definitions
-# Source: src/digital-letters-events/failure-codes.ts
-# Build: make build / make generate (runs generate-dependencies)
 resource "aws_s3_object" "failure_codes" {
   bucket       = module.s3bucket_reporting.bucket
   key          = "reference-data/failure_codes/failure_codes.csv"

lambdas/mesh-download/mesh_download/__tests__/test_processor.py

@@ -70,6 +70,58 @@ def create_sqs_record(cloud_event=None):
         'body': json.dumps({'detail': cloud_event})
     }
 
+def create_fhir_content():
+    """
+    Create a mock FHIR JSON content for testing
+    """
+    return json.dumps({
+        "resourceType": "DocumentReference",
+        "id": "82bfb7f3-4889-4e15-b308-bbe4e3cd431f",
+        "status": "current",
+        "docStatus": "final",
+        "type": {
+            "coding": [
+            {
+                "system": "http://snomed.info/sct",
+                "code": "308540004",
+                "display": "Appointment"
+            }
+            ]
+        },
+        "subject": {
+            "identifier": {
+            "system": "https://fhir.nhs.uk/Id/nhs-number",
+            "value": "9876543210"
+            }
+        },
+        "author": [
+            {
+            "identifier": {
+                "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+                "value": "RX809"
+            },
+            "display": "Example NHS Trust"
+            }
+        ],
+        "custodian": {
+            "identifier": {
+            "system": "https://fhir.nhs.uk/Id/ods-organization-code",
+            "value": "C4L8E"
+            },
+            "display": "NHS ENGLAND: NHS NOTIFY"
+        },
+        "date": "2025-11-19T14:30:00Z",
+        "description": "Appointment notification letter for outpatient consultation",
+        "content": [
+            {
+                "attachment": {
+                    "contentType": "application/pdf",
+                    "title": "Appointment Letter - November 2025",
+                    "data": "base64here=="
+                    }
+            }
+        ]
+    })
 
 def create_mesh_message(message_id='test_123', sender='SENDER_001', local_id='ref_001'):
     """
@@ -82,8 +134,9 @@ def create_mesh_message(message_id='test_123', sender='SENDER_001', local_id='re
     message.subject = 'test_document.pdf'
     message.workflow_id = 'TEST_WORKFLOW'
     message.message_type = 'DATA'
-    message.read.return_value = b'Test message content'
+    message.read.return_value = create_fhir_content()
     message.acknowledge = Mock()
+
     return message
 
 
@@ -148,7 +201,7 @@ def test_process_sqs_message_success(self, mock_datetime):
             sender_id='TEST-SENDER',
             message_reference='ref-001',
             mesh_message_id='test-message-123',
-            content=b'Test message content'
+            content=create_fhir_content()
         )
 
         mesh_message.acknowledge.assert_called_once()
@@ -183,6 +236,76 @@ def test_process_sqs_message_success(self, mock_datetime):
         assert event_data['messageUri'] == 's3://test-pii-bucket/document-reference/SENDER-001/ref-001_test-message-123'
         assert set(event_data.keys()) == {'senderId', 'messageReference', 'messageUri', 'meshMessageId'}
 
+    @patch('mesh_download.processor.datetime')
+    def test_process_sqs_message_invalid_fhir_content(self, mock_datetime):
+        from mesh_download.processor import MeshDownloadProcessor
+
+        config, log, event_publisher, document_store = setup_mocks()
+
+        fixed_time = datetime(2025, 11, 19, 15, 30, 45, tzinfo=timezone.utc)
+        mock_datetime.now.return_value = fixed_time
+
+        document_store.store_document.return_value = 'document-reference/SENDER_001_ref_001'
+
+        event_publisher.send_events.return_value = []
+
+        processor = MeshDownloadProcessor(
+            config=config,
+            log=log,
+            mesh_client=config.mesh_client,
+            download_metric=config.download_metric,
+            duplicate_download_metric=config.duplicate_download_metric,
+            document_store=document_store,
+            event_publisher=event_publisher
+        )
+
+        mesh_message = create_mesh_message()
+        mesh_message.read.return_value = '{}'  # invalid FHIR content (empty JSON)}
+        config.mesh_client.retrieve_message.return_value = mesh_message
+
+        sqs_record = create_sqs_record()
+
+        processor.process_sqs_message(sqs_record)
+
+        config.mesh_client.retrieve_message.assert_called_once_with('test-message-123')
+
+        mesh_message.read.assert_called_once()
+
+        document_store.store_document.assert_not_called()
+
+        mesh_message.acknowledge.assert_called_once()
+
+        config.download_metric.record.assert_not_called()
+
+        event_publisher.send_events.assert_called_once()
+
+        # Verify the published event content
+        published_events = event_publisher.send_events.call_args[0][0]
+        assert len(published_events) == 1
+
+        published_event = published_events[0]
+
+        # Verify CloudEvent envelope fields
+        assert published_event['type'] == 'uk.nhs.notify.digital.letters.mesh.inbox.message.invalid.v1'
+        assert published_event['source'] == '/nhs/england/notify/development/primary/data-plane/digitalletters/mesh'
+        assert published_event['subject'] == 'customer/00000000-0000-0000-0000-000000000000/recipient/00000000-0000-0000-0000-000000000000'
+        assert published_event['time'] == '2025-11-19T15:30:45+00:00'
+        assert 'id' in published_event
+        assert 'tracestate' not in published_event
+        assert 'partitionkey' not in published_event
+        assert 'sequence' not in published_event
+        assert 'dataclassification' not in published_event
+        assert 'dataregulation' not in published_event
+        assert 'datacategory' not in published_event
+
+        # Verify CloudEvent data payload
+        event_data = published_event['data']
+        assert event_data['senderId'] == 'TEST-SENDER'
+        assert event_data['messageReference'] == 'ref-001'
+        assert event_data['meshMessageId'] == 'test-message-123'
+        assert event_data['failureCode'] == 'DL_CLIV_005'
+        assert set(event_data.keys()) == {'senderId', 'messageReference', 'meshMessageId', 'failureCode'}
+
     def test_process_sqs_message_validation_failure(self):
         """Malformed CloudEvents should be rejected by pydantic and not trigger downloads"""
         from mesh_download.processor import MeshDownloadProcessor

lambdas/mesh-download/mesh_download/processor.py

@@ -3,9 +3,10 @@
 from uuid import uuid4
 
 from pydantic import ValidationError
-from digital_letters_events import MESHInboxMessageDownloaded, MESHInboxMessageReceived
+from digital_letters_events import MESHInboxMessageDownloaded, MESHInboxMessageReceived, MESHInboxMessageInvalid
 from mesh_download.errors import MeshMessageNotFound
 from mesh_download.document_store import DocumentAlreadyExistsError
+from nhs_notify_letters_onboarding import validate
 
 
 class MeshDownloadProcessor:
@@ -54,6 +55,10 @@ def _parse_and_validate_event(self, sqs_record):
             )
             raise
 
+    def _validate_fhir_content(self, content):
+        json_content = json.loads(content)
+        validate(json_content)
+
     def _handle_download(self, event, logger):
         data = event.data
 
@@ -74,6 +79,18 @@ def _handle_download(self, event, logger):
         content = message.read()
         logger.info("Downloaded MESH message content")
 
+        try:
+            self._validate_fhir_content(content)
+        except Exception as e:
+            logger.error("FHIR content is invalid", error=str(e))
+
+            self._publish_message_invalid_event(incoming_event=event)
+
+            message.acknowledge()
+            logger.info("Acknowledged message")
+
+            return
+
         duplicate = False
         try:
             uri = self._store_message_content(
@@ -155,3 +172,39 @@ def _publish_downloaded_event(self, incoming_event, message_uri):
             message_uri=message_uri,
             message_reference=incoming_event.data.messageReference
         )
+
+    def _publish_message_invalid_event(self, incoming_event):
+        """
+        Publishes a MESHInboxMessageInvalid event.
+        """
+        now = datetime.now(timezone.utc).isoformat()
+
+        cloud_event = {
+            **incoming_event.model_dump(exclude_none=True),
+            'id': str(uuid4()),
+            'time': now,
+            'recordedtime': now,
+            'type': 'uk.nhs.notify.digital.letters.mesh.inbox.message.invalid.v1',
+            'dataschema': (
+                'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10-draft/data/'
+                'digital-letters-mesh-inbox-message-invalid-data.schema.json'
+            ),
+            'data': {
+                'senderId': incoming_event.data.senderId,
+                'meshMessageId': incoming_event.data.meshMessageId,
+                'failureCode': 'DL_CLIV_005',
+                'messageReference': incoming_event.data.messageReference,
+            }
+        }
+
+        failed = self.__event_publisher.send_events([cloud_event], MESHInboxMessageInvalid)
+        if failed:
+            msg = f"Failed to publish MESHInboxMessageInvalid event: {failed}"
+            self.__log.error(msg, failed_count=len(failed))
+            raise RuntimeError(msg)
+
+        self.__log.info(
+            "Published MESHInboxMessageInvalid event",
+            sender_id=incoming_event.data.senderId,
+            message_reference=incoming_event.data.messageReference
+        )

lambdas/mesh-download/requirements.txt

@@ -8,6 +8,7 @@ urllib3>=1.26.19,<2.0.0
 idna>=3.7
 requests>=2.32.0
 pyopenssl>=24.2.1
+nhs-notify-digital-letters-onboarding @ git+https://github.com/NHSDigital/nhs-notify-digital-letters-onboarding@0.1.0
 -e ../../src/digital-letters-events
 -e ../../utils/py-mock-mesh
 -e ../../utils/py-utils

src/cloudevents/domains/digital-letters/2025-10-draft/data/digital-letters-mesh-inbox-message-invalid-data.schema.yaml

@@ -10,6 +10,8 @@ properties:
     $ref: ../defs/requests.schema.yaml#/properties/senderId
   failureCode:
     $ref: ../defs/requests.schema.yaml#/properties/failureCode
+  messageReference:
+    $ref: ../defs/requests.schema.yaml#/properties/messageReference
 required:
   - meshMessageId
   - senderId

tests/playwright/constants/backend-constants.ts

@@ -61,7 +61,6 @@ export const REPORTING_S3_BUCKET_NAME = `nhs-${process.env.AWS_ACCOUNT_ID}-${REG
 export const PREFIX_DL_FILES = `${CSI}/`;
 
 // Cloudwatch
-export const MESH_DOWNLOAD_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-mesh-download`;
 export const PDM_UPLOADER_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-pdm-uploader`;
 export const PDM_POLL_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-pdm-poll`;
 export const CORE_NOTIFIER_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-core-notifier`;
@@ -70,6 +69,7 @@ export const PRINT_STATUS_HANDLER_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-pr
 export const PRINT_ANALYSER_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-print-analyser`;
 export const PRINT_SENDER_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-print-sender`;
 export const MOVE_SCANNED_FILES_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-move-scanned-files`;
+export const MESH_DOWNLOAD_LAMBDA_LOG_GROUP_NAME = `/aws/lambda/${CSI}-mesh-download`;
 
 // Data Firehose
 export const FIREHOSE_STREAM_NAME = `${CSI}-to-s3-reporting`;