New: [AEA-5280] - Push notify data to dynamo (#1598)

## Summary

- Routine Change

### Details

When the notify processor pulls a message off the queue, log that
activity in a dynamo data store.

Author: wildjames

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

@@ -128,6 +128,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,12 @@ 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,
+  clearCompletedSQSMessages,
+  drainQueue,
+  PSUDataItemMessage
+} from "./utils"
 
 const logger = new Logger({serviceName: "nhsNotify"})
 
@@ -21,7 +25,7 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
 
   logger.info("NHS Notify lambda triggered by scheduler", {event})
 
-  let messages
+  let messages: Array<PSUDataItemMessage>
   try {
     messages = await drainQueue(logger, 100)
 
@@ -30,19 +34,9 @@ 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)
-
-    const toNotify = items.map((m) => ({
-      RequestID: m.RequestID,
-      TaskId: m.TaskID,
+    const toNotify = messages.map((m) => ({
+      RequestID: m.PSUDataItem.RequestID,
+      TaskId: m.PSUDataItem.TaskID,
       Message: "Notification Required"
     }))
     logger.info("Fetched prescription notification messages", {count: toNotify.length, toNotify})
@@ -58,10 +52,17 @@ export const lambdaHandler = async (event: EventBridgeEvent<string, string>): Pr
     throw err
   }
 
+  try {
+    await addPrescriptionMessagesToNotificationStateStore(logger, messages)
+  } 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, messages)
   } catch (err) {
     logger.error("Error while deleting successfully processed messages from SQS", {error: err})
     throw err