New: [AEA-5296] - Log message for ITOC monitoring (#1787)

## Summary

- ✨ New Feature

### Details

Log the queue attributes at the start of the processor run. 
- timestamp
- Number of messages on queue
- Number of invisible messages
- Number of delayed message
There should always be 0 delayed messages.

Author: wildjames

packages/nhsNotifyLambda/src/nhsNotifyLambda.ts

@@ -12,6 +12,7 @@ import {
   addPrescriptionMessagesToNotificationStateStore,
   checkCooldownForUpdate,
   removeSQSMessages,
+  reportQueueStatus,
   drainQueue,
   makeBatchNotifyRequest,
   NotifyDataItemMessage
@@ -108,6 +109,7 @@ export const lambdaHandler = async (
   logger.info("NHS Notify lambda triggered by scheduler", {event})
   logger.info("Routing Plan ID:", {routingId})
 
+  await reportQueueStatus(logger)
   await drainAndProcess(routingId)
 }
 

packages/nhsNotifyLambda/src/utils.ts

@@ -3,7 +3,8 @@ import {
   SQSClient,
   ReceiveMessageCommand,
   DeleteMessageBatchCommand,
-  Message
+  Message,
+  GetQueueAttributesCommand
 } from "@aws-sdk/client-sqs"
 import {DynamoDBClient} from "@aws-sdk/client-dynamodb"
 import {DynamoDBDocumentClient, GetCommand, PutCommand} from "@aws-sdk/lib-dynamodb"
@@ -58,6 +59,37 @@ function chunkArray<T>(arr: Array<T>, size: number): Array<Array<T>> {
   return chunks
 }
 
+export async function reportQueueStatus(logger: Logger): Promise<void> {
+  if (!sqsUrl) {
+    logger.error("Notifications SQS URL not configured")
+    throw new Error("NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set")
+  }
+
+  const attrsCmd = new GetQueueAttributesCommand({
+    QueueUrl: sqsUrl,
+    AttributeNames: [
+      "ApproximateNumberOfMessages",
+      "ApproximateNumberOfMessagesNotVisible",
+      "ApproximateNumberOfMessagesDelayed"
+    ]
+  })
+  const {Attributes} = await sqs.send(attrsCmd)
+
+  // Fall back to a negative value so missing data can be identified
+  const ApproximateNumberOfMessages = parseInt(Attributes?.ApproximateNumberOfMessages ?? "-1")
+  const ApproximateNumberOfMessagesNotVisible = parseInt(Attributes?.ApproximateNumberOfMessagesNotVisible ?? "-1")
+  const ApproximateNumberOfMessagesDelayed = parseInt(Attributes?.ApproximateNumberOfMessagesDelayed ?? "-1")
+
+  logger.info(
+    "Current queue attributes (if a value failed to fetch, it will be reported as -1):",
+    {
+      ApproximateNumberOfMessages,
+      ApproximateNumberOfMessagesNotVisible,
+      ApproximateNumberOfMessagesDelayed
+    }
+  )
+}
+
 // This is an extension of the SQS message interface, which explicitly parses the PSUDataItem
 // and helps track the nhs notify results
 export interface NotifyDataItemMessage extends Message {
@@ -78,7 +110,7 @@ export interface NotifyDataItemMessage extends Message {
  *  - messages the array of parsed NotifyDataItemMessage
  *  - isEmpty: true if the last receive returned fewer than 5 messages (or none),
  *             indicating the queue is effectively drained.
- */
+*/
 export async function drainQueue(
   logger: Logger,
   maxTotal = 100

packages/nhsNotifyLambda/tests/testNhsNotifyLambda.test.ts

@@ -19,6 +19,7 @@ jest.unstable_mockModule(
 
 const mockAddPrescriptionMessagesToNotificationStateStore = jest.fn()
 const mockRemoveSQSMessages = jest.fn()
+const mockReportQueueStatus = jest.fn()
 const mockDrainQueue = jest.fn()
 const mockCheckCooldownForUpdate = jest.fn()
 const mockMakeBatchNotifyRequest = jest.fn()
@@ -27,6 +28,7 @@ jest.unstable_mockModule(
   "../src/utils",
   async () => ({
     __esModule: true,
+    reportQueueStatus: mockReportQueueStatus,
     drainQueue: mockDrainQueue,
     addPrescriptionMessagesToNotificationStateStore: mockAddPrescriptionMessagesToNotificationStateStore,
     removeSQSMessages: mockRemoveSQSMessages,

packages/nhsNotifyLambda/tests/testUtils.test.ts

@@ -4,7 +4,7 @@ import nock from "nock"
 
 import {Logger} from "@aws-lambda-powertools/logger"
 import {DynamoDBDocumentClient, GetCommand, PutCommand} from "@aws-sdk/lib-dynamodb"
-import {DeleteMessageBatchCommand, Message} from "@aws-sdk/client-sqs"
+import {GetQueueAttributesCommand, DeleteMessageBatchCommand, Message} from "@aws-sdk/client-sqs"
 
 import {constructMessage, constructPSUDataItemMessage, mockSQSClient} from "./testHelpers"
 
@@ -36,8 +36,9 @@ jest.unstable_mockModule(
 
 const {
   addPrescriptionMessagesToNotificationStateStore,
-  removeSQSMessages: clearCompletedSQSMessages,
+  removeSQSMessages,
   checkCooldownForUpdate,
+  reportQueueStatus,
   drainQueue,
   makeBatchNotifyRequest
 } = await import("../src/utils")
@@ -155,7 +156,7 @@ describe("NHS notify lambda helper functions", () => {
     })
   })
 
-  describe("clearCompletedSQSMessages", () => {
+  describe("removeSQSMessages", () => {
     let logger: Logger
     let errorSpy: SpiedFunction<(msg: string, ...meta: Array<unknown>) => void>
     let infoSpy: SpiedFunction<(msg: string, ...meta: Array<unknown>) => void>
@@ -179,7 +180,7 @@ describe("NHS notify lambda helper functions", () => {
       // successful delete (no .Failed)
       sqsMockSend.mockImplementationOnce(() => Promise.resolve({}))
 
-      await expect(clearCompletedSQSMessages(logger, messages))
+      await expect(removeSQSMessages(logger, messages))
         .resolves
         .toBeUndefined()
 
@@ -205,7 +206,7 @@ describe("NHS notify lambda helper functions", () => {
       // succeed both batches
       sqsMockSend.mockImplementation(() => Promise.resolve({}))
 
-      await clearCompletedSQSMessages(logger, messages)
+      await removeSQSMessages(logger, messages)
       expect(sqsMockSend).toHaveBeenCalledTimes(2)
 
       // first batch of 10
@@ -234,7 +235,7 @@ describe("NHS notify lambda helper functions", () => {
       // partial failure
       sqsMockSend.mockImplementationOnce(() => Promise.resolve({Failed: failedEntries}))
 
-      await clearCompletedSQSMessages(logger, messages)
+      await removeSQSMessages(logger, messages)
 
       expect(errorSpy).toHaveBeenCalledWith(
         "Some messages failed to delete in this batch",
@@ -756,4 +757,63 @@ describe("NHS notify lambda helper functions", () => {
       })
     })
   })
+
+  describe("reportQueueStatus", () => {
+    let logger: Logger
+    let infoSpy: SpiedFunction<(msg: string, ...meta: Array<unknown>) => void>
+    let errorSpy: SpiedFunction<(msg: string, ...meta: Array<unknown>) => void>
+
+    beforeEach(() => {
+      jest.resetModules()
+      jest.clearAllMocks()
+
+      process.env = {...ORIGINAL_ENV}
+      logger = new Logger({serviceName: "test-service"})
+      infoSpy = jest.spyOn(logger, "info")
+      errorSpy = jest.spyOn(logger, "error")
+    })
+
+    it("logs current queue attributes when SQS returns attributes", async () => {
+      const attrs = {
+        ApproximateNumberOfMessages: "7",
+        ApproximateNumberOfMessagesNotVisible: "4",
+        ApproximateNumberOfMessagesDelayed: "1"
+      }
+
+      sqsMockSend.mockImplementationOnce((cmd) => {
+        expect(cmd).toBeInstanceOf(GetQueueAttributesCommand)
+        expect((cmd as GetQueueAttributesCommand).input).toEqual({
+          QueueUrl: process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL,
+          AttributeNames: [
+            "ApproximateNumberOfMessages",
+            "ApproximateNumberOfMessagesNotVisible",
+            "ApproximateNumberOfMessagesDelayed"
+          ]
+        })
+        return Promise.resolve({Attributes: attrs})
+      })
+
+      await reportQueueStatus(logger)
+
+      expect(infoSpy).toHaveBeenCalledWith(
+        "Current queue attributes (if a value failed to fetch, it will be reported as -1):",
+        {
+          ApproximateNumberOfMessages: 7,
+          ApproximateNumberOfMessagesNotVisible: 4,
+          ApproximateNumberOfMessagesDelayed: 1
+        }
+      )
+    })
+
+    it("throws if the SQS URL is not configured", async () => {
+      delete process.env.NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL
+      const {reportQueueStatus: rqs} = await import("../src/utils")
+
+      await expect(rqs(logger)).rejects.toThrow(
+        "NHS_NOTIFY_PRESCRIPTIONS_SQS_QUEUE_URL not set"
+      )
+      expect(errorSpy).toHaveBeenCalledWith("Notifications SQS URL not configured")
+    })
+  })
+
 })

packages/nhsNotifyUpdateCallback/src/lambdaHandler.ts

@@ -35,12 +35,23 @@ const lambdaHandler = async (event: APIGatewayProxyEvent): Promise<APIGatewayPro
   let payload: MessageStatusResponse
   try {
     payload = JSON.parse(event.body)
-    logger.info("Payload parsed", {payload})
   } catch (error) {
     logger.error("Failed to parse payload", {error, payload: event.body})
     return response(400, {message: "Request body failed to parse"})
   }
 
+  payload.data.forEach(m => {
+    logger.info(
+      "Message state updated",
+      {
+        messageStatus: m.attributes.messageStatus,
+        messageReference: m.attributes.messageReference,
+        messageId: m.attributes.messageId,
+        receivedTimestamp: m.attributes.timestamp
+      }
+    )
+  })
+
   try {
     await updateNotificationsTable(logger, payload)
   } catch (error) {

packages/nhsNotifyUpdateCallback/tests/testNhsNotifyCallbackLambda.test.ts

@@ -90,7 +90,19 @@ describe("NHS Notify update callback lambda handler", () => {
   })
 
   it("returns 500 if updateNotificationsTable throws", async () => {
-    const payload = {status: "foo"}
+    const payload = {
+      status: "foo",
+      data: [
+        {
+          attributes: {
+            messageStatus: "messageStatus",
+            messageReference: "messageReference",
+            messageId: "messageId",
+            timestamp: "timestamp"
+          }
+        }
+      ]
+    }
     const event = generateMockEvent(payload)
     event.headers["x-request-id"] = "abc"
     mockCheckSignature.mockImplementation(() => undefined)
@@ -110,7 +122,19 @@ describe("NHS Notify update callback lambda handler", () => {
   })
 
   it("returns 202 and 'OK' when everything succeeds", async () => {
-    const payload = {status: "ok"}
+    const payload = {
+      status: "ok",
+      data: [
+        {
+          attributes: {
+            messageStatus: "messageStatus",
+            messageReference: "messageReference",
+            messageId: "messageId",
+            timestamp: "timestamp"
+          }
+        }
+      ]
+    }
     const event = generateMockEvent(payload)
     event.headers["x-request-id"] = "abc"
     mockCheckSignature.mockImplementation(() => undefined)