update processor to handle challenge updates

Author: ThomasKranitsas

.circleci/config.yml

@@ -79,4 +79,4 @@ workflows:
           context : org-global
           filters:
             branches:
-              only: master
+              only: master

.gitignore

@@ -1,3 +1,4 @@
 node_modules
-
 .env
+coverage
+.nyc_output

build.sh

@@ -20,4 +20,4 @@ fi
 if [ "$UPDATE_CACHE" == 1 ]
 then
   docker cp app:/$APP_NAME/node_modules .
-fi
+fi

config/default.js

@@ -14,8 +14,9 @@ module.exports = {
   // Kafka group id
   KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'autopilot-processor',
   NOTIFICATION_CREATE_TOPIC: process.env.NOTIFICATION_CREATE_TOPIC || 'challenge.notification.create',
+  NOTIFICATION_UPDATE_TOPIC: process.env.NOTIFICATION_UPDATE_TOPIC || 'challenge.notification.update',
 
-  SCHEDULE_API_URL: process.env.SCHEDULE_API_URL,
+  SCHEDULE_API_URL: process.env.SCHEDULE_API_URL || 'https://api.topcoder-dev.com/v5/schedules',
   CHALLENGE_API_URL: process.env.CHALLENGE_API_URL || 'https://api.topcoder-dev.com/v5/challenges',
 
   AUTH0_URL: process.env.AUTH0_URL,

docker-kafka/docker-compose.yml

@@ -1,4 +1,4 @@
-version: '3'
+version: "3"
 services:
   zookeeper:
     image: wurstmeister/zookeeper
@@ -12,5 +12,5 @@ services:
       - "9092:9092"
     environment:
       KAFKA_ADVERTISED_HOST_NAME: localhost
-      KAFKA_CREATE_TOPICS: "challenge.notification.create:1:1"
+      KAFKA_CREATE_TOPICS: "challenge.notification.create:1:1,challenge.notification.update:1:1"
       KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181

src/app.js

@@ -20,8 +20,7 @@ const consumer = new Kafka.GroupConsumer(helper.getKafkaOptions())
  */
 const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, async (m) => {
   const message = m.message.value.toString('utf8')
-  logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${
-    m.offset}; Message: ${message}.`)
+  logger.info(`Handle Kafka event message; Topic: ${topic}; Partition: ${partition}; Offset: ${m.offset}; Message: ${message}.`)
   let messageJSON
   try {
     messageJSON = JSON.parse(message)
@@ -43,8 +42,11 @@ const dataHandler = (messageSet, topic, partition) => Promise.each(messageSet, a
   }
 
   try {
-    await ProcessorService.processCreate(messageJSON)
-
+    if (topic === config.NOTIFICATION_CREATE_TOPIC) {
+      await ProcessorService.processCreate(messageJSON)
+    } else {
+      await ProcessorService.processUpdate(messageJSON)
+    }
     logger.debug('Successfully processed message')
   } catch (err) {
     logger.logFullError(err)
@@ -68,7 +70,7 @@ const check = () => {
   return connected
 }
 
-const topics = [config.NOTIFICATION_CREATE_TOPIC]
+const topics = [config.NOTIFICATION_CREATE_TOPIC, config.NOTIFICATION_UPDATE_TOPIC]
 
 logger.info('Starting kafka consumer')
 consumer

src/common/helper.js

@@ -53,6 +53,7 @@ function getErrorWithStatus (message, statusCode) {
 /**
  * Get challenge by id
  * @param challengeId the challenge id
+ * @returns {object} challenge
  */
 async function getChallenge (challengeId) {
   const url = `${config.CHALLENGE_API_URL}/${challengeId}`
@@ -76,30 +77,67 @@ async function getChallenge (challengeId) {
   }
 }
 
+/**
+ * Get challenge by id from Schedule Api
+ * @param challengeId the challenge id
+ * @returns {Array} array of events
+ */
+async function getEventsFromScheduleApi (challengeId) {
+  const url = `${config.SCHEDULE_API_URL}?externalId=${challengeId}`
+
+  logger.debug(`request GET ${url}`)
+  try {
+    const res = await axios.get(url)
+    return res.data
+  } catch (err) {
+    logger.error(err.message)
+
+    if (err.response) {
+      if (err.response.status === 404) {
+        logger.error(`The Challenge with the id: ${challengeId} not exist`)
+        throw getErrorWithStatus('[resource_not_found_exception]', 404)
+      }
+    }
+    throw Error(`get ${url} failed`)
+  }
+}
+
 /**
  * Create events from challenge object
  * @param challenge the challenge object
  */
 function getEventsFromPhases (challenge) {
   const events = []
-  // for each phase, create 2 events for the scheduledStartDate and scheduledEndDate respectively
+  const dateBasedEvents = {}
+
+
   for (const phase of challenge.phases) {
-    const event = {
-      phaseId: phase.phaseId,
-      challengeId: challenge.id
+    if (!dateBasedEvents[phase.scheduledStartDate]) {
+      dateBasedEvents[phase.scheduledStartDate] = []
     }
-
-    events.push({
-      ...event,
-      status: 'starting',
-      scheduleTime: phase.scheduledStartDate
+    if (!dateBasedEvents[phase.scheduledEndDate]) {
+      dateBasedEvents[phase.scheduledEndDate] = []
+    }
+    dateBasedEvents[phase.scheduledStartDate].push({
+      phaseId: phase.phaseId,
+      isOpen: true
     })
-    events.push({
-      ...event,
-      status: 'closing',
-      scheduleTime: phase.scheduledEndDate
+    dateBasedEvents[phase.scheduledEndDate].push({
+      phaseId: phase.phaseId,
+      isOpen: false
     })
   }
+
+  _.each(dateBasedEvents, (eventData, scheduleTime) => {
+    events.push({
+      externalId: challenge.id,
+      scheduleTime,
+      payload: {
+        phases: eventData
+      }
+    })
+  })
+
   return events
 }
 
@@ -115,19 +153,15 @@ async function createEventsInExecutor (events) {
     for (const event of events) {
       // schedule executor api payload
       const executorPayload = {
-        url: `${config.CHALLENGE_API_URL}/${event.challengeId}`,
+        url: `${config.CHALLENGE_API_URL}/${event.externalId}`,
+        externalId: event.externalId,
         method: 'patch',
         scheduleTime: event.scheduleTime,
         headers: {
           'content-type': 'application/json',
           Authorization: `Bearer ${token}`
         },
-        payload: JSON.stringify({
-          phases: [{
-            phaseId: event.phaseId,
-            isOpen: event.status === 'starting'
-          }]
-        })
+        payload: JSON.stringify(event.payload)
       }
 
       // call executor api
@@ -140,10 +174,37 @@ async function createEventsInExecutor (events) {
   }
 }
 
+/**
+ * Delete events in executor app
+ * @param events the events array
+ */
+async function deleteEventsInExecutor (events) {
+  const url = config.SCHEDULE_API_URL
+  try {
+    for (const event of events) {
+      // schedule executor api payload
+      const executorPayload = {
+        id: event.id
+      }
+
+      // call executor api
+      logger.debug(`request DELETE ${url}`)
+      await axios.delete(`${url}`, { data: executorPayload })
+    }
+  } catch (err) {
+    logger.error(err.message)
+    throw err
+  }
+}
+
 module.exports = {
   getKafkaOptions,
   getTopcoderM2Mtoken,
   getChallenge,
+  getEventsFromScheduleApi,
   getEventsFromPhases,
-  createEventsInExecutor
+  compareEvents,
+  updateChallenge,
+  createEventsInExecutor,
+  deleteEventsInExecutor
 }

src/services/ProcessorService.js

@@ -12,7 +12,7 @@ Joi.id = () => Joi.string().uuid().required()
  * @param {Object} message the kafka message
  * @returns {Promise}
  */
-async function processCreate (message) {
+async function processCreate(message) {
   // get challenge
   const challenge = await helper.getChallenge(message.payload.id)
   // create events
@@ -23,6 +23,22 @@ async function processCreate (message) {
   logger.info(`processing of the record completed, id: ${message.payload.id}`)
 }
 
+/**
+ * Process update entity message
+ * @param {Object} message the kafka message
+ * @returns {Promise}
+ */
+async function processUpdate(message) {
+  const sourceChallenge = await helper.getChallenge(message.payload.id)
+  const newEvents = helper.getEventsFromPhases(sourceChallenge)
+  const oldEvents = await helper.getEventsFromScheduleApi(message.payload.id)
+  logger.info(`Deleting existing events for challenge ${message.payload.id}`)
+  await helper.deleteEventsInExecutor(oldEvents)
+  logger.info(`Creating events for challenge ${message.payload.id}`)
+  await helper.createEventsInExecutor(newEvents)
+  logger.info(`processing of the record completed, id: ${message.payload.id}`)
+}
+
 processCreate.schema = {
   message: Joi.object().keys({
     topic: Joi.string().required(),
@@ -34,9 +50,21 @@ processCreate.schema = {
     }).required().unknown(true)
   }).required()
 }
+processUpdate.schema = {
+  message: Joi.object().keys({
+    topic: Joi.string().required(),
+    originator: Joi.string().required(),
+    timestamp: Joi.date().required(),
+    'mime-type': Joi.string().required(),
+    payload: Joi.object().keys({
+      id: Joi.id()
+    }).required().unknown(true)
+  }).required()
+}
 
 module.exports = {
-  processCreate
+  processCreate,
+  processUpdate
 }
 
 logger.buildService(module.exports)