more event related fixes

Author: Parth Shah

src/constants.js

@@ -39,7 +39,8 @@ export const EVENT = {
     PROJECT_LAUNCHED: 'project.launched',
     PROJECT_UPDATED: 'project.updated',
     PROJECT_CANCELLED: 'project.cancelled',
-    PROJECT_COMPLETED: 'project.completed'
+    PROJECT_COMPLETED: 'project.completed',
+    PROJECT_DELETED: 'project.deleted'
   }
 }
 EVENT.INTERNAL = _.mapValues(EVENT.ROUTING_KEY, (a) => { return `internal.${a}` })

src/events/projectMembers/index.js

@@ -16,16 +16,18 @@ module.exports = (app, logger) => {
 
   // Publish messages to the queue
   _.map(internalEvents, (evt) => {
-    app.on(evt, (member) => {
+    app.on(evt, ({payload, props}) => {
       logger.debug('handling ', evt)
       let key = evt.substring(evt.indexOf('.') + 1)
-      return app.services.pubsub.publish(key, member)
+      return app.services.pubsub.publish(key, payload, props)
     })
   })
 
 
   // EXTERNAL events
   app.on(EVENT.EXTERNAL.PROJECT_MEMBER_ADDED, (msg, next) => {
+    const origRequestId = msg.properties.correlationId
+    logger = logger.child({requestId: origRequestId})
     let newMember = JSON.parse(msg.content.toString())
     logger.debug(`received msg '${EVENT.EXTERNAL.PROJECT_MEMBER_ADDED}'`, newMember)
 
@@ -38,13 +40,16 @@ module.exports = (app, logger) => {
             return util.getSystemUserToken(logger)
               .then(token => {
                 const req = {
-                  id: 1,
+                  id: origRequestId,
                   log: logger,
                   headers: { authorization: `Bearer ${token}` }
                 }
                 return  directProject.addCopilot(req, directProjectId, {
                   copilotUserId: newMember.userId
                 })
+                  .then((resp) => {
+                    next()
+                  })
               })
               .catch(err => {
                 logger.error('Error caught while adding co-pilot from direct', err)
@@ -62,7 +67,9 @@ module.exports = (app, logger) => {
   })
 
   app.on(EVENT.EXTERNAL.PROJECT_MEMBER_REMOVED, (msg, next) => {
+    const origRequestId = msg.properties.correlationId
     const member = JSON.parse(msg.content.toString())
+    logger = logger.child({requestId: origRequestId})
     logger.debug(`received msg '${EVENT.EXTERNAL.PROJECT_MEMBER_REMOVED}'`, member)
 
     if (member.role === PROJECT_MEMBER_ROLE.COPILOT) {
@@ -74,6 +81,7 @@ module.exports = (app, logger) => {
             return util.getSystemUserToken(logger)
               .then(token => {
                 const req = {
+                  id: origRequestId,
                   log: logger,
                   headers: { authorization: `Bearer ${token}` }
                 }

src/events/projects/index.js

@@ -11,19 +11,22 @@ module.exports = (app, logger) => {
     EVENT.INTERNAL.PROJECT_LAUNCHED,
     EVENT.INTERNAL.PROJECT_UPDATED,
     EVENT.INTERNAL.PROJECT_CANCELLED,
-    EVENT.INTERNAL.PROJECT_COMPLETED
+    EVENT.INTERNAL.PROJECT_COMPLETED,
+    EVENT.INTERNAL.PROJECT_DELETED
   ]
 
   // Publish messages to the queue
   _.map(internalEvents, (evt) => {
-    app.on(evt, (project) => {
+    app.on(evt, ({payload, props}) => {
       logger.debug('handling ' + evt)
       let key = evt.substring(evt.indexOf('.') + 1)
-      return app.services.pubsub.publish(key, project)
+      return app.services.pubsub.publish(key, payload, props)
     })
   })
 
   app.on(EVENT.EXTERNAL.PROJECT_DRAFT_CREATED, (msg, next) => {
+    // log with correlationId (request Id)
+    logger = logger.child({requestId: msg.properties.correlationId})
     let project = JSON.parse(msg.content.toString())
     logger.debug('received msg \'project.draft-created\'', project.id)
 

src/routes/projectMembers/create.js

@@ -64,7 +64,10 @@ module.exports = [
       .then(_newMember => {
         newMember = _newMember.get({plain: true})
         // fire event
-        req.app.emit(EVENT.INTERNAL.PROJECT_MEMBER_ADDED, newMember)
+        req.app.emit(EVENT.INTERNAL.PROJECT_MEMBER_ADDED, {
+          payload: newMember,
+          props: { correlationId: req.id }
+        })
         res.status(201).json(util.wrapResponse(req.id, newMember))
       })
       .catch((err) => {

src/routes/projectMembers/delete.js

@@ -36,7 +36,10 @@ module.exports = [
           })
           .then(member => {
             // fire event
-            req.app.emit(EVENT.INTERNAL.PROJECT_MEMBER_REMOVED, member.get({plain:true}))
+            req.app.emit(EVENT.INTERNAL.PROJECT_MEMBER_REMOVED, {
+              payload: member.get({plain:true}),
+              props: { correlationId: req.id }
+            })
             res.status(204).json({})
           })
           .catch(err => next(err))

src/routes/projectMembers/update.js

@@ -81,6 +81,7 @@ module.exports = [
           return Promise.all(operations)
         })
         .then(() => {
+          // TODO move this to an event
           // if copilot role is added or removed should invoke related direct project service
           if(previousValue.role !== newValue.role && (previousValue.role === PROJECT_MEMBER_ROLE.COPILOT
               || newValue.role === PROJECT_MEMBER_ROLE.COPILOT)) {

src/routes/projects/create.js

@@ -125,7 +125,10 @@ module.exports = [
             newProject = _.omit(newProject, ['deletedAt', 'utm'])
             // add an empty attachments array
             newProject.attachments = []
-            req.app.emit(EVENT.INTERNAL.PROJECT_DRAFT_CREATED, newProject)
+            req.app.emit(EVENT.INTERNAL.PROJECT_DRAFT_CREATED, {
+              payload: newProject,
+              props: { correlationId: req.id }
+            })
             res.status(201).json(util.wrapResponse(req.id, newProject))
           })
           .catch((err) => {

src/routes/projects/delete.js

@@ -32,6 +32,10 @@ module.exports = [
             err.status = 404
             next(err)
           } else {
+            req.app.emit(EVENT.INTERNAL.PROJECT_DELETED, {
+              payload: { id: projectId },
+              props: { correlationId: req.id }
+            })
             res.status(204).json({})
           }
         })

src/routes/projects/update.js

@@ -165,8 +165,8 @@ module.exports = [
           previousValue =  _.omit(previousValue, ['deletedAt'])
           // emit original and updated project information
           req.app.emit(EVENT.INTERNAL.PROJECT_UPDATED, {
-            original: previousValue,
-            updated: project
+            payload: { original: previousValue, updated: project },
+            props: { correlationId: req.id }
           })
           // check context for project members
           project.members = req.context.currentProjectMembers

src/services/directProject.js

@@ -18,7 +18,7 @@ function _getHttpClient(req){
     httpClient.defaults.baseURL = config.get('directProjectServiceEndpoint')
     httpClient.defaults.timeout = 4000
     httpClient.interceptors.response.use((resp) => {
-        req.log.debug('resp: ', JSON.stringify(resp.data, null, 2))
+        // req.log.debug('resp: ', JSON.stringify(resp.data, null, 2))
         if (resp.status !== 200 || resp.data.result.status !== 200) {
           req.log.error('error resp: ', JSON.stringify(resp.data, null, 2))
           return Promise.reject(new Error(resp.data.result.content.message))

src/services/rabbitmq.js

@@ -138,7 +138,7 @@ module.exports = class RabbitMQService extends EventEmitter{
    * @param  {string} key     routing key
    * @param  {object} payload message payload
    */
-  publish(key, payload) {
+  publish(key, payload, props={}) {
     var channel = null
     var self = this
       // first create a channel - this is a lightweight connection
@@ -151,9 +151,12 @@ module.exports = class RabbitMQService extends EventEmitter{
         })
       }).then(() => {
         // publish the message
-        channel.publish(self.exchangeName, key,
+        props = _.defaults(props, { contentType: 'application/json'})
+        channel.publish(
+          self.exchangeName,
+          key,
           new Buffer(JSON.stringify(payload)),
-          { contentType: 'application/json'}
+          props
         )
         self.logger.debug('Sent %s: %s', self.exchangeName, payload)
         return channel.close()