feat(csi/551): add transfer state change for proxied fxTransfer (#1087)

* feat(csi/551): add transfer state change for proxied fxTransfer

* remove

* add case

* dep

* unit tests

* int tests

* chore(snapshot): 17.8.0-snapshot.16

package.json

@@ -1,6 +1,6 @@
 {
   "name": "@mojaloop/central-ledger",
-  "version": "17.8.0-snapshot.15",
+  "version": "17.8.0-snapshot.16",
   "description": "Central ledger hosted by a scheme to record and settle transfers",
   "license": "Apache-2.0",
   "author": "ModusBox",
@@ -92,7 +92,7 @@
     "@mojaloop/central-services-health": "15.0.0",
     "@mojaloop/central-services-logger": "11.5.1",
     "@mojaloop/central-services-metrics": "12.0.8",
-    "@mojaloop/central-services-shared": "18.7.2",
+    "@mojaloop/central-services-shared": "18.7.3",
     "@mojaloop/central-services-stream": "11.3.1",
     "@mojaloop/database-lib": "11.0.6",
     "@mojaloop/event-sdk": "14.1.1",
@@ -133,7 +133,7 @@
     "jsdoc": "4.0.3",
     "jsonpath": "1.1.1",
     "nodemon": "3.1.4",
-    "npm-check-updates": "17.1.0",
+    "npm-check-updates": "17.1.1",
     "nyc": "17.0.0",
     "pre-commit": "1.2.2",
     "proxyquire": "2.1.3",

src/domain/fx/index.js

@@ -54,6 +54,22 @@ const handleFulfilResponse = async (transferId, payload, action, fspiopError) =>
   }
 }
 
+const forwardedFxPrepare = async (commitRequestId) => {
+  const histTimerTransferServicePrepareEnd = Metrics.getHistogram(
+    'fx_domain_transfer',
+    'prepare - Metrics for fx transfer domain',
+    ['success', 'funcName']
+  ).startTimer()
+  try {
+    const result = await FxTransferModel.fxTransfer.updateFxPrepareReservedForwarded(commitRequestId)
+    histTimerTransferServicePrepareEnd({ success: true, funcName: 'forwardedFxPrepare' })
+    return result
+  } catch (err) {
+    histTimerTransferServicePrepareEnd({ success: false, funcName: 'forwardedFxPrepare' })
+    throw ErrorHandler.Factory.reformatFSPIOPError(err)
+  }
+}
+
 // TODO: Need to implement this for fxTransferError
 // /**
 //  * @function LogFxTransferError
@@ -82,6 +98,8 @@ const handleFulfilResponse = async (transferId, payload, action, fspiopError) =>
 
 const TransferService = {
   handleFulfilResponse,
+  forwardedFxPrepare,
+  getByIdLight: FxTransferModel.fxTransfer.getByIdLight,
   // logFxTransferError,
   Cyril
 }

src/handlers/transfers/FxFulfilService.js

@@ -31,7 +31,7 @@ const ErrorHandler = require('@mojaloop/central-services-error-handling')
 
 const { Type, Action } = Enum.Events.Event
 const { SOURCE, DESTINATION } = Enum.Http.Headers.FSPIOP
-const { TransferState } = Enum.Transfers
+const { TransferState, TransferInternalState } = Enum.Transfers
 
 const consumerCommit = true
 const fromSwitch = true
@@ -255,7 +255,8 @@ class FxFulfilService {
   }
 
   async validateTransferState(transfer, functionality) {
-    if (transfer.transferState !== TransferState.RESERVED) {
+    if (transfer.transferState !== TransferInternalState.RESERVED &&
+        transfer.transferState !== TransferInternalState.RESERVED_FORWARDED) {
       const fspiopError = fspiopErrorFactory.fxTransferNonReservedState()
       const apiFSPIOPError = fspiopError.toApiErrorObject(this.Config.ERROR_HANDLING)
       const eventDetail = {

src/handlers/transfers/dto.js

@@ -16,11 +16,11 @@ const prepareInputDto = (error, messages) => {
   if (!message) throw new Error('No input kafka message')
 
   const payload = decodePayload(message.value.content.payload)
-  const isForwarded = message.value.metadata.event.action === Action.FORWARDED
-  const isFx = !payload.transferId && !isForwarded
+  const isForwarded = message.value.metadata.event.action === Action.FORWARDED || message.value.metadata.event.action === Action.FX_FORWARDED
+  const isFx = !payload.transferId
 
   const { action } = message.value.metadata.event
-  const isPrepare = [Action.PREPARE, Action.FX_PREPARE, Action.FORWARDED].includes(action)
+  const isPrepare = [Action.PREPARE, Action.FX_PREPARE, Action.FORWARDED, Action.FX_FORWARDED].includes(action)
 
   const actionLetter = isPrepare
     ? Enum.Events.ActionLetter.prepare

src/handlers/transfers/handler.js

@@ -108,7 +108,8 @@ const fulfil = async (error, messages) => {
       TransferEventAction.FX_COMMIT,
       TransferEventAction.FX_RESERVE,
       TransferEventAction.FX_REJECT,
-      TransferEventAction.FX_ABORT
+      TransferEventAction.FX_ABORT,
+      TransferEventAction.FX_FORWARDED
     ]
 
     if (fxActions.includes(action)) {
@@ -677,7 +678,8 @@ const processFxFulfilMessage = async (message, functionality, span) => {
     TransferEventAction.FX_RESERVE,
     TransferEventAction.FX_COMMIT,
     // TransferEventAction.FX_REJECT,
-    TransferEventAction.FX_ABORT
+    TransferEventAction.FX_ABORT,
+    TransferEventAction.FX_FORWARDED
   ]
   if (!validActions.includes(action)) {
     const errorMessage = ERROR_MESSAGES.fxActionIsNotAllowed(action)