Merge pull request #184 from topcoder-platform/feature/milestone_phase_events_separation_squashed

Separated the raising of project and phase events from timeline/milestone changes

Author: vikasrohit

.circleci/config.yml

@@ -76,7 +76,7 @@ workflows:
             - test
           filters:
             branches:
-              only: ['dev', 'feature/timeline-milestone']
+              only: ['dev']
       - deployProd:
           requires:
             - test

src/constants.js

@@ -101,11 +101,14 @@ export const BUS_API_EVENT = {
 
   // When milestone is added/deleted to/from the phase,
   // When milestone is updated for duration/startDate/endDate/status
-  TIMELINE_MODIFIED: 'notifications.connect.project.phase.timelineModified',
+  TIMELINE_ADJUSTED: 'notifications.connect.project.phase.timeline.adjusted',
 
   // When specification of a product is modified
   PROJECT_PRODUCT_SPECIFICATION_MODIFIED: 'notifications.connect.project.productSpecificationModified',
 
+  MILESTONE_ADDED: 'notifications.connect.project.phase.milestone.added',
+  MILESTONE_REMOVED: 'notifications.connect.project.phase.milestone.removed',
+  MILESTONE_UPDATED: 'notifications.connect.project.phase.milestone.updated',
   // When milestone is marked as active
   MILESTONE_TRANSITION_ACTIVE: 'notifications.connect.project.phase.milestone.transition.active',
   // When milestone is marked as completed

src/events/busApi.js

@@ -474,10 +474,22 @@ module.exports = (app, logger) => {
    * @param {Object} original the original milestone
    * @param {Object} updated the updated milestone
    * @param {Object} project the project
+   * @param {Object} timeline the updated timeline
    * @returns {Promise<void>} void
    */
-  function sendMilestoneNotification(req, original, updated, project) {
+  function sendMilestoneNotification(req, original, updated, project, timeline) {
     logger.debug('sendMilestoneNotification', original, updated);
+    // throw generic milestone updated bus api event
+    createEvent(BUS_API_EVENT.MILESTONE_UPDATED, {
+      projectId: project.id,
+      projectName: project.name,
+      projectUrl: connectProjectUrl(project.id),
+      timeline,
+      originalMilestone: original,
+      updatedMilestone: updated,
+      userId: req.authUser.userId,
+      initiatorUserId: req.authUser.userId,
+    }, logger);
     // Send transition events
     if (original.status !== updated.status) {
       let event;
@@ -492,8 +504,7 @@ module.exports = (app, logger) => {
           projectId: project.id,
           projectName: project.name,
           projectUrl: connectProjectUrl(project.id),
-          timelineId: req.timeline.id,
-          timelineName: req.timeline.name,
+          timeline,
           originalMilestone: original,
           updatedMilestone: updated,
           userId: req.authUser.userId,
@@ -510,8 +521,7 @@ module.exports = (app, logger) => {
         projectId: project.id,
         projectName: project.name,
         projectUrl: connectProjectUrl(project.id),
-        timelineId: req.timeline.id,
-        timelineName: req.timeline.name,
+        timeline,
         originalMilestone: original,
         updatedMilestone: updated,
         userId: req.authUser.userId,
@@ -533,15 +543,16 @@ module.exports = (app, logger) => {
     })
       .then((project) => {
         if (project) {
-          createEvent(BUS_API_EVENT.PROJECT_PLAN_UPDATED, {
+          createEvent(BUS_API_EVENT.MILESTONE_ADDED, {
             projectId,
             projectName: project.name,
             projectUrl: connectProjectUrl(projectId),
+            addedMilestone: created,
             userId: req.authUser.userId,
             initiatorUserId: req.authUser.userId,
           }, logger);
         }
-        sendMilestoneNotification(req, {}, created, project);
+        // sendMilestoneNotification(req, {}, created, project);
       })
       .catch(err => null);    // eslint-disable-line no-unused-vars
   });
@@ -551,60 +562,54 @@ module.exports = (app, logger) => {
   */
   // eslint-disable-next-line no-unused-vars
   app.on(EVENT.ROUTING_KEY.MILESTONE_UPDATED, ({ req, original, updated, cascadedUpdates }) => {
-    logger.debug('receive MILESTONE_UPDATED event');
+    logger.debug(`receive MILESTONE_UPDATED event for milestone ${original.id}`);
 
     const projectId = _.parseInt(req.params.projectId);
+    const timeline = _.omit(req.timeline.toJSON(), 'deletedAt', 'deletedBy');
 
     models.Project.findOne({
       where: { id: projectId },
     })
-      .then((project) => {
-        // send PROJECT_UPDATED Kafka message when one of the specified below properties changed
-        const watchProperties = ['startDate', 'endDate', 'duration', 'details', 'status', 'order'];
-        if (!_.isEqual(_.pick(original, watchProperties),
-                      _.pick(updated, watchProperties))) {
-          createEvent(BUS_API_EVENT.PROJECT_PLAN_UPDATED, {
-            projectId,
-            projectName: project.name,
-            projectUrl: connectProjectUrl(projectId),
-            userId: req.authUser.userId,
-            initiatorUserId: req.authUser.userId,
-          }, logger);
-        }
-        sendMilestoneNotification(req, original, updated, project);
+    .then((project) => {
+      logger.debug(`Found project with id ${projectId}`);
+      return models.Milestone.getTimelineDuration(timeline.id)
+      .then(({ duration, progress }) => {
+        timeline.duration = duration;
+        timeline.progress = progress;
+        sendMilestoneNotification(req, original, updated, project, timeline);
 
         logger.debug('cascadedUpdates', cascadedUpdates);
         if (cascadedUpdates && cascadedUpdates.milestones && cascadedUpdates.milestones.length > 0) {
           _.each(cascadedUpdates.milestones, cascadedUpdate =>
-            sendMilestoneNotification(req, cascadedUpdate.original, cascadedUpdate.updated, project),
+            sendMilestoneNotification(req, cascadedUpdate.original, cascadedUpdate.updated, project, timeline),
           );
         }
 
         // if timeline is modified
         if (cascadedUpdates && cascadedUpdates.timeline) {
-          const timeline = cascadedUpdates.timeline;
-          // if endDate of the timeline is modified, raise TIMELINE_MODIFIED event
-          if (timeline.original.endDate !== timeline.updated.endDate) {
+          const cTimeline = cascadedUpdates.timeline;
+          // if endDate of the timeline is modified, raise TIMELINE_ADJUSTED event
+          if (cTimeline.original.endDate !== cTimeline.updated.endDate) {
             // Raise Timeline changed event
-            createEvent(BUS_API_EVENT.TIMELINE_MODIFIED, {
+            createEvent(BUS_API_EVENT.TIMELINE_ADJUSTED, {
               projectId: project.id,
               projectName: project.name,
               projectUrl: connectProjectUrl(project.id),
-              original: timeline.original,
-              updated: timeline.updated,
+              originalTimeline: cTimeline.original,
+              updatedTimeline: cTimeline.updated,
               userId: req.authUser.userId,
               initiatorUserId: req.authUser.userId,
             }, logger);
           }
         }
-      })
-      .catch(err => null);    // eslint-disable-line no-unused-vars
+      });
+    }).catch(err => null);    // eslint-disable-line no-unused-vars
   });
 
  /**
   * MILESTONE_REMOVED.
   */
-  app.on(EVENT.ROUTING_KEY.MILESTONE_REMOVED, ({ req }) => {
+  app.on(EVENT.ROUTING_KEY.MILESTONE_REMOVED, ({ req, deleted }) => {
     logger.debug('receive MILESTONE_REMOVED event');
     // req.params.projectId is set by validateTimelineIdParam middleware
     const projectId = _.parseInt(req.params.projectId);
@@ -614,10 +619,11 @@ module.exports = (app, logger) => {
     })
     .then((project) => {
       if (project) {
-        createEvent(BUS_API_EVENT.PROJECT_PLAN_UPDATED, {
+        createEvent(BUS_API_EVENT.MILESTONE_REMOVED, {
           projectId,
           projectName: project.name,
           projectUrl: connectProjectUrl(projectId),
+          removedMilestone: deleted,
           userId: req.authUser.userId,
           initiatorUserId: req.authUser.userId,
         }, logger);
@@ -639,10 +645,12 @@ module.exports = (app, logger) => {
       })
       .then((project) => {
         if (project) {
-          createEvent(BUS_API_EVENT.PROJECT_PLAN_UPDATED, {
+          createEvent(BUS_API_EVENT.TIMELINE_ADJUSTED, {
             projectId,
             projectName: project.name,
             projectUrl: connectProjectUrl(projectId),
+            originalTimeline: original,
+            updatedTimeline: updated,
             userId: req.authUser.userId,
             initiatorUserId: req.authUser.userId,
           }, logger);

src/events/index.js

@@ -10,8 +10,18 @@ import { projectPhaseAddedHandler, projectPhaseRemovedHandler,
   projectPhaseUpdatedHandler } from './projectPhases';
 import { phaseProductAddedHandler, phaseProductRemovedHandler,
   phaseProductUpdatedHandler } from './phaseProducts';
-import { timelineAddedHandler, timelineUpdatedHandler, timelineRemovedHandler } from './timelines';
-import { milestoneAddedHandler, milestoneUpdatedHandler, milestoneRemovedHandler } from './milestones';
+import {
+  timelineAddedHandler,
+  timelineUpdatedHandler,
+  timelineRemovedHandler,
+  timelineAdjustedKafkaHandler,
+} from './timelines';
+import {
+  milestoneAddedHandler,
+  milestoneUpdatedHandler,
+  milestoneRemovedHandler,
+  milestoneUpdatedKafkaHandler,
+} from './milestones';
 
 export const rabbitHandlers = {
   'project.initial': projectCreatedHandler,
@@ -56,4 +66,8 @@ export const kafkaHandlers = {
   [BUS_API_EVENT.TOPIC_UPDATED]: projectUpdatedKafkaHandler,
   [BUS_API_EVENT.POST_CREATED]: projectUpdatedKafkaHandler,
   [BUS_API_EVENT.POST_UPDATED]: projectUpdatedKafkaHandler,
+
+  // Events coming from timeline/milestones (considering it as a separate module/service in future)
+  [BUS_API_EVENT.MILESTONE_TRANSITION_COMPLETED]: milestoneUpdatedKafkaHandler,
+  [BUS_API_EVENT.TIMELINE_ADJUSTED]: timelineAdjustedKafkaHandler,
 };

src/events/milestones/index.js

@@ -3,8 +3,12 @@
  */
 import config from 'config';
 import _ from 'lodash';
+import Joi from 'joi';
 import Promise from 'bluebird';
 import util from '../../util';
+// import { createEvent } from '../../services/busApi';
+import { EVENT, TIMELINE_REFERENCES, MILESTONE_STATUS, REGEX } from '../../constants';
+import models from '../../models';
 
 const ES_TIMELINE_INDEX = config.get('elasticsearchConfig.timelineIndexName');
 const ES_TIMELINE_TYPE = config.get('elasticsearchConfig.timelineDocType');
@@ -154,9 +158,105 @@ const milestoneRemovedHandler = Promise.coroutine(function* (logger, msg, channe
   }
 });
 
+/**
+ * Kafka event handlers
+ */
+
+const payloadSchema = Joi.object().keys({
+  projectId: Joi.number().integer().positive().required(),
+  projectName: Joi.string().optional(),
+  projectUrl: Joi.string().regex(REGEX.URL).optional(),
+  userId: Joi.number().integer().positive().required(),
+  initiatorUserId: Joi.number().integer().positive().required(),
+}).unknown(true).required();
+
+const findProjectPhaseProduct = function (logger, productId, raw = true) { // eslint-disable-line func-names
+  let product;
+  return models.PhaseProduct.findOne({
+    where: { id: productId },
+    raw,
+  }).then((_product) => {
+    logger.debug('_product', _product);
+    if (_product) {
+      product = _product;
+      const phaseId = product.phaseId;
+      const projectId = product.projectId;
+      return Promise.all([
+        models.ProjectPhase.findOne({
+          where: { id: phaseId, projectId },
+          raw,
+        }),
+        models.Project.findOne({
+          where: { id: projectId },
+          raw,
+        }),
+      ]);
+    }
+    return Promise.reject('Unable to find product');
+  }).then((projectAndPhase) => {
+    logger.debug('projectAndPhase', projectAndPhase);
+    if (projectAndPhase) {
+      const phase = projectAndPhase[0];
+      const project = projectAndPhase[1];
+      return Promise.resolve({ product, phase, project });
+    }
+    return Promise.reject('Unable to find phase/project');
+  });
+};
+
+/**
+ * Raises the project plan modified event
+ * @param   {Object}  app       Application object used to interact with RMQ service
+ * @param   {String}  topic     Kafka topic
+ * @param   {Object}  payload   Message payload
+ * @return  {Promise} Promise
+ */
+async function milestoneUpdatedKafkaHandler(app, topic, payload) {
+  app.logger.info(`Handling Kafka event for ${topic}`);
+  // Validate payload
+  const result = Joi.validate(payload, payloadSchema);
+  if (result.error) {
+    throw new Error(result.error);
+  }
+
+  const timeline = payload.timeline;
+  // process only if timeline is related to a product reference
+  if (timeline && timeline.reference === TIMELINE_REFERENCES.PRODUCT) {
+    const productId = timeline.referenceId;
+    const original = payload.originalMilestone;
+    const updated = payload.updatedMilestone;
+    app.logger.debug('Calling findProjectPhaseProduct');
+    const { project, phase } = await findProjectPhaseProduct(app.logger, productId, false);
+    app.logger.debug('Successfully fetched project, phase and product');
+    if (original.status !== updated.status) {
+      if (updated.status === MILESTONE_STATUS.COMPLETED) {
+        app.logger.debug('Found milestone status to be completed');
+        app.logger.debug(`Duration: ${timeline.duration}`);
+        if (!isNaN(timeline.duration) && !isNaN(timeline.progress)) {
+          app.logger.debug(`Current phase progress ${phase.progress} and duration ${phase.duration}`);
+          const updatedPhase = await phase.update({
+            progress: timeline.progress,
+            duration: timeline.duration,
+          }, ['progress', 'duration']);
+          app.logger.debug(`Updated phase progress ${timeline.progress} and duration ${timeline.duration}`);
+          app.logger.debug('Raising node event for PROJECT_PHASE_UPDATED');
+          app.emit(EVENT.ROUTING_KEY.PROJECT_PHASE_UPDATED, {
+            req: {
+              params: { projectId: project.id, phaseId: phase.id },
+              authUser: { userId: payload.userId },
+            },
+            original: phase,
+            updated: _.omit(updatedPhase.toJSON(), 'deletedAt', 'deletedBy'),
+          });
+        }
+      }
+    }
+  }
+}
 
 module.exports = {
   milestoneAddedHandler,
   milestoneRemovedHandler,
   milestoneUpdatedHandler,
+  milestoneUpdatedKafkaHandler,
 };