DOC-1762 send messages to trigger alerts

app/config/index.js

@@ -30,6 +30,9 @@ value.submitSubscription = mqConfig.submitSubscription
 value.returnSubscription = mqConfig.returnSubscription
 value.statementTopic = mqConfig.statementTopic
 value.statementDataSubscription = mqConfig.statementDataSubscription
+value.processingSubscriptionFailed = mqConfig.processingSubscriptionFailed
+value.submitSubscriptionFailed = mqConfig.submitSubscriptionFailed
+value.returnSubscriptionFailed = mqConfig.returnSubscriptionFailed
 
 value.dbConfig = dbConfig
 value.processingConfig = processingConfig

app/config/message.js

@@ -35,6 +35,18 @@ const mqSchema = Joi.object({
     address: Joi.string(),
     source: Joi.string()
   },
+  processingSubscriptionFailed: {
+    address: Joi.string(),
+    source: Joi.string()
+  },
+  submitSubscriptionFailed: {
+    address: Joi.string(),
+    source: Joi.string()
+  },
+  returnSubscriptionFailed: {
+    address: Joi.string(),
+    source: Joi.string()
+  },
   idleCheckBatchSize: Joi.number().default(number250),
   idleCheckMaxDeliveryCount: Joi.number().default(number1),
   idleCheckInterval: Joi.number().default(number10000),
@@ -73,6 +85,18 @@ const mqConfig = {
     address: process.env.STATEMENT_TOPIC_ADDRESS,
     source: 'ffc-doc-statement-constructor'
   },
+  processingSubscriptionFailed: {
+    address: process.PROCESSING_SUBSCRIPTION_ERROR_ADDRESS,
+    source: 'ffc-doc-statement-constructor'
+  },
+  submitSubscriptionFailed: {
+    address: process.env.SUBMIT_SUBSCRIPTION_ERROR_ADDRESS,
+    source: 'ffc-doc-statement-constructor'
+  },
+  returnSubscriptionFailed: {
+    address: process.RETURN_SUBSCRIPTION_ERROR_ADDRESS,
+    source: 'ffc-doc-statement-constructor'
+  },
   idleCheckBatchSize: process.env.IDLE_CHECK_BATCH_SIZE,
   idleCheckMaxDeliveryCount: process.env.IDLE_CHECK_MAX_DELIVERY_COUNT,
   idleCheckInterval: process.env.IDLE_CHECK_INTERVAL,

app/constants/message-types.js

@@ -0,0 +1,5 @@
+module.exports = {
+  PROCESSING_SUBCRIPTION_FAILED: 'uk.gov.defra.ffc.doc.warning.processing.subscription.failed',
+  SUBMIT_SUBCRIPTION_FAILED: 'uk.gov.defra.ffc.doc.warning.submit.subscription.failed',
+  RETURN_SUBCRIPTION_FAILED: 'uk.gov.defra.ffc.doc.warning.return.subscription.failed'
+}

app/messaging/index.js

@@ -10,16 +10,15 @@ let returnReceiver
 let statementDataReceiver
 
 const start = async () => {
-
-  if(config.paymentLinkActive) {
+  if (config.paymentLinkActive) {
     const processingAction = message => processProcessingMessage(message, processingReceiver)
     processingReceiver = new MessageReceiver(config.processingSubscription, processingAction)
     await processingReceiver.subscribe()
-  
+
     const submitAction = message => processSubmitMessage(message, submitReceiver)
     submitReceiver = new MessageReceiver(config.submitSubscription, submitAction)
     await submitReceiver.subscribe()
-  
+
     const returnAction = message => processReturnMessage(message, returnReceiver)
     returnReceiver = new MessageReceiver(config.returnSubscription, returnAction)
     await returnReceiver.subscribe()

app/messaging/process-processing-message.js

@@ -1,5 +1,6 @@
 const util = require('util')
 const { processProcessingPaymentRequest } = require('../inbound')
+const publishProcessSubscriptionFailed = require('./publish-process-subscription-failed')
 
 const processProcessingMessage = async (message, receiver) => {
   try {
@@ -9,6 +10,7 @@ const processProcessingMessage = async (message, receiver) => {
     await receiver.completeMessage(message)
   } catch (err) {
     console.error('Unable to process processing message:', err)
+    await publishProcessSubscriptionFailed(message.body, err)
     await receiver.deadLetterMessage(message)
   }
 }

app/messaging/process-return-message.js

@@ -1,13 +1,15 @@
 const util = require('util')
-
 const { processReturnSettlement } = require('../inbound')
+const publishReturnSubscriptionFailed = require('./publish-return-subscription-failed')
+
 const processReturnMessage = async (message, receiver) => {
   try {
     const settlement = message.body
     console.log('Processing return settlement:', util.inspect(settlement, false, null, true))
     await processReturnSettlement(settlement)
     await receiver.completeMessage(message)
   } catch (err) {
+    await publishReturnSubscriptionFailed(message.body, err)
     console.error('Unable to process return settlement:', err)
   }
 }

app/messaging/process-submit-message.js

@@ -1,5 +1,6 @@
 const util = require('util')
 const { processSubmitPaymentRequest } = require('../inbound')
+const publishSubmitSubscriptionFailed = require('./publish-submit-subscription-failed')
 
 const processSubmitMessage = async (message, receiver) => {
   try {
@@ -9,6 +10,7 @@ const processSubmitMessage = async (message, receiver) => {
     await receiver.completeMessage(message)
   } catch (err) {
     console.error('Unable to process submit message:', err)
+    await publishSubmitSubscriptionFailed(message.body, err)
     await receiver.deadLetterMessage(message)
   }
 }

app/messaging/publish-process-subscription-failed.js

@@ -0,0 +1,18 @@
+const util = require('util')
+const { PROCESSING_SUBCRIPTION_FAILED } = require('../constants/message-types')
+const sendMessage = require('../messaging/send-message')
+const config = require('../config')
+
+const publishProcessingSubscriptionFailed = async (paymentRequest, error) => {
+  const body = {
+    data: {
+      message: error.message,
+      ...paymentRequest
+    }
+  }
+
+  await sendMessage(body, PROCESSING_SUBCRIPTION_FAILED, config.processingSubscriptionFailed)
+  console.log('Message sent:', util.inspect(body, false, null, true))
+}
+
+module.exports = publishProcessingSubscriptionFailed

app/messaging/publish-return-subscription-failed.js

@@ -0,0 +1,18 @@
+const util = require('util')
+const { RETURN_SUBCRIPTION_FAILED } = require('../constants/message-types')
+const sendMessage = require('../messaging/send-message')
+const config = require('../config')
+
+const publishReturnSubscriptionFailed = async (paymentRequest, error) => {
+  const body = {
+    data: {
+      message: error.message,
+      ...paymentRequest
+    }
+  }
+
+  await sendMessage(body, RETURN_SUBCRIPTION_FAILED, config.returnSubscriptionFailed)
+  console.log('Message sent:', util.inspect(body, false, null, true))
+}
+
+module.exports = publishReturnSubscriptionFailed

app/messaging/publish-submit-subscription-failed.js

@@ -0,0 +1,18 @@
+const util = require('util')
+const { SUBMIT_SUBCRIPTION_FAILED } = require('../constants/message-types')
+const sendMessage = require('../messaging/send-message')
+const config = require('../config')
+
+const publishSubmitSubscriptionFailed = async (paymentRequest, error) => {
+  const body = {
+    data: {
+      message: error.message,
+      ...paymentRequest
+    }
+  }
+
+  await sendMessage(body, SUBMIT_SUBCRIPTION_FAILED, config.submitSubscriptionFailed)
+  console.log('Message sent:', util.inspect(body, false, null, true))
+}
+
+module.exports = publishSubmitSubscriptionFailed

docker-compose.yaml

@@ -22,6 +22,9 @@ services:
       DATA_TOPIC_ADDRESS: ${DATA_TOPIC_ADDRESS:-ffc-doc-statement-data}${MESSAGE_QUEUE_SUFFIX}
       DATA_SUBSCRIPTION_ADDRESS: ${DATA_SUBSCRIPTION_ADDRESS:-ffc-doc-statement-constructor}${MESSAGE_QUEUE_SUFFIX}
       STATEMENT_TOPIC_ADDRESS: ${STATEMENT_TOPIC_ADDRESS:-ffc-doc-statements}${MESSAGE_QUEUE_SUFFIX}
+      PROCESSING_TOPIC_FAILED_ADDRESS: ${PROCESSING_TOPIC_FAILED_ADDRESS:-ffc-doc-statements}${MESSAGE_QUEUE_SUFFIX}
+      SUBMIT_TOPIC_FAILED_ADDRESS: ${SUBMIT_TOPIC_FAILED_ADDRESS:-ffc-doc-statements}${MESSAGE_QUEUE_SUFFIX}
+      RETURN_TOPIC_FAILED_ADDRESS: ${RETURN_TOPIC_FAILED_ADDRESS:-ffc-doc-statements}${MESSAGE_QUEUE_SUFFIX}
       POSTGRES_DB: ffc_doc_statement_constructor
       POSTGRES_HOST: ${POSTGRES_HOST:-ffc-doc-statement-constructor-postgres}
       POSTGRES_PASSWORD: ${POSTGRES_SCHEMA_PASSWORD:-ppp}

helm/ffc-doc-statement-constructor/templates/config-map.yaml

@@ -8,12 +8,18 @@ data:
   PROCESSING_TOPIC_ADDRESS: {{ .Values.container.processingTopicAddress }}-{{ .Values.environmentCode }}
   SUBMIT_TOPIC_ADDRESS: {{ .Values.container.submitTopicAddress }}-{{ .Values.environmentCode }}
   RETURN_TOPIC_ADDRESS: {{ .Values.container.returnTopicAddress }}-{{ .Values.environmentCode }}
+  PROCESSING_TOPIC_FAILED_ADDRESS: {{ .Values.container.processingTopicErrorAddress }}
+  SUBMIT_TOPIC_FAILED_ADDRESS: {{ .Values.container.submitTopicFailedAddress }}
+  RETURN_TOPIC_FAILED_ADDRESS: {{ .Values.container.returnTopicFailedAddress }}
   DATA_TOPIC_ADDRESS: {{ .Values.container.dataTopicAddress }}-{{ .Values.environmentCode }}
   STATEMENT_TOPIC_ADDRESS: {{ .Values.container.statementTopicAddress }}-{{ .Values.environmentCode }}
   {{ else }}
   PROCESSING_TOPIC_ADDRESS: {{ .Values.container.processingTopicAddress }}
   SUBMIT_TOPIC_ADDRESS: {{ .Values.container.submitTopicAddress }}
   RETURN_TOPIC_ADDRESS: {{ .Values.container.returnTopicAddress }}
+  PROCESSING_TOPIC_FAILED_ADDRESS: {{ .Values.container.processingTopicErrorAddress }}
+  SUBMIT_TOPIC_FAILED_ADDRESS: {{ .Values.container.submitTopicFailedAddress }}
+  RETURN_TOPIC_FAILED_ADDRESS: {{ .Values.container.returnTopicFailedAddress }}
   DATA_TOPIC_ADDRESS: {{ .Values.container.dataTopicAddress }}
   STATEMENT_TOPIC_ADDRESS: {{ .Values.container.statementTopicAddress }}
   {{- end }}

helm/ffc-doc-statement-constructor/values.yaml

@@ -40,6 +40,9 @@ container:
   dataTopicAddress: ffc-doc-statement-data
   dataSubscriptionAddress: ffc-doc-statement-constructor
   statementTopicAddress: ffc-doc-statements
+  processingSubscriptionAddressError: ffc-doc-statement-constructor
+  submitSubscriptionAddressError: ffc-doc-statement-constructor
+  returnSubscriptionAddressError: ffc-doc-statement-constructor  
   settlementProcessingInterval: 60000
   statementConstructionActive: false
   scheduleConstructionActive: false

test/unit/messaging/index.test.js

@@ -33,4 +33,4 @@ describe('messaging', () => {
     expect(MessageReceiver).toHaveBeenCalledTimes(1)
     expect(MessageReceiver).toHaveBeenCalledWith(config.statementDataSubscription, expect.any(Function))
   })
-})
+})

test/unit/messaging/process-processing-message.test.js

@@ -1,8 +1,9 @@
 jest.mock('ffc-messaging')
-
 jest.mock('../../../app/inbound')
-const { processProcessingPaymentRequest } = require('../../../app/inbound')
+jest.mock('../../../app/messaging/publish-process-subscription-failed')
 
+const { processProcessingPaymentRequest } = require('../../../app/inbound')
+const publishProcessSubscriptionFailed = require('../../../app/messaging/publish-process-subscription-failed')
 const processProcessingMessage = require('../../../app/messaging/process-processing-message')
 
 let receiver
@@ -12,6 +13,7 @@ let message
 describe('process processing message', () => {
   beforeEach(() => {
     processProcessingPaymentRequest.mockReturnValue(undefined)
+    publishProcessSubscriptionFailed.mockReturnValue(undefined)
 
     paymentRequest = JSON.parse(JSON.stringify(require('../../mock-objects/mock-payment-request').processingPaymentRequest))
 
@@ -29,7 +31,8 @@ describe('process processing message', () => {
 
   test('should call processProcessingPaymentRequest when nothing throws', async () => {
     await processProcessingMessage(message, receiver)
-    expect(processProcessingPaymentRequest).toHaveBeenCalled()
+    expect(processProcessingPaymentRequest).toHaveBeenCalledWith(paymentRequest)
+    expect(receiver.completeMessage).toHaveBeenCalledWith(message)
   })
 
   test('should call processProcessingPaymentRequest once when nothing throws', async () => {
@@ -103,4 +106,14 @@ describe('process processing message', () => {
 
     expect(wrapper).not.toThrow()
   })
+
+  test('should call publishProcessSubscriptionFailed and deadLetterMessage on error', async () => {
+    const error = new Error('Test error')
+    processProcessingPaymentRequest.mockRejectedValue(error)
+
+    await processProcessingMessage(message, receiver)
+    expect(processProcessingPaymentRequest).toHaveBeenCalledWith(paymentRequest)
+    expect(publishProcessSubscriptionFailed).toHaveBeenCalledWith(paymentRequest, error)
+    expect(receiver.deadLetterMessage).toHaveBeenCalledWith(message)
+  })
 })

test/unit/messaging/process-return-message.test.js

@@ -1,8 +1,10 @@
 jest.mock('ffc-messaging')
 
 jest.mock('../../../app/inbound/return/process-return-settlement')
-const processReturnSettlement = require('../../../app/inbound/return/process-return-settlement')
+jest.mock('../../../app/messaging/publish-return-subscription-failed')
 
+const processReturnSettlement = require('../../../app/inbound/return/process-return-settlement')
+const publishReturnSubscriptionFailed = require('../../../app/messaging/publish-return-subscription-failed')
 const processReturnMessage = require('../../../app/messaging/process-return-message')
 
 let receiver
@@ -12,6 +14,7 @@ let message
 describe('process return message', () => {
   beforeEach(() => {
     processReturnSettlement.mockReturnValue(undefined)
+    publishReturnSubscriptionFailed.mockReturnValue(undefined)
     settlement = JSON.parse(JSON.stringify(require('../../mock-objects/mock-settlement')))
     receiver = {
       completeMessage: jest.fn()
@@ -88,4 +91,13 @@ describe('process return message', () => {
 
     expect(wrapper).not.toThrow()
   })
+
+  test('should call publishReturnSubscriptionFailed and deadLetterMessage on error', async () => {
+    const error = new Error('Test error')
+    processReturnSettlement.mockRejectedValue(error)
+
+    await processReturnMessage(message, receiver)
+    expect(processReturnSettlement).toHaveBeenCalledWith(settlement)
+    expect(publishReturnSubscriptionFailed).toHaveBeenCalledWith(settlement, error)
+  })
 })

test/unit/messaging/process-submit-message.test.js

@@ -1,8 +1,10 @@
 jest.mock('ffc-messaging')
 
 jest.mock('../../../app/inbound')
-const { processSubmitPaymentRequest } = require('../../../app/inbound')
+jest.mock('../../../app/messaging/publish-submit-subscription-failed')
 
+const { processSubmitPaymentRequest } = require('../../../app/inbound')
+const publishSubmitSubscriptionFailed = require('../../../app/messaging/publish-submit-subscription-failed')
 const processSubmitMessage = require('../../../app/messaging/process-submit-message')
 
 let receiver
@@ -12,6 +14,7 @@ let message
 describe('process submit message', () => {
   beforeEach(() => {
     processSubmitPaymentRequest.mockReturnValue(undefined)
+    publishSubmitSubscriptionFailed.mockReturnValue(undefined)
 
     paymentRequest = JSON.parse(JSON.stringify(require('../../mock-objects/mock-payment-request').submitPaymentRequest))
 
@@ -29,7 +32,8 @@ describe('process submit message', () => {
 
   test('should call processSubmitPaymentRequest when nothing throws', async () => {
     await processSubmitMessage(message, receiver)
-    expect(processSubmitPaymentRequest).toHaveBeenCalled()
+    expect(processSubmitPaymentRequest).toHaveBeenCalledWith(paymentRequest)
+    expect(receiver.completeMessage).toHaveBeenCalledWith(message)
   })
 
   test('should call processSubmitPaymentRequest once when nothing throws', async () => {
@@ -103,4 +107,14 @@ describe('process submit message', () => {
 
     expect(wrapper).not.toThrow()
   })
+
+  test('should call publishSubmitSubscriptionFailed and deadLetterMessage on error', async () => {
+    const error = new Error('Test error')
+    processSubmitPaymentRequest.mockRejectedValue(error)
+
+    await processSubmitMessage(message, receiver)
+    expect(processSubmitPaymentRequest).toHaveBeenCalledWith(paymentRequest)
+    expect(publishSubmitSubscriptionFailed).toHaveBeenCalledWith(paymentRequest, error)
+    expect(receiver.deadLetterMessage).toHaveBeenCalledWith(message)
+  })
 })