feat(mojaloop/#3427): add oracle endpoint db caching and oracle request caching

config/default.json

@@ -36,16 +36,21 @@
   },
   "DISPLAY_ROUTES": true,
   "RUN_MIGRATIONS": true,
-  "ENDPOINT_CACHE_CONFIG": {
+  "CENTRAL_SHARED_ENDPOINT_CACHE_CONFIG": {
     "expiresIn": 180000,
     "generateTimeout": 30000,
     "getDecoratedValue": true
   },
-  "PARTICIPANT_CACHE_CONFIG": {
+  "CENTRAL_SHARED_PARTICIPANT_CACHE_CONFIG": {
     "expiresIn": 61000,
     "generateTimeout": 30000,
     "getDecoratedValue": true
   },
+  "ALS_GENERAL_CACHE_CONFIG": {
+    "CACHE_ENABLED": false,
+    "MAX_BYTE_SIZE": 10000000,
+    "EXPIRES_IN_MS": 61000
+  },
   "ERROR_HANDLING": {
     "includeCauseExtension": false,
     "truncateExtensions": true

docker/account-lookup-service/default.json

@@ -20,7 +20,7 @@
   },
   "DISPLAY_ROUTES": true,
   "RUN_MIGRATIONS": true,
-  "ENDPOINT_CACHE_CONFIG": {
+  "CENTRAL_SHARED_ENDPOINT_CACHE_CONFIG": {
     "expiresIn": 180000,
     "generateTimeout": 30000
   },

package.json

@@ -1,7 +1,7 @@
 {
   "name": "account-lookup-service",
   "description": "Account Lookup Service is used to validate Party and Participant lookups.",
-  "version": "15.1.0",
+  "version": "15.2.0-snapshot.0",
   "license": "Apache-2.0",
   "author": "ModusBox",
   "contributors": [
@@ -77,17 +77,18 @@
   "dependencies": {
     "@hapi/basic": "7.0.2",
     "@hapi/boom": "10.0.1",
+    "@hapi/catbox-memory": "^6.0.1",
     "@hapi/good": "9.0.1",
     "@hapi/hapi": "21.3.2",
     "@hapi/inert": "7.1.0",
     "@hapi/joi": "17.1.1",
     "@hapi/vision": "7.0.3",
-    "@mojaloop/database-lib": "11.0.3",
     "@mojaloop/central-services-error-handling": "12.0.7",
     "@mojaloop/central-services-health": "14.0.2",
     "@mojaloop/central-services-logger": "11.2.2",
     "@mojaloop/central-services-metrics": "12.0.8",
     "@mojaloop/central-services-shared": "18.2.0",
+    "@mojaloop/database-lib": "11.0.3",
     "@mojaloop/event-sdk": "14.0.0",
     "@mojaloop/sdk-standard-components": "17.1.3",
     "@now-ims/hapi-now-auth": "2.1.0",

src/domain/oracle/oracle.js

@@ -33,6 +33,7 @@ const currency = require('../../models/currency')
 const ErrorHandler = require('@mojaloop/central-services-error-handling')
 const Config = require('../../lib/config')
 const Metrics = require('@mojaloop/central-services-metrics')
+const cachedOracleEndpoint = require('../../models/oracle/oracleEndpointCached')
 
 /**
  * @function createOracle
@@ -109,11 +110,11 @@ exports.getOracle = async (query) => {
       isType = true
     }
     if (isCurrency && isType) {
-      oracleEndpointModelList = await oracleEndpoint.getOracleEndpointByTypeAndCurrency(query.type, query.currency)
+      oracleEndpointModelList = await cachedOracleEndpoint.getOracleEndpointByTypeAndCurrency(query.type, query.currency)
     } else if (isCurrency && !isType) {
-      oracleEndpointModelList = await oracleEndpoint.getOracleEndpointByCurrency(query.currency)
+      oracleEndpointModelList = await cachedOracleEndpoint.getOracleEndpointByCurrency(query.currency)
     } else if (isType && !isCurrency) {
-      oracleEndpointModelList = await oracleEndpoint.getOracleEndpointByType(query.type)
+      oracleEndpointModelList = await cachedOracleEndpoint.getOracleEndpointByType(query.type)
     } else {
       oracleEndpointModelList = await oracleEndpoint.getAllOracleEndpoint()
     }

src/domain/parties/parties.js

@@ -52,7 +52,7 @@ const Config = require('../../lib/config')
  * @param {object} query - uri query parameters of the http request
  * @param {object} span
  */
-const getPartiesByTypeAndID = async (headers, params, method, query, span = undefined) => {
+const getPartiesByTypeAndID = async (headers, params, method, query, span = undefined, cache) => {
   const histTimerEnd = Metrics.getHistogram(
     'getPartiesByTypeAndID',
     'Get party by Type and Id',
@@ -95,7 +95,7 @@ const getPartiesByTypeAndID = async (headers, params, method, query, span = unde
         return
       }
 
-      const response = await oracle.oracleRequest(headers, method, params, query)
+      const response = await oracle.oracleRequest(headers, method, params, query, cache)
 
       if (response && response.data && Array.isArray(response.data.partyList) && response.data.partyList.length > 0) {
         // Oracle's API is a standard rest-style end-point Thus a GET /party on the oracle will return all participant-party records. We must filter the results based on the callbackEndpointType to make sure we remove records containing partySubIdOrType when we are in FSPIOP_CALLBACK_URL_PARTIES_GET mode:

src/handlers/endpointcache.js

@@ -56,7 +56,7 @@ module.exports = {
     }, EventSdk.AuditEventAction.start)
     try {
       await ParticipantEndpointCache.stopCache()
-      await ParticipantEndpointCache.initializeCache(Config.ENDPOINT_CACHE_CONFIG)
+      await ParticipantEndpointCache.initializeCache(Config.CENTRAL_SHARED_ENDPOINT_CACHE_CONFIG)
       histTimerEnd({ success: true })
     } catch (err) {
       histTimerEnd({ success: false })

src/handlers/participants/{Type}/{ID}.js

@@ -26,9 +26,7 @@
 'use strict'
 
 const Enum = require('@mojaloop/central-services-shared').Enum
-const ErrorHandler = require('@mojaloop/central-services-error-handling')
 const EventSdk = require('@mojaloop/event-sdk')
-const Logger = require('@mojaloop/central-services-logger')
 const Metrics = require('@mojaloop/central-services-metrics')
 const LibUtil = require('../../../lib/util')
 const participants = require('../../../domain/participants')
@@ -58,16 +56,11 @@ module.exports = {
       payload: request.payload
     }, EventSdk.AuditEventAction.start)
     const metadata = `${request.method} ${request.path}`
-    try {
-      participants.getParticipantsByTypeAndID(request.headers, request.params, request.method, request.query, span).catch(err => {
-        request.server.log(['error'], `ERROR - getParticipantsByTypeAndID:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
-      })
-      histTimerEnd({ success: true })
-    } catch (err) {
-      Logger.isErrorEnabled && Logger.error(`ERROR - ${metadata}: ${err.stack}`)
-      histTimerEnd({ success: false })
-      throw ErrorHandler.Factory.reformatFSPIOPError(err)
-    }
+    participants.getParticipantsByTypeAndID(request.headers, request.params, request.method, request.query, span).catch(err => {
+      request.server.log(['error'], `ERROR - getParticipantsByTypeAndID:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
+    })
+    histTimerEnd({ success: true })
+
     return h.response().code(202)
   },
   /**
@@ -84,16 +77,10 @@ module.exports = {
       ['success']
     ).startTimer()
     const metadata = `${request.method} ${request.path}`
-    try {
-      participants.putParticipantsByTypeAndID(request.headers, request.params, request.method, request.payload).catch(err => {
-        request.server.log(['error'], `ERROR - putParticipantsByTypeAndID:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
-      })
-      histTimerEnd({ success: true })
-    } catch (err) {
-      Logger.isErrorEnabled && Logger.error(`ERROR - ${metadata}: ${err.stack}`)
-      histTimerEnd({ success: false })
-      throw ErrorHandler.Factory.reformatFSPIOPError(err)
-    }
+    participants.putParticipantsByTypeAndID(request.headers, request.params, request.method, request.payload).catch(err => {
+      request.server.log(['error'], `ERROR - putParticipantsByTypeAndID:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
+    })
+    histTimerEnd({ success: true })
     return h.response().code(200)
   },
   /**
@@ -117,16 +104,10 @@ module.exports = {
       payload: request.payload
     }, EventSdk.AuditEventAction.start)
     const metadata = `${request.method} ${request.path}`
-    try {
-      participants.postParticipants(request.headers, request.method, request.params, request.payload, span).catch(err => {
-        request.server.log(['error'], `ERROR - postParticipants:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
-      })
-      histTimerEnd({ success: true })
-    } catch (err) {
-      Logger.isErrorEnabled && Logger.error(`ERROR - ${metadata}: ${err.stack}`)
-      histTimerEnd({ success: false })
-      throw ErrorHandler.Factory.reformatFSPIOPError(err)
-    }
+    participants.postParticipants(request.headers, request.method, request.params, request.payload, span).catch(err => {
+      request.server.log(['error'], `ERROR - postParticipants:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
+    })
+    histTimerEnd({ success: true })
     return h.response().code(202)
   },
   /**
@@ -143,16 +124,10 @@ module.exports = {
       ['success']
     ).startTimer()
     const metadata = `${request.method} ${request.path}`
-    try {
-      participants.deleteParticipants(request.headers, request.params, request.method, request.query).catch(err => {
-        request.server.log(['error'], `ERROR - deleteParticipants:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
-      })
-      histTimerEnd({ success: true })
-    } catch (err) {
-      Logger.isErrorEnabled && Logger.error(`ERROR - ${metadata}: ${err.stack}`)
-      histTimerEnd({ success: false })
-      throw ErrorHandler.Factory.reformatFSPIOPError(err)
-    }
+    participants.deleteParticipants(request.headers, request.params, request.method, request.query).catch(err => {
+      request.server.log(['error'], `ERROR - deleteParticipants:${metadata}: ${LibUtil.getStackOrInspect(err)}`)
+    })
+    histTimerEnd({ success: true })
     return h.response().code(202)
   }
 }

src/handlers/parties/{Type}/{ID}.js

@@ -56,7 +56,7 @@ module.exports = {
     }, EventSdk.AuditEventAction.start)
     // Here we call an async function- but as we send an immediate sync response, _all_ errors
     // _must_ be handled by getPartiesByTypeAndID.
-    parties.getPartiesByTypeAndID(request.headers, request.params, request.method, request.query, span).catch(err => {
+    parties.getPartiesByTypeAndID(request.headers, request.params, request.method, request.query, span, request.server.app.cache).catch(err => {
       request.server.log(['error'], `ERROR - getPartiesByTypeAndID: ${LibUtil.getStackOrInspect(err)}`)
     })
     histTimerEnd({ success: true })

src/lib/cache.js

@@ -0,0 +1,101 @@
+'use strict'
+
+const CatboxMemory = require('@hapi/catbox-memory')
+const Config = require('../lib/config')
+
+let enabled = true
+let ttl
+let catboxMemoryClient = null
+
+class CacheClient {
+  constructor (meta) {
+    this.meta = meta
+  }
+
+  getMeta () {
+    return this.meta
+  }
+
+  createKey (id) {
+    return {
+      segment: this.meta.id,
+      id
+    }
+  }
+
+  get (key) {
+    if (enabled) {
+      return catboxMemoryClient.get(key)
+    }
+    return null
+  }
+
+  set (key, value) {
+    catboxMemoryClient.set(key, value, parseInt(ttl))
+  }
+
+  drop (key) {
+    catboxMemoryClient.drop(key)
+  }
+}
+
+/*
+  Each client should register itself during module load.
+  The client meta should be:
+  {
+    id [MANDATORY]
+    preloadCache() [OPTIONAL]
+      this will be called to preload data
+  }
+*/
+let cacheClients = {}
+
+const registerCacheClient = (clientMeta) => {
+  const newClient = new CacheClient(clientMeta)
+  cacheClients[clientMeta.id] = newClient
+  return newClient
+}
+
+const initCache = async function () {
+  // Read config
+  ttl = Config.ALS_GENERAL_CACHE_CONFIG.EXPIRES_IN_MS
+  enabled = Config.ALS_GENERAL_CACHE_CONFIG.CACHE_ENABLED
+
+  // Init catbox.
+  catboxMemoryClient = new CatboxMemory.Engine({
+    maxByteSize: Config.ALS_GENERAL_CACHE_CONFIG.MAX_BYTE_SIZE
+  })
+  catboxMemoryClient.start()
+
+  for (const clientId in cacheClients) {
+    const clientMeta = cacheClients[clientId].getMeta()
+    await clientMeta.preloadCache()
+  }
+}
+
+const destroyCache = async function () {
+  catboxMemoryClient.stop()
+  catboxMemoryClient = null
+}
+
+const dropClients = function () {
+  cacheClients = {}
+}
+
+const isCacheEnabled = function () {
+  return enabled
+}
+
+module.exports = {
+  // Clients registration
+  registerCacheClient,
+
+  // Init & destroy the cache
+  initCache,
+  destroyCache,
+  isCacheEnabled,
+
+  // exposed for tests
+  CatboxMemory,
+  dropClients
+}

src/lib/config.js

@@ -142,8 +142,9 @@ const config = {
   DISPLAY_ROUTES: RC.DISPLAY_ROUTES,
   RUN_MIGRATIONS: RC.RUN_MIGRATIONS,
   ADMIN_PORT: RC.ADMIN_PORT,
-  ENDPOINT_CACHE_CONFIG: RC.ENDPOINT_CACHE_CONFIG,
-  PARTICIPANT_CACHE_CONFIG: RC.PARTICIPANT_CACHE_CONFIG,
+  CENTRAL_SHARED_ENDPOINT_CACHE_CONFIG: RC.CENTRAL_SHARED_ENDPOINT_CACHE_CONFIG,
+  CENTRAL_SHARED_PARTICIPANT_CACHE_CONFIG: RC.CENTRAL_SHARED_PARTICIPANT_CACHE_CONFIG,
+  ALS_GENERAL_CACHE_CONFIG: RC.ALS_GENERAL_CACHE_CONFIG,
   ERROR_HANDLING: RC.ERROR_HANDLING,
   SWITCH_ENDPOINT: RC.SWITCH_ENDPOINT,
   INSTRUMENTATION_METRICS_DISABLED: RC.INSTRUMENTATION.METRICS.DISABLED,