Issue #SB-9183 feat: Single message implemenation"

app/app.ts

@@ -1,12 +1,17 @@
 import * as express from "express";
 import * as bodyParser from "body-parser";
+import { AppConfig } from "./config"
 import { Routes } from "./routes";
-import { CassandraAdapter } from "./modules/shared";
+import { CassandraAdapter, KafkaProducerAdapter, KafkaConsumerAdapter } from "./modules/shared";
+import { MessageProcessService } from "./modules/notification-delivery-manager";
 class App {
 
     public app: express.Application;
     public router = express.Router();
     public cassandraAdapter: CassandraAdapter;
+    public kafkaProducerAdapter: KafkaProducerAdapter;
+    public kafkaConsumerAdapter: KafkaConsumerAdapter;
+    public messageProcessService:MessageProcessService;
     constructor() {
         this.app = express();
         this.config();
@@ -20,6 +25,12 @@ class App {
         this.router = Routes.configure();
         this.app.use("/api/v1", this.router);
         this.cassandraAdapter = CassandraAdapter.connect();
+        this.kafkaProducerAdapter = KafkaProducerAdapter.connect();
+        this.messageProcessService = new MessageProcessService();
+        this.kafkaConsumerAdapter = new KafkaConsumerAdapter(AppConfig.KAFKA_TOPICS.split(','), (data) => {
+            console.log("kafka consumer data",data);
+            this.messageProcessService.processMessage(JSON.parse(data.value.toString()));
+        })
 
     }
 

app/config/appConfig.ts

@@ -1,5 +1,10 @@
 
 const env = process.env
 export abstract class AppConfig {    
-    public static PORT = env.PORT || 3000    
+    public static PORT = env.sunbird_notification_service_port || 3000    
+    public static KAFKA_TOPICS = env.sunbird_notification_service_kafka_topics || 'normal,immediate'
+    public static SMTP_CONFIG = {
+        USERNAME: env.sunbird_notification_service_smtp_username,
+        PASSWORD: env.sunbird_notification_service_smtp_password
+    }
 }

app/config/appConstants.ts

@@ -0,0 +1,17 @@
+export abstract class AppConstants {
+    public static API_PREFIX = 'sunbird.notification.';
+    public static API_VERSION = 'v1';
+    public static API_IDS = {
+        CREATE_NOTIFICATION: 'create'
+    }
+    public static RESPONSE_CODES = {
+        SERVER_ERROR: 'SERVER_ERROR',
+        CLIENT_ERROR: 'CLIENT_ERROR',
+        BAD_REQUEST: 'BAD_REQUEST',
+        OK:'OK'
+    }
+    public static ERROR_CODES = {
+        REQUIRED_PARAMS_MISSING : 'REQUIRED_PARAMS_MISSING',
+        SOMETHING_WENT_WRONG:'SOMETHING_WENT_WRONG'
+    }
+}

app/config/index.ts

@@ -1 +1,2 @@
 export { AppConfig } from "./appConfig";
+export { AppConstants } from "./appConstants";

app/modules/notification-delivery-manager/adapters/email-adapter.ts

@@ -1,12 +1,34 @@
-export class EmailAdapter{
-
-    constructor(){
+import * as nodemailer from 'nodemailer';
+
+import { AppConfig } from '../../../config';
+export class EmailAdapter {
+
+    constructor() {
 
     }
 
-    public sendMessage(){
+    public sendMessage(messageData) {
         return new Promise((resolve, reject) => {
-
+            let transporter = nodemailer.createTransport({
+                service: 'gmail',
+                auth: {
+                    user: AppConfig.SMTP_CONFIG.USERNAME,
+                    pass: AppConfig.SMTP_CONFIG.PASSWORD
+                }
+            });
+            const mailOptions = {
+                from: AppConfig.SMTP_CONFIG.USERNAME, // sender address
+                to: messageData.to, // list of receivers
+                subject: messageData.subject, // Subject line
+                html: messageData.message// plain text body
+            };
+            transporter.sendMail(mailOptions, function (err, info) {
+                if (err)
+                    console.log(err)
+                else
+                    console.log(info);
+            });
+            resolve(true)
         });
     }
 }

app/modules/notification-delivery-manager/index.ts

@@ -0,0 +1 @@
+export * from "./service/message-process-service"

app/modules/notification-delivery-manager/service/email-delivery-service.ts

@@ -1,5 +1,6 @@
 import { DeliveryService } from "./delivery-service";
 import { EmailAdapter } from "../adapters/email-adapter";
+import { parse, Compile } from 'velocityjs';
 export class EmailDeliveryService extends DeliveryService {
     public emailAdapter: EmailAdapter;
     public emailData: Object;
@@ -10,12 +11,19 @@ export class EmailDeliveryService extends DeliveryService {
     }
     public sendEmail() {
         let message = this.getComposedMessage(this.emailData);
-        //super class method
-        return this.deliverMessage(message, this.emailAdapter);
+        let email = {
+            to: this.emailData['to'],
+            subject: this.emailData['subject'],
+            message: message
+        }
+        return this.emailAdapter.sendMessage(email);
     }
     private getComposedMessage(emailData) {
+        const asts = parse(emailData.templateText);
+        const data = JSON.parse(emailData.templateSubData);
+        const msg = (new Compile(asts)).render(data);
         // use velocity js to substitute appropriate details and return composed message object
-        return {};
+        return msg;
     }
 
 }

app/modules/notification-delivery-manager/service/message-process-service.ts

@@ -1,54 +1,45 @@
 import { EmailDeliveryService } from "./email-delivery-service";
+import { CassandraAdapter } from "../../shared";
 export class MessageProcessService {
     public emailDeliveryService: EmailDeliveryService;
     constructor() {
 
     }
     public processMessage(msgData) {
 
-        let receipients = this.getReceipients(msgData.receipientRefIds,msgData.receipientRefType, "email");
-        // if the receipient is single send
-        if (receipients.length == 1) {
-            this.emailDeliveryService = new EmailDeliveryService({});
-            this.emailDeliveryService.sendEmail().then((result) => {
-                // based on success status upadate message delivery status to db
-            }, (error) => {
-                // if error add the message back to queue with retries count
-            });
-        } else {
-            this.processGroupMessages();
+        let models = CassandraAdapter.connect()
+        if (models.instance.Messages) {
+            models.instance.Messages.findOne({ id: models.uuidFromString(msgData.messageId) }, { raw: true, allow_filtering: true }, (err, message) => {
+                if (err) {
+                }
+                switch (message.broadcast_type) {
+                    case "email":
+                        let reciepients = this.getReceipients(message.recipient_refid, message.recipientRefType, "email")
+                        models.instance.Templates.findOne({ name: message.template_name }, { raw: true, allow_filtering: true }, (err, templateData) => {
+                            if (err) {
+
+                            }
+                            let emailData = {
+                                subject: "Batch notification update",
+                                templateSubData: message.message_data,
+                                templateText: templateData.template,
+                                to: reciepients
+                            }
+                            this.emailDeliveryService = new EmailDeliveryService(emailData)
+                            this.emailDeliveryService.sendEmail().then((result)=>{
+                             console.log("result",result)   
+                            })
+                        })
+                        break
+                }
+            })
         }
-
-    }
-    private getReceipients(refIds:Array<String>,refType: String, fieldToFetch: String): Array<String> {
-        let receipients = [];
-        // based on type get associated users from the provided reference ids and return list of user emails or mobile numbers
-        switch (refType) {
-            case 'BATCH': {
-
-            }
-                break;
-            case 'USER': {
-
-            }
-                break;
+        else {
+            console.log("Cassandra models not initialised.Try again !!!")
         }
-        return receipients;
-    }
-    private processGroupMessages() {
-        // fanout group messages as individual and push back to queue and database
-    }
 
-    private saveMessageToDB(msg) {
-        // use 'CassandraAdapter' to insert message details into DB
     }
-
-    private addMessageToQueue(msg) {
-        // use 'KafkaProducerAdapter' to insert message details into DB
+    private getReceipients(refIds: String, refType: String, fieldToFetch: String) {
+        return "[email protected]";
     }
-
-    private updateDeliveryStatus(msg) {
-        // use 'CassandraAdapter' to update message delivery status in DB
-    }
-
 }

app/modules/notification-scheduler/middleware/request-middleware.ts

@@ -1,11 +1,39 @@
 import * as express from "express";
+import * as Joi from "joi";
+import * as _ from "lodash";
+import * as HttpStatus from "http-status-codes";
+import { ResponseUtil } from '../../shared';
+import { AppConstants } from '../../../config';
 export class RequestMiddleware {
-    constructor(){
+    constructor() {
 
     }
-    public static validateRequest(req:express.Request,res:express.Response,next:express.NextFunction){
+    public static validateRequest(req: express.Request, res: express.Response, next: express.NextFunction) {
         // based on diferent request write validation rules here
         next();
-
+
+    }
+
+    public static validateCreateNotificationRequest(req: express.Request, res: express.Response, next: express.NextFunction) {        
+        const requestSchema = Joi.object().keys({
+            request: Joi.object().keys({
+                broadcastType: Joi.string().valid('email').required(),
+                messageData:Joi.object().required(),
+                messageType: Joi.string().valid('immediate','normal').required(),
+                recipientRefId: Joi.string().required(),
+                recipientRefType: Joi.string().valid('individual','batch').required(),
+                templateName: Joi.string().required()
+            }).required()
+        })
+        const result = Joi.validate(req.body, requestSchema)
+        if (result.error && !_.isEmpty(result.error.message)) {
+            let err = result.error.message
+            let responseUtil = new ResponseUtil(AppConstants.API_IDS.CREATE_NOTIFICATION);
+            res.status(HttpStatus.BAD_REQUEST);
+            res.send(responseUtil.prepareErrorResponse(AppConstants.RESPONSE_CODES.CLIENT_ERROR,
+                AppConstants.ERROR_CODES.REQUIRED_PARAMS_MISSING, err))
+        } else {
+            next();
+        }
     }
 }

app/modules/notification-scheduler/scheduler-routes.ts

@@ -1,5 +1,7 @@
 import * as express from "express";
 import {SchedulerService} from "./service/schedule-service";
 import {RequestMiddleware} from "./middleware/request-middleware";
+let schedulerServiceInstance = new SchedulerService()
 export let SchedulerRouter = express.Router({mergeParams: true});
-SchedulerRouter.get('/hello',RequestMiddleware.validateRequest,SchedulerService.helloWorld);
+SchedulerRouter.get('/hello',RequestMiddleware.validateRequest,schedulerServiceInstance.helloWorld);
+SchedulerRouter.post('/create',RequestMiddleware.validateCreateNotificationRequest,schedulerServiceInstance.createNotification);

app/modules/notification-scheduler/service/schedule-service.ts

@@ -1,12 +1,88 @@
 import * as express from "express";
+import * as moment from "moment";
+import { CassandraAdapter, ResponseUtil, KafkaProducerAdapter } from "../../shared";
+import * as HttpStatus from "http-status-codes";
+import { AppConstants } from '../../../config';
 export class SchedulerService {
-    public static helloWorld(req:express.Request,res:express.Response):void{
-      res.send({status:"success",message:"hello world"});
-    }
 
-    public static createNotification(req:express.Request,res:express.Response):void{
-      // first save the notification details to cassandra and then push to notification queue and return response
-    }
+  constructor() {
+
+  }
+
+  public saveNotificationToDB(reqObj) {
+
+
+  }
+
+
+  public helloWorld(req: express.Request, res: express.Response): void {
+    res.send({ status: "success", message: "hello world" });
+  }
+
+  public createNotification(req: express.Request, res: express.Response): void {
+
+    let reqObj = req.body.request;
+
+    // validate the template data here
+
+    // store to cassandra
+    let models = CassandraAdapter.connect()
+    let currentTimestamp = moment().valueOf()
+    let message = new models.instance.Messages({
+      broadcast_type: reqObj.broadcastType,
+      created_on: currentTimestamp,
+      message_data: JSON.stringify(reqObj.messageData),
+      message_type: reqObj.messageType,
+      recipient_refid: reqObj.recipientRefId,
+      recipient_reftype: reqObj.recipientRefType,
+      status: 'pending',
+      template_name: reqObj.templateName,
+      updated_on: currentTimestamp
+    });
+    message.save((err) => {
+      if (err) {
+        let responseUtil = new ResponseUtil(AppConstants.API_IDS.CREATE_NOTIFICATION);
+        res.status(HttpStatus.INTERNAL_SERVER_ERROR);
+        res.send(responseUtil.prepareErrorResponse(AppConstants.RESPONSE_CODES.SERVER_ERROR,
+          AppConstants.RESPONSE_CODES.SERVER_ERROR, err))
+      } else {
+
+        models.instance.Messages.findOne({
+          recipient_refid: message.recipient_refid,
+          recipient_reftype: message.recipient_reftype, status: message.status, created_on: currentTimestamp
+        }, { raw: true, allow_filtering: true },  (err, messageInfo) => {
+          if (err) {
+            let responseUtil = new ResponseUtil(AppConstants.API_IDS.CREATE_NOTIFICATION);
+            res.status(HttpStatus.INTERNAL_SERVER_ERROR);
+            res.send(responseUtil.prepareErrorResponse(AppConstants.RESPONSE_CODES.SERVER_ERROR,
+              AppConstants.RESPONSE_CODES.SERVER_ERROR, err))
+          }
+          let kafkaProducerAdapter = KafkaProducerAdapter.connect()
+          KafkaProducerAdapter.pushMessageToBroker(kafkaProducerAdapter, {
+            topic: messageInfo.message_type,
+            message: {
+              messageId: messageInfo.id
+            }
+          }, (err, status) => {
+            if (err) {
+              let responseUtil = new ResponseUtil(AppConstants.API_IDS.CREATE_NOTIFICATION);
+              res.status(HttpStatus.INTERNAL_SERVER_ERROR);
+              res.send(responseUtil.prepareErrorResponse(AppConstants.RESPONSE_CODES.SERVER_ERROR,
+                AppConstants.RESPONSE_CODES.SERVER_ERROR, err))
+            } else {
+              let responseUtil = new ResponseUtil(AppConstants.API_IDS.CREATE_NOTIFICATION);
+              res.status(HttpStatus.OK);
+              res.send(responseUtil.prepareSuccessResponse(AppConstants.RESPONSE_CODES.OK, { id: messageInfo.id }))
+            }
+
+          })
+
+
+        });
+      }
+    })
+  }
+
 
 
 }