feat(int): working on testConsumer to monitor kafka events

package.json

@@ -91,7 +91,7 @@
     "@mojaloop/central-services-health": "13.0.0",
     "@mojaloop/central-services-logger": "10.6.2",
     "@mojaloop/central-services-metrics": "11.0.0",
-    "@mojaloop/central-services-shared": "15.0.1",
+    "@mojaloop/central-services-shared": "git+https://github.com/vessels-tech/central-services-shared.git#feat/2553-patch-notification-3",
     "@mojaloop/central-services-stream": "10.7.0",
     "@mojaloop/event-sdk": "10.7.1",
     "@mojaloop/ml-number": "11.2.1",

src/handlers/transfers/handler.js

@@ -490,55 +490,77 @@ const fulfil = async (error, messages) => {
       // emit an extra message -  RESERVED_ABORTED if action === TransferEventAction.RESERVE
       if (action === TransferEventAction.RESERVE) {
         Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `callbackReservedAborted--${actionLetter}1`))
-        const metadataState = Util.StreamingProtocol.createEventState(
-          Enum.Events.EventStatus.FAILURE.status,
-          apiFspiopError.errorInformation.errorCode,
-          apiFspiopError.errorInformation.errorDescription
-        )
-        const metadata = {
-          event: {
-            // TODO: These seem to have no effect - they are overridden by
-            // some crazy logic in Kafka.produceGeneralMessage
-            type: TransferEventType.FULFIL,
-            action: TransferEventAction.RESERVED_ABORTED,
-            state: metadataState
-          }
+        const eventDetail = { functionality: TransferEventType.NOTIFICATION, action: TransferEventAction.RESERVED_ABORTED  }
+        const abortParams = {
+          // TODO: can we just use the original message?
+          message,
+          kafkaTopic,
+          span,
+          consumer: Consumer, 
+          producer: Producer,
         }
         const reservedAbortedPayload = {
           transferId: transfer.id,
+          //TODO: get this time from the db - don't be lazy
           completedTimestamp: Util.Time.getUTCString(new Date()),
           transferState: TransferState.ABORTED
         }
-
-        // clone the headers, and change the FSPIOP-Source and FSPIOP-Destination
-        const reservedAbortedHeaders = JSON.parse(JSON.stringify(headers))
-        reservedAbortedHeaders[Enum.Http.Headers.FSPIOP.DESTINATION] = transfer.payeeFsp
-        reservedAbortedHeaders[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
-        const reservedAbortedMessage = Util.StreamingProtocol.createMessage(
-          transferId,
-          transfer.payeeFsp, 
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
-          metadata,
-          reservedAbortedHeaders,
-          reservedAbortedPayload,
-        )
-
-        console.log('Kafka.produceGeneralMessage - ', JSON.stringify(reservedAbortedMessage))
-        await Kafka.produceGeneralMessage(
-          Config.KAFKA_CONFIG,
-          Producer,
-          Enum.Events.Event.Type.NOTIFICATION,
-          // TODO: should this be EVENT? or should we make a new topic for RESERVED_ABORTED?
-          // I can't seem to find how to send a message to the `topic-notification-event` topic
-          // that has the reservedAbortedMessage.metadata.event.action of RESERVED_ABORTED
-          // TransferEventAction.RESERVED_ABORTED,
-          // 
-          // produceGeneralMessage overrides the reservedAbortedMessage.metadata.event.action
-          // for some reason.
-          TransferEventAction.EVENT,
-          reservedAbortedMessage,
-          metadataState
-        )
+        message.value.content.payload = reservedAbortedPayload
+        await Kafka.proceed(Config.KAFKA_CONFIG, abortParams, { consumerCommit, eventDetail, toDestination: transfer.payeeFsp, fromSwitch: true })
+
+
+        // Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `callbackReservedAborted--${actionLetter}1`))
+        // const metadataState = Util.StreamingProtocol.createEventState(
+        //   Enum.Events.EventStatus.FAILURE.status,
+        //   apiFspiopError.errorInformation.errorCode,
+        //   apiFspiopError.errorInformation.errorDescription
+        // )
+        // //TODO: replace this call with createEventMetadata
+        // const metadata = {
+        //   event: {
+        //     // TODO: These seem to have no effect - they are overridden by
+        //     // some crazy logic in Kafka.produceGeneralMessage
+        //     type: TransferEventType.FULFIL,
+        //     action: TransferEventAction.RESERVED_ABORTED,
+        //     state: metadataState
+        //     // TODO: add span/trace stuff in here?
+        //   }
+        // }
+
+
+        // // clone the headers, and change the FSPIOP-Source and FSPIOP-Destination
+        // const reservedAbortedHeaders = Util.clone(headers)
+        // reservedAbortedHeaders[Enum.Http.Headers.FSPIOP.DESTINATION] = transfer.payeeFsp
+        // reservedAbortedHeaders[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        // const reservedAbortedMessage = Util.StreamingProtocol.createMessage(
+        //   // TODO: should this be the transferId or a messageId
+        //   transferId,
+        //   transfer.payeeFsp, 
+        //   Enum.Http.Headers.FSPIOP.SWITCH.value,
+        //   metadata,
+        //   reservedAbortedHeaders,
+        //   reservedAbortedPayload,
+        // )
+
+        // console.log('Kafka.produceGeneralMessage - ', JSON.stringify(reservedAbortedMessage))
+        // await Kafka.produceGeneralMessage(
+        //   Config.KAFKA_CONFIG,
+        //   Producer,
+        //   Enum.Events.Event.Type.NOTIFICATION,
+        //   // TODO: should this be EVENT? or should we make a new topic for RESERVED_ABORTED?
+        //   // I can't seem to find how to send a message to the `topic-notification-event` topic
+        //   // that has the reservedAbortedMessage.metadata.event.action of RESERVED_ABORTED
+        //   // TransferEventAction.RESERVED_ABORTED,
+        //   // 
+        //   // produceGeneralMessage overrides the reservedAbortedMessage.metadata.event.action
+        //   // for some reason.
+        //   TransferEventAction.RESERVED_ABORTED,
+        //   // TransferEventAction.EVENT,
+        //   reservedAbortedMessage,
+        //   metadataState
+
+        //   // TODO: missing tracing here - pass the span object in
+        // )
       }
       throw fspiopError
     } 

src/models/transfer/transferDuplicateCheck.js

@@ -52,7 +52,6 @@ const getTransferDuplicateCheck = async (transferId) => {
   Logger.isDebugEnabled && Logger.debug(`get transferDuplicateCheck (transferId=${transferId})`)
   try {
     const result = Db.from('transferDuplicateCheck').findOne({ transferId })
-    console.log("getTransferDuplicateCheck", result)
     histTimerGetTransferDuplicateCheckEnd({ success: true, queryName: 'transferDuplicateCheck_getTransferDuplicateCheck' })
     return result
   } catch (err) {

test/integration/handlers/handlers.test.js

@@ -388,23 +388,20 @@ Test('Handlers test', async handlersTest => {
 
       const transferStateIsAbortedError = await wrapWithRetries(async () => {
         const transfer = await TransferService.getById(td.messageProtocolPrepare.content.payload.transferId)
-        if (transfer.transferState !== 'ABORTED_ERROR') {
-          return false
-        }
-        return true
+        return transfer.transferState === 'ABORTED_ERROR'
       })
       test.equal(transferStateIsAbortedError, true, 'Transfer is in ABORTED_ERROR state')
 
       // Assert
       // 3. Check that we sent 2 notifications to kafka - one for the Payee, one for the Payer
-      await currentEventLoopEnd()
-      const payerAbortNotification = testConsumer.getEventsForFilter({ topicFilter: 'topic-notification-event', action: 'abort-validation' })[0]
-      // TODO: I don't think this action should be event
-      // There seem to be some inconsistiences with how different kafka messages are produced
-      const payeeAbortNotification = testConsumer.getEventsForFilter({ topicFilter: 'topic-notification-event', action: 'event' })[0]
-      console.log('payerAbortNotification are', JSON.stringify(payerAbortNotification))
-      console.log('payeeAbortNotification are', JSON.stringify(payeeAbortNotification))
-
+      const payerAbortNotification = (await wrapWithRetries(
+        () => testConsumer.getEventsForFilter({ topicFilter: 'topic-notification-event', action: 'abort-validation' }))
+      )[0]
+      const payeeAbortNotification = (await wrapWithRetries(
+        () => testConsumer.getEventsForFilter({ topicFilter: 'topic-notification-event', action: 'reserved-aborted' }))
+        )[0]
+      test.ok(payerAbortNotification, 'Payer Abort notification sent')
+      test.ok(payeeAbortNotification, 'Payee Abort notification sent')
 
       // Cleanup
       testConsumer.clearEvents()