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": {

package.json

@@ -1,6 +1,6 @@
 {
  "name": "@mojaloop/central-ledger",
- "version": "17.4.1",
+ "version": "17.5.0-snapshot.0",
  "description": "Central ledger hosted by a scheme to record and settle transfers",
  "license": "Apache-2.0",
  "author": "ModusBox",
@@ -111,7 +111,7 @@
  "hapi-auth-bearer-token": "8.0.0",
  "hapi-swagger": "17.2.0",
  "ilp-packet": "2.2.0",
- "knex": "3.0.1",
+ "knex": "3.1.0",
  "lodash": "4.17.21",
  "moment": "2.29.4",
  "mongo-uri-builder": "^4.0.0",

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,28 @@ const processBins = async (bins, trx) => {
  ...settlementCurrencyIds.map(pc => pc.participantCurrencyId)
  ])
 
+ const latestTransferInfoByTransferId = await BatchPositionModel.getTransferInfoList(
+ 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 isSubset = (array1, array2) =>
+ array2.every((element) => array1.includes(element))
+ // If non-prepare/non-commit action found, log error
+ // We need to remove this once we implement all the actions
+ if (!isSubset(['prepare', 'commit'], actions)) {
+ Logger.isErrorEnabled && Logger.error('Only prepare/commit actions are allowed in a batch')
+ // throw new Error('Only prepare action is allowed in a batch')
+ }
 
  const settlementParticipantPosition = positions[accountIdMap[accountID].settlementCurrencyId].value
  const settlementModel = currencyIdMap[accountIdMap[accountID].currencyId].settlementModel
@@ -146,12 +165,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,