feat(mojaloop/#3524): add position fulfil to binprocessor

.circleci/config.yml

@@ -35,7 +35,8 @@ defaults_machine_Dependencies: &defaults_machine_Dependencies |
   sudo add-apt-repository -y "deb https://packages.confluent.io/clients/deb $(lsb_release -cs) main"
 
   ## Install deps
-  sudo apt install -y librdkafka-dev curl bash
+  sudo apt install -y librdkafka-dev curl bash musl-dev libsasl2-dev
+  sudo ln -s /usr/lib/x86_64-linux-musl/libc.so /lib/libc.musl-x86_64.so.1
 
 defaults_awsCliDependencies: &defaults_awsCliDependencies |
     apk --no-cache add aws-cli
@@ -345,14 +346,15 @@ jobs:
           command: |
             # Set Node version to default (Note: this is needed on Ubuntu)
             nvm use default
+            npm ci
 
             echo "Running integration tests...."
             bash ./test/scripts/test-integration.sh
           environment:
             ENDPOINT_URL: http://localhost:4545/notification
             UV_THREADPOOL_SIZE: 12
             WAIT_FOR_REBALANCE: 20
-            TEST_INT_RETRY_COUNT: 20
+            TEST_INT_RETRY_COUNT: 30
             TEST_INT_RETRY_DELAY: 2
             TEST_INT_REBALANCE_DELAY: 20000
       - store_artifacts:

.circleci/curl-retry-cl-health.sh

@@ -6,7 +6,7 @@
 url="http://localhost:3001/health"
 
 # Number of retries
-retries=10
+retries=30
 
 # Sleep between retries
 sleepwait=1

README.md

@@ -96,13 +96,14 @@ diverges from the defaults.
 You can configure the customized topic names in the config. Each position action key
 refers to position messages with associated actions.
 
-NOTE: Only POSITION.PREPARE is supported at this time, with additional event-type-actions being added later when required.
+NOTE: Only POSITION.PREPARE and POSITION.COMMIT is supported at this time, with additional event-type-actions being added later when required.
 
 ```
   "KAFKA": {
     "EVENT_TYPE_ACTION_TOPIC_MAP" : {
       "POSITION":{
-        "PREPARE": "topic-transfer-position-batch"
+        "PREPARE": "topic-transfer-position-batch",
+        "COMMIT": "topic-transfer-position-batch"
       }
     }
   }
@@ -123,7 +124,10 @@ Batch processing can be enabled in the transfer execution flow. Follow the steps
       "EVENT_TYPE_ACTION_TOPIC_MAP" : {
         "POSITION":{
           "PREPARE": "topic-transfer-position-batch",
-          "BULK_PREPARE": "topic-transfer-position"
+          "BULK_PREPARE": "topic-transfer-position",
+          "COMMIT": "topic-transfer-position-batch",
+          "BULK_COMMIT": "topic-transfer-position",
+          "RESERVE": "topic-transfer-position",
         }
       }
     }
@@ -243,7 +247,10 @@ env "CLEDG_KAFKA__EVENT_TYPE_ACTION_TOPIC_MAP__POSITION__PREPARE=topic-transfer-
 ```
 - Additionally, run position batch handler in a new terminal
 ```
-env "CLEDG_KAFKA__EVENT_TYPE_ACTION_TOPIC_MAP__POSITION__PREPARE=topic-transfer-position-batch" "CLEDG_HANDLERS__API__DISABLED=true" node src/handlers/index.js handler --positionbatch
+export CLEDG_KAFKA__EVENT_TYPE_ACTION_TOPIC_MAP__POSITION__PREPARE=topic-transfer-position-batch
+export CLEDG_KAFKA__EVENT_TYPE_ACTION_TOPIC_MAP__POSITION__COMMIT=topic-transfer-position-batch
+export CLEDG_HANDLERS__API__DISABLED=true
+node src/handlers/index.js handler --positionbatch
 ```
 - Run tests using `npx tape 'test/integration-override/**/handlerBatch.test.js'`
 

config/default.json

@@ -88,7 +88,10 @@
     "EVENT_TYPE_ACTION_TOPIC_MAP" : {
       "POSITION":{
         "PREPARE": null,
-        "BULK_PREPARE": null
+        "BULK_PREPARE": null,
+        "COMMIT": null,
+        "BULK_COMMIT": null,
+        "RESERVE": null
       }
     },
     "TOPIC_TEMPLATES": {

scripts/_wait4_all.js

@@ -21,7 +21,7 @@ const expectedContainers = [
 ]
 
 let retries = 40
-const waitTimeMs = 60000
+const waitTimeMs = 30000
 
 async function main () {
   const waitingMap = {}

src/domain/position/binProcessor.js

@@ -34,6 +34,7 @@ const Logger = require('@mojaloop/central-services-logger')
 const BatchPositionModel = require('../../models/position/batch')
 const BatchPositionModelCached = require('../../models/position/batchCached')
 const PositionPrepareDomain = require('./prepare')
+const PositionFulfilDomain = require('./fulfil')
 const SettlementModelCached = require('../../models/settlement/settlementModelCached')
 const Enum = require('@mojaloop/central-services-shared').Enum
 const ErrorHandler = require('@mojaloop/central-services-error-handling')
@@ -52,9 +53,12 @@ const participantFacade = require('../../models/participant/facade')
  */
 const processBins = async (bins, trx) => {
   const transferIdList = []
-  iterateThroughBins(bins, (_accountID, _action, item) => {
+  await iterateThroughBins(bins, (_accountID, _action, item) => {
     if (item.decodedPayload?.transferId) {
       transferIdList.push(item.decodedPayload.transferId)
+    // get transferId from uriParams for fulfil messages
+    } else if (item.message?.value?.content?.uriParams?.id) {
+      transferIdList.push(item.message.value.content.uriParams.id)
     }
   })
   // Pre fetch latest transferStates for all the transferIds in the account-bin
@@ -120,13 +124,20 @@ const processBins = async (bins, trx) => {
     ...settlementCurrencyIds.map(pc => pc.participantCurrencyId)
   ])
 
+  const latestTransferInfoByTransferId = await BatchPositionModel.getTransferInfosToChangePosition(
+    trx,
+    transferIdList,
+    Enum.Accounts.TransferParticipantRoleType.PAYEE_DFSP,
+    Enum.Accounts.LedgerEntryType.PRINCIPLE_VALUE
+  )
+
   let notifyMessages = []
   let limitAlarms = []
 
   // For each account-bin in the list
   for (const accountID in bins) {
     const accountBin = bins[accountID]
-    const actions = Object.keys(accountBin)
+    // const actions = Object.keys(accountBin)
 
     const settlementParticipantPosition = positions[accountIdMap[accountID].settlementCurrencyId].value
     const settlementModel = currencyIdMap[accountIdMap[accountID].currencyId].settlementModel
@@ -146,12 +157,24 @@ const processBins = async (bins, trx) => {
     let accumulatedTransferStateChanges = []
     let accumulatedPositionChanges = []
 
-    // If non-prepare action found, log error
-    // We need to remove this once we implement all the actions
-    if (actions.length > 1 || (actions.length === 1 && actions[0] !== 'prepare')) {
-      Logger.isErrorEnabled && Logger.error('Only prepare action is allowed in a batch')
-      // throw new Error('Only prepare action is allowed in a batch')
-    }
+    // If fulfil action found then call processPositionPrepareBin function
+    const fulfilActionResult = await PositionFulfilDomain.processPositionFulfilBin(
+      accountBin.commit,
+      accumulatedPositionValue,
+      accumulatedPositionReservedValue,
+      accumulatedTransferStates,
+      latestTransferInfoByTransferId
+    )
+
+    // Update accumulated values
+    accumulatedPositionValue = fulfilActionResult.accumulatedPositionValue
+    accumulatedPositionReservedValue = fulfilActionResult.accumulatedPositionReservedValue
+    accumulatedTransferStates = fulfilActionResult.accumulatedTransferStates
+    // Append accumulated arrays
+    accumulatedTransferStateChanges = accumulatedTransferStateChanges.concat(fulfilActionResult.accumulatedTransferStateChanges)
+    accumulatedPositionChanges = accumulatedPositionChanges.concat(fulfilActionResult.accumulatedPositionChanges)
+    notifyMessages = notifyMessages.concat(fulfilActionResult.notifyMessages)
+
     // If prepare action found then call processPositionPrepareBin function
     const prepareActionResult = await PositionPrepareDomain.processPositionPrepareBin(
       accountBin.prepare,

src/domain/position/fulfil.js

@@ -0,0 +1,153 @@
+const { Enum } = require('@mojaloop/central-services-shared')
+const ErrorHandler = require('@mojaloop/central-services-error-handling')
+const Config = require('../../lib/config')
+const Utility = require('@mojaloop/central-services-shared').Util
+const MLNumber = require('@mojaloop/ml-number')
+const Logger = require('@mojaloop/central-services-logger')
+
+/**
+ * @function processPositionFulfilBin
+ *
+ * @async
+ * @description This is the domain function to process a bin of position-fulfil messages of a single participant account.
+ *
+ * @param {array} binItems - an array of objects that contain a position fulfil message and its span. {message, span}
+ * @param {number} accumulatedPositionValue - value of position accumulated so far from previous bin processing
+ * @param {number} accumulatedPositionReservedValue - value of position reserved accumulated so far, not used but kept for consistency
+ * @param {object} accumulatedTransferStates - object with transfer id keys and transfer state id values. Used to check if transfer is in correct state for processing. Clone and update states for output.
+ * @param {object} accumulatedTransferInfo - object with transfer id keys and transfer info values. Used to pass transfer info to domain function.
+ * @returns {object} - Returns an object containing accumulatedPositionValue, accumulatedPositionReservedValue, accumulatedTransferStateChanges, accumulatedTransferStates, resultMessages, limitAlarms or throws an error if failed
+ */
+const processPositionFulfilBin = async (
+  binItems,
+  accumulatedPositionValue,
+  accumulatedPositionReservedValue,
+  accumulatedTransferStates,
+  accumulatedTransferInfo
+) => {
+  const transferStateChanges = []
+  const participantPositionChanges = []
+  const resultMessages = []
+  const accumulatedTransferStatesCopy = Object.assign({}, accumulatedTransferStates)
+  let runningPosition = new MLNumber(accumulatedPositionValue)
+
+  if (binItems && binItems.length > 0) {
+    for (const binItem of binItems) {
+      let transferStateId
+      let reason
+      let resultMessage
+      const transferId = binItem.message.value.content.uriParams.id
+      const payeeFsp = binItem.message.value.from
+      const payerFsp = binItem.message.value.to
+      const transfer = binItem.decodedPayload
+      Logger.isDebugEnabled && Logger.debug(`processPositionFulfilBin::transfer:processingMessage: ${JSON.stringify(transfer)}`)
+      Logger.isDebugEnabled && Logger.debug(`accumulatedTransferStates: ${JSON.stringify(accumulatedTransferStates)}`)
+      // Inform payee dfsp if transfer is not in RECEIVED_FULFIL state, skip making any transfer state changes
+      if (accumulatedTransferStates[transferId] !== Enum.Transfers.TransferInternalState.RECEIVED_FULFIL) {
+        // forward same headers from the prepare message, except the content-length header
+        // set destination to payeefsp and source to switch
+        const headers = { ...binItem.message.value.content.headers }
+        headers[Enum.Http.Headers.FSPIOP.DESTINATION] = payeeFsp
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        delete headers['content-length']
+
+        const fspiopError = ErrorHandler.Factory.createInternalServerFSPIOPError(
+          `Invalid State: ${accumulatedTransferStates[transferId]} - expected: ${Enum.Transfers.TransferInternalState.RECEIVED_FULFIL}`
+        ).toApiErrorObject(Config.ERROR_HANDLING)
+        const state = Utility.StreamingProtocol.createEventState(
+          Enum.Events.EventStatus.FAILURE.status,
+          fspiopError.errorInformation.errorCode,
+          fspiopError.errorInformation.errorDescription
+        )
+
+        const metadata = Utility.StreamingProtocol.createMetadataWithCorrelatedEvent(
+          transferId,
+          Enum.Kafka.Topics.NOTIFICATION,
+          Enum.Events.Event.Action.FULFIL,
+          state
+        )
+
+        resultMessage = Utility.StreamingProtocol.createMessage(
+          transferId,
+          payeeFsp,
+          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          metadata,
+          headers,
+          fspiopError,
+          { id: transferId },
+          'application/json'
+        )
+      } else {
+        const transferInfo = accumulatedTransferInfo[transferId]
+
+        // forward same headers from the prepare message, except the content-length header
+        const headers = { ...binItem.message.value.content.headers }
+        delete headers['content-length']
+
+        const state = Utility.StreamingProtocol.createEventState(
+          Enum.Events.EventStatus.SUCCESS.status,
+          null,
+          null
+        )
+        const metadata = Utility.StreamingProtocol.createMetadataWithCorrelatedEvent(
+          transferId,
+          Enum.Kafka.Topics.TRANSFER,
+          Enum.Events.Event.Action.COMMIT,
+          state
+        )
+
+        resultMessage = Utility.StreamingProtocol.createMessage(
+          transferId,
+          payerFsp,
+          payeeFsp,
+          metadata,
+          headers,
+          transfer,
+          { id: transferId },
+          'application/json'
+        )
+
+        transferStateId = Enum.Transfers.TransferState.COMMITTED
+        // Amounts in `transferParticipant` for the payee are stored as negative values
+        runningPosition = new MLNumber(runningPosition.add(transferInfo.amount).toFixed(Config.AMOUNT.SCALE))
+
+        const participantPositionChange = {
+          transferId, // Need to delete this in bin processor while updating transferStateChangeId
+          transferStateChangeId: null, // Need to update this in bin processor while executing queries
+          value: runningPosition.toNumber(),
+          reservedValue: accumulatedPositionReservedValue
+        }
+        participantPositionChanges.push(participantPositionChange)
+        binItem.result = { success: true }
+      }
+
+      resultMessages.push({ binItem, message: resultMessage })
+
+      if (transferStateId) {
+        const transferStateChange = {
+          transferId,
+          transferStateId,
+          reason
+        }
+        transferStateChanges.push(transferStateChange)
+        Logger.isDebugEnabled && Logger.debug(`processPositionFulfilBin::transferStateChange: ${JSON.stringify(transferStateChange)}`)
+
+        accumulatedTransferStatesCopy[transferId] = transferStateId
+        Logger.isDebugEnabled && Logger.debug(`processPositionFulfilBin::accumulatedTransferStatesCopy:finalizedTransferState ${JSON.stringify(transferStateId)}`)
+      }
+    }
+  }
+
+  return {
+    accumulatedPositionValue: runningPosition.toNumber(),
+    accumulatedTransferStates: accumulatedTransferStatesCopy, // finalized transfer state after fulfil processing
+    accumulatedPositionReservedValue, // not used but kept for consistency
+    accumulatedTransferStateChanges: transferStateChanges, // transfer state changes to be persisted in order
+    accumulatedPositionChanges: participantPositionChanges, // participant position changes to be persisted in order
+    notifyMessages: resultMessages // array of objects containing bin item and result message. {binItem, message}
+  }
+}
+
+module.exports = {
+  processPositionFulfilBin
+}

src/handlers/transfers/handler.js

@@ -665,8 +665,25 @@ const fulfil = async (error, messages) => {
         await TransferService.handlePayeeResponse(transferId, payload, action)
         const eventDetail = { functionality: TransferEventType.POSITION, action }
         // Key position fulfil message with payee account id
+        let topicNameOverride
+        if (action === TransferEventAction.COMMIT) {
+          topicNameOverride = Config.KAFKA_CONFIG.EVENT_TYPE_ACTION_TOPIC_MAP?.POSITION?.COMMIT
+        } else if (action === TransferEventAction.RESERVE) {
+          topicNameOverride = Config.KAFKA_CONFIG.EVENT_TYPE_ACTION_TOPIC_MAP?.POSITION?.RESERVE
+        } else if (action === TransferEventAction.BULK_COMMIT) {
+          topicNameOverride = Config.KAFKA_CONFIG.EVENT_TYPE_ACTION_TOPIC_MAP?.POSITION?.BULK_COMMIT
+        }
         const payeeAccount = await Participant.getAccountByNameAndCurrency(transfer.payeeFsp, transfer.currency, Enum.Accounts.LedgerAccountType.POSITION)
-        await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, eventDetail, messageKey: payeeAccount.participantCurrencyId.toString() })
+        await Kafka.proceed(
+          Config.KAFKA_CONFIG,
+          params,
+          {
+            consumerCommit,
+            eventDetail,
+            messageKey: payeeAccount.participantCurrencyId.toString(),
+            topicNameOverride
+          }
+        )
         histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
         return true
       }

src/models/position/batch.js

@@ -104,6 +104,37 @@ const updateParticipantPosition = async (trx, participantPositionId, participant
     })
 }
 
+const getTransferInfosToChangePosition = async (trx, transferIds, transferParticipantRoleTypeId, ledgerEntryTypeId) => {
+  try {
+    const knex = await Db.getKnex()
+    const transferInfos = await knex('transferParticipant')
+      .transacting(trx)
+      .innerJoin('transferStateChange AS tsc', 'tsc.transferId', 'transferParticipant.transferId')
+      .where({
+        'transferParticipant.transferParticipantRoleTypeId': transferParticipantRoleTypeId,
+        'transferParticipant.ledgerEntryTypeId': ledgerEntryTypeId
+      })
+      .whereIn('transferParticipant.transferId', transferIds)
+      .select(
+        'transferParticipant.*',
+        'tsc.transferStateId',
+        'tsc.reason'
+      )
+      .orderBy('tsc.transferStateChangeId', 'desc')
+    const info = {}
+    // This should key the transfer info with the latest transferStateChangeId
+    for (const transferInfo of transferInfos) {
+      if (!(transferInfo.transferId in info)) {
+        info[transferInfo.transferId] = transferInfo
+      }
+    }
+    return info
+  } catch (err) {
+    Logger.isErrorEnabled && Logger.error(err)
+    throw err
+  }
+}
+
 const bulkInsertTransferStateChanges = async (trx, transferStateChangeList) => {
   const knex = await Db.getKnex()
   return await knex.batchInsert('transferStateChange', transferStateChangeList).transacting(trx)
@@ -121,5 +152,6 @@ module.exports = {
   updateParticipantPosition,
   bulkInsertTransferStateChanges,
   bulkInsertParticipantPositionChanges,
-  getAllParticipantCurrency
+  getAllParticipantCurrency,
+  getTransferInfosToChangePosition
 }