Merge branch 'develop'

Author: James Cori

.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,64 @@ 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) {
+    return []
+  }
+}
+
 /**
  * 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] = []
+    }
+    if (!dateBasedEvents[phase.scheduledEndDate]) {
+      dateBasedEvents[phase.scheduledEndDate] = []
+    }
+    if (new Date(phase.scheduledEndDate).getTime() >= Date.now() && !phase.isOpen) {
+      dateBasedEvents[phase.scheduledStartDate].push({
+        phaseId: phase.phaseId,
+        isOpen: true
+      })
+    }
+    if (new Date(phase.scheduledStartDate).getTime() <= Date.now() && phase.isOpen) {
+      dateBasedEvents[phase.scheduledEndDate].push({
+        phaseId: phase.phaseId,
+        isOpen: false
+      })
     }
-
-    events.push({
-      ...event,
-      status: 'starting',
-      scheduleTime: phase.scheduledStartDate
-    })
-    events.push({
-      ...event,
-      status: 'closing',
-      scheduleTime: phase.scheduledEndDate
-    })
   }
+
+  _.each(dateBasedEvents, (eventData, scheduleTime) => {
+    if (eventData.length > 0) {
+      events.push({
+        externalId: challenge.id,
+        scheduleTime,
+        payload: {
+          phases: eventData
+        }
+      })
+    }
+  })
+
   return events
 }
 
@@ -111,39 +146,59 @@ async function createEventsInExecutor (events) {
   const url = config.SCHEDULE_API_URL
   const token = await getTopcoderM2Mtoken()
 
-  try {
-    for (const event of events) {
+  for (const event of events) {
+    try {
       // 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
       logger.debug(`request POST ${url}`)
       await axios.post(`${url}`, executorPayload)
+    } catch (err) {
+      logger.warn(`Failed to create event for external ID ${event.externalId}`)
+      logger.error(err.message)
+    }
+  }
+}
+
+/**
+ * Delete events in executor app
+ * @param events the events array
+ */
+async function deleteEventsInExecutor (events) {
+  const url = config.SCHEDULE_API_URL
+  for (const event of events) {
+    // schedule executor api payload
+    const executorPayload = {
+      id: event.id
+    }
+    try {
+      // call executor api
+      logger.debug(`request DELETE ${url}`)
+      await axios.delete(`${url}`, { data: executorPayload })
+    } catch (err) {
+      logger.warn(`Failed to delete event ${event.id}`)
+      logger.error(err.message)
     }
-  } catch (err) {
-    logger.error(err.message)
-    throw err
   }
 }
 
 module.exports = {
   getKafkaOptions,
   getTopcoderM2Mtoken,
   getChallenge,
+  getEventsFromScheduleApi,
   getEventsFromPhases,
-  createEventsInExecutor
+  createEventsInExecutor,
+  deleteEventsInExecutor
 }

src/services/ProcessorService.js

@@ -2,10 +2,12 @@
  * Processor Service
  */
 const Joi = require('@hapi/joi')
+const _ = require('lodash')
 const logger = require('../common/logger')
 const helper = require('../common/helper')
 
 Joi.id = () => Joi.string().uuid().required()
+const VALID_CHALLENGE_STATUSES = ['Active']
 
 /**
  * Process create entity message
@@ -15,6 +17,14 @@ Joi.id = () => Joi.string().uuid().required()
 async function processCreate (message) {
   // get challenge
   const challenge = await helper.getChallenge(message.payload.id)
+  if (!VALID_CHALLENGE_STATUSES.includes(challenge.status)) {
+    logger.info(`Not creating events for challenge status ${challenge.status}...`)
+    return
+  }
+  if (!_.get(challenge, 'legacy.useSchedulingAPI')) {
+    logger.info(`The legacy.useSchedulingAPI is not set on challenge ${challenge.id}...`)
+    return
+  }
   // create events
   const events = helper.getEventsFromPhases(challenge)
   // call the executor api
@@ -23,6 +33,30 @@ 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)
+  if (!VALID_CHALLENGE_STATUSES.includes(sourceChallenge.status)) {
+    logger.info(`Not creating events for challenge status ${sourceChallenge.status}...`)
+    return
+  }
+  if (!_.get(sourceChallenge, 'legacy.useSchedulingAPI')) {
+    logger.info(`The legacy.useSchedulingAPI is not set on challenge ${sourceChallenge.id}...`)
+    return
+  }
+  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 +68,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)