Partially implement the business logic

Author: wildjames

packages/postDatedLambda/src/businessLogic.ts

@@ -1,9 +1,40 @@
 import {Logger} from "@aws-lambda-powertools/logger"
 
-import {PostDatedSQSMessageWithExistingRecords} from "./types"
+import {PSUDataItem} from "@psu-common/commonTypes"
 
+import {PostDatedSQSMessageWithExistingRecords, PostDatedProcessingResult} from "./types"
+
+// defaults to false
 const POST_DATED_OVERRIDE = process.env.POST_DATED_OVERRIDE === "true"
-const POST_DATED_OVERRIDE_VALUE = process.env.POST_DATED_OVERRIDE_VALUE === "true"
+
+// set from environment variable POST_DATED_OVERRIDE_VALUE
+const POST_DATED_OVERRIDE_VALUE_ENV = process.env.POST_DATED_OVERRIDE_VALUE ?? "ignore"
+let POST_DATED_OVERRIDE_VALUE: PostDatedProcessingResult
+switch (POST_DATED_OVERRIDE_VALUE_ENV.toLowerCase()) {
+  case "matured":
+    POST_DATED_OVERRIDE_VALUE = PostDatedProcessingResult.MATURED
+    break
+  case "immature":
+    POST_DATED_OVERRIDE_VALUE = PostDatedProcessingResult.IMMATURE
+    break
+  default:
+    POST_DATED_OVERRIDE_VALUE = PostDatedProcessingResult.IGNORE
+    break
+}
+
+export function getMostRecentRecord(
+  existingRecords: Array<PSUDataItem>
+): PSUDataItem {
+  return existingRecords.reduce((latest, record) => {
+    const latestTimestamp = latest.PostDatedLastModifiedSetAt
+      ? new Date(latest.PostDatedLastModifiedSetAt)
+      : new Date(latest.LastModified)
+    const recordTimestamp = record.PostDatedLastModifiedSetAt
+      ? new Date(record.PostDatedLastModifiedSetAt)
+      : new Date(record.LastModified)
+    return recordTimestamp > latestTimestamp ? record : latest
+  }, existingRecords[0])
+}
 
 /**
  * Process a single post-dated prescription message.
@@ -13,15 +44,14 @@ const POST_DATED_OVERRIDE_VALUE = process.env.POST_DATED_OVERRIDE_VALUE === "tru
  * @param message - The SQS message containing post-dated prescription data and existing records
  * @returns Promise<boolean> - true if the post-dated prescription has matured, and false otherwise
  */
-export async function processMessage(
+export function processMessage(
   logger: Logger,
   message: PostDatedSQSMessageWithExistingRecords
-): Promise<boolean> {
-  logger.info("Processing post-dated prescription message (dummy)", {
+): string {
+  logger.info("Processing post-dated prescription message", {
     messageId: message.MessageId,
     prescriptionData: message.prescriptionData,
-    existingRecordsCount: message.existingRecords.length,
-    existingRecordTaskIds: message.existingRecords.map((r) => r.TaskID)
+    existingRecords: message.existingRecords
   })
   if (POST_DATED_OVERRIDE) {
     logger.info("Post-dated override is enabled, returning override value", {
@@ -30,24 +60,56 @@ export async function processMessage(
     return POST_DATED_OVERRIDE_VALUE
   }
 
-  // TODO: Implement actual business logic for post-dated prescription processing
   // The existingRecords array contains all records from the DynamoDB table
   // that match this prescription's PrescriptionID
 
   // NOTE: It is technically possible for the array to be empty if no existing records are found
-  // This SHOULD never happen in practice, but the code should handle it gracefully just in case
+  // This SHOULD never happen in practice, but catch it anyway
+  if (message.existingRecords.length === 0) {
+    logger.error("No existing records found for post-dated prescription, cannot process", {
+      badMessage: message
+    })
+
+    // throw new Error("No existing records found for post-dated prescription") // maybe?
+    return PostDatedProcessingResult.IGNORE
+  }
+
+  // We only care about the most recent submission for this prescription
+  // If PostDatedLastModifiedSetAt IS set, it is the timestamp we received the submission
+  // If it is NOT set, then LastModified is the timestamp we received the submission
+  const mostRecentRecord = getMostRecentRecord(message.existingRecords)
+
+  logger.info("Most recent NPPTS record for post-dated processing", {
+    mostRecentRecord
+  })
+
+  // Is it post-dated?
+  if (!mostRecentRecord.PostDatedLastModifiedSetAt) {
+    logger.info(
+      "Most recent record is not marked as post-dated, and will have been processed " +
+      "by the standard logic already. Marking as to be ignored by the post-dated notifications lambda."
+    )
+    return PostDatedProcessingResult.IGNORE
+  }
+
+  // Is it still RTC?
+  const mostRecentStatus = mostRecentRecord.Status.toLowerCase()
+  const notifiableStatuses: Array<string> = ["ready to collect", "ready to collect - partial"]
+  if (!notifiableStatuses.includes(mostRecentStatus)) {
+    logger.info("Most recent status in the NPPTS data store is not a notifiable status, so will be ignored", {
+      mostRecentStatus: mostRecentStatus
+    })
+    return PostDatedProcessingResult.IGNORE
+  }
 
-  const mostRecentRecord = message.existingRecords.reduce((latest, record) => {
-    return new Date(record.LastModified) > new Date(latest.LastModified) ? record : latest
-  }, message.existingRecords[0])
   const mostRecentLastModified = new Date(mostRecentRecord.LastModified)
-  const desiredTransitionTime = new Date(mostRecentRecord.PostDatedLastModifiedSetAt as string)
+  const desiredTransitionTime = new Date(mostRecentRecord.PostDatedLastModifiedSetAt)
   const currentTime = new Date()
   logger.info("Post-dated prescription timing details", {
     mostRecentLastModified: mostRecentLastModified.toISOString(),
     desiredTransitionTime: desiredTransitionTime.toISOString(),
     currentTime: currentTime.toISOString()
   })
 
-  return true
+  return PostDatedProcessingResult.MATURED
 }

packages/postDatedLambda/src/orchestration.ts

@@ -3,7 +3,7 @@ import {Logger} from "@aws-lambda-powertools/logger"
 import {processMessage} from "./businessLogic"
 import {enrichMessagesWithExistingRecords} from "./databaseClient"
 import {receivePostDatedSQSMessages, reportQueueStatus, handleProcessedMessages} from "./sqs"
-import {BatchProcessingResult, PostDatedSQSMessage} from "./types"
+import {BatchProcessingResult, PostDatedProcessingResult, PostDatedSQSMessage} from "./types"
 
 export const MAX_QUEUE_RUNTIME = 14 * 60 * 1000 // 14 minutes, to avoid Lambda timeout issues (timeout is 15 minutes)
 const MIN_RECEIVED_THRESHOLD = 3 // If fewer than this number of messages are received, consider the queue empty
@@ -21,24 +21,22 @@ export async function processMessages(
   messages: Array<PostDatedSQSMessage>,
   logger: Logger
 ): Promise<BatchProcessingResult> {
-  if (messages.length === 0) {
-    logger.info("No messages to process in batch")
-    return {maturedPrescriptionUpdates: [], immaturePrescriptionUpdates: []}
-  }
-
   // Enrich messages with existing records from DynamoDB
   const enrichedMessages = await enrichMessagesWithExistingRecords(messages, logger)
 
   const maturedPrescriptionUpdates: Array<PostDatedSQSMessage> = []
   const immaturePrescriptionUpdates: Array<PostDatedSQSMessage> = []
+  const ignoredPrescriptionUpdates: Array<PostDatedSQSMessage> = []
 
   for (const message of enrichedMessages) {
     try {
-      const success = await processMessage(logger, message)
-      if (success) {
+      const action = processMessage(logger, message)
+      if (action === PostDatedProcessingResult.MATURED) {
         maturedPrescriptionUpdates.push(message)
-      } else {
+      } else if (action === PostDatedProcessingResult.IMMATURE) {
         immaturePrescriptionUpdates.push(message)
+      } else {
+        ignoredPrescriptionUpdates.push(message)
       }
     } catch (error) {
       logger.error("Error processing message", {
@@ -52,10 +50,11 @@ export async function processMessages(
   logger.info("Batch processing complete", {
     totalMessages: messages.length,
     maturedPrescriptionUpdatesCount: maturedPrescriptionUpdates.length,
-    immaturePrescriptionUpdatesCount: immaturePrescriptionUpdates.length
+    immaturePrescriptionUpdatesCount: immaturePrescriptionUpdates.length,
+    ignoredPrescriptionUpdatesCount: ignoredPrescriptionUpdates.length
   })
 
-  return {maturedPrescriptionUpdates, immaturePrescriptionUpdates}
+  return {maturedPrescriptionUpdates, immaturePrescriptionUpdates, ignoredPrescriptionUpdates}
 }
 
 /**

packages/postDatedLambda/src/sqs.ts

@@ -243,7 +243,8 @@ export async function sendSQSMessagesToNotificationQueue(
   messages: Array<PostDatedSQSMessage>
 ): Promise<Array<string>> {
   if (messages.length === 0) {
-    logger.info("No matured post-dated messages to forward to notifications queue")
+    // exit early so we don't send a SendMessageBatch with no entries
+    logger.info("No messages to forward to notifications queue")
     return []
   }
 
@@ -272,6 +273,7 @@ export async function removeSQSMessages(
   messages: Array<Message>
 ): Promise<void> {
   if (messages.length === 0) {
+    // exit early so we don't send a DeleteMessageBatch with no entries
     logger.info("No messages to delete")
     return
   }
@@ -319,6 +321,7 @@ export async function returnMessagesToQueue(
   messages: Array<Message>
 ): Promise<void> {
   if (messages.length === 0) {
+    // exit early so we don't send a ChangeMessageVisibilityBatch with no entries
     logger.info("No messages to return to queue")
     return
   }
@@ -377,16 +380,15 @@ export async function handleProcessedMessages(
   result: BatchProcessingResult,
   logger: Logger
 ): Promise<void> {
-  const {maturedPrescriptionUpdates, immaturePrescriptionUpdates} = result
+  const {maturedPrescriptionUpdates, immaturePrescriptionUpdates, ignoredPrescriptionUpdates} = result
 
   // Move matured messages to the notification queue and remove them from the post-dated queue
-  if (maturedPrescriptionUpdates.length > 0) {
-    await sendSQSMessagesToNotificationQueue(logger, maturedPrescriptionUpdates)
-    await removeSQSMessages(logger, maturedPrescriptionUpdates)
-  }
+  await sendSQSMessagesToNotificationQueue(logger, maturedPrescriptionUpdates)
+  await removeSQSMessages(logger, maturedPrescriptionUpdates)
 
   // Return failed messages to the queue
-  if (immaturePrescriptionUpdates.length > 0) {
-    await returnMessagesToQueue(logger, immaturePrescriptionUpdates)
-  }
+  await returnMessagesToQueue(logger, immaturePrescriptionUpdates)
+
+  // Remove ignored messages from the queue, so they are not reprocessed
+  await removeSQSMessages(logger, ignoredPrescriptionUpdates)
 }

packages/postDatedLambda/src/types.ts

@@ -29,10 +29,18 @@ export interface PostDatedSQSMessageWithExistingRecords extends PostDatedSQSMess
   existingRecords: Array<PSUDataItem>
 }
 
+// Enum of strings, "matured", "immature", "ignore"
+export enum PostDatedProcessingResult {
+  MATURED = "matured",
+  IMMATURE = "immature",
+  IGNORE = "ignore"
+}
+
 /**
  * Result of processing a batch of messages.
  */
 export interface BatchProcessingResult {
   maturedPrescriptionUpdates: Array<PostDatedSQSMessage>
   immaturePrescriptionUpdates: Array<PostDatedSQSMessage>
+  ignoredPrescriptionUpdates: Array<PostDatedSQSMessage>
 }