adding member api call and publishing to another topic

Author: Sachin Maheshwari

config/default.js

@@ -10,10 +10,14 @@ module.exports = {
   // for the local Kafka, they are not needed
   KAFKA_CLIENT_CERT: process.env.KAFKA_CLIENT_CERT,
   KAFKA_CLIENT_CERT_KEY: process.env.KAFKA_CLIENT_CERT_KEY,
-
+  KAFKA_MESSAGE_ORIGINATOR: process.env.KAFKA_MESSAGE_ORIGINATOR || 'u-bahn-user-reconciliation-processor',
+ 
   // Kafka group id
   KAFKA_GROUP_ID: process.env.KAFKA_GROUP_ID || 'u-bahn-user-reconciliation-processor',
   USER_RECONCILATION_TOPIC: process.env.USER_RECONCILATION_TOPIC || 'backgroundjob.reconcile.user',
+  PUBLISH_TOPIC: process.env.PUBLISH_TOPIC_TOPIC || 'legacy.sync.user',
+
+  V5_API_URL: process.env.V5_API_URL || 'https://api.topcoder-dev.com/v5',
 
   UBAHN_API_URL: process.env.UBAHN_API_URL || 'https://api.topcoder-dev.com/v5',
   MEMBERS_API_URL: process.env.MEMBERS_API_URL || 'https://api.topcoder-dev.com/v5/members',

src/common/helper.js

@@ -19,7 +19,7 @@ const topcoderM2M = m2mAuth({ ...topcoderM2MConfig, AUTH0_AUDIENCE: topcoderM2MC
  * js version of sleep()
  * @param {Number} ms Timeout in ms
  */
-async function sleep (ms) {
+async function sleep(ms) {
   if (!ms) {
     ms = config.SLEEP_TIME
   }
@@ -33,33 +33,63 @@ async function sleep (ms) {
  * (U-Bahn APIs only)
  * @returns {Promise}
  */
-async function getUbahnM2Mtoken () {
+async function getUbahnM2Mtoken() {
   return ubahnM2M.getMachineToken(config.AUTH0_CLIENT_ID, config.AUTH0_CLIENT_SECRET)
 }
 
 /* Function to get M2M token
  * (Topcoder APIs only)
  * @returns {Promise}
  */
-async function getTopcoderM2Mtoken () {
+async function getTopcoderM2Mtoken() {
   return topcoderM2M.getMachineToken(config.AUTH0_CLIENT_ID, config.AUTH0_CLIENT_SECRET)
 }
 
 /**
  * Get Kafka options
  * @return {Object} the Kafka options
  */
-function getKafkaOptions () {
+function getKafkaOptions() {
   const options = { connectionString: config.KAFKA_URL, groupId: config.KAFKA_GROUP_ID }
   if (config.KAFKA_CLIENT_CERT && config.KAFKA_CLIENT_CERT_KEY) {
     options.ssl = { cert: config.KAFKA_CLIENT_CERT, key: config.KAFKA_CLIENT_CERT_KEY }
   }
   return options
 }
 
+/**
+ * Returns the member details
+ * 
+ * @param {String} handle The member handle
+ */
+async function getMember(handle, token) {
+  const res = await axios.get(`${config.MEMBERS_API_URL}/${qs.escape(handle)}`, { headers: { Authorization: `Bearer ${token}` } })
+  return _.get(res, 'data', {})
+}
+
+/**
+ * Send Kafka event message
+ * @params {String} topic the topic name
+ * @params {Object} payload the payload
+ */
+async function postEvent (topic, payload, token) {
+  logger.debug(`Posting event to Kafka topic ${topic}, ${JSON.stringify(payload, null, 2)}`)
+  const message = {
+    topic,
+    originator: config.KAFKA_MESSAGE_ORIGINATOR,
+    timestamp: new Date().toISOString(),
+    'mime-type': 'application/json',
+    payload
+  }
+  
+  await axios.post(`${config.V5_API_URL}/bus/events`, message, { headers: { Authorization: `Bearer ${token}` } })
+  logger.debug(`Posted event to Kafka topic`)
+}
 
 module.exports = {
   getKafkaOptions,
   getTopcoderM2Mtoken,
-  getUbahnM2Mtoken
+  getUbahnM2Mtoken,
+  getMember,
+  postEvent
 }

src/services/ProcessorService.js

@@ -13,10 +13,28 @@ const helper = require('../common/helper')
  * @param {Object} message the kafka message
  * @returns {Promise}
  */
-async function processSync (message) {
-  const { handle } = message.payload
-  const tcToken = await helper.getTopcoderM2Mtoken()
-  const ubToken = await helper.getUbahnM2Mtoken()
+async function processSync(message) {
+  try {
+    const { handle } = message.payload
+    const tcToken = await helper.getTopcoderM2Mtoken()
+    const member = await helper.getMember(handle, tcToken)
+    const location = member.homeCountryCode || member.competitionCountryCode || null
+    const payload = {
+      id : member.id,
+      handle: handle,
+      firstName : member.firstName,
+      lastName: member.lastName,
+      email: member.email,
+      country : {
+        isoAlpha3Code: location
+      },
+      active : (member.status === 'ACTIVE') ? true : false 
+    }
+    await postEvent(config.PUBLISH_TOPIC, payload, tcToken)
+  } catch (e) {
+    logger.error(`unable to process the message, error : ${JSON.stringify(e)}`)
+  }
+
 }
 
 processSync.schema = {