feat(csi-22): add proxy lib to handlers (#1060)

* feat(csi-22): add proxy lib to handlers * diff * add * int tests * fix hanging int tests * fixes? * unit fixes? * coverage * feat: refactor proxy cache integration * feat: restore default * feat: minor optimization * test: update coverage * test: remove try-catch * fix: fix disconnect error * feat: proxy cache update (#1061) * addressed comments --------- Co-authored-by: Steven Oderayi <[email protected]>
mojaloop · Jul 18, 2024 · eaa0ce0 · eaa0ce0
1 parent fef1a57
commit eaa0ce0
Show file tree

Hide file tree

Showing 52 changed files with 757 additions and 59 deletions.
diff --git a/config/default.json b/config/default.json
@@ -84,6 +84,14 @@
     "MAX_BYTE_SIZE": 10000000,
     "EXPIRES_IN_MS": 1000
   },
+  "PROXY_CACHE": {
+    "enabled": true,
+    "type": "redis",
+    "proxyConfig": {
+      "host": "localhost",
+      "port": 6379
+    }
+  },
   "API_DOC_ENDPOINTS_ENABLED": true,
   "KAFKA": {
     "EVENT_TYPE_ACTION_TOPIC_MAP" : {

diff --git a/docker-compose.yml b/docker-compose.yml
@@ -219,3 +219,16 @@ services:
       - cl-mojaloop-net
     environment:
       - KAFKA_BROKERS=kafka:29092
+
+  redis:
+    image: redis:6.2.4-alpine
+    restart: "unless-stopped"
+    environment:
+      - ALLOW_EMPTY_PASSWORD=yes
+      - REDIS_PORT=6379
+      - REDIS_REPLICATION_MODE=master
+      - REDIS_TLS_ENABLED=no
+    ports:
+      - "6379:6379"
+    networks:
+      - cl-mojaloop-net
diff --git a/docker/central-ledger/default.json b/docker/central-ledger/default.json
@@ -82,6 +82,14 @@
     "MAX_BYTE_SIZE": 10000000,
     "EXPIRES_IN_MS": 1000
   },
+  "PROXY_CACHE": {
+    "enabled": true,
+    "type": "redis",
+    "proxyConfig": {
+      "host": "localhost",
+      "port": 6379
+    }
+  },
   "KAFKA": {
     "TOPIC_TEMPLATES": {
       "PARTICIPANT_TOPIC_TEMPLATE": {

diff --git a/docker/config-modifier/configs/central-ledger.js b/docker/config-modifier/configs/central-ledger.js
@@ -12,6 +12,14 @@ module.exports = {
     PASSWORD: '',
     DATABASE: 'mlos'
   },
+  PROXY_CACHE: {
+    enabled: true,
+    type: 'redis',
+    proxyConfig: {
+      host: 'redis',
+      port: 6379
+    }
+  },
   KAFKA: {
     EVENT_TYPE_ACTION_TOPIC_MAP: {
       POSITION: {

diff --git a/package-lock.json b/package-lock.json
diff --git a/package.json b/package.json
@@ -81,13 +81,13 @@
     "wait-4-docker": "node ./scripts/_wait4_all.js"
   },
   "dependencies": {
+    "@hapi/basic": "7.0.2",
+    "@hapi/catbox-memory": "6.0.2",
     "@hapi/good": "9.0.1",
     "@hapi/hapi": "21.3.10",
-    "@hapi/basic": "7.0.2",
     "@hapi/inert": "7.1.0",
     "@hapi/joi": "17.1.1",
     "@hapi/vision": "7.0.3",
-    "@hapi/catbox-memory": "6.0.2",
     "@mojaloop/central-services-error-handling": "13.0.1",
     "@mojaloop/central-services-health": "15.0.0",
     "@mojaloop/central-services-logger": "11.3.1",
@@ -96,10 +96,11 @@
     "@mojaloop/central-services-stream": "11.3.1",
     "@mojaloop/database-lib": "11.0.6",
     "@mojaloop/event-sdk": "14.1.1",
+    "@mojaloop/inter-scheme-proxy-cache-lib": "^1.4.0",
     "@mojaloop/ml-number": "11.2.4",
     "@mojaloop/object-store-lib": "12.0.3",
     "@now-ims/hapi-now-auth": "2.1.0",
-    "ajv": "8.16.0",
+    "ajv": "8.17.1",
     "ajv-keywords": "5.1.0",
     "base64url": "3.0.1",
     "blipp": "4.0.2",

diff --git a/src/api/root/handler.js b/src/api/root/handler.js
@@ -30,13 +30,23 @@ const { defaultHealthHandler } = require('@mojaloop/central-services-health')
 const packageJson = require('../../../package.json')
 const {
   getSubServiceHealthDatastore,
-  getSubServiceHealthBroker
+  getSubServiceHealthBroker,
+  getSubServiceHealthProxyCache
 } = require('../../lib/healthCheck/subServiceHealth')
+const Config = require('../../lib/config')
 
-const healthCheck = new HealthCheck(packageJson, [
-  getSubServiceHealthDatastore,
-  getSubServiceHealthBroker
-])
+const subServiceChecks = Config.PROXY_CACHE_CONFIG.enabled
+  ? [
+      getSubServiceHealthDatastore,
+      getSubServiceHealthBroker,
+      getSubServiceHealthProxyCache
+    ]
+  : [
+      getSubServiceHealthDatastore,
+      getSubServiceHealthBroker
+    ]
+
+const healthCheck = new HealthCheck(packageJson, subServiceChecks)
 
 /**
  * @function getHealth

diff --git a/src/lib/config.js b/src/lib/config.js
@@ -23,6 +23,7 @@ module.exports = {
   HANDLERS_TIMEOUT_TIMEXP: RC.HANDLERS.TIMEOUT.TIMEXP,
   HANDLERS_TIMEOUT_TIMEZONE: RC.HANDLERS.TIMEOUT.TIMEZONE,
   CACHE_CONFIG: RC.CACHE,
+  PROXY_CACHE_CONFIG: RC.PROXY_CACHE,
   KAFKA_CONFIG: RC.KAFKA,
   PARTICIPANT_INITIAL_POSITION: RC.PARTICIPANT_INITIAL_POSITION,
   RUN_MIGRATIONS: !RC.MIGRATIONS.DISABLED,

diff --git a/src/lib/healthCheck/subServiceHealth.js b/src/lib/healthCheck/subServiceHealth.js
@@ -26,7 +26,7 @@
 const { statusEnum, serviceName } = require('@mojaloop/central-services-shared').HealthCheck.HealthCheckEnums
 const Logger = require('@mojaloop/central-services-logger')
 const Consumer = require('@mojaloop/central-services-stream').Util.Consumer
-
+const ProxyCache = require('../proxyCache')
 const MigrationLockModel = require('../../models/misc/migrationLock')
 
 /**
@@ -82,7 +82,17 @@ const getSubServiceHealthDatastore = async () => {
   }
 }
 
+const getSubServiceHealthProxyCache = async () => {
+  const proxyCache = ProxyCache.getCache()
+  const healthCheck = await proxyCache.healthCheck()
+  return {
+    name: 'proxyCache',
+    status: healthCheck ? statusEnum.OK : statusEnum.DOWN
+  }
+}
+
 module.exports = {
   getSubServiceHealthBroker,
-  getSubServiceHealthDatastore
+  getSubServiceHealthDatastore,
+  getSubServiceHealthProxyCache
 }
diff --git a/src/lib/proxyCache.js b/src/lib/proxyCache.js
@@ -0,0 +1,48 @@
+'use strict'
+const { createProxyCache } = require('@mojaloop/inter-scheme-proxy-cache-lib')
+const Config = require('./config.js')
+const ParticipantService = require('../../src/domain/participant')
+
+let proxyCache
+
+const connect = async () => {
+  return getCache().connect()
+}
+
+const disconnect = async () => {
+  return proxyCache?.isConnected && proxyCache.disconnect()
+}
+
+const getCache = () => {
+  if (!proxyCache) {
+    proxyCache = Object.freeze(createProxyCache(
+      Config.PROXY_CACHE_CONFIG.type,
+      Config.PROXY_CACHE_CONFIG.proxyConfig
+    ))
+  }
+  return proxyCache
+}
+
+const getFSPProxy = async (dfspId) => {
+  const participant = await ParticipantService.getByName(dfspId)
+  return {
+    inScheme: !!participant,
+    proxyId: !participant ? await getCache().lookupProxyByDfspId(dfspId) : null
+  }
+}
+
+const checkSameCreditorDebtorProxy = async (debtorDfspId, creditorDfspId) => {
+  const [debtorProxyId, creditorProxyId] = await Promise.all([
+    await getCache().lookupProxyByDfspId(debtorDfspId),
+    await getCache().lookupProxyByDfspId(creditorDfspId)
+  ])
+  return debtorProxyId && creditorProxyId ? debtorProxyId === creditorProxyId : false
+}
+
+module.exports = {
+  connect,
+  disconnect,
+  getCache,
+  getFSPProxy,
+  checkSameCreditorDebtorProxy
+}
diff --git a/src/shared/setup.js b/src/shared/setup.js
@@ -36,6 +36,7 @@
 const Hapi = require('@hapi/hapi')
 const Migrator = require('../lib/migrator')
 const Db = require('../lib/db')
+const ProxyCache = require('../lib/proxyCache')
 const ObjStoreDb = require('@mojaloop/object-store-lib').Db
 const Plugins = require('./plugins')
 const Config = require('../lib/config')
@@ -265,6 +266,9 @@ const initialize = async function ({ service, port, modules = [], runMigrations
     await connectDatabase()
     await connectMongoose()
     await initializeCache()
+    if (Config.PROXY_CACHE_CONFIG.enabled) {
+      await ProxyCache.connect()
+    }
 
     let server
     switch (service) {
@@ -303,6 +307,9 @@ const initialize = async function ({ service, port, modules = [], runMigrations
     Logger.isErrorEnabled && Logger.error(`Error while initializing ${err}`)
 
     await Db.disconnect()
+    if (Config.PROXY_CACHE_CONFIG.enabled) {
+      await ProxyCache.disconnect()
+    }
     process.exit(1)
   }
 }

diff --git a/test/integration-override/handlers/positions/handlerBatch.test.js b/test/integration-override/handlers/positions/handlerBatch.test.js
@@ -28,6 +28,7 @@ const Test = require('tape')
 const { randomUUID } = require('crypto')
 const Logger = require('@mojaloop/central-services-logger')
 const Config = require('#src/lib/config')
+const ProxyCache = require('#src/lib/proxyCache')
 const Db = require('@mojaloop/database-lib').Db
 const Cache = require('#src/lib/cache')
 const Producer = require('@mojaloop/central-services-stream').Util.Producer
@@ -1838,6 +1839,7 @@ Test('Handlers test', async handlersTest => {
       await testConsumer.destroy() // this disconnects the consumers
 
       await Producer.disconnect()
+      await ProxyCache.disconnect()
 
       if (debug) {
         const elapsedTime = Math.round(((new Date()) - startTime) / 100) / 10

diff --git a/test/integration-override/handlers/transfers/fxFulfil.test.js b/test/integration-override/handlers/transfers/fxFulfil.test.js
@@ -30,6 +30,7 @@ const { Producer } = require('@mojaloop/central-services-stream').Kafka
 
 const Config = require('#src/lib/config')
 const Cache = require('#src/lib/cache')
+const ProxyCache = require('#src/lib/proxyCache')
 const fspiopErrorFactory = require('#src/shared/fspiopErrorFactory')
 const ParticipantCached = require('#src/models/participant/participantCached')
 const ParticipantCurrencyCached = require('#src/models/participant/participantCurrencyCached')
@@ -275,6 +276,7 @@ Test('FxFulfil flow Integration Tests -->', async fxFulfilTest => {
       producer.disconnect(),
       testConsumer.destroy()
     ])
+    await ProxyCache.disconnect()
     await new Promise(resolve => setTimeout(resolve, 5_000))
     t.pass('teardown is finished')
     t.end()

diff --git a/test/integration-override/handlers/transfers/fxTimeout.test.js b/test/integration-override/handlers/transfers/fxTimeout.test.js
@@ -30,6 +30,7 @@ const Logger = require('@mojaloop/central-services-logger')
 const Config = require('#src/lib/config')
 const Db = require('@mojaloop/database-lib').Db
 const Cache = require('#src/lib/cache')
+const ProxyCache = require('#src/lib/proxyCache')
 const Producer = require('@mojaloop/central-services-stream').Util.Producer
 const Utility = require('@mojaloop/central-services-shared').Util.Kafka
 const Enum = require('@mojaloop/central-services-shared').Enum
@@ -773,6 +774,7 @@ Test('Handlers test', async handlersTest => {
       await testConsumer.destroy() // this disconnects the consumers
 
       await Producer.disconnect()
+      await ProxyCache.disconnect()
 
       if (debug) {
         const elapsedTime = Math.round(((new Date()) - startTime) / 100) / 10

diff --git a/test/integration-override/handlers/transfers/handlers.test.js b/test/integration-override/handlers/transfers/handlers.test.js
@@ -30,6 +30,7 @@ const Logger = require('@mojaloop/central-services-logger')
 const Config = require('#src/lib/config')
 const Db = require('@mojaloop/database-lib').Db
 const Cache = require('#src/lib/cache')
+const ProxyCache = require('#src/lib/proxyCache')
 const Producer = require('@mojaloop/central-services-stream').Util.Producer
 const Utility = require('@mojaloop/central-services-shared').Util.Kafka
 const Enum = require('@mojaloop/central-services-shared').Enum
@@ -753,6 +754,7 @@ Test('Handlers test', async handlersTest => {
       await testConsumer.destroy() // this disconnects the consumers
 
       await Producer.disconnect()
+      await ProxyCache.disconnect()
 
       if (debug) {
         const elapsedTime = Math.round(((new Date()) - startTime) / 100) / 10