Merge pull request #177 from muttaqin1/Feat-notification-system

Feat:notification-system

backend-app/app.ts

@@ -17,6 +17,7 @@ import cookieParser from 'cookie-parser';
 // import routesVersioning from 'express-routes-versioning';
 // import indexRouter from './routes/index';
 import { RegisterRoutes } from './routes';
+import notification_controller from './controllers/notification_controller';
 
 const app = express();
 
@@ -70,6 +71,8 @@ app.use(handleAPIVersion);
 // handle bearer token
 app.use(bearerToken());
 
+app.get('/streams', notification_controller);
+
 app.get('/', (_req: Request, res: Response) => {
     res.status(200).json({
         message: 'Welcome to the backend app',

backend-app/controllers/notification_controller.ts

@@ -0,0 +1,29 @@
+import { IReq } from '@root/interfaces/vendors';
+import { expressAuthentication } from '@root/middlewares/authentications';
+import TaskEmitter from '@root/utils/TaskEmitter';
+import { NextFunction, Response } from 'express';
+
+export default [
+    async (req: IReq, res: Response, next: NextFunction) => {
+        try {
+            await expressAuthentication(req, 'jwt');
+            next();
+        } catch (err) {
+            res.json(err.message);
+        }
+    },
+    (req: IReq, res: Response) => {
+        TaskEmitter.registerClient(req, res);
+        res.writeHead(200, {
+            'Content-Type': 'text/event-stream',
+            'Cache-Control': 'no-cache',
+            'Content-Encoding': 'none',
+            Connection: 'keep-alive',
+        });
+        TaskEmitter.listenIncomingNotification();
+        res.on('close', () => {
+            TaskEmitter.removeConnectedClient(String(req.user._id));
+            res.end();
+        });
+    },
+];

backend-app/middlewares/api_version_controll.ts

@@ -19,12 +19,16 @@ const handleAPIVersion = (req: Request, res: Response, next: NextFunction) => {
     if (
         req.url.includes('/docs') ||
         req.url.includes('/docs-json') ||
-        req.url.includes(`/${req.headers['api-version']}`)
+        req.url.includes(`/${req.headers['api-version']}`) ||
+        req.url.includes('/streams')
     )
         return next();
 
     // validate the API version
-    if (!/^v[0-9]+$/.test(String(req.headers['api-version'])) || req.headers['api-version'] > API_VERSION) {
+    if (
+        !/^v[0-9]+$/.test(String(req.headers['api-version'])) ||
+        req.headers['api-version'] > API_VERSION
+    ) {
         throw new AppError(400, 'Invalid API version');
     }
 

backend-app/models/notifications/notification_model.ts

@@ -2,28 +2,28 @@ import mongoose, { Document, Model, Schema } from 'mongoose';
 import validator from 'validator';
 import metaData from '@constants/meta_data';
 
-enum SendThrough {
-    SMS = "sms",
-    Gmail = "gmail",
-    Notification = "notification"
+export enum SendThrough {
+    SMS = 'sms',
+    Gmail = 'gmail',
+    Notification = 'notification',
 }
-enum CategoryType{
-    Message = "message",
-    Alert = "alert",
-    Reminder = "reminder",
-    Other = "other"
+export enum CategoryType {
+    Message = 'message',
+    Alert = 'alert',
+    Reminder = 'reminder',
+    Other = 'other',
 }
 
-interface INotification extends Document {
+export interface INotification extends Document {
     id: string;
     sentDate: Date;
     receivedDate: Date;
     seen: boolean;
     content: string;
-    category: CategoryType;
+    category: `${CategoryType}`;
     sender: string | mongoose.Types.ObjectId;
     receiver: mongoose.Types.ObjectId;
-    sendThrough: SendThrough;
+    sendThrough: `${SendThrough}`;
 }
 
 const notificationSchema: Schema = new mongoose.Schema<INotification>(
@@ -103,7 +103,7 @@ const notificationSchema: Schema = new mongoose.Schema<INotification>(
 );
 
 // add meta data to the schema
-metaData.apply(notificationSchema)
+metaData.apply(notificationSchema);
 
 notificationSchema.pre('find', function () {
     this.where({ deleted: false });
@@ -113,6 +113,8 @@ notificationSchema.pre('findOne', function () {
     this.where({ deleted: false });
 });
 
-const Notification: Model<INotification> = mongoose.model<INotification>("Notification", notificationSchema);
+const Notification: Model<INotification> = mongoose.model<INotification>(
+    'Notification',
+    notificationSchema
+);
 export default Notification;
-

backend-app/server.ts

@@ -5,7 +5,7 @@ import fs from 'fs';
 import { DATABASE, PORT } from './config/app_config';
 import createRoles from './utils/authorization/roles/create_roles';
 import createDefaultUser from './utils/create_default_user';
-
+import TaskEmitter from './utils/TaskEmitter';
 process.on('uncaughtException', (err) => {
     logger.error('UNCAUGHT EXCEPTION!!!  shutting down ...');
     logger.error(`${err}, ${err.message}, ${err.stack}`);
@@ -14,6 +14,16 @@ process.on('uncaughtException', (err) => {
 
 import app from './app';
 
+// let cnt = 0;
+// setInterval(() => {
+//     TaskEmitter.emitNotification({
+//         category: 'alert',
+//         sender: '6532ae0cf63bfdb633eb5f2b',
+//         receiver: '668e42b6b8833839371fd0d1' as unknown as any,
+//         content: `My name is Muhammad Muttaqin ${cnt++}`,
+//     });
+// }, 1000 * 40);
+
 mongoose.set('strictQuery', true);
 
 let expServer: Promise<import('http').Server>;

backend-app/utils/TaskEmitter.ts

@@ -0,0 +1,64 @@
+import { IReq } from '@root/interfaces/vendors';
+import { EventEmitter } from 'events';
+import { Response } from 'express';
+import NotificationHandler, {
+    NotificationInput,
+} from './notification/notification';
+import {
+    INotification,
+    SendThrough,
+} from '@root/models/notifications/notification_model';
+import Logger from '@utils/logger';
+
+class TaskEmitter extends EventEmitter {
+    public connectedClients: Map<string, Response> = new Map();
+    constructor() {
+        super();
+    }
+
+    public emitEvent<T>(event: string, data?: T): void {
+        this.emit(event, data);
+    }
+    public registerClient(req: IReq, res: Response): void {
+        let userId = String(req.user._id);
+        if (this.connectedClients.has(userId)) {
+            this.connectedClients.delete(userId);
+            this.connectedClients.set(userId, res);
+        } else {
+            this.connectedClients.set(userId, res);
+        }
+    }
+    public listenIncomingNotification(): void {
+        this.on('live-notification', (notification: INotification) => {
+            if (this.connectedClients.has(notification.receiver.toString())) {
+                const resObj = this.connectedClients.get(
+                    notification.receiver.toString()
+                );
+                const jsonNotificationString = JSON.stringify(notification);
+                resObj.write(`data: ${jsonNotificationString}\n\n`);
+            }
+        });
+    }
+    public async emitNotification(
+        payload: Omit<NotificationInput, 'sendThrough'>
+    ): Promise<void> {
+        try {
+            if (payload.sender.toString() === payload.receiver.toString())
+                return;
+            const notificationInstance =
+                await NotificationHandler.createNotification({
+                    ...payload,
+                    sendThrough: SendThrough.Notification,
+                });
+            this.emitEvent('live-notification', notificationInstance);
+        } catch (err) {
+            Logger.error('Failed to send Notification.');
+            Logger.error(err);
+        }
+    }
+    public removeConnectedClient(userId: string) {
+        this.connectedClients.delete(userId);
+    }
+}
+
+export default new TaskEmitter();

backend-app/utils/notification/notification.ts

@@ -2,14 +2,14 @@ import Notification, {
     INotification,
 } from '@root/models/notifications/notification_model';
 import logger from '../logger';
-type notificationInput = Pick<
+export type NotificationInput = Pick<
     INotification,
     'content' | 'sender' | 'receiver' | 'category' | 'sendThrough'
 >;
 // type notificationOutput = Required<INotification>;
 export default class NotificationHandler {
     public static async createNotification(
-        data: notificationInput
+        data: NotificationInput
     ): Promise<INotification> {
         const { content, sender, receiver, category, sendThrough } = data;
 
@@ -33,9 +33,6 @@ export default class NotificationHandler {
             return null;
         }
     }
-    // public static async sendNotification(notification: notificationOutput) {
-    //     //send with SSE
-    // }
     public static async seenNotification(id: string) {
         try {
             await Notification.findOneAndUpdate({ id }, { seen: true });
@@ -61,4 +58,3 @@ export default class NotificationHandler {
         return notifications;
     }
 }
-