Read groupIds from ES

Author: ThomasKranitsas

npm-shrinkwrap.json

@@ -23,8 +23,10 @@
   "dependencies": {
     "@hapi/joi": "^15.0.2",
     "async-mutex": "^0.2.4",
+    "aws-sdk": "^2.694.0",
     "bluebird": "^3.7.2",
     "config": "^3.3.1",
+    "elasticsearch": "^16.7.1",
     "get-parameter-names": "^0.3.0",
     "gulp": "^4.0.2",
     "http-json-response": "^1.0.1",

package.json

@@ -2,6 +2,8 @@
  * Contains generic helper methods
  */
 
+const elasticsearch = require('elasticsearch')
+const AWS = require('aws-sdk')
 const _ = require('lodash')
 const config = require('config')
 const ifxnjs = require('ifxnjs')
@@ -15,9 +17,41 @@ const Pool = ifxnjs.Pool
 const pool = Promise.promisifyAll(new Pool())
 pool.setMaxPoolSize(config.get('INFORMIX.POOL_MAX_SIZE'))
 
+AWS.config.region = config.get('V4_ES.AWS_REGION')
+// ES Client
+let esClient
+
 // Bus API Client
 let busApiClient
 
+/**
+ * Get ES Client
+ * @return {Object} Elastic Host Client Instance
+ */
+function getESClient () {
+  const esHost = config.get('V4_ES.HOST')
+  if (!esClient) {
+    // AWS ES configuration is different from other providers
+    if (/.*amazonaws.*/.test(esHost)) {
+      esClient = elasticsearch.Client({
+        apiVersion: config.get('V4_ES.API_VERSION'),
+        hosts: esHost,
+        connectionClass: require('http-aws-es'), // eslint-disable-line global-require
+        amazonES: {
+          region: config.get('V4_ES.AWS_REGION'),
+          credentials: new AWS.EnvironmentCredentials('AWS')
+        }
+      })
+    } else {
+      esClient = new elasticsearch.Client({
+        apiVersion: config.get('V4_ES.API_VERSION'),
+        hosts: esHost
+      })
+    }
+  }
+  return esClient
+}
+
 /**
  * Get Informix connection using the configured parameters
  * @return {Object} Informix connection
@@ -158,5 +192,6 @@ module.exports = {
   getRequest,
   putRequest,
   postRequest,
-  postBusEvent
+  postBusEvent,
+  getESClient
 }

src/common/helper.js

@@ -120,7 +120,7 @@ async function getLegacyTrackInformation (trackId, typeId, tags, m2mToken) {
  * @param {Object} payload the Kafka message payload
  * @param {String} m2mToken the m2m token
  * @param {Boolean} isCreated flag indicate the DTO is used in creating challenge
- * @param {Object} existingV4Challenge the existing V4 challenge
+ * @param {Object} existingV4Challenge the existing V4 challenge from ES
  * @returns the DTO for saving a draft contest.(refer SaveDraftContestDTO in ap-challenge-microservice)
  */
 async function parsePayload (payload, m2mToken, isCreated = true, existingV4Challenge) {
@@ -374,6 +374,9 @@ async function processUpdate (message) {
   try {
     // ensure challenge existed
     challenge = await getChallengeById(m2mToken, message.payload.legacyId)
+    if (!challenge) {
+      throw new Error(`Could not find challenge ${message.payload.legacyId}`)
+    }
   } catch (e) {
     // postponne kafka event
     logger.info('Challenge does not exist yet. Will post the same message back to the bus API')
@@ -386,13 +389,46 @@ async function processUpdate (message) {
     return
   }
 
-  const saveDraftContestDTO = await parsePayload(message.payload, m2mToken, false, challenge)
-  logger.debug('Parsed Payload', saveDraftContestDTO)
+  let challengeV4FromEs
   try {
-    if (!challenge) {
-      throw new Error(`Could not find challenge ${message.payload.legacyId}`)
+    // Search with constructed query
+    const esQuery = {
+      index: config.get('ES.ES_INDEX'),
+      type: config.get('ES.ES_TYPE'),
+      size: 1,
+      from: 0,
+      body: {
+        query: {
+          match_phrase: {
+            _id: message.payload.legacyId
+          }
+        }
+      }
     }
+    const docs = await helper.getESClient().search(esQuery)
+    // Extract data from hits
+    if (docs.hits.total === 0) {
+      throw new Error('Challenge does not exist yet on ES')
+    }
+    challengeV4FromEs = _.map(docs.hits.hits, item => item._source)[0]
+    if (!challengeV4FromEs) {
+      throw new Error(`Could not find challenge ${message.payload.legacyId} on ES`)
+    }
+  } catch (e) {
+    // postponne kafka event
+    logger.info('Challenge does not exist yet on ES. Will post the same message back to the bus API')
+    await new Promise((resolve) => {
+      setTimeout(async () => {
+        await helper.postBusEvent(config.UPDATE_CHALLENGE_TOPIC, message.payload)
+        resolve()
+      }, config.RETRY_TIMEOUT)
+    })
+    return
+  }
 
+  const saveDraftContestDTO = await parsePayload(message.payload, m2mToken, false, challengeV4FromEs)
+  logger.debug('Parsed Payload', saveDraftContestDTO)
+  try {
     await helper.putRequest(`${config.V4_CHALLENGE_API_URL}/${message.payload.legacyId}`, { param: _.omit(saveDraftContestDTO, ['groupsToBeAdded', 'groupsToBeDeleted']) }, m2mToken)
     await associateChallengeGroups(saveDraftContestDTO.groupsToBeAdded, saveDraftContestDTO.groupsToBeDeleted, message.payload.legacyId)