Merge pull request #471 from DEFRA/feat/df-912-dlq

feat/df-912: Added DLQ capabilities

Author: jbarnsley10

package-lock.json

@@ -59,6 +59,14 @@
  * @typedef {{ Params: { formId: string }}} GenerateFeedbackSubmissionsFile
  */
 
+/**
+ * @typedef {{ Params: { dlq: string }}} DeadLetterQueueRequest
+ */
+
+/**
+ * @typedef {{ Params: { dlq: string, receiptHandle: string }}} DeadLetterQueueAndHandleRequest
+ */
+
 /**
  * @typedef {object} UploadPayload
  * @property {('initiated'|'pending'|'ready')} uploadStatus - Have all scans completed, can be initiated, pending or ready

src/api/types.js

@@ -265,6 +265,18 @@ export const config = convict({
     default: '',
     env: 'SUBMISSION_QUEUE_URL'
   },
+  sqsFormSubmissionsDlqArn: {
+    doc: 'SQS deadletter queue ARN for form submission events',
+    format: String,
+    default: '',
+    env: 'FORM_SUBMISSIONS_SQS_DLQ_ARN'
+  },
+  sqsSaveAndExitDlqArn: {
+    doc: 'SQS deadletter queue ARN for save-and-exit events',
+    format: String,
+    default: '',
+    env: 'SAVE_AND_EXIT_SQS_DLQ_ARN'
+  },
   receiveMessageTimeout: {
     doc: 'The wait time between each poll in milliseconds',
     format: Number,

src/config/index.js

@@ -1,6 +1,7 @@
 import {
   DeleteMessageCommand,
-  ReceiveMessageCommand
+  ReceiveMessageCommand,
+  StartMessageMoveTaskCommand
 } from '@aws-sdk/client-sqs'
 
 import { config } from '~/src/config/index.js'
@@ -11,6 +12,15 @@ export const receiveMessageTimeout = config.get('receiveMessageTimeout')
 const maxNumberOfMessages = config.get('maxNumberOfMessages')
 const visibilityTimeout = config.get('visibilityTimeout')
 
+/**
+ * @param {string} dlqName
+ */
+function getDeadLetterQueueUrl(dlqName) {
+  return dlqName === 'save-and-exit'
+    ? `${config.get('saveAndExitQueueUrl')}-deadletter`
+    : `${config.get('submissionQueueUrl')}-deadletter`
+}
+
 /**
  * Receive event messages
  * @param {string} queueUrl - the SQS queue url
@@ -30,6 +40,53 @@ export function receiveMessages(queueUrl) {
   return sqsClient.send(command)
 }
 
+/**
+ * Receive dead-letter queue messages
+ * @param {string} dlq - the SQS deadletter queue identifier
+ * @returns {Promise<ReceiveMessageResult>}
+ */
+export function receiveDlqMessages(dlq) {
+  const queueUrl = getDeadLetterQueueUrl(dlq)
+  const command = new ReceiveMessageCommand({
+    QueueUrl: queueUrl,
+    MaxNumberOfMessages: 10,
+    VisibilityTimeout: 1,
+    WaitTimeSeconds: 0
+  })
+  return sqsClient.send(command)
+}
+
+/**
+ * Redrive all messages from the dead-letter queue to the main queue
+ * @param {string} dlq - the SQS deadletter queue ARN
+ * @returns {Promise<StartMessageMoveTaskResult>}
+ */
+export function redriveDlqMessages(dlq) {
+  const queueArn =
+    dlq === 'save-and-exit'
+      ? config.get('sqsSaveAndExitDlqArn')
+      : config.get('sqsFormSubmissionsDlqArn')
+  const command = new StartMessageMoveTaskCommand({
+    SourceArn: queueArn
+  })
+  return sqsClient.send(command)
+}
+
+/**
+ * Delete the specified message from the dead-letter queue
+ * @param {string} dlq - the SQS deadletter queue ARN
+ * @param {string} receiptHandle - the message receipt handle (not the same as the message id)
+ * @returns {Promise<DeleteMessageCommandOutput>}
+ */
+export function deleteDlqMessage(dlq, receiptHandle) {
+  const queueUrl = getDeadLetterQueueUrl(dlq)
+  const command = new DeleteMessageCommand({
+    QueueUrl: queueUrl,
+    ReceiptHandle: receiptHandle
+  })
+  return sqsClient.send(command)
+}
+
 /**
  * Delete event message
  * @param {string} queueUrl - the SQS queue url
@@ -46,5 +103,5 @@ export function deleteMessage(queueUrl, message) {
 }
 
 /**
- * @import { ReceiveMessageCommandInput, ReceiveMessageResult, DeleteMessageCommandOutput, Message } from '@aws-sdk/client-sqs'
+ * @import { ReceiveMessageCommandInput, ReceiveMessageResult, DeleteMessageCommandOutput, Message, StartMessageMoveTaskResult } from '@aws-sdk/client-sqs'
  */

src/messaging/event.js

@@ -1,12 +1,19 @@
 import {
   DeleteMessageCommand,
   ReceiveMessageCommand,
-  SQSClient
+  SQSClient,
+  StartMessageMoveTaskCommand
 } from '@aws-sdk/client-sqs'
 import { mockClient } from 'aws-sdk-client-mock'
 
 import 'aws-sdk-client-mock-jest'
-import { deleteMessage, receiveMessages } from '~/src/messaging/event.js'
+import {
+  deleteDlqMessage,
+  deleteMessage,
+  receiveDlqMessages,
+  receiveMessages,
+  redriveDlqMessages
+} from '~/src/messaging/event.js'
 
 jest.mock('~/src/helpers/logging/logger.js')
 
@@ -51,8 +58,90 @@ describe('event', () => {
       })
     })
   })
+
+  describe('receiveDlqMessages', () => {
+    it('should receive dead-letter queue messages from form-submissions', async () => {
+      const receivedMessage = {
+        Messages: [messageStub]
+      }
+
+      snsMock.on(ReceiveMessageCommand).resolves(receivedMessage)
+      await receiveDlqMessages('form-submissions')
+      expect(snsMock).toHaveReceivedCommandWith(ReceiveMessageCommand, {
+        QueueUrl: expect.any(String),
+        VisibilityTimeout: 1,
+        WaitTimeSeconds: 0
+      })
+    })
+
+    it('should receive dead-letter queue messages from save-and-exit', async () => {
+      const receivedMessage = {
+        Messages: [messageStub]
+      }
+
+      snsMock.on(ReceiveMessageCommand).resolves(receivedMessage)
+      await receiveDlqMessages('save-and-exit')
+      expect(snsMock).toHaveReceivedCommandWith(ReceiveMessageCommand, {
+        QueueUrl: expect.any(String),
+        VisibilityTimeout: 1,
+        WaitTimeSeconds: 0
+      })
+    })
+  })
+
+  describe('redriveDlqMessages', () => {
+    it('should redrive dead-letter queue messages from form-submissions', async () => {
+      /**
+       * @type {StartMessageMoveTaskCommandOutput}
+       */
+      const redriveResult = {
+        TaskHandle: '123',
+        $metadata: {}
+      }
+
+      snsMock.on(StartMessageMoveTaskCommand).resolves(redriveResult)
+      await redriveDlqMessages('form-submissions')
+      expect(snsMock).toHaveReceivedCommandWith(StartMessageMoveTaskCommand, {
+        SourceArn: expect.any(String)
+      })
+    })
+
+    it('should redrive dead-letter queue messages from save-and-exit', async () => {
+      /**
+       * @type {StartMessageMoveTaskCommandOutput}
+       */
+      const redriveResult = {
+        TaskHandle: '123',
+        $metadata: {}
+      }
+
+      snsMock.on(StartMessageMoveTaskCommand).resolves(redriveResult)
+      await redriveDlqMessages('save-and-exit')
+      expect(snsMock).toHaveReceivedCommandWith(StartMessageMoveTaskCommand, {
+        SourceArn: expect.any(String)
+      })
+    })
+  })
+
+  describe('deleteDlqMessage', () => {
+    it('should delete event message', async () => {
+      /**
+       * @type {DeleteMessageCommandOutput}
+       */
+      const deleteResult = {
+        $metadata: {}
+      }
+
+      snsMock.on(DeleteMessageCommand).resolves(deleteResult)
+      await deleteDlqMessage('save-and-exit', messageStub.ReceiptHandle)
+      expect(snsMock).toHaveReceivedCommandWith(DeleteMessageCommand, {
+        QueueUrl: expect.any(String),
+        ReceiptHandle: receiptHandle
+      })
+    })
+  })
 })
 
 /**
- * @import { DeleteMessageCommandOutput } from '@aws-sdk/client-sqs'
+ * @import { DeleteMessageCommandOutput, StartMessageMoveTaskCommandOutput } from '@aws-sdk/client-sqs'
  */

src/messaging/event.test.js

@@ -46,3 +46,7 @@ export const resetSaveAndExitLinkResponseSchema = Joi.object({
   recordFound: Joi.boolean().required(),
   recordUpdated: Joi.boolean().required()
 }).label('resetSaveAndExitLinkResponseSchema')
+
+export const dqlSchema = Joi.string().valid('form-submissions', 'save-and-exit')
+
+export const receiptHandleSchema = Joi.string()