Added logic for creating/deleting a topic for a phase created/deleted for a project.

Author: Vikas Agarwal

config/custom-environment-variables.json

@@ -31,6 +31,7 @@
   "VALID_ISSUERS": "VALID_ISSUERS",
   "jwksUri": "JWKS_URI",
   "busApiUrl": "BUS_API_URL",
+  "messageApiUrl": "MESSAGE_SERVICE_URL",
   "AUTH0_URL" : "AUTH0_URL",
   "AUTH0_CLIENT_ID": "AUTH0_CLIENT_ID",
   "AUTH0_CLIENT_SECRET": "AUTH0_CLIENT_SECRET",

config/default.json

@@ -36,6 +36,7 @@
   "validIssuers": "[\"https:\/\/topcoder-newauth.auth0.com\/\",\"https:\/\/api.topcoder-dev.com\"]",
   "jwksUri": "",
   "busApiUrl": "http://api.topcoder-dev.com/v5",
+  "messageApiUrl": "http://api.topcoder-dev.com/v5",
   "busApiToken": "eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJuYW1lIjoicHJvamVjdC1zZXJ2aWNlIiwiaWF0IjoxNTEyNzQ3MDgyLCJleHAiOjE1MjEzODcwODJ9.PHuNcFDaotGAL8RhQXQMdpL8yOKXxjB5DbBIodmt7RE",
   "HEALTH_CHECK_URL": "_health",
   "maxPhaseProductCount": 1,

src/events/busApi.js

@@ -257,6 +257,7 @@ module.exports = (app, logger) => {
           projectName: project.name,
           userId: req.authUser.userId,
           initiatorUserId: req.authUser.userId,
+          phase: created,
         }, logger);
       }).catch(err => null);    // eslint-disable-line no-unused-vars
   });

src/events/projectPhases/index.js

@@ -7,20 +7,21 @@ import config from 'config';
 import _ from 'lodash';
 import Promise from 'bluebird';
 import util from '../../util';
+import messageService from '../../services/messageService';
 
 const ES_PROJECT_INDEX = config.get('elasticsearchConfig.indexName');
 const ES_PROJECT_TYPE = config.get('elasticsearchConfig.docType');
 
 const eClient = util.getElasticSearchClient();
 
 /**
- * Handler for project phase creation event
+ * Indexes the project phase in the elastic search.
+ *
  * @param  {Object} logger  logger to log along with trace id
  * @param  {Object} msg     event payload
- * @param  {Object} channel channel to ack, nack
  * @returns {undefined}
  */
-const projectPhaseAddedHandler = Promise.coroutine(function* (logger, msg, channel) { // eslint-disable-line func-names
+const indexProjectPhase = Promise.coroutine(function* (logger, msg) { // eslint-disable-line func-names
   try {
     const data = JSON.parse(msg.content.toString());
     const doc = yield eClient.get({ index: ES_PROJECT_INDEX, type: ES_PROJECT_TYPE, id: data.projectId });
@@ -29,6 +30,50 @@ const projectPhaseAddedHandler = Promise.coroutine(function* (logger, msg, chann
     const merged = _.assign(doc._source, { phases }); // eslint-disable-line no-underscore-dangle
     yield eClient.update({ index: ES_PROJECT_INDEX, type: ES_PROJECT_TYPE, id: data.projectId, body: { doc: merged } });
     logger.debug('project phase added to project document successfully');
+  } catch (error) {
+    logger.error('Error handling indexing the project phase', error);
+    // throw the error back to nack the bus
+    throw error;
+  }
+});
+
+/**
+ * Creates a new phase topic in message api.
+ *
+ * @param  {Object} logger  logger to log along with trace id
+ * @param  {Object} msg     event payload
+ * @returns {undefined}
+ */
+const createPhaseTopic = Promise.coroutine(function* (logger, msg) { // eslint-disable-line func-names
+  try {
+    const phase = JSON.parse(msg.content.toString());
+    const topic = yield messageService.createTopic({
+      reference: 'project',
+      referenceId: `${phase.projectId}`,
+      tag: `phase#${phase.id}`,
+      title: phase.name,
+      body: 'Welcome!!! Please use this channel for communication around the phase.',
+    }, logger);
+    logger.debug('topic for the phase created successfully');
+    logger.debug(topic);
+  } catch (error) {
+    logger.error('Error in creating topic for the project phase', error);
+    // don't throw the error back to nack the bus, because we don't want to get multiple topics per phase
+    // we can create topic for a phase manually, if somehow it fails
+  }
+});
+
+/**
+ * Handler for project phase creation event
+ * @param  {Object} logger  logger to log along with trace id
+ * @param  {Object} msg     event payload
+ * @param  {Object} channel channel to ack, nack
+ * @returns {undefined}
+ */
+const projectPhaseAddedHandler = Promise.coroutine(function* (logger, msg, channel) { // eslint-disable-line func-names
+  try {
+    yield indexProjectPhase(logger, msg, channel);
+    yield createPhaseTopic(logger, msg);
     channel.ack(msg);
   } catch (error) {
     logger.error('Error handling project.phase.added event', error);
@@ -73,13 +118,13 @@ const projectPhaseUpdatedHandler = Promise.coroutine(function* (logger, msg, cha
 });
 
 /**
- * Handler for project phase deleted event
+ * Removes the project phase from the elastic search.
+ *
  * @param  {Object} logger  logger to log along with trace id
  * @param  {Object} msg     event payload
- * @param  {Object} channel channel to ack, nack
  * @returns {undefined}
  */
-const projectPhaseRemovedHandler = Promise.coroutine(function* (logger, msg, channel) { // eslint-disable-line func-names
+const removePhaseFromIndex = Promise.coroutine(function* (logger, msg) { // eslint-disable-line func-names
   try {
     const data = JSON.parse(msg.content.toString());
     const doc = yield eClient.get({ index: ES_PROJECT_INDEX, type: ES_PROJECT_TYPE, id: data.projectId });
@@ -94,6 +139,45 @@ const projectPhaseRemovedHandler = Promise.coroutine(function* (logger, msg, cha
       },
     });
     logger.debug('project phase removed from project document successfully');
+  } catch (error) {
+    logger.error('Error in removing project phase from index', error);
+    // throw the error back to nack the bus
+    throw error;
+  }
+});
+
+/**
+ * Removes the phase topic from the message api.
+ *
+ * @param  {Object} logger  logger to log along with trace id
+ * @param  {Object} msg     event payload
+ * @returns {undefined}
+ */
+const removePhaseTopic = Promise.coroutine(function* (logger, msg) { // eslint-disable-line func-names
+  try {
+    const phase = JSON.parse(msg.content.toString());
+    const phaseTopic = yield messageService.getPhaseTopic(phase.projectId, phase.id, logger);
+    yield messageService.deletePosts(phaseTopic.id, phaseTopic.postIds, logger);
+    yield messageService.deleteTopic(phaseTopic.id, logger);
+    logger.debug('topic for the phase removed successfully');
+  } catch (error) {
+    logger.error('Error in removing topic for the project phase', error);
+    // don't throw the error back to nack the bus
+    // we can delete topic for a phase manually, if somehow it fails
+  }
+});
+
+/**
+ * Handler for project phase deleted event
+ * @param  {Object} logger  logger to log along with trace id
+ * @param  {Object} msg     event payload
+ * @param  {Object} channel channel to ack, nack
+ * @returns {undefined}
+ */
+const projectPhaseRemovedHandler = Promise.coroutine(function* (logger, msg, channel) { // eslint-disable-line func-names
+  try {
+    yield removePhaseFromIndex(logger, msg, channel);
+    yield removePhaseTopic(logger, msg);
     channel.ack(msg);
   } catch (error) {
     logger.error('Error fetching project document from elasticsearch', error);

src/services/messageService.js

@@ -0,0 +1,153 @@
+import config from 'config';
+import _ from 'lodash';
+
+const Promise = require('bluebird');
+const axios = require('axios');
+const tcCoreLibAuth = require('tc-core-library-js').auth;
+
+const m2m = tcCoreLibAuth.m2m(config);
+
+let client = null;
+
+/**
+ * Get Http client to bus api
+ * @param {Object} logger object
+ * @return {Object} Http Client to bus api
+ */
+async function getClient(logger) {
+  if (client) return client;
+  const msgApiUrl = config.get('messageApiUrl');
+  try {
+    const token = await m2m.getMachineToken(config.AUTH0_CLIENT_ID, config.AUTH0_CLIENT_SECRET);
+    client = axios.create({ baseURL: msgApiUrl });
+
+    // Alter defaults after instance has been created
+    client.defaults.headers.common.Authorization = `Bearer ${token}`;
+
+    // Add a response interceptor
+    client.interceptors.response.use(function (res) { // eslint-disable-line
+      return res;
+    }, function (error) { // eslint-disable-line
+      if (error.response) {
+        // The request was made and the server responded with a status code
+        // that falls out of the range of 2xx
+        logger.debug(error.response.data);
+        logger.debug(error.response.status);
+        logger.debug(error.response.headers);
+      } else if (error.request) {
+        // The request was made but no response was received
+        // `error.request` is an instance of XMLHttpRequest in the browser and an instance of
+        // http.ClientRequest in node.js
+        logger.debug(error.request);
+      } else {
+        // Something happened in setting up the request that triggered an Error
+        logger.debug(error.message);
+      }
+      logger.debug(error.config);
+      // Ingore response errors
+      return Promise.reject(error);
+    });
+
+    return client;
+  } catch (err) {
+    return Promise.reject(`Bus api calling - Error in genearting m2m token : ${err.message}`);
+  }
+}
+
+/**
+ * Creates a new topic in message api
+ *
+ * @param {Object} topic the topic, should be a JSON object
+ * @param {Object} logger object
+ * @return {Promise} new topic promise
+ */
+function createTopic(topic, logger) {
+  logger.debug(`createTopic for topic: ${JSON.stringify(topic)}`);
+  return getClient(logger).then((msgClient) => {
+    logger.debug('calling message service');
+    return msgClient.post('/topics/create', topic)
+      .then((resp) => {
+        logger.debug('Topic created successfully');
+        logger.debug(`Topic created successfully [status]: ${resp.status}`);
+        logger.debug(`Topic created successfully [data]: ${resp.data}`);
+      })
+      .catch((error) => {
+        logger.debug('Error creating topic');
+        logger.error(error);
+         // eslint-disable-line
+      });
+  }).catch((errMessage) => {
+    logger.debug(errMessage);
+  });
+}
+
+/**
+ * Deletes the given posts for the given topic.
+ *
+ * @param {Integer} topicId id of the topic
+ * @param {Array} postIds array of post ids to be deleted, array of integers
+ * @param {Object} logger object
+ * @return {Promise} delete posts promise
+ */
+function deletePosts(topicId, postIds, logger) {
+  logger.debug(`deletePosts for topicId: ${topicId} and postIds: ${postIds}`);
+  const promises = [];
+  if (postIds && postIds.length > 0) {
+    postIds.forEach((postId) => {
+      promises.push(getClient(logger).then((msgClient) => {
+        logger.debug(`calling message service for deleting post#${postId}`);
+        return msgClient.delete(`/topics/${topicId}/posts/${postId}/remove`);
+      }));
+    });
+  }
+  if (promises.length > 0) {
+    return Promise.all(promises).then(() => logger.debug(`All posts deleted for topic ${topicId}`));
+  }
+  return Promise.resolve();
+}
+
+/**
+ * Fetches the topic of given phase of the project.
+ *
+ * @param {Integer} projectId id of the project
+ * @param {Integer} phaseId id of the phase of the project
+ * @param {Object} logger object
+ * @return {Promise} topic promise
+ */
+function getPhaseTopic(projectId, phaseId, logger) {
+  logger.debug(`getPhaseTopic for phaseId: ${phaseId}`);
+  return getClient(logger).then((msgClient) => {
+    logger.debug(`calling message service for fetching phaseId#${phaseId}`);
+    return msgClient.get('/topics/list', {
+      params: { filter: `reference=project&referenceId=${projectId}&tag=phase#${phaseId}` },
+    }).then((resp) => {
+      const topics = _.get(resp.data, 'result.content', []);
+      if (topics && topics.length > 0) {
+        return topics[0];
+      }
+      return null;
+    });
+  });
+}
+
+/**
+ * Deletes the given topic.
+ *
+ * @param {Integer} topicId id of the topic
+ * @param {Object} logger object
+ * @return {Promise} delete topic promise
+ */
+function deleteTopic(topicId, logger) {
+  logger.debug(`deleteTopic for topicId: ${topicId}`);
+  return getClient(logger).then((msgClient) => {
+    logger.debug(`calling message service for deleting topic#${topicId}`);
+    return msgClient.delete(`/topics/${topicId}/remove`);
+  });
+}
+
+module.exports = {
+  createTopic,
+  deletePosts,
+  getPhaseTopic,
+  deleteTopic,
+};