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

* feat: added skeleton and comments for prepare bin implementation * chore: some changes * feat: added new position handler v2 * feat: added some functionality to new position handler v2 * feat: added some todos * feat: changed new handler name * fix: lint * feat: add some changes * fix: lint * chore: added copyright header * feat: added some logic * fix: lint * feat: added unit tests * feat: some improvements * feat: added bin processor logic * fix: lint * feat; refactor * feat: added limit * fix: unit tests * feat: added integration test * feat: refactor * fix: unit tests * fix: integration tests * feat: add position prepare bin logic (#970) * feat: add position prepare bin logic * edit * lint * dep * chore: changes * chore: commet * lint * chore: refactor * chore: unit tests * chore: change * chore: comment * chore: test * chore: more test * chore: test? * chore: remove * chore: rework domain logic * chore: test * fix: position calculation * fix: position change data * fix; prepare domain function * chore: fix --------- Co-authored-by: Vijay <vijaya.guthi@infitx.com> * chore: add tests (#972) * chore: add tests * chore: unit * sleep * chore: clear * chore: test * enum * fix: deadlock in tests --------- Co-authored-by: Vijay <vijaya.guthi@infitx.com> * fix: metrics * fix: some issue with bin processor * feat(mojaloop/#3489): add cached calls for participant currency, update tests (#973) * chore: add tests * chore: unit * sleep * chore: clear * chore: test * enum * feat: add cached calls for participant currency * chore: change * chore: unit * chore: unit * lock * test * fix: unit test --------- Co-authored-by: Vijay <vijaya.guthi@infitx.com> * chore: improve coverage and edit metric and query name (#974) * chore: coverage * lint * chore: coverage * more coverage * chore: test * chore: coverage * chore: metric name * chore: metric name * chore: update query name * chore: remove log * chore(snapshot): 17.3.0-snapshot.0 * fix: test downgrade stream lib * chore(snapshot): 17.3.0-snapshot.1 * fix: notification event action * chore(snapshot): 17.3.0-snapshot.2 * fix: notification event * chore(snapshot): 17.3.0-snapshot.3 * fix: notification messages * chore(snapshot): 17.3.0-snapshot.4 * chore: updated conf * chore: update deps * chore(snapshot): 17.3.0-snapshot.5 * fix(mojaloop/#3529): fix high latency (#981) * fix: upgrade central-services-stream library * chore(snapshot): v17.2.1-snapshot.0 * chore(snapshot): 17.2.1-snapshot.1 * chore(snapshot): 17.2.1-snapshot.2 * chore(release): 17.2.1 [skip ci] * feat(mojaloop/#3489): add negative integration tests (#975) * chore: add support for unique transfers and dual currencies in batch * check in for a negative test case * uncommented tests * added test for mixed currencies * updated test for mixed currencies * chore: dep and lint --------- Co-authored-by: Kevin Leyow <kleyow@gmail.com> * Merge branch main of https://github.com/mojaloop/central-ledger into feature/position-prepare-binning * chore: foreach to for loop * chore: cleanup * Update src/handlers/index.js Co-authored-by: Miguel de Barros <miguel@debarros.me> * chore: added readme doc * fix: lint * fix: unit tests * fix: audit * chore: cleanup * fix(mojaloop/#3533): helm v15.2.0-rc fixes (#982) fix(mojaloop/#3533): helm v15.2.0-rc fixes - mojaloop/project#3533 List of fixes: - fix(mojaloop/#3580): missing toDestination on handling the fspiop source/destiation headers failing match validation on fulfil - regression on #2697 - v17.0.0...v17.2.0#diff-3a2d4aabbde0cd9517dd372f6ae6001ad607d005b5316785c8698fe25160aa92L393 - mojaloop/project#3580 Fixes currently resolve regression failures on these tests: - p2p_money_transfer_put_notifications - payee receives no Notification with ABORTED status after sending invalid FSPIOP-Destination header with transferStatus=COMMITTED, file path: golden_path/bug fixes /Test for Bugfix #2697 - Central-Ledger Fulfil Handler does not correctly invalidate requests with an incorrect-non-existent FSP-ID in the FSPIOP-Destination header.json - p2p_money_transfer_patch_notifications - payee receives PATCH Notification with ABORTED status after sending invalid FSPIOP-Destination header with transferStatus=RESERVED, file path: golden_path/bug fixes/Test for Bugfix #2697 - Central-Ledger Fulfil Handler does not correctly invalidate requests with an incorrect-non-existent FSP-ID in the FSPIOP-Destination header.json * chore(release): 17.3.0 [skip ci] * chore: remove uuid4 (#976) chore: remove uuid4 - Remove uuid4 and use native randomUUID. - Bump some deps are quested by commit hook * chore(release): 17.3.1 [skip ci] * fix(mojaloop/#3615): upgrade dependencies (#985) * chore(release): 17.3.2 [skip ci] * fix: remove unneeded async/await * feat: add validation for participantCurrencyIds / accountIds mapping * chore: remove resolved TODOs * test: Add test coverage * chore: remove whitespaces * fix: fixrebase merge conflict * chore: update dependencies * doc: fix typo * chore(snapshot): 17.4.0-snapshot.0 * chore(snapshot): 17.4.0-snapshot.1 * test: update unit tests * chore(snapshot): 17.4.0-snapshot.2 * fix: route error callback for batch correctly * chore(snapshot): 17.4.0-snapshot.3 * fix(batch): fix error callback message routing. update tests * chore(snapshot): 17.4.0-snapshot.4 * fix(batch): fix message routing / properties for more error cases * chore(snapshot): 17.4.0-snapshot.5 * test: update test coverage * chore(snapshot): 17.4.0-snapshot.6 * fix: route bulk-prepare messages to non-batch prepare handler * chore(snapshot): 17.4.0-snapshot.7 * doc: update comment * chore: update deps. update README for batch. * doc: Address TODO regarding participanLimits query optimization * stash * chore: domain * chore: test * test * chore: update bin processor tests * chore: coverage * log * chore: update * chore: up retries * chore: test * ci * test * install * test * test * test * script * chore: fix order * time * test * time * time * state * time * chore: check transfer state * test * lint * chore: await * address comments * chore: dep * chore: address comments * coverage * chore(snapshot): 17.5.0-snapshot.0 * remove --------- Co-authored-by: Vijay <vijaya.guthi@infitx.com> Co-authored-by: Aarón Reynoza <aaron.alejandro.reynoza@gmail.com> Co-authored-by: mojaloopci <info@mojaloop.io> Co-authored-by: Sridevi Miriyala <38143240+sri-miriyala@users.noreply.github.com> Co-authored-by: vijayg10 <33152110+vijayg10@users.noreply.github.com> Co-authored-by: Miguel de Barros <miguel@debarros.me> Co-authored-by: Marco Ippolito <marcoippolito54@gmail.com> Co-authored-by: Steven Oderayi <oderayi@gmail.com>
mojaloop · Dec 13, 2023 · dddedf6 · dddedf6
1 parent bd7ce1f
commit dddedf6
Show file tree

Hide file tree

Showing 19 changed files with 1,496 additions and 56 deletions.
diff --git a/.circleci/config.yml b/.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:

diff --git a/.circleci/curl-retry-cl-health.sh b/.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

diff --git a/README.md b/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'`
 

diff --git a/config/default.json b/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": {

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/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",

diff --git a/scripts/_wait4_all.js b/scripts/_wait4_all.js
@@ -21,7 +21,7 @@ const expectedContainers = [
 ]
 
 let retries = 40
-const waitTimeMs = 60000
+const waitTimeMs = 30000
 
 async function main () {
  const waitingMap = {}

diff --git a/src/domain/position/binProcessor.js b/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,