New: [AEA-5206] - Check cooldown when sending a notification

.vscode/eps-prescription-status-update-api.code-workspace

@@ -74,6 +74,7 @@
       "Codeable",
       "codeinline",
       "codesystem",
+      "Cooldown",
       "cpsu",
       "dbaeumer",
       "devcontainer",
@@ -128,6 +129,7 @@
       "sourcetype",
       "timonwong",
       "Truststore",
+      "Upserted",
       "URID",
       "URPID",
       "uuidv4",

SAMtemplates/tables/main.yaml

@@ -417,7 +417,7 @@ Resources:
               - kms:Encrypt
               - kms:ReEncrypt*
               - kms:Decrypt
-            Resource: !GetAtt PrescriptionStatusUpdatesKMSKey.Arn
+            Resource: !GetAtt PrescriptionNotificationStateKMSKey.Arn
 
   PrescriptionNotificationStateTable:
     Type: AWS::DynamoDB::Table
@@ -426,29 +426,19 @@ Resources:
       PointInTimeRecoverySpecification:
         PointInTimeRecoveryEnabled: true
       AttributeDefinitions:
-        - AttributeName: PrescriptionID
-          AttributeType: S
         - AttributeName: NHSNumber
           AttributeType: S
+        - AttributeName: ODSCode
+          AttributeType: S
       KeySchema:
-        - AttributeName: PrescriptionID
-          KeyType: HASH
+        - AttributeName: NHSNumber
+          KeyType: HASH       # Partition key
+        - AttributeName: ODSCode
+          KeyType: RANGE      # Sort key
       BillingMode: !If
         - EnableDynamoDBAutoScalingCondition
         - PROVISIONED
         - PAY_PER_REQUEST
-      GlobalSecondaryIndexes:
-        - IndexName: NotificationNHSNumberIndex
-          KeySchema:
-            - AttributeName: NHSNumber
-              KeyType: HASH
-          Projection:
-            ProjectionType: ALL
-          ProvisionedThroughput: !If
-            - EnableDynamoDBAutoScalingCondition
-            - ReadCapacityUnits: 1
-              WriteCapacityUnits: !Ref MinWritePrescriptionNotificationStateCapacity
-            - !Ref "AWS::NoValue"
       ProvisionedThroughput: !If
         - EnableDynamoDBAutoScalingCondition
         - ReadCapacityUnits: 1
@@ -524,59 +514,6 @@ Resources:
         PredefinedMetricSpecification:
           PredefinedMetricType: DynamoDBReadCapacityUtilization
 
-  # Scaling for the indexes
-  NotificationNHSNumberIndexScalingWriteTarget:
-    Type: AWS::ApplicationAutoScaling::ScalableTarget
-    DependsOn: PrescriptionNotificationStateTable
-    Condition: EnableDynamoDBAutoScalingCondition
-    Properties:
-      MinCapacity: !Ref MinWritePrescriptionNotificationStateCapacity
-      MaxCapacity: !Ref MaxWritePrescriptionNotificationStateCapacity
-      ResourceId: !Sub table/${PrescriptionNotificationStateTable}/index/NotificationNHSNumberIndex
-      RoleARN: !GetAtt DynamoDbScalingRole.Arn
-      ScalableDimension: "dynamodb:index:WriteCapacityUnits"
-      ServiceNamespace: dynamodb
-
-  NotificationNHSNumberIndexScalingWritePolicy:
-    Type: AWS::ApplicationAutoScaling::ScalingPolicy
-    Condition: EnableDynamoDBAutoScalingCondition
-    Properties:
-      PolicyName: NotificationNHSNumberIndexScalingWritePolicy
-      PolicyType: TargetTrackingScaling
-      ScalingTargetId: !Ref NotificationNHSNumberIndexScalingWriteTarget
-      TargetTrackingScalingPolicyConfiguration:
-        TargetValue: 50
-        ScaleInCooldown: 600
-        ScaleOutCooldown: 0
-        PredefinedMetricSpecification:
-          PredefinedMetricType: DynamoDBWriteCapacityUtilization
-
-  NotificationNHSNumberIndexScalingReadTarget:
-    Type: AWS::ApplicationAutoScaling::ScalableTarget
-    DependsOn: PrescriptionNotificationStateTable
-    Condition: EnableDynamoDBAutoScalingCondition
-    Properties:
-      MinCapacity: 1
-      MaxCapacity: 100
-      ResourceId: !Sub table/${PrescriptionNotificationStateTable}/index/NotificationNHSNumberIndex
-      RoleARN: !GetAtt DynamoDbScalingRole.Arn
-      ScalableDimension: "dynamodb:index:ReadCapacityUnits"
-      ServiceNamespace: dynamodb
-
-  NotificationNHSNumberIndexScalingReadPolicy:
-    Type: AWS::ApplicationAutoScaling::ScalingPolicy
-    Condition: EnableDynamoDBAutoScalingCondition
-    Properties:
-      PolicyName: NotificationNHSNumberIndexReadScalingPolicy
-      PolicyType: TargetTrackingScaling
-      ScalingTargetId: !Ref NotificationNHSNumberIndexScalingReadTarget
-      TargetTrackingScalingPolicyConfiguration:
-        TargetValue: 70
-        ScaleInCooldown: 60
-        ScaleOutCooldown: 60
-        PredefinedMetricSpecification:
-          PredefinedMetricType: DynamoDBReadCapacityUtilization
-
 Outputs:
   PrescriptionStatusUpdatesTableName:
     Description: PrescriptionStatusUpdates table name

packages/nhsNotifyLambda/jest.config.ts

@@ -3,8 +3,15 @@ import type {JestConfigWithTsJest} from "ts-jest"
 
 const jestConfig: JestConfigWithTsJest = {
   ...defaultConfig,
-  "rootDir": "./",
-  setupFiles: ["<rootDir>/.jest/setEnvVars.js"]
+  rootDir: "./",
+  setupFiles: ["<rootDir>/.jest/setEnvVars.js"],
+  coveragePathIgnorePatterns: ["<rootDir>/tests/"],
+  coverageReporters: [
+    "clover",
+    "json",
+    "text",
+    ["lcov", {projectRoot: "../../"}]
+  ]
 }
 
 export default jestConfig

packages/nhsNotifyLambda/src/nhsNotifyLambda.ts

@@ -6,8 +6,13 @@ import middy from "@middy/core"
 import inputOutputLogger from "@middy/input-output-logger"
 import errorHandler from "@nhs/fhir-middy-error-handler"
 
-import {PSUDataItem} from "@PrescriptionStatusUpdate_common/commonTypes"
-import {clearCompletedSQSMessages, drainQueue} from "./utils"
+import {
+  addPrescriptionMessagesToNotificationStateStore,
+  checkCooldownForUpdate,
+  clearCompletedSQSMessages,
+  drainQueue,
+  PSUDataItemMessage
+} from "./utils"
 
 const logger = new Logger({serviceName: "nhsNotify"})
 
@@ -21,7 +26,8 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
 
   logger.info("NHS Notify lambda triggered by scheduler", {event})
 
-  let messages
+  let messages: Array<PSUDataItemMessage>
+  let processed: Array<PSUDataItemMessage>
   try {
     messages = await drainQueue(logger, 100)
 
@@ -30,38 +36,63 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
       return
     }
 
-    // parse & log each PSUDataItem as a placeholder for now.
-    const items = messages.map((m) => {
-      try {
-        return JSON.parse(m.Body!) as PSUDataItem
-      } catch (err) {
-        logger.error("Failed to parse message body", {body: m.Body, error: err})
-        return null
-      }
-    }).filter((i): i is PSUDataItem => i !== null)
+    // Filter messages by checkCooldownForUpdate. This is done in two stages so we can check in parallel
+    const eligibility = await Promise.all(
+      messages.map(async (m) => ({
+        message: m,
+        allowed: await checkCooldownForUpdate(logger, m.PSUDataItem)
+      }))
+    )
+    const toProcess = eligibility
+      .filter((e) => e.allowed)
+      .map((e) => e.message)
 
-    const toNotify = items.map((m) => ({
-      RequestID: m.RequestID,
-      TaskId: m.TaskID,
-      Message: "Notification Required"
-    }))
+    // Log the results of checking the cooldown
+    const suppressedCount = messages.length - toProcess.length
+    if (toProcess.length === 0) {
+      logger.info("All messages suppressed by cooldown; nothing to notify",
+        {
+          suppressedCount,
+          totalFetched: messages.length
+        })
+      return
+    } else if (suppressedCount > 0) {
+      logger.info(`Suppressed ${suppressedCount} messages due to cooldown`,
+        {
+          suppressedCount,
+          totalFetched: messages.length
+        }
+      )
+    }
+
+    // Just for diagnostics for now
+    const toNotify = toProcess
+      .map((m) => ({
+        RequestID: m.PSUDataItem.RequestID,
+        TaskId: m.PSUDataItem.TaskID,
+        Message: "Notification Required"
+      }))
     logger.info("Fetched prescription notification messages", {count: toNotify.length, toNotify})
 
-    // TODO: Notifications logic will be done here.
-    // - query PrescriptionNotificationState
-    // - process prescriptions, build NHS notify payload
-    // - Make NHS notify request
-    // Don't forget to make appropriate logs!
+    // TODO: Notifications request will be done here.
+    processed = toProcess
 
   } catch (err) {
     logger.error("Error while draining SQS queue", {error: err})
     throw err
   }
 
+  try {
+    await addPrescriptionMessagesToNotificationStateStore(logger, processed)
+  } catch (err) {
+    logger.error("Error while pushing data to the PSU notification state data store", {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)
+    await clearCompletedSQSMessages(logger, processed)
   } catch (err) {
     logger.error("Error while deleting successfully processed messages from SQS", {error: err})
     throw err