Merge pull request #10 from imcaizheng/update-job-candidate-status-recruit-crm

Update the status of candidates in RecruitCRM

Author: nkumar-topcoder

README.md

@@ -47,8 +47,10 @@ The following parameters can be set in config files or in env variables:
 
 - `zapier.ZAPIER_COMPANYID_SLUG`: your company id in zapier; numeric value
 - `zapier.ZAPIER_CONTACTID_SLUG`: your contact id in zapier; numeric value
-- `zapier.ZAPIER_SWITCH`: decides whether posting message to zapier or not; possible values are `ON` and `OFF`, default is `OFF`
-- `zapier.ZAPIER_WEBHOOK`: the remote zapier zap webhook url for posting message
+- `zapier.ZAPIER_SWITCH`: decides whether posting job related message to zapier or not; possible values are `ON` and `OFF`, default is `OFF`
+- `zapier.ZAPIER_WEBHOOK`: the remote zapier zap webhook url for posting job related message
+- `zapier.ZAPIER_JOB_CANDIDATE_SWITCH`: decides whether posting job candidate related message to zapier or not; possible values are `ON` and `OFF`, default is `OFF`
+- `zapier.ZAPIER_JOB_CANDIDATE_WEBHOOK`: the remote zapier zap webhook url for posting job candidate related message
 
 ## Local Kafka and ElasticSearch setup
 

config/default.js

@@ -58,6 +58,8 @@ module.exports = {
     ZAPIER_CONTACTID_SLUG: process.env.ZAPIER_CONTACTID_SLUG,
     ZAPIER_SWITCH: process.env.ZAPIER_SWITCH || 'OFF',
     ZAPIER_WEBHOOK: process.env.ZAPIER_WEBHOOK,
+    ZAPIER_JOB_CANDIDATE_SWITCH: process.env.ZAPIER_JOB_CANDIDATE_SWITCH || 'OFF',
+    ZAPIER_JOB_CANDIDATE_WEBHOOK: process.env.ZAPIER_JOB_CANDIDATE_WEBHOOK,
     TOPCODER_API_URL: process.env.TOPCODER_API_URL || 'http://api.topcoder-dev.com/v5'
   }
 }

src/app.js

@@ -21,9 +21,9 @@ const eventEmitter = new events.EventEmitter()
 process.env.PORT = config.PORT
 
 const localLogger = {
-  'info': (message) => logger.info({ component: 'app', message }),
-  'debug': (message) => logger.debug({ component: 'app', message }),
-  'error': (message) => logger.error({ component: 'app', message })
+  info: (message) => logger.info({ component: 'app', message }),
+  debug: (message) => logger.debug({ component: 'app', message }),
+  error: (message) => logger.error({ component: 'app', message })
 }
 
 const topicServiceMapping = {
@@ -47,7 +47,7 @@ localLogger.info('Starting kafka consumer')
 const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
 
 let count = 0
-let mutex = new Mutex()
+const mutex = new Mutex()
 
 async function getLatestCount () {
   const release = await mutex.acquire()
@@ -71,7 +71,7 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   localLogger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${
     m.offset}; Message: ${message}.`)
   let messageJSON
-  let messageCount = await getLatestCount()
+  const messageCount = await getLatestCount()
 
   localLogger.debug(`Current message count: ${messageCount}`)
   try {

src/bootstrap.js

@@ -15,6 +15,7 @@ const zapierSwitch = Joi.string().label('ZAPIER_SWITCH').valid(...Object.values(
 // validate configuration
 try {
   Joi.attempt(config.zapier.ZAPIER_SWITCH, zapierSwitch)
+  Joi.attempt(config.zapier.ZAPIER_JOB_CANDIDATE_SWITCH, zapierSwitch)
 } catch (err) {
   console.error(err.message)
   process.exit(1)

src/common/constants.js

@@ -12,7 +12,9 @@ module.exports = {
     },
     MessageType: {
       JobCreate: 'job:create',
-      JobUpdate: 'job:update'
+      JobUpdate: 'job:update',
+      JobCandidateCreate: 'jobcandidate:create',
+      JobCandidateUpdate: 'jobcandidate:update'
     }
   }
 }

src/common/helper.js

@@ -10,7 +10,6 @@ const elasticsearch = require('@elastic/elasticsearch')
 const _ = require('lodash')
 const { Mutex } = require('async-mutex')
 const m2mAuth = require('tc-core-library-js').auth.m2m
-const constants = require('./constants')
 
 AWS.config.region = config.esConfig.AWS_REGION
 
@@ -154,33 +153,19 @@ async function getM2MToken () {
 /**
  * Post message to zapier via webhook url.
  *
- * @param {Object} message the message object
+ * @param {String} webhook the webhook url
+ * @param {Object} message the message data
  * @returns {undefined}
  */
-async function postMessageToZapier ({ type, payload }) {
-  if (config.zapier.ZAPIER_SWITCH === constants.Zapier.Switch.OFF) {
-    logger.debug({ component: 'helper', context: 'postMessageToZapier', message: 'Zapier Switch off via config, no messages sent' })
-    return
-  }
-  const requestBody = {
-    type,
-    payload,
-    companySlug: config.zapier.ZAPIER_COMPANYID_SLUG,
-    contactSlug: config.zapier.ZAPIER_CONTACTID_SLUG
-  }
-  if (type === constants.Zapier.MessageType.JobCreate) {
-    const token = await getM2MToken()
-    requestBody.authToken = token
-    requestBody.topcoderApiUrl = config.zapier.TOPCODER_API_URL
-  }
-  logger.debug({ component: 'helper', context: 'postMessageToZapier', message: `request body: ${JSON.stringify(requestBody)}` })
-  await request.post(config.zapier.ZAPIER_WEBHOOK)
-    .send(requestBody)
+async function postMessageViaWebhook (webhook, message) {
+  logger.debug({ component: 'helper', context: 'postMessageToZapier', message: `message: ${JSON.stringify(message)}` })
+  await request.post(webhook).send(message)
 }
 
 module.exports = {
   getKafkaOptions,
   getESClient,
   checkEsMutexRelease,
-  postMessageToZapier
+  getM2MToken,
+  postMessageViaWebhook
 }

src/services/JobCandidateProcessorService.js

@@ -11,6 +11,64 @@ const config = require('config')
 
 const esClient = helper.getESClient()
 
+const localLogger = {
+  debug: ({ context, message }) => logger.debug({ component: 'JobCandidateProcessorService', context, message })
+}
+
+/**
+ * Update job candidate status in recruit CRM.
+ *
+ * @param {Object} message the message object
+ * @returns {undefined}
+ */
+async function updateCandidateStatus ({ type, payload }) {
+  if (!payload.status) {
+    localLogger.debug({ context: 'updateCandidateStatus', message: 'status not updated' })
+    return
+  }
+  if (!['rejected', 'shortlist'].includes(payload.status)) {
+    localLogger.debug({ context: 'updateCandidateStatus', message: `not interested status: ${payload.status}` })
+    return
+  }
+  const { body: jobCandidate } = await esClient.getSource({
+    index: config.get('esConfig.ES_INDEX_JOB_CANDIDATE'),
+    id: payload.id
+  })
+  if (!jobCandidate.externalId) {
+    localLogger.debug({ context: 'updateCandidateStatus', message: `id: ${jobCandidate.id} candidate without externalId - ignored` })
+    return
+  }
+  const { body: job } = await esClient.getSource({
+    index: config.get('esConfig.ES_INDEX_JOB'),
+    id: jobCandidate.jobId
+  })
+  const message = {
+    type,
+    status: jobCandidate.status,
+    jobCandidateSlug: jobCandidate.externalId,
+    jobSlug: job.externalId
+  }
+  await helper.postMessageViaWebhook(config.zapier.ZAPIER_JOB_CANDIDATE_WEBHOOK, message)
+}
+
+/**
+ * Post message to zapier for JobCandidate.
+ *
+ * @param {Object} message the message object
+ * @returns {undefined}
+ */
+async function postMessageToZapier ({ type, payload }) {
+  if (config.zapier.ZAPIER_JOB_CANDIDATE_SWITCH === constants.Zapier.Switch.OFF) {
+    localLogger.debug({ context: 'postMessageToZapier', message: 'Zapier Switch off via config, no messages sent' })
+    return
+  }
+  if (type === constants.Zapier.MessageType.JobCandidateUpdate) {
+    await updateCandidateStatus({ type, payload })
+    return
+  }
+  throw new Error(`unrecognized message type: ${type}`)
+}
+
 /**
  * Process create entity message
  * @param {Object} message the kafka message
@@ -63,6 +121,10 @@ async function processUpdate (message, transactionId) {
     },
     refresh: constants.esRefreshOption
   })
+  await postMessageToZapier({
+    type: constants.Zapier.MessageType.JobCandidateUpdate,
+    payload: data
+  })
 }
 
 processUpdate.schema = {

src/services/JobProcessorService.js

@@ -11,6 +11,35 @@ const config = require('config')
 
 const esClient = helper.getESClient()
 
+const localLogger = {
+  debug: ({ context, message }) => logger.debug({ component: 'JobProcessorService', context, message })
+}
+
+/**
+ * Post message to zapier for Job.
+ *
+ * @param {Object} message the message object
+ * @returns {undefined}
+ */
+async function postMessageToZapier ({ type, payload }) {
+  if (config.zapier.ZAPIER_SWITCH === constants.Zapier.Switch.OFF) {
+    localLogger.debug({ context: 'postMessageToZapier', message: 'Zapier Switch off via config, no messages sent' })
+    return
+  }
+  const message = {
+    type,
+    payload,
+    companySlug: config.zapier.ZAPIER_COMPANYID_SLUG,
+    contactSlug: config.zapier.ZAPIER_CONTACTID_SLUG
+  }
+  if (type === constants.Zapier.MessageType.JobCreate) {
+    const token = await helper.getM2MToken()
+    message.authToken = token
+    message.topcoderApiUrl = config.zapier.TOPCODER_API_URL
+  }
+  await helper.postMessageViaWebhook(config.zapier.ZAPIER_WEBHOOK, message)
+}
+
 /**
  * Process create entity message
  * @param {Object} message the kafka message
@@ -25,7 +54,7 @@ async function processCreate (message, transactionId) {
     body: _.omit(job, 'id'),
     refresh: constants.esRefreshOption
   })
-  await helper.postMessageToZapier({
+  await postMessageToZapier({
     type: constants.Zapier.MessageType.JobCreate,
     payload: job
   })
@@ -74,7 +103,7 @@ async function processUpdate (message, transactionId) {
     },
     refresh: constants.esRefreshOption
   })
-  await helper.postMessageToZapier({
+  await postMessageToZapier({
     type: constants.Zapier.MessageType.JobUpdate,
     payload: data
   })