Review yesterdays work. Leave some todo notes for the business logic

Author: wildjames

packages/postDatedLambda/src/businessLogic.ts

@@ -8,7 +8,7 @@ import {PostDatedSQSMessageWithExistingRecords} from "./types"
  *
  * @param logger - The AWS Lambda Powertools logger instance
  * @param message - The SQS message containing post-dated prescription data and existing records
- * @returns Promise<boolean> - true if processing succeeded, false if it failed
+ * @returns Promise<boolean> - true if the post-dated prescription has matured, and false otherwise
  */
 export async function processMessage(
   logger: Logger,

packages/postDatedLambda/src/databaseClient.ts

@@ -19,11 +19,13 @@ function createPrescriptionLookupKey(prescriptionID: string, pharmacyODSCode: st
 
 /**
  * Query the PrescriptionStatusUpdates table for all records matching a given prescription ID and ODS code.
+ * There should always be at least one result, but there may be multiple if the prescription has been
+ * updated multiple times.
  *
  * @param prescriptionID - The prescription ID to query for
  * @param pharmacyODSCode - The pharmacy ODS code to query for
  * @param logger - The AWS Lambda Powertools logger instance
- * @returns Array of PSUDataItem records matching the prescription ID
+ * @returns Array of PSUDataItem records matching the prescription ID. Sorted by LastModified descending.
  */
 export async function getExistingRecordsByPrescriptionID(
   prescriptionID: string,
@@ -76,8 +78,8 @@ export async function getExistingRecordsByPrescriptionID(
       recordCount: items.length
     })
 
-    // Sort by LastModified ascending so most recent is last
-    items.sort((a, b) => new Date(a.LastModified).valueOf() - new Date(b.LastModified).valueOf())
+    // Sort by LastModified ascending so most recent is first
+    items.sort((a, b) => new Date(b.LastModified).valueOf() - new Date(a.LastModified).valueOf())
 
     return items
   } catch (err) {
@@ -96,7 +98,8 @@ export async function getExistingRecordsByPrescriptionID(
  *
  * @param postDatedItems - Array of post-dated prescription data items
  * @param logger - The AWS Lambda Powertools logger instance
- * @returns Array of objects containing both the post-dated data and existing records
+ * @returns Array of objects containing both the post-dated data and existing records.
+ *   Existing records are sorted by LastModified descending.
  */
 export async function fetchExistingRecordsForPrescriptions(
   postDatedItems: Array<PostDatedNotifyDataItem>,
@@ -106,6 +109,8 @@ export async function fetchExistingRecordsForPrescriptions(
     prescriptionCount: postDatedItems.length
   })
 
+  // The data table is indexed by both PrescriptionID and PharmacyODSCode, so build a map keyed by these
+  // in combination. Should avoid duplicate queries this way.
   const uniquePrescriptionLookups = new Map<string, {prescriptionID: string; pharmacyODSCode: string}>()
   for (const item of postDatedItems) {
     const lookupKey = createPrescriptionLookupKey(item.PrescriptionID, item.PharmacyODSCode)
@@ -117,10 +122,7 @@ export async function fetchExistingRecordsForPrescriptions(
     }
   }
 
-  // Create a map of prescription ID to existing records
   const existingRecordsMap = new Map<string, Array<PSUDataItem>>()
-
-  // Fetch existing records for each unique prescription ID
   await Promise.all(
     Array.from(uniquePrescriptionLookups.entries()).map(async ([lookupKey, {prescriptionID, pharmacyODSCode}]) => {
       try {
@@ -133,6 +135,7 @@ export async function fetchExistingRecordsForPrescriptions(
           error
         })
         // Store empty array on error to allow processing to continue
+        // TODO: Make sure later, that if existingRecords is empty, we handle that
         existingRecordsMap.set(lookupKey, [])
       }
     })
@@ -183,10 +186,15 @@ export async function enrichMessagesWithExistingRecords(
     existingRecords: recordsMap.get(message.prescriptionData.PrescriptionID) ?? []
   }))
 
-  logger.info("Enriched messages with existing records from DynamoDB", {
-    messageCount: messages.length,
-    messagesWithRecords: enrichedMessages.filter((m) => m.existingRecords.length > 0).length
-  })
+  for (const msg of enrichedMessages) {
+    logger.info("Prescription and most recent existing record", {
+      prescriptionID: msg.prescriptionData.PrescriptionID,
+      existingRecordCount: msg.existingRecords.length,
+      mostRecentExistingRecord: msg.existingRecords.length > 0
+        ? msg.existingRecords[0]
+        : null
+    })
+  }
 
   return enrichedMessages
 }

packages/postDatedLambda/src/orchestration.ts

@@ -23,39 +23,39 @@ export async function processMessages(
 ): Promise<BatchProcessingResult> {
   if (messages.length === 0) {
     logger.info("No messages to process in batch")
-    return {successful: [], failed: []}
+    return {maturedPrescriptionUpdates: [], immaturePrescriptionUpdates: []}
   }
 
   // Enrich messages with existing records from DynamoDB
   const enrichedMessages = await enrichMessagesWithExistingRecords(messages, logger)
 
-  const successful: Array<PostDatedSQSMessage> = []
-  const failed: Array<PostDatedSQSMessage> = []
+  const maturedPrescriptionUpdates: Array<PostDatedSQSMessage> = []
+  const immaturePrescriptionUpdates: Array<PostDatedSQSMessage> = []
 
   for (const message of enrichedMessages) {
     try {
       const success = await processMessage(logger, message)
       if (success) {
-        successful.push(message)
+        maturedPrescriptionUpdates.push(message)
       } else {
-        failed.push(message)
+        immaturePrescriptionUpdates.push(message)
       }
     } catch (error) {
       logger.error("Error processing message", {
         messageId: message.MessageId,
         error
       })
-      failed.push(message)
+      immaturePrescriptionUpdates.push(message)
     }
   }
 
   logger.info("Batch processing complete", {
     totalMessages: messages.length,
-    successfulCount: successful.length,
-    failedCount: failed.length
+    maturedPrescriptionUpdatesCount: maturedPrescriptionUpdates.length,
+    immaturePrescriptionUpdatesCount: immaturePrescriptionUpdates.length
   })
 
-  return {successful, failed}
+  return {maturedPrescriptionUpdates, immaturePrescriptionUpdates}
 }
 
 /**

packages/postDatedLambda/src/sqs.ts

@@ -13,7 +13,6 @@ import {PostDatedNotifyDataItem} from "@psu-common/commonTypes"
 import {BatchProcessingResult, PostDatedSQSMessage, ReceivedPostDatedSQSResult} from "./types"
 
 const sqs = new SQSClient({region: process.env.AWS_REGION})
-const FAILED_MESSAGE_VISIBILITY_TIMEOUT = 300 // 5 minutes in seconds
 
 /**
  * Get the SQS queue URL from environment variables.
@@ -210,34 +209,39 @@ export async function removeSQSMessages(
 }
 
 /**
- * Return failed messages to the queue with a visibility timeout.
+ * Edit failed that are on the queue to update their visibility timeout.
  * This makes the messages invisible for the specified duration before they can be processed again.
+ * This does not delete the messages, or post new ones; it only alters their visibility.
  *
  * @param logger - The logging object
  * @param messages - The messages that failed processing and should be returned to the queue
- * @param visibilityTimeoutSeconds - The time in seconds before messages become visible again (default: 300 = 5 minutes)
  */
 export async function returnMessagesToQueue(
   logger: Logger,
-  messages: Array<Message>,
-  visibilityTimeoutSeconds = 300
+  messages: Array<Message>
 ): Promise<void> {
   if (messages.length === 0) {
-    logger.info("No failed messages to return to queue")
+    logger.info("No messages to return to queue")
     return
   }
 
   const sqsUrl = getQueueUrl(logger)
 
+  // TODO: Each message needs to have an appropriate visibility timeout based on when it is due to be retried.
+  // For now, use a fixed 5 minute timeout for all messages.
+  const visibilityTimeoutSeconds = 300
   const entries = messages.map((m) => ({
     Id: m.MessageId!,
     ReceiptHandle: m.ReceiptHandle!,
     VisibilityTimeout: visibilityTimeoutSeconds
   }))
 
   logger.info(
-    `Returning messages to queue with ${visibilityTimeoutSeconds}s timeout`,
-    {numberOfMessages: entries.length, messageIds: entries.map((e) => e.Id)}
+    `Returning messages to queue with timeouts`,
+    {
+      numberOfMessages: entries.length,
+      idAndTimeouts: entries.map((e) => ({id: e.Id, visibilityTimeout: e.VisibilityTimeout}))
+    }
   )
 
   const changeVisibilityCmd = new ChangeMessageVisibilityBatchCommand({
@@ -258,8 +262,8 @@ export async function returnMessagesToQueue(
 
 /**
  * Handle the results of message processing:
- * - Delete successful messages from the queue
- * - Return failed messages to the queue with a visibility timeout
+ * - Delete matured messages from the queue
+ * - Return immature messages to the queue with a visibility timeout
  * Does not alter the input result object, only performs side effects.
  *
  * @param result - The batch processing result
@@ -269,15 +273,16 @@ export async function handleProcessedMessages(
   result: BatchProcessingResult,
   logger: Logger
 ): Promise<void> {
-  const {successful, failed} = result
+  const {maturedPrescriptionUpdates, immaturePrescriptionUpdates} = result
 
-  // Delete successful messages
-  if (successful.length > 0) {
-    await removeSQSMessages(logger, successful)
+  // Delete matured messages
+  if (maturedPrescriptionUpdates.length > 0) {
+    // TODO: Also need to send messages to the notification queue here (do that first, then delete)
+    await removeSQSMessages(logger, maturedPrescriptionUpdates)
   }
 
-  // Return failed messages to the queue with a 5 minute timeout
-  if (failed.length > 0) {
-    await returnMessagesToQueue(logger, failed, FAILED_MESSAGE_VISIBILITY_TIMEOUT)
+  // Return failed messages to the queue
+  if (immaturePrescriptionUpdates.length > 0) {
+    await returnMessagesToQueue(logger, immaturePrescriptionUpdates)
   }
 }

packages/postDatedLambda/src/types.ts

@@ -33,8 +33,8 @@ export interface PostDatedSQSMessageWithExistingRecords extends PostDatedSQSMess
  * Result of processing a batch of messages.
  */
 export interface BatchProcessingResult {
-  successful: Array<PostDatedSQSMessage>
-  failed: Array<PostDatedSQSMessage>
+  maturedPrescriptionUpdates: Array<PostDatedSQSMessage>
+  immaturePrescriptionUpdates: Array<PostDatedSQSMessage>
 }
 
 /**