adding broadcast processor

Author: Sachin Maheshwari

config/default.js

@@ -98,6 +98,10 @@ module.exports = {
         },
       },
     ],
+    'admin.notification.broadcast' : [{
+      handleBulkNotification: {}
+    }
+    ]
     //'notifications.community.challenge.created': ['handleChallengeCreated'],
     //'notifications.community.challenge.phasewarning': ['handleChallengePhaseWarning'],
   },

src/hooks/hookBulkMessage.js

@@ -11,8 +11,12 @@ const logger = require('../common/logger')
 const models = require('../models')
 const logPrefix = "BulkNotificationHook: "
 
-models.BulkMessages.sync()
-models.BulkMessageUserRefs.sync()
+/**
+ * CREATE NEW TABLES IF NOT EXISTS
+ */
+models.BulkMessages.sync().then((t)=> {
+    models.BulkMessageUserRefs.sync()
+})
 
 /**
  * Main function 
@@ -115,7 +119,7 @@ function createNotificationForUser(userId, bulkMessage) {
         type: bulkMessage.type,
         contents: {
             id: bulkMessage.id, /** broadcast message id  */
-            name: bulkMessage.contents, /** broadcast message */
+            message: bulkMessage.message, /** broadcast message */
             group: 'broadcast',
             title: 'Broadcast Message',
         },

src/models/BulkMessages.js

@@ -13,8 +13,9 @@
 module.exports = (sequelize, DataTypes) => sequelize.define('bulk_messages', {
     id: { type: DataTypes.BIGINT, primaryKey: true, autoIncrement: true },
     type: { type: DataTypes.STRING, allowNull: false },
-    contents: { type: DataTypes.JSONB, allowNull: false },
-    recipient_group: { type: DataTypes.STRING, allowNull: false }
+    message: { type: DataTypes.TEXT, allowNull: false },
+    recipients: { type: DataTypes.JSONB, allowNull: false },
+    rules: {type: DataTypes.JSONB, allowNull: true}
   }, {});
 
   // sequelize will generate and manage createdAt, updatedAt fields

src/processors/broadcast/bulkNotificationHandler.js

@@ -0,0 +1,35 @@
+/**
+ * Bulk notification handler.
+ */
+const co = require('co');
+const models = require('../../models');
+const logger = require('../../common/logger')
+
+/**
+ * Handle Kafka JSON message of broadcast.
+ *
+ * @param {Object} message the Kafka JSON message
+ * @param {Object} ruleSets
+ *
+ * @return {Promise} promise resolved to notifications
+ */
+const handle = (message, ruleSets) => co(function* () {
+  return new Promise(function(resolve, reject){
+     models.BulkMessages.create({
+          type: message.topic,
+          message: message.payload.message,
+          recipients: message.payload.recipients,
+          rules: message.payload.rules || null,
+      }).then((bm) => {
+        logger.info("Broadcast message recieved and inserted in db with id:", bm.id)
+        resolve([]) // no notification need to insert at this point 
+     }).catch((e) => {
+         logger.error("Broadcast processor failed in db operation. Error: ", e)
+         reject(e)
+     }) 
+  })
+});
+
+module.exports = {
+  handle,
+};

src/processors/index.js

@@ -8,6 +8,7 @@ const ChallengePhaseWarningHandler = require('./challenge/ChallengePhaseWarningH
 const ChallengeHandler = require('./challenge/ChallengeHandler');
 const AutoPilotHandler = require('./challenge/AutoPilotHandler');
 const SubmissionHandler = require('./challenge/SubmissionHandler');
+const BulkNotificationHandler = require('./broadcast/bulkNotificationHandler');
 
 // Exports
 module.exports = {
@@ -16,4 +17,5 @@ module.exports = {
   handleChallenge: ChallengeHandler.handle,
   handleAutoPilot: AutoPilotHandler.handle,
   handleSubmission: SubmissionHandler.handle,
+  handleBulkNotification: BulkNotificationHandler.handle,
 };