CCM-12875: update so invalid events are removed before attempting an upload

Author: Ian-Hodges

lambdas/pdm-uploader-lambda/src/__tests__/apis/sqs-trigger-lambda.test.ts

@@ -186,6 +186,34 @@ describe('sqs-trigger-lambda', () => {
       );
     });
 
+    it('should handle unexpected error in validateRecord', async () => {
+      const handler = createHandler({
+        uploadToPdm: mockUploadToPdm,
+        eventPublisher: mockEventPublisher,
+        logger: mockLogger,
+      });
+      const sqsEvent = createValidSQSEvent({
+        Records: [
+          {
+            ...createValidSQSEvent().Records[0],
+            body: 'I-am-not-json',
+          },
+        ],
+      });
+
+      const result = await handler(sqsEvent);
+
+      expect(result.batchItemFailures).toEqual([{ itemIdentifier: 'msg-1' }]);
+      expect(mockLogger.error).toHaveBeenCalledWith(
+        expect.objectContaining({
+          description: 'Error parsing SQS record',
+          err: expect.objectContaining({
+            message: expect.stringContaining('Unexpected token'),
+          }),
+        }),
+      );
+    });
+
     it('should handle invalid message body', async () => {
       const handler = createHandler({
         uploadToPdm: mockUploadToPdm,

lambdas/pdm-uploader-lambda/src/__tests__/infra/pdm-api-client.test.ts

@@ -155,10 +155,7 @@ describe('PdmClient', () => {
       conditionalRetry.mockRejectedValue(mockError);
 
       await expect(
-        pdmClient.createDocumentReference(
-          mockFhirRequest,
-          mockRequestId,
-        ),
+        pdmClient.createDocumentReference(mockFhirRequest, mockRequestId),
       ).rejects.toThrow('Network error');
 
       expect(mockLogger.error).toHaveBeenCalledWith({

lambdas/pdm-uploader-lambda/src/apis/sqs-trigger-lambda.ts

@@ -12,7 +12,10 @@ import type {
 import messageDownloadedValidator from 'digital-letters-events/MESHInboxMessageDownloaded.js';
 import pdmResourceSubmittedValidator from 'digital-letters-events/PDMResourceSubmitted.js';
 import pdmResourceSubmissionRejectedValidator from 'digital-letters-events/PDMResourceSubmissionRejected.js';
-import { MESHInboxMessageDownloaded, PDMResourceSubmitted } from 'digital-letters-events';
+import {
+  MESHInboxMessageDownloaded,
+  PDMResourceSubmitted,
+} from 'digital-letters-events';
 import { EventPublisher, Logger } from 'utils';
 
 interface ProcessingResult {
@@ -26,12 +29,15 @@ interface CreateHandlerDependencies {
   logger: Logger;
 }
 
-async function processRecord(
+interface ValidatedRecord {
+  messageId: string;
+  event: MESHInboxMessageDownloaded;
+}
+
+function validateRecord(
   { body, messageId }: { body: string; messageId: string },
-  uploadToPdm: UploadToPdm,
   logger: Logger,
-  batchItemFailures: SQSBatchItemFailure[],
-): Promise<ProcessingResult> {
+): ValidatedRecord | null {
   try {
     const sqsEventBody = JSON.parse(body);
     const sqsEventDetail = sqsEventBody.detail;
@@ -42,20 +48,34 @@ async function processRecord(
         err: messageDownloadedValidator.errors,
         description: 'Error parsing queue entry',
       });
-      batchItemFailures.push({ itemIdentifier: messageId });
-      return { result: { outcome: 'failed' }, item: sqsEventDetail };
+      return null;
     }
 
-    const messageDownloadedEvent: MESHInboxMessageDownloaded = sqsEventDetail;
+    return { messageId, event: sqsEventDetail };
+  } catch (error) {
+    logger.error({
+      err: error,
+      description: 'Error parsing SQS record',
+    });
+    return null;
+  }
+}
 
-    const result = await uploadToPdm.send(messageDownloadedEvent);
+async function processRecord(
+  { event, messageId }: ValidatedRecord,
+  uploadToPdm: UploadToPdm,
+  logger: Logger,
+  batchItemFailures: SQSBatchItemFailure[],
+): Promise<ProcessingResult> {
+  try {
+    const result = await uploadToPdm.send(event);
 
     if (result.outcome === 'failed') {
       batchItemFailures.push({ itemIdentifier: messageId });
-      return { result: { outcome: 'failed' }, item: sqsEventDetail };
+      return { result: { outcome: 'failed' }, item: event };
     }
 
-    return { result, item: sqsEventDetail };
+    return { result, item: event };
   } catch (error) {
     logger.error({
       err: error,
@@ -120,24 +140,25 @@ async function publishSuccessfulEvents(
   if (successfulItems.length === 0) return;
 
   try {
-    const submittedFailedEvents = await eventPublisher.sendEvents<PDMResourceSubmitted>(
-      successfulItems.map(({ event, resourceId }) => ({
-        ...event,
-        id: randomUUID(),
-        time: new Date().toISOString(),
-        recordedtime: new Date().toISOString(),
-        type: 'uk.nhs.notify.digital.letters.pdm.resource.submitted.v1',
-        dataschema:
-          'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10-draft/data/digital-letters-pdm-resource-submitted-data.schema.json',
-        source: event.source.replace(/\/mesh$/, '/pdm'),
-        data: {
-          messageReference: event.data.messageReference,
-          senderId: event.data.senderId,
-          resourceId,
-        },
-      })),
-      pdmResourceSubmittedValidator,
-    );
+    const submittedFailedEvents =
+      await eventPublisher.sendEvents<PDMResourceSubmitted>(
+        successfulItems.map(({ event, resourceId }) => ({
+          ...event,
+          id: randomUUID(),
+          time: new Date().toISOString(),
+          recordedtime: new Date().toISOString(),
+          type: 'uk.nhs.notify.digital.letters.pdm.resource.submitted.v1',
+          dataschema:
+            'https://notify.nhs.uk/cloudevents/schemas/digital-letters/2025-10-draft/data/digital-letters-pdm-resource-submitted-data.schema.json',
+          source: event.source.replace(/\/mesh$/, '/pdm'),
+          data: {
+            messageReference: event.data.messageReference,
+            senderId: event.data.senderId,
+            resourceId,
+          },
+        })),
+        pdmResourceSubmittedValidator,
+      );
     if (submittedFailedEvents.length > 0) {
       logger.warn({
         description: 'Some successful events failed to publish',
@@ -203,7 +224,17 @@ export const createHandler = ({
   async function handler(sqsEvent: SQSEvent): Promise<SQSBatchResponse> {
     const batchItemFailures: SQSBatchItemFailure[] = [];
 
-    const promises = sqsEvent.Records.map((record) =>
+    const validatedRecords: ValidatedRecord[] = [];
+    for (const record of sqsEvent.Records) {
+      const validated = validateRecord(record, logger);
+      if (validated) {
+        validatedRecords.push(validated);
+      } else {
+        batchItemFailures.push({ itemIdentifier: record.messageId });
+      }
+    }
+
+    const promises = validatedRecords.map((record) =>
       processRecord(record, uploadToPdm, logger, batchItemFailures),
     );
 

lambdas/pdm-uploader-lambda/src/app/upload-to-pdm.ts

@@ -18,7 +18,7 @@ export class UploadToPdm {
   async send(event: MESHInboxMessageDownloaded): Promise<UploadToPdmResult> {
     try {
       const fhirRequest = await getS3ObjectFromUri(event.data.messageUri);
-      const messageReference = event.data.messageReference;
+      const { messageReference } = event.data;
 
       const response = await this.pdmClient.createDocumentReference(
         fhirRequest,