Feature/perf3.6 enhancements

seeds/ledgerAccountType.js

@@ -34,7 +34,7 @@ const ledgerAccountTypes = [
     'description': 'Reflects the individual DFSP Settlement Accounts as held at the Settlement Bank'
   },
   {
-    'name': 'HUB_SETTLEMENT',
+    'name': 'HUB_LIABILITY',
     'description': 'A single account for each currency with which the hub operates. The account is "held" by the Participant representing the hub in the switch'
   }
 ]

seeds/ledgerEntryType.js

@@ -58,12 +58,12 @@ const ledgerEntryTypes = [
     'description': 'Participant is settlement net sender'
   },
   {
-    'name': 'SETTLEMENT_ACCOUNT_DEPOSIT',
-    'description': 'Settlement account deposit'
+    'name': 'RECORD_FUNDS_IN',
+    'description': 'Settlement account funds in'
   },
   {
-    'name': 'SETTLEMENT_ACCOUNT_WITHDRAWAL',
-    'description': 'Settlement account withdrawal'
+    'name': 'RECORD_FUNDS_OUT',
+    'description': 'Settlement account funds out'
   }
 ]
 

src/admin/participants/handler.js

@@ -208,6 +208,15 @@ const getPositions = async function (request, h) {
   }
 }
 
+const getAccounts = async function (request, h) {
+  Sidecar.logRequest(request)
+  try {
+    return Participant.getAccounts(request.params.name, request.query)
+  } catch (err) {
+    throw Boom.badRequest()
+  }
+}
+
 module.exports = {
   create,
   getAll,
@@ -218,5 +227,6 @@ module.exports = {
   addLimitAndInitialPosition,
   getLimits,
   adjustLimits,
-  getPositions
+  getPositions,
+  getAccounts
 }

src/admin/participants/routes.js

@@ -222,5 +222,23 @@ module.exports = [
         }
       }
     }
+  },
+  {
+    method: 'GET',
+    path: '/participants/{name}/accounts',
+    handler: Handler.getAccounts,
+    options: {
+      id: 'participants_accounts_get',
+      tags: tags,
+      description: 'View participant accounts balances',
+      validate: {
+        params: {
+          name: nameValidator
+        },
+        query: {
+          currency: currencyValidator
+        }
+      }
+    }
   }
 ]

src/domain/participant/index.js

@@ -119,6 +119,8 @@ const getParticipantCurrencyById = async (participantCurrencyId) => {
 const destroyByName = async (name) => {
   try {
     let participant = await ParticipantModel.getByName(name)
+    await ParticipantLimitModel.destroyByParticipantId(participant.participantId)
+    await ParticipantPositionModel.destroyByParticipantId(participant.participantId)
     await ParticipantCurrencyModel.destroyByParticipantId(participant.participantId)
     return await ParticipantModel.destroyByName(name)
   } catch (err) {
@@ -254,16 +256,18 @@ const addLimitAndInitialPosition = async (participantName, limitAndInitialPositi
   try {
     const participant = await ParticipantFacade.getByNameAndCurrency(participantName, limitAndInitialPositionObj.currency, Enum.LedgerAccountType.POSITION)
     participantExists(participant)
+    const settlementAccount = await ParticipantFacade.getByNameAndCurrency(participantName, limitAndInitialPositionObj.currency, Enum.LedgerAccountType.SETTLEMENT)
     const existingLimit = await ParticipantLimitModel.getByParticipantCurrencyId(participant.participantCurrencyId)
     const existingPosition = await ParticipantPositionModel.getByParticipantCurrencyId(participant.participantCurrencyId)
-    if (existingLimit || existingPosition) {
+    const existingSettlementPosition = await ParticipantPositionModel.getByParticipantCurrencyId(settlementAccount.participantCurrencyId)
+    if (existingLimit || existingPosition || existingSettlementPosition) {
       throw new Error('Participant Limit or Initial Position already set')
     }
     let limitAndInitialPosition = limitAndInitialPositionObj
     if (limitAndInitialPosition.initialPosition == null) {
       limitAndInitialPosition.initialPosition = Config.PARTICIPANT_INITIAL_POSTITION
     }
-    return ParticipantFacade.addLimitAndInitialPosition(participant.participantCurrencyId, limitAndInitialPosition)
+    return ParticipantFacade.addLimitAndInitialPosition(participant.participantCurrencyId, settlementAccount.participantCurrencyId, limitAndInitialPosition)
   } catch (err) {
     throw err
   }
@@ -468,27 +472,27 @@ const getPositions = async (name, query) => {
     if (query.currency) {
       const participant = await ParticipantFacade.getByNameAndCurrency(name, query.currency, Enum.LedgerAccountType.POSITION)
       participantExists(participant)
-      const result = await PositionFacade.getByNameAndCurrency(name, query.currency, Enum.LedgerAccountType.POSITION)
+      const result = await PositionFacade.getByNameAndCurrency(name, Enum.LedgerAccountType.POSITION, query.currency, Enum.LedgerAccountType.POSITION)
       let position = {}
       if (Array.isArray(result) && result.length > 0) {
         position = {
           currency: result[0].currencyId,
           value: result[0].value,
-          updatedTime: result[0].changedDate
+          changedDate: result[0].changedDate
         }
       }
       return position
     } else {
       const participant = await ParticipantModel.getByName(name)
       participantExists(participant)
-      const result = await await PositionFacade.getByNameAndCurrency(name, null, Enum.LedgerAccountType.POSITION)
+      const result = await await PositionFacade.getByNameAndCurrency(name, Enum.LedgerAccountType.POSITION)
       let positions = []
       if (Array.isArray(result) && result.length > 0) {
         result.forEach(item => {
           positions.push({
             currency: item.currencyId,
             value: item.value,
-            updatedTime: item.changedDate
+            changedDate: item.changedDate
           })
         })
       }
@@ -499,6 +503,30 @@ const getPositions = async (name, query) => {
   }
 }
 
+const getAccounts = async (name, query) => {
+  try {
+    const participant = await ParticipantModel.getByName(name)
+    participantExists(participant)
+    const result = await PositionFacade.getAllByNameAndCurrency(name, query.currency)
+    let positions = []
+    if (Array.isArray(result) && result.length > 0) {
+      result.forEach(item => {
+        positions.push({
+          id: item.participantCurrencyId,
+          ledgerAccountType: item.ledgerAccountType,
+          currency: item.currencyId,
+          value: item.value,
+          reservedValue: item.reservedValue,
+          changedDate: item.changedDate
+        })
+      })
+    }
+    return positions
+  } catch (err) {
+    throw err
+  }
+}
+
 module.exports = {
   create,
   getAll,
@@ -520,5 +548,6 @@ module.exports = {
   destroyParticipantLimitByNameAndCurrency,
   getLimits,
   adjustLimits,
-  getPositions
+  getPositions,
+  getAccounts
 }

src/handlers/positions/handler.js

@@ -78,20 +78,20 @@ const positions = async (error, messages) => {
       message = Object.assign({}, messages)
     }
     Logger.info('PositionHandler::positions')
-    let consumer = {}
+    let consumer
     let kafkaTopic
     const payload = message.value.content && message.value.content.payload || {}
     payload.transferId = message.value.id
+    kafkaTopic = message.topic
+    try {
+      consumer = Kafka.Consumer.getConsumer(kafkaTopic)
+    } catch (e) {
+      Logger.info(`No consumer found for topic ${kafkaTopic}`)
+      Logger.error(e)
+      return true
+    }
     if (message.value.metadata.event.type === TransferEventType.POSITION && message.value.metadata.event.action === TransferEventAction.PREPARE) {
       Logger.info('PositionHandler::positions::prepare')
-      kafkaTopic = message.topic
-      try {
-        consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-      } catch (e) {
-        Logger.info(`No consumer found for topic ${kafkaTopic}`)
-        Logger.error(e)
-        return true
-      }
       const {preparedMessagesList, limitAlarms} = await PositionService.calculatePreparePositionsBatch(prepareBatch)
       for (let prepareMessage of preparedMessagesList) {
         const {transferState, rawMessage} = prepareMessage
@@ -111,12 +111,6 @@ const positions = async (error, messages) => {
       return true
     } else if (message.value.metadata.event.type === TransferEventType.POSITION && message.value.metadata.event.action === TransferEventAction.COMMIT) {
       Logger.info('PositionHandler::positions::commit')
-      kafkaTopic = message.topic
-      consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-      if (!consumer) {
-        Logger.info(`No consumer found for topic ${kafkaTopic}`)
-        return true
-      }
       // Check current transfer state
       const transferInfo = await TransferService.getTransferInfoToChangePosition(payload.transferId, Enum.TransferParticipantRoleType.PAYEE_DFSP, Enum.LedgerEntryType.PRINCIPLE_VALUE)
       if (transferInfo.transferStateId !== TransferState.RECEIVED_FULFIL) {
@@ -131,22 +125,10 @@ const positions = async (error, messages) => {
         }
         await PositionService.changeParticipantPosition(transferInfo.participantCurrencyId, isReversal, transferInfo.amount, transferStateChange)
       }
-      if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
-        await consumer.commitMessageSync(message)
-      }
       // Will follow framework flow in future
       await Utility.produceGeneralMessage(TransferEventType.NOTIFICATION, TransferEventAction.COMMIT, message.value, Utility.ENUMS.STATE.SUCCESS)
-      return true
     } else if (message.value.metadata.event.type === TransferEventType.POSITION && message.value.metadata.event.action === TransferEventAction.REJECT) {
       Logger.info('PositionHandler::positions::reject')
-      kafkaTopic = message.topic
-      try {
-        consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-      } catch (e) {
-        Logger.info(`No consumer found for topic ${kafkaTopic}`)
-        Logger.error(e)
-        return true
-      }
       const transferInfo = await TransferService.getTransferInfoToChangePosition(payload.transferId, Enum.TransferParticipantRoleType.PAYER_DFSP, Enum.LedgerEntryType.PRINCIPLE_VALUE)
       if (transferInfo.transferStateId !== TransferState.REJECTED) {
         Logger.info('PositionHandler::positions::reject::validationFailed::notRejectedState')
@@ -161,18 +143,13 @@ const positions = async (error, messages) => {
         }
         await PositionService.changeParticipantPosition(transferInfo.participantCurrencyId, isReversal, transferInfo.amount, transferStateChange)
       }
-      if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
-        await consumer.commitMessageSync(message)
-      }
       // Will follow framework flow in future
       await Utility.produceGeneralMessage(TransferEventType.NOTIFICATION, TransferEventAction.REJECT, message.value, Utility.ENUMS.STATE.SUCCESS)
-      return true
     } else if (message.value.metadata.event.type === TransferEventType.POSITION && message.value.metadata.event.action === TransferEventAction.TIMEOUT_RESERVED) {
       Logger.info('PositionHandler::positions::timeout')
       const transferInfo = await TransferService.getTransferInfoToChangePosition(payload.transferId, Enum.TransferParticipantRoleType.PAYER_DFSP, Enum.LedgerEntryType.PRINCIPLE_VALUE)
       if (transferInfo.transferStateId !== TransferState.RESERVED_TIMEOUT) {
         Logger.info('PositionHandler::positions::commit::validationFailed::notReceivedFulfilState')
-        // throw Error 2001
         throw new Error('Internal server error')
       } else { // transfer state check success
         const isReversal = true
@@ -185,44 +162,15 @@ const positions = async (error, messages) => {
         await PositionService.changeParticipantPosition(transferInfo.participantCurrencyId, isReversal, transferInfo.amount, transferStateChange)
         let newMessage = Object.assign({}, message)
         newMessage.value.content.payload = Utility.createPrepareErrorStatus(3303, reason, newMessage.value.content.payload.extensionList)
-        kafkaTopic = message.topic
-        try {
-          consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-        } catch (e) {
-          Logger.info(`No consumer found for topic ${kafkaTopic}`)
-          Logger.error(e)
-          return true
-        }
-        if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
-          await consumer.commitMessageSync(message)
-        }
         await Utility.produceGeneralMessage(TransferEventType.NOTIFICATION, TransferEventAction.ABORT, newMessage.value, Utility.createState(Utility.ENUMS.STATE.FAILURE.status, 4001, transferStateChange.reason))
-        return true
       }
-      // TODO: Need to understand the purpose of this branch.
-      // } else if (message.value.metadata.event.type === TransferEventType.POSITION && message.value.metadata.event.action === TransferEventAction.FAIL) {
-      //   Logger.info('PositionHandler::positions::fail')
-      //   kafkaTopic = Utility.transformAccountToTopicName(message.value.from, TransferEventType.POSITION, TransferEventAction.ABORT)
-      //   consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-      //   if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
-      //     await consumer.commitMessageSync(message)
-      //   }
-      //   throw new Error('Position Fail messaged received - What do we do here??')
     } else {
       Logger.info('PositionHandler::positions::invalidEventTypeOrAction')
-      kafkaTopic = message.topic
-      try {
-        consumer = Kafka.Consumer.getConsumer(kafkaTopic)
-      } catch (e) {
-        Logger.info(`No consumer found for topic ${kafkaTopic}`)
-        Logger.error(e)
-        return true
-      }
-      if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
-        await consumer.commitMessageSync(message)
-      }
-      throw new Error('Event type or action is invalid')
     }
+    if (!Kafka.Consumer.isConsumerAutoCommitEnabled(kafkaTopic)) {
+      await consumer.commitMessageSync(message)
+    }
+    return true
   } catch (error) {
     Logger.error(error)
     throw error

src/handlers/timeouts/handler.js

@@ -129,7 +129,7 @@ const isRunning = async () => {
  */
 const stop = async () => {
   if (isRegistered) {
-    await timeoutJob.destroy()
+    await timeoutJob.stop()
     isRegistered = undefined
   }
 }
@@ -183,7 +183,9 @@ const registerAllHandlers = async () => {
 }
 
 module.exports = {
+  timeout,
   registerAllHandlers,
   registerTimeoutHandler,
-  isRunning
+  isRunning,
+  stop
 }