Pass sequence and primary term during update to ensure that we are updating sequentially

Author: callmekatootie

src/common/helper.js

@@ -66,6 +66,16 @@ let esClient = {
     }
   },
 
+  async get () {
+    const release = await this.mutex.acquire()
+
+    try {
+      return this.client.get(...arguments)
+    } finally {
+      release()
+    }
+  },
+
   indices: undefined
 }
 
@@ -127,26 +137,35 @@ function validProperties (payload, keys) {
 /**
  * Function to get user from es
  * @param {String} userId
+ * @param {Boolean} sourceOnly
  * @returns {Object} user
  */
-async function getUser (userId) {
+async function getUser (userId, sourceOnly = true) {
   const client = await getESClient()
-  return client.getSource({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
+
+  if (sourceOnly) {
+    return client.getSource({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
+  }
+
+  return client.get({ index: config.get('ES.USER_INDEX'), type: config.get('ES.USER_TYPE'), id: userId })
 }
 
 /**
  * Function to update es user
  * @param {String} userId
+ * @param {Number} seqNo
+ * @param {Number} primaryTerm
  * @param {Object} body
  */
-async function updateUser (userId, body) {
+async function updateUser (userId, body, seqNo, primaryTerm) {
   const client = await getESClient()
   await client.update({
     index: config.get('ES.USER_INDEX'),
     type: config.get('ES.USER_TYPE'),
     id: userId,
     body: { doc: body },
-    refresh: 'true'
+    if_seq_no: seqNo,
+    if_primary_term: primaryTerm
   })
 }
 

src/services/ProcessorService.js

@@ -112,7 +112,10 @@ async function processUpdate (message) {
     logger.info(`Processing user level resource: ${resource}:${relateId}`)
     userResource.validate(message.payload)
     logger.info(`Resource validated for ${relateId}`)
-    const user = await helper.getUser(message.payload.userId)
+    let user = await helper.getUser(message.payload.userId, false)
+    const seqNo = user._seq_no
+    const primaryTerm = user._primary_term
+    user = user._source
     logger.info(`User fetched ${user.id} and ${relateId}`)
     // const relateId = message.payload[userResource.relateKey]
 
@@ -124,7 +127,7 @@ async function processUpdate (message) {
       const updateIndex = _.findIndex(user[userResource.propertyName], [userResource.relateKey, relateId])
       user[userResource.propertyName].splice(updateIndex, 1, _.omit(message.payload, 'resource'))
       logger.info(`Updating ${user.id} and ${relateId}`)
-      await helper.updateUser(message.payload.userId, user)
+      await helper.updateUser(message.payload.userId, user, seqNo, primaryTerm)
       logger.info(`Updated ${user.id} and ${relateId}`)
     }
   } else if (_.includes(_.keys(organizationResources), resource)) {