feat: parameterize switch id

docker/ml-api-adapter/default.json

@@ -1,4 +1,8 @@
 {
+  "HUB_PARTICIPANT": {
+    "ID": 1,
+    "NAME": "Hub"
+  },
   "PORT": 3000,
   "HOSTNAME": "http://ml-api-adapter",
   "ENDPOINT_SOURCE_URL": "http://host.docker.internal:3001",
@@ -13,7 +17,6 @@
     },
     "JWS": {
       "JWS_SIGN": false,
-      "FSPIOP_SOURCE_TO_SIGN": "switch",
       "JWS_SIGNING_KEY_PATH": "secrets/jwsSigningKey.key"
     }
   },

package.json

@@ -101,7 +101,7 @@
     "@mojaloop/central-services-health": "15.0.0",
     "@mojaloop/central-services-logger": "11.3.1",
     "@mojaloop/central-services-metrics": "12.0.8",
-    "@mojaloop/central-services-shared": "18.4.0-snapshot.17",
+    "@mojaloop/central-services-shared": "18.5.0-snapshot.2",
     "@mojaloop/central-services-stream": "11.3.1",
     "@mojaloop/database-lib": "11.0.5",
     "@mojaloop/event-sdk": "14.1.1",
@@ -118,7 +118,7 @@
     "docdash": "2.0.2",
     "event-stream": "4.0.1",
     "five-bells-condition": "5.0.1",
-    "glob": "10.4.1",
+    "glob": "10.4.2",
     "hapi-auth-basic": "5.0.0",
     "hapi-auth-bearer-token": "8.0.0",
     "hapi-swagger": "17.2.1",
@@ -139,7 +139,7 @@
     "get-port": "5.1.1",
     "jsdoc": "4.0.3",
     "jsonpath": "1.1.1",
-    "nodemon": "3.1.3",
+    "nodemon": "3.1.4",
     "npm-check-updates": "16.14.20",
     "nyc": "17.0.0",
     "pre-commit": "1.2.2",

src/domain/position/fulfil.js

@@ -129,7 +129,7 @@ const _handleIncorrectTransferState = (binItem, payeeFsp, transferId, accumulate
   // 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
+  headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
   delete headers['content-length']
 
   const fspiopError = ErrorHandler.Factory.createInternalServerFSPIOPError(
@@ -151,7 +151,7 @@ const _handleIncorrectTransferState = (binItem, payeeFsp, transferId, accumulate
   return Utility.StreamingProtocol.createMessage(
     transferId,
     payeeFsp,
-    Enum.Http.Headers.FSPIOP.SWITCH.value,
+    Config.HUB_NAME,
     metadata,
     headers,
     fspiopError,

src/domain/position/fx-fulfil.js

@@ -41,7 +41,7 @@ const processPositionFxFulfilBin = async (
         // set destination to counterPartyFsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = counterPartyFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         // TODO: Confirm if this setting transferStateId to ABORTED_REJECTED is correct. There is no such logic in the fulfil handler.
@@ -67,7 +67,7 @@ const processPositionFxFulfilBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           commitRequestId,
           counterPartyFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,

src/domain/position/fx-prepare.js

@@ -65,7 +65,7 @@ const processFxPositionPrepareBin = async (
         // set destination to initiatingFsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = fxTransfer.initiatingFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -88,7 +88,7 @@ const processFxPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           fxTransfer.commitRequestId,
           fxTransfer.initiatingFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,
@@ -107,7 +107,7 @@ const processFxPositionPrepareBin = async (
         // set destination to payerfsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = fxTransfer.initiatingFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -130,7 +130,7 @@ const processFxPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           fxTransfer.commitRequestId,
           fxTransfer.initiatingFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,
@@ -149,7 +149,7 @@ const processFxPositionPrepareBin = async (
         // set destination to payerfsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = fxTransfer.initiatingFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -172,7 +172,7 @@ const processFxPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           fxTransfer.commitRequestId,
           fxTransfer.initiatingFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,

src/domain/position/prepare.js

@@ -67,7 +67,7 @@ const processPositionPrepareBin = async (
         // set destination to payerfsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = transfer.payerFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -90,7 +90,7 @@ const processPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           transfer.transferId,
           transfer.payerFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,
@@ -109,7 +109,7 @@ const processPositionPrepareBin = async (
         // set destination to payerfsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = transfer.payerFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -132,7 +132,7 @@ const processPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           transfer.transferId,
           transfer.payerFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,
@@ -151,7 +151,7 @@ const processPositionPrepareBin = async (
         // set destination to payerfsp and source to switch
         const headers = { ...binItem.message.value.content.headers }
         headers[Enum.Http.Headers.FSPIOP.DESTINATION] = transfer.payerFsp
-        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Enum.Http.Headers.FSPIOP.SWITCH.value
+        headers[Enum.Http.Headers.FSPIOP.SOURCE] = Config.HUB_NAME
         delete headers['content-length']
 
         const fspiopError = ErrorHandler.Factory.createFSPIOPError(
@@ -174,7 +174,7 @@ const processPositionPrepareBin = async (
         resultMessage = Utility.StreamingProtocol.createMessage(
           transfer.transferId,
           transfer.payerFsp,
-          Enum.Http.Headers.FSPIOP.SWITCH.value,
+          Config.HUB_NAME,
           metadata,
           headers,
           fspiopError,

src/handlers/bulk/fulfil/handler.js

@@ -110,7 +110,7 @@ const bulkFulfil = async (error, messages) => {
       Logger.isErrorEnabled && Logger.error(Util.breadcrumb(location, `callbackErrorModified--${actionLetter}2`))
       const fspiopError = ErrorHandler.Factory.createFSPIOPError(ErrorHandler.Enums.FSPIOPErrorCodes.MODIFIED_REQUEST)
       const eventDetail = { functionality: Enum.Events.Event.Type.NOTIFICATION, action }
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
       throw fspiopError
     }
 
@@ -134,7 +134,7 @@ const bulkFulfil = async (error, messages) => {
         const eventDetail = { functionality: Enum.Events.Event.Type.NOTIFICATION, action }
         params.message.value.content.uriParams = { id: bulkTransferId }
 
-        await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+        await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
         throw fspiopError
       }
       try {
@@ -240,15 +240,15 @@ const bulkFulfil = async (error, messages) => {
         const eventDetail = { functionality: Enum.Events.Event.Type.NOTIFICATION, action }
         params.message.value.content.uriParams = { id: bulkTransferId }
 
-        await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+        await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
         throw fspiopError
       }
       Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `callbackErrorGeneric--${actionLetter}8`))
 
       const eventDetail = { functionality: Enum.Events.Event.Type.NOTIFICATION, action }
       params.message.value.content.uriParams = { id: bulkTransferId }
 
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: validationFspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: validationFspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
       throw validationFspiopError
     }
   } catch (err) {
@@ -293,7 +293,7 @@ const sendIndividualTransfer = async (message, messageId, kafkaTopic, headers, p
     value: Util.StreamingProtocol.createMessage(messageId, headers[Enum.Http.Headers.FSPIOP.DESTINATION], headers[Enum.Http.Headers.FSPIOP.SOURCE], metadata, headers, dataUri, { id: transferId })
   }
   params = { message: msg, kafkaTopic, consumer: Consumer, producer: Producer }
-  await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, histTimerEnd, eventDetail })
+  await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, histTimerEnd, eventDetail, hubName: Config.HUB_NAME })
   histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
 }
 

src/handlers/bulk/get/handler.js

@@ -88,7 +88,7 @@ const getBulkTransfer = async (error, messages) => {
 
     if (!(await Validator.validateParticipantByName(message.value.from)).isValid) {
       Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `breakParticipantDoesntExist--${actionLetter}1`))
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, histTimerEnd })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, histTimerEnd, hubName: Config.HUB_NAME })
       histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
       return true
     }
@@ -97,7 +97,7 @@ const getBulkTransfer = async (error, messages) => {
     if (!bulkTransferLight) {
       Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `callbackErrorBulkTransferNotFound--${actionLetter}3`))
       const fspiopError = ErrorHandler.Factory.createFSPIOPError(ErrorHandler.Enums.FSPIOPErrorCodes.BULK_TRANSFER_ID_NOT_FOUND, 'Provided Bulk Transfer ID was not found on the server.')
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
       throw fspiopError
     }
     // The SD says this should be 404 response which I think will not be constent with single transfers
@@ -106,7 +106,7 @@ const getBulkTransfer = async (error, messages) => {
     if (![participants.payeeFsp, participants.payerFsp].includes(message.value.from)) {
       Logger.isInfoEnabled && Logger.info(Util.breadcrumb(location, `callbackErrorNotBulkTransferParticipant--${actionLetter}2`))
       const fspiopError = ErrorHandler.Factory.createFSPIOPError(ErrorHandler.Enums.FSPIOPErrorCodes.CLIENT_ERROR)
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(Config.ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
       throw fspiopError
     }
     const isPayeeRequest = participants.payeeFsp === message.value.from
@@ -129,9 +129,9 @@ const getBulkTransfer = async (error, messages) => {
     }
     message.value.content.payload = payload
     if (fspiopError) {
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(ERROR_HANDLING), eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, fspiopError: fspiopError.toApiErrorObject(ERROR_HANDLING), eventDetail, fromSwitch, hubName: Config.HUB_NAME })
     } else {
-      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, eventDetail, fromSwitch })
+      await Kafka.proceed(Config.KAFKA_CONFIG, params, { consumerCommit, eventDetail, fromSwitch, hubName: Config.HUB_NAME })
     }
     histTimerEnd({ success: true, fspId: Config.INSTRUMENTATION_METRICS_LABELS.fspId })
     return true