test(csi/551): add integration test for fx-forwarded message

docker-compose.yml

@@ -70,9 +70,9 @@ services:
 
   central-ledger:
     # image: mojaloop/central-ledger:latest
-    image: mojaloop/central-ledger:v17.8.0.160-snapshot.4
+    image: mojaloop/central-ledger:v17.8.0.551-snapshot.16
     container_name: ml_central-ledger
-    command: sh -c "/opt/app/wait4/wait4.js central-ledger && node src/api/index.js"
+    command: sh -c "/opt/app/wait4/wait4.js central-ledger && npm run migrate && node src/api/index.js"
     links:
       - mysql
       - kafka
@@ -97,6 +97,36 @@ services:
       retries: 10
       interval: 30s
 
+  central-handler-position-batch:
+    image: mojaloop/central-ledger:v17.8.0.551-snapshot.16
+    command: sh -c " /opt/app/wait4/wait4.js central-ledger && "CLEDG_HANDLERS__API__DISABLED=true" node src/handlers/index.js handler --positionbatch"
+    ports:
+      - "3002:3001"
+    links:
+      - mysql
+      - kafka
+    volumes:
+      - ./docker/central-ledger/default.json:/opt/app/config/default.json
+      - ./docker/wait4:/opt/app/wait4
+    environment:
+      - LOG_LEVEL=info
+      - CLEDG_MONGODB__DISABLED=true
+      - EVENT_SDK_TRACEID_PER_VENDOR=false
+    networks:
+      - ml-mojaloop-net
+    depends_on:
+      central-ledger:
+        condition: service_healthy
+      kafka:
+        condition: service_started
+    healthcheck:
+      test: wget -q http://localhost:3001/health -O /dev/null || exit 1
+      timeout: 20s
+      retries: 30
+      interval: 30s
+      start_period: 30s
+    user: root
+
   mysql:
     image: mysql/mysql-server
     container_name: ml_mysql
@@ -182,6 +212,22 @@ services:
       retries: 10
       interval: 30s
 
+  ## Debug utilities
+  kafka-debug-console:
+    # image: quay.io/cloudhut/kowl:v1.4.0
+    image: docker.redpanda.com/redpandadata/console:latest
+    deploy:
+      replicas: 1
+    restart: on-failure
+    hostname: kafka-debug-console
+    ports:
+    - "9080:8080"
+    networks:
+      - ml-mojaloop-net
+    environment:
+      - KAFKA_BROKERS=kafka:29092
+
+
   redis-node-0:
     <<: *REDIS_NODE
     environment:

docker/central-ledger/default.json

@@ -2,7 +2,7 @@
   "PORT": 3001,
   "HOSTNAME": "http://central-ledger",
   "MIGRATIONS": {
-    "DISABLED": false,
+    "DISABLED": true,
     "RUN_DATA_MIGRATIONS": true
   },
   "DATABASE": {
@@ -84,6 +84,15 @@
   },
   "INTERNAL_TRANSFER_VALIDITY_SECONDS": "432000",
   "ENABLE_ON_US_TRANSFERS": false,
+  "PROXY_CACHE": {
+    "enabled": true,
+    "type": "redis-cluster",
+    "proxyConfig": {
+      "cluster": [
+        { "host": "redis-node-0", "port": 6379 }
+      ]
+    }
+  },
   "CACHE": {
     "CACHE_ENABLED": false,
     "MAX_BYTE_SIZE": 10000000,
@@ -93,11 +102,17 @@
   "KAFKA": {
     "EVENT_TYPE_ACTION_TOPIC_MAP" : {
       "POSITION":{
-        "PREPARE": null,
+        "PREPARE": "topic-transfer-position-batch",
+        "FX_PREPARE": "topic-transfer-position-batch",
         "BULK_PREPARE": null,
         "COMMIT": null,
         "BULK_COMMIT": null,
-        "RESERVE": null
+        "RESERVE": null,
+        "FX_RESERVE": "topic-transfer-position-batch",
+        "TIMEOUT_RESERVED": "topic-transfer-position-batch",
+        "FX_TIMEOUT_RESERVED": "topic-transfer-position-batch",
+        "ABORT": "topic-transfer-position-batch",
+        "FX_ABORT": "topic-transfer-position-batch"
       }
     },
     "TOPIC_TEMPLATES": {
@@ -323,7 +338,7 @@
             "rdkafkaConf": {
               "client.id": "cl-con-transfer-position-batch",
               "group.id": "cl-group-transfer-position-batch",
-              "metadata.broker.list": "kafka:9092",
+              "metadata.broker.list": "kafka:29092",
               "socket.keepalive.enable": true,
               "allow.auto.create.topics": true,
               "partition.assignment.strategy": "cooperative-sticky",

package.json

@@ -96,7 +96,7 @@
     "@mojaloop/event-sdk": "14.1.1",
     "@mojaloop/sdk-standard-components": "18.4.0",
     "@now-ims/hapi-now-auth": "2.1.0",
-    "axios": "1.7.5",
+    "axios": "1.7.7",
     "blipp": "4.0.2",
     "commander": "12.1.0",
     "docdash": "2.0.2",
@@ -119,7 +119,7 @@
     "license-checker": "25.0.1",
     "nodemon": "3.1.4",
     "npm-audit-resolver": "3.0.0-RC.0",
-    "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",

test/integration/handlers/notification/index.test.js

@@ -206,6 +206,64 @@ Test('Notification Handler', notificationHandlerTest => {
       test.end()
     })
 
+    notificationTest.test('consume a FX_PREPARE message and send POST callback to proxy', async test => {
+      await proxy.addDfspIdToProxyMapping('proxied2', 'fxp1') // simulate proxy mapping
+      const commitRequestId = Uuid()
+      const payload = {
+        commitRequestId,
+        determiningTransferId: Uuid(),
+        initiatingFsp: 'dfsp1',
+        counterPartyFsp: 'proxied2',
+        amountType: 'SEND',
+        sourceAmount: { amount: 100, currency: 'USD' },
+        targetAmount: { amount: 200, currency: 'USD' },
+        condition: 'uU0nuZNNPgilLlLX2n2r-sSE7-N6U4DukIj3rOLvze1',
+        expiration: new Date((new Date()).getTime() + (24 * 60 * 60 * 1000)).toISOString(), // tomorrow
+        ilpPacket: 'AQAAAAAAAABkEGcuZXdwMjEuaWQuODAwMjCCAhd7InRyYW5zYWN0aW9uSWQiOiJmODU0NzdkYi0xMzVkLTRlMDgtYThiNy0xMmIyMmQ4MmMwZDYiLCJxdW90ZUlkIjoiOWU2NGYzMjEtYzMyNC00ZDI0LTg5MmYtYzQ3ZWY0ZThkZTkxIiwicGF5ZWUiOnsicGFydHlJZEluZm8iOnsicGFydHlJZFR5cGUiOiJNU0lTRE4iLCJwYXJ0eUlkZW50aWZpZXIiOiIyNTYxMjM0NTYiLCJmc3BJZCI6IjIxIn19LCJwYXllciI6eyJwYXJ0eUlkSW5mbyI6eyJwYXJ0eUlkVHlwZSI6Ik1TSVNETiIsInBhcnR5SWRlbnRpZmllciI6IjI1NjIwMTAwMDAxIiwiZnNwSWQiOiIyMCJ9LCJwZXJzb25hbEluZm8iOnsiY29tcGxleE5hbWUiOnsiZmlyc3ROYW1lIjoiTWF0cyIsImxhc3ROYW1lIjoiSGFnbWFuIn0sImRhdGVPZkJpcnRoIjoiMTk4My0xMC0yNSJ9fSwiYW1vdW50Ijp7ImFtb3VudCI6IjEwMCIsImN1cnJlbmN5IjoiVVNEIn0sInRyYW5zYWN0aW9uVHlwZSI6eyJzY2VuYXJpbyI6IlRSQU5TRkVSIiwiaW5pdGlhdG9yIjoiUEFZRVIiLCJpbml0aWF0b3JUeXBlIjoiQ09OU1VNRVIifSwibm90ZSI6ImhlaiJ9'
+      }
+      await prepare(
+        {
+          'fspiop-source': payload.initiatingFsp,
+          'fspiop-destination': payload.counterPartyFsp
+        },
+        encodePayload(JSON.stringify(payload), 'application/vnd.interoperability.fxTransfers+json;version=1.1'),
+        payload,
+        { injectContextToMessage: msg => msg }
+      )
+      const messageProtocol = Fixtures.createMessageProtocol(
+        Action.PREPARE,
+        Action.FX_PREPARE,
+        payload,
+        payload.initiatingFsp,
+        payload.counterPartyFsp
+      )
+      const { kafkaConfig, topicConfig } = Fixtures.createProducerConfig(
+        Config.KAFKA_CONFIG, EventTypes.TRANSFER, EventActions.PREPARE,
+        GeneralTopicTemplate, EventTypes.NOTIFICATION, EventActions.EVENT
+      )
+      await new Promise(resolve => setTimeout(resolve, 10000)) // wait for RESERVED
+      const response = await testNotification(messageProtocol, 'post', commitRequestId, kafkaConfig, topicConfig, undefined, undefined, 'fxp1')
+      await new Promise(resolve => setTimeout(resolve, 5000)) // wait for RESERVED_FORWARDED
+      await db.connect({
+        client: centralLedgerConfig.DATABASE.DIALECT,
+        connection: {
+          host: 'localhost',
+          port: centralLedgerConfig.DATABASE.PORT,
+          user: centralLedgerConfig.DATABASE.USER,
+          password: centralLedgerConfig.DATABASE.PASSWORD,
+          database: centralLedgerConfig.DATABASE.SCHEMA
+        }
+      })
+      try {
+        const stateChange = await db.from('fxTransferStateChange').findOne({ commitRequestId, transferStateId: Enum.Transfers.TransferInternalState.RESERVED_FORWARDED })
+        test.equal(stateChange.transferStateId, Enum.Transfers.TransferInternalState.RESERVED_FORWARDED, 'Fx Transfer state changed to RESERVED_FORWARDED')
+      } finally {
+        await db.disconnect()
+      }
+      test.deepEqual(response.payload, messageProtocol.content.payload, 'Notification sent successfully to FXP')
+      test.end()
+    })
+
     notificationTest.test('consume a PREPARE message and send PUT callback on error', async test => {
       const transferId = Uuid()
       const messageProtocol = Fixtures.createMessageProtocol(

test/scripts/env.sh

@@ -6,7 +6,7 @@ DEFAULT_CONFIG_FILE="$BASE_DIR/../../docker/central-ledger/default.json"
 export HUB_NAME=$(cat "$DEFAULT_CONFIG_FILE" | jq -r '.HUB_PARTICIPANT.NAME')
 
 export FSPList=("dfsp1" "dfsp2" "proxied2")
-export DEFAULT_NET_DEBIT_CAP=1000
+export DEFAULT_NET_DEBIT_CAP=5000
 export CENTRAL_LEDGER_ADMIN_URI_PREFIX=http
 export CENTRAL_LEDGER_ADMIN_HOST=127.0.0.1
 export CENTRAL_LEDGER_ADMIN_PORT=3001

test/scripts/populateTestData.sh

@@ -101,8 +101,6 @@ curl -i -X POST "${CENTRAL_LEDGER_ADMIN_URI_PREFIX}://${CENTRAL_LEDGER_ADMIN_HOS
       \"isProxy\": $ISPROXY
     }"
 
-if ! isProxy $FSP; then
-
   echo
   echo "Setting limits and initial position for '$FSP'"
   echo "---------------------------------------------------------------------"
@@ -155,8 +153,6 @@ if ! isProxy $FSP; then
   echo "---------------------------------------------------------------------"
   curl -X GET "${CENTRAL_LEDGER_ADMIN_URI_PREFIX}://${CENTRAL_LEDGER_ADMIN_HOST}:${CENTRAL_LEDGER_ADMIN_PORT}${CENTRAL_LEDGER_ADMIN_BASE}participants/${FSP}/limits" -H 'Cache-Control: no-cache'
 
-fi
-
 
   echo
   echo "Get accounts list for '$FSP' to show balances"