enhancement: encapsulate old value to the event payload as well

Author: imcaizheng

src/common/helper.js

@@ -282,8 +282,9 @@ async function getUserId (userId) {
  * Send Kafka event message
  * @params {String} topic the topic name
  * @params {Object} payload the payload
+ * @params {Object} options the extra options to control the function
  */
-async function postEvent (topic, payload) {
+async function postEvent (topic, payload, options = {}) {
   logger.debug({ component: 'helper', context: 'postEvent', message: `Posting event to Kafka topic ${topic}, ${JSON.stringify(payload)}` })
   const client = getBusApiClient()
   const message = {
@@ -294,7 +295,7 @@ async function postEvent (topic, payload) {
     payload
   }
   await client.postEvent(message)
-  await eventDispatcher.handleEvent(topic, payload)
+  await eventDispatcher.handleEvent(topic, { value: payload, options })
 }
 
 /**

src/eventHandlers/JobEventHandler.js

@@ -16,17 +16,25 @@ const ResourceBookingService = require('../services/ResourceBookingService')
  * @returns {undefined}
  */
 async function cancelJob (payload) {
-  if (payload.status !== 'cancelled') {
+  if (payload.value.status === payload.options.oldValue.status) {
+    logger.debug({
+      component: 'JobEventHandler',
+      context: 'cancelJob',
+      message: 'status not changed'
+    })
+    return
+  }
+  if (payload.value.status !== 'cancelled') {
     logger.info({
       component: 'JobEventHandler',
       context: 'cancelJob',
-      message: `not interested job - status: ${payload.status}`
+      message: `not interested job - status: ${payload.value.status}`
     })
     return
   }
   // pull data from db instead of directly extract data from the payload
   // since the payload may not contain all fields when it is from partically update operation.
-  const job = await models.Job.findById(payload.id)
+  const job = await models.Job.findById(payload.value.id)
   const candidates = await models.JobCandidate.findAll({
     where: {
       jobId: job.id,

src/eventHandlers/ResourceBookingEventHandler.js

@@ -81,15 +81,23 @@ async function assignJob (job) {
  * @returns {undefined}
  */
 async function processUpdate (payload) {
-  if (payload.status !== 'assigned') {
+  if (payload.value.status === payload.options.oldValue.status) {
+    logger.debug({
+      component: 'ResourceBookingEventHandler',
+      context: 'processUpdate',
+      message: 'status not changed'
+    })
+    return
+  }
+  if (payload.value.status !== 'assigned') {
     logger.info({
       component: 'ResourceBookingEventHandler',
       context: 'processUpdate',
-      message: `not interested resource booking - status: ${payload.status}`
+      message: `not interested resource booking - status: ${payload.value.status}`
     })
     return
   }
-  const resourceBooking = await models.ResourceBooking.findById(payload.id)
+  const resourceBooking = await models.ResourceBooking.findById(payload.value.id)
   const jobs = await models.Job.findAll({
     where: {
       projectId: resourceBooking.projectId,

src/services/JobService.js

@@ -192,6 +192,7 @@ async function updateJob (currentUser, id, data) {
     await _validateSkills(data.skills)
   }
   let job = await Job.findById(id)
+  const oldValue = job.toJSON()
   const ubhanUserId = await helper.getUserId(currentUser.userId)
   if (!currentUser.hasManagePermission && !currentUser.isMachine) {
     if (currentUser.isConnectManager) {
@@ -209,7 +210,7 @@ async function updateJob (currentUser, id, data) {
   data.updatedBy = ubhanUserId
 
   await job.update(data)
-  await helper.postEvent(config.TAAS_JOB_UPDATE_TOPIC, { id, ...data })
+  await helper.postEvent(config.TAAS_JOB_UPDATE_TOPIC, { id, ...data }, { oldValue: oldValue })
   job = await Job.findById(id, true)
   job.dataValues.candidates = _.map(job.dataValues.candidates, (c) => helper.clearObject(c.dataValues))
   return helper.clearObject(job.dataValues)

src/services/ResourceBookingService.js

@@ -142,14 +142,13 @@ async function updateResourceBooking (currentUser, id, data) {
   }
 
   const resourceBooking = await ResourceBooking.findById(id)
-  const isDiffStatus = resourceBooking.status !== data.status
+  const oldValue = resourceBooking.toJSON()
 
   data.updatedAt = new Date()
   data.updatedBy = await helper.getUserId(currentUser.userId)
 
   await resourceBooking.update(data)
-  const eventPayload = { id, ...(isDiffStatus ? data : _.omit(data, ['status'])) } // send data with status only if status is changed
-  await helper.postEvent(config.TAAS_RESOURCE_BOOKING_UPDATE_TOPIC, eventPayload)
+  await helper.postEvent(config.TAAS_RESOURCE_BOOKING_UPDATE_TOPIC, { id, ...data }, { oldValue: oldValue })
   const result = helper.clearObject(_.assign(resourceBooking.dataValues, data))
   return result
 }