New: [AEA-5274] - Deduplicate notification queue by NHS Number (#1562)

## Summary

- ✨ New Feature

### Details

We deduplicate using SQS FIFO deduplication IDs. This is a salted hash
of the patient NHS number, and the ODS code. So, one notification per
patient, per dispensing site, per 5 minute SQS deduplication window.

Note that I'm technically not using a proper salted hash, since that
would use a random salt. Since we need hashes to be deterministic,
instead we use a secret value that's generated once during deployment
and re-used for all hashes thereafter.

Author: wildjames

SAMtemplates/functions/main.yaml

@@ -60,6 +60,17 @@ Conditions:
     - !Ref DeployCheckPrescriptionStatusUpdate
 
 Resources:
+  SQSSaltSecret:
+    Type: AWS::SecretsManager::Secret
+    Properties:
+      Name: !Sub ${StackName}-SqsSalt
+      Description: Auto-generated salt for SQS_SALT
+      GenerateSecretString:
+        SecretStringTemplate: "{}"
+        GenerateStringKey: salt
+        PasswordLength: 32
+        ExcludePunctuation: true
+
   UpdatePrescriptionStatus:
     Type: AWS::Serverless::Function
     Properties:
@@ -71,6 +82,7 @@ Resources:
         Variables:
           TABLE_NAME: !Ref PrescriptionStatusUpdatesTableName
           NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL: !Ref NHSNotifyPrescriptionsSQSQueueUrl
+          SQS_SALT: !Sub "{{resolve:secretsmanager:${SQSSaltSecret}:SecretString:salt}}"
           LOG_LEVEL: !Ref LogLevel
           ENVIRONMENT: !Ref Environment
           TEST_PRESCRIPTIONS_1: "None"

SAMtemplates/messaging/main.yaml

@@ -48,7 +48,9 @@ Resources:
   NHSNotifyPrescriptionsSQSQueue:
     Type: AWS::SQS::Queue
     Properties:
-      QueueName: !Sub ${StackName}-NHSNotifyPrescriptions
+      QueueName: !Sub ${StackName}-NHSNotifyPrescriptions.fifo
+      FifoQueue: true
+      ContentBasedDeduplication: false
       KmsMasterKeyId: !Ref NotificationSQSQueueKMSKeyAlias
       MessageRetentionPeriod: 86400    # 1 day in seconds
       RedrivePolicy:
@@ -59,7 +61,9 @@ Resources:
   NHSNotifyPrescriptionsDeadLetterQueue:
     Type: AWS::SQS::Queue
     Properties:
-      QueueName: !Sub ${StackName}-NHSNotifyPrescriptionsDeadLetter
+      QueueName: !Sub ${StackName}-NHSNotifyPrescriptionsDeadLetter.fifo
+      FifoQueue: true
+      ContentBasedDeduplication: false
       KmsMasterKeyId: !Ref NotificationSQSQueueKMSKeyAlias
       MessageRetentionPeriod: 604800   # 1 week in seconds
       VisibilityTimeout: 300
@@ -80,7 +84,7 @@ Resources:
               - kms:GenerateDataKey
               - kms:Decrypt
             Resource: !GetAtt NHSNotifyPrescriptionsSQSQueue.Arn
-            
+
   WriteNHSNotifyPrescriptionsSQSQueuePolicy:
     Type: AWS::IAM::ManagedPolicy
     Properties:
@@ -96,7 +100,7 @@ Resources:
               - kms:GenerateDataKey
               - kms:Decrypt
             Resource: !GetAtt NHSNotifyPrescriptionsSQSQueue.Arn
-            
+
 Outputs:
   NHSNotifyPrescriptionsSQSQueueUrl:
     Description: The URL of the NHS Notify Prescriptions SQS Queue

packages/nhsNotifyLambda/src/nhsNotifyLambda.ts

@@ -7,7 +7,7 @@ import inputOutputLogger from "@middy/input-output-logger"
 import errorHandler from "@nhs/fhir-middy-error-handler"
 
 import {PSUDataItem} from "@PrescriptionStatusUpdate_common/commonTypes"
-import {drainQueue} from "./utils"
+import {clearCompletedSQSMessages, drainQueue} from "./utils"
 
 const logger = new Logger({serviceName: "nhsNotify"})
 
@@ -21,8 +21,9 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
 
   logger.info("NHS Notify lambda triggered by scheduler", {event})
 
+  let messages
   try {
-    const messages = await drainQueue(logger, 100)
+    messages = await drainQueue(logger, 100)
 
     if (messages.length === 0) {
       logger.info("No messages to process")
@@ -56,6 +57,15 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
     logger.error("Error while draining SQS queue", {error: err})
     throw err
   }
+
+  // By waiting until a message is successfully processed before deleting it from SQS,
+  // failed messages will eventually be retried by subsequent notify consumers.
+  try {
+    await clearCompletedSQSMessages(messages, logger)
+  } catch (err) {
+    logger.error("Error while deleting successfully processed messages from SQS", {error: err})
+    throw err
+  }
 }
 
 export const handler = middy(lambdaHandler)

packages/nhsNotifyLambda/src/utils.ts

@@ -22,7 +22,7 @@ const docClient = DynamoDBDocumentClient.from(dynamo)
  * Pulls up to `maxTotal` messages off the queue (in batches of up to 10),
  * logs them, and deletes them.
  */
-export async function drainQueue(logger: Logger, maxTotal = 100) {
+export async function drainQueue(logger: Logger, maxTotal = 100): Promise<Array<Message>> {
   let receivedSoFar = 0
   const allMessages: Array<Message> = []
 
@@ -47,25 +47,43 @@ export async function drainQueue(logger: Logger, maxTotal = 100) {
 
     allMessages.push(...Messages)
     receivedSoFar += Messages.length
+  }
 
-    // delete this batch of messages from the queue
-    const deleteEntries = Messages.map((m) => ({
-      Id: m.MessageId!,
-      ReceiptHandle: m.ReceiptHandle!
-    }))
-    const deleteCmd = new DeleteMessageBatchCommand({
-      QueueUrl: sqsUrl,
-      Entries: deleteEntries
-    })
-    const delResult = await sqs.send(deleteCmd)
+  return allMessages
+}
 
-    if (delResult.Failed) {
-      logger.error("Some messages failed to delete", {failed: delResult.Failed})
-      throw new Error("Failed to delete fetched messages from SQS")
-    }
+/**
+ * For each message given, delete it from the notifications SQS. Throws an error if it fails
+ *
+ * @param messages - The messages that were received from SQS, and are to be deleted.
+ * @param logger - the logging object
+ */
+export async function clearCompletedSQSMessages(
+  messages: Array<Message>,
+  logger: Logger
+): Promise<void> {
+  if (!sqsUrl) {
+    logger.error("Notifications SQS URL not configured")
+    throw new Error("NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set")
   }
 
-  return allMessages
+  const deleteMessages = messages.map((m) => ({
+    Id: m.MessageId!,
+    ReceiptHandle: m.ReceiptHandle!
+  }))
+
+  const deleteCmd = new DeleteMessageBatchCommand({
+    QueueUrl: sqsUrl,
+    Entries: deleteMessages
+  })
+  const delResult = await sqs.send(deleteCmd)
+
+  if (delResult.Failed) {
+    logger.error("Some messages failed to delete", {failed: delResult.Failed})
+    throw new Error("Failed to delete fetched messages from SQS")
+  }
+
+  logger.info("Successfully deleted messages from SQS", {result: delResult})
 }
 
 export async function addPrescriptionToNotificationStateStore(logger: Logger, dataArray: Array<PSUDataItem>) {

packages/nhsNotifyLambda/tests/testNhsNotifyLambda.test.ts

@@ -7,11 +7,13 @@ import {
 } from "@jest/globals"
 
 const mockDrainQueue = jest.fn()
+const mockClearCompletedSQSMessages = jest.fn()
 jest.unstable_mockModule(
   "../src/utils",
   async () => ({
     __esModule: true,
-    drainQueue: mockDrainQueue
+    drainQueue: mockDrainQueue,
+    clearCompletedSQSMessages: mockClearCompletedSQSMessages
   })
 )
 
@@ -57,6 +59,51 @@ describe("Unit test for NHS Notify lambda handler", () => {
     expect(mockInfo).toHaveBeenCalledWith("No messages to process")
   })
 
+  it("Clears completed messages after successful processing", async () => {
+    const item1 = {TaskID: "t1", RequestID: "r1"}
+    const item2 = {TaskID: "t2", RequestID: "r2"}
+    const msg1 = {Body: JSON.stringify(item1)}
+    const msg2 = {Body: JSON.stringify(item2)}
+    // drainQueue returns two messages
+    mockDrainQueue.mockImplementationOnce(() => Promise.resolve([msg1, msg2]))
+    // deletion succeeds
+    mockClearCompletedSQSMessages.mockImplementationOnce(() => Promise.resolve(undefined))
+
+    await expect(lambdaHandler(mockEventBridgeEvent)).resolves.not.toThrow()
+
+    // ensure we logged the fetched notifications
+    expect(mockInfo).toHaveBeenCalledWith(
+      "Fetched prescription notification messages",
+      {
+        count: 2,
+        toNotify: [
+          {RequestID: "r1", TaskId: "t1", Message: "Notification Required"},
+          {RequestID: "r2", TaskId: "t2", Message: "Notification Required"}
+        ]
+      }
+    )
+    // ensure clearCompletedSQSMessages was called with the original messages array
+    expect(mockClearCompletedSQSMessages).toHaveBeenCalledWith(
+      [msg1, msg2],
+      expect.any(Object) // the logger instance
+    )
+  })
+
+  it("Throws and logs if clearCompletedSQSMessages fails", async () => {
+    const item = {TaskID: "tx", RequestID: "rx"}
+    const msg = {Body: JSON.stringify(item)}
+    mockDrainQueue.mockImplementationOnce(() => Promise.resolve([msg]))
+    const deletionError = new Error("Delete failed")
+    mockClearCompletedSQSMessages.mockImplementationOnce(() => Promise.reject(deletionError))
+
+    await expect(lambdaHandler(mockEventBridgeEvent)).rejects.toThrow("Delete failed")
+
+    expect(mockError).toHaveBeenCalledWith(
+      "Error while deleting successfully processed messages from SQS",
+      {error: deletionError}
+    )
+  })
+
   it("When drainQueue returns only valid JSON messages, all are processed", async () => {
     const validItem = {
       prescriptionId: "abc123",

packages/nhsNotifyLambda/tests/testUtils.test.ts

@@ -3,13 +3,13 @@ import {SpiedFunction} from "jest-mock"
 
 import {Logger} from "@aws-lambda-powertools/logger"
 import {DynamoDBDocumentClient, PutCommand} from "@aws-sdk/lib-dynamodb"
-import {Message} from "@aws-sdk/client-sqs"
+import {DeleteMessageBatchCommand, Message} from "@aws-sdk/client-sqs"
 
 import {constructPSUDataItem, mockSQSClient} from "./testHelpers"
 
 const {mockSend: sqsMockSend} = mockSQSClient()
 
-const {addPrescriptionToNotificationStateStore, drainQueue} = await import("../src/utils")
+const {addPrescriptionToNotificationStateStore, clearCompletedSQSMessages, drainQueue} = await import("../src/utils")
 
 const ORIGINAL_ENV = {...process.env}
 
@@ -31,13 +31,10 @@ describe("NHS notify lambda helper functions", () => {
     it("Does not throw an error when the SQS fetch succeeds", async () => {
       const payload = {Messages: Array.from({length: 10}, () => (constructPSUDataItem() as Message))}
 
-      // Mock once for the fetch, and once for the delete
-      sqsMockSend
-        .mockImplementationOnce(() => Promise.resolve(payload))
-        .mockImplementationOnce(() => Promise.resolve({Successful: []}))
+      sqsMockSend.mockImplementationOnce(() => Promise.resolve(payload))
 
       const messages = await drainQueue(logger, 10)
-      expect(sqsMockSend).toHaveBeenCalledTimes(2)
+      expect(sqsMockSend).toHaveBeenCalledTimes(1)
       expect(messages).toStrictEqual(payload.Messages)
     })
 
@@ -47,39 +44,92 @@ describe("NHS notify lambda helper functions", () => {
       const messages = await drainQueue(logger, 5)
       expect(messages).toEqual([])
       expect(sqsMockSend).toHaveBeenCalledTimes(1)
-      // no deletion attempted
     })
 
     it("Throws an error if the SQS fetch fails", async () => {
       sqsMockSend.mockImplementation(() => Promise.reject(new Error("Fetch failed")))
       await expect(drainQueue(logger, 10)).rejects.toThrow("Fetch failed")
     })
 
-    it("Throws an error if the delete batch operation fails", async () => {
-      const msg = constructPSUDataItem() as Message
-      // first call: fetch, second call: delete
-      sqsMockSend
-        .mockImplementationOnce(() =>
-          Promise.resolve({Messages: [msg]})
-        )
-        .mockImplementationOnce(() =>
-          Promise.resolve({
-            Failed: [{Id: msg.MessageId!, Message: "del-error", Code: "500"}]
-          })
-        )
-
-      await expect(drainQueue(logger, 1)).rejects.toThrow("Failed to delete fetched messages from SQS")
+    it("Throws an error if the SQS URL is not configured", async () => {
+      delete process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL
+      const {drainQueue} = await import("../src/utils")
+
+      await expect(drainQueue(logger)).rejects.toThrow("NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set")
+      expect(errorSpy).toHaveBeenCalledWith("Notifications SQS URL not configured")
+    })
+  })
+
+  describe("clearCompletedSQSMessages", () => {
+    let logger: Logger
+    let errorSpy: SpiedFunction<(msg: string, ...meta: Array<unknown>) => void>
+
+    beforeEach(() => {
+      jest.resetModules()
+      jest.clearAllMocks()
+
+      process.env = {...ORIGINAL_ENV}
+      logger = new Logger({serviceName: "test-service"})
+      errorSpy = jest.spyOn(logger, "error")
+    })
+
+    it("deletes messages successfully without error", async () => {
+      const messages: Array<Message> = [
+        {MessageId: "msg1", ReceiptHandle: "rh1"},
+        {MessageId: "msg2", ReceiptHandle: "rh2"}
+      ]
+
+      // successful delete (no .Failed)
+      sqsMockSend.mockImplementationOnce(() => Promise.resolve({}))
+
+      await expect(clearCompletedSQSMessages(messages, logger))
+        .resolves
+        .toBeUndefined()
+
+      expect(sqsMockSend).toHaveBeenCalledTimes(1)
+
+      const cmd = sqsMockSend.mock.calls[0][0]
+
+      expect(cmd).toBeInstanceOf(DeleteMessageBatchCommand)
+      expect((cmd as DeleteMessageBatchCommand).input).toEqual({
+        QueueUrl: process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL,
+        Entries: [
+          {Id: "msg1", ReceiptHandle: "rh1"},
+          {Id: "msg2", ReceiptHandle: "rh2"}
+        ]
+      })
+
+      expect(errorSpy).not.toHaveBeenCalled()
+    })
+
+    it("logs and throws if some deletions fail", async () => {
+      const messages: Array<Message> = [
+        {MessageId: "msg1", ReceiptHandle: "rh1"}
+      ]
+      const failedEntries = [
+        {Id: "msg1", SenderFault: true, Code: "Error", Message: "fail"}
+      ]
+
+      // partial failure
+      sqsMockSend.mockImplementationOnce(() => Promise.resolve({Failed: failedEntries}))
+
+      await expect(clearCompletedSQSMessages(messages, logger))
+        .rejects
+        .toThrow("Failed to delete fetched messages from SQS")
+
       expect(errorSpy).toHaveBeenCalledWith(
         "Some messages failed to delete",
-        {failed: expect.any(Array)}
+        {failed: failedEntries}
       )
     })
 
     it("Throws an error if the SQS URL is not configured", async () => {
       delete process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL
-      const {drainQueue} = await import("../src/utils")
+      const {clearCompletedSQSMessages} = await import("../src/utils")
 
-      await expect(drainQueue(logger)).rejects.toThrow("NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set")
+      await expect(clearCompletedSQSMessages([], logger))
+        .rejects
+        .toThrow("NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set")
       expect(errorSpy).toHaveBeenCalledWith("Notifications SQS URL not configured")
     })
   })

packages/updatePrescriptionStatus/.jest/setEnvVars.js

@@ -1,3 +1,4 @@
 /* eslint-disable no-undef */
 process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL = "dummy_notify_sqs";
 process.env.AWS_REGION = "eu-west-2";
+process.env.SQS_SALT = "the quick brown fox something something"