committed

services/analytics.js

@@ -0,0 +1,44 @@
+const express = require('express');
+const router = express.Router();
+const Notification = require('../models/notification');
+
+router.get('/analytics', async (req, res) => {
+  try {
+    const totalSent = await Notification.countDocuments({ status: 'sent' });
+    const totalFailed = await Notification.countDocuments({ status: 'failed' });
+    const totalRetried = await Notification.countDocuments({ status: 'retried' });
+
+    const deliveryStats = {
+      totalSent,
+      totalFailed,
+      totalRetried
+    };
+
+    
+    const sentNotifications = await Notification.find({ status: 'sent' });
+    const totalDeliveryTime = sentNotifications.reduce((sum, notification) => {
+      return sum + (notification.sentAt - notification.createdAt);
+    }, 0);
+    const averageDeliveryTime = totalDeliveryTime / sentNotifications.length;
+
+    
+    const totalResponses = await Notification.countDocuments({ userResponded: true });
+    const responseRate = totalResponses / totalSent;
+
+    const userEngagement = {
+      averageDeliveryTime,
+      responseRate
+    };
+
+    res.json({
+      deliveryStats,
+      userEngagement
+    });
+  } catch (error) {
+    console.error('Error generating analytics:', error);
+    res.status(500).json({ error: 'Internal server error' });
+  }
+});
+
+module.exports = router;
+

services/notificationDelivery.js

@@ -0,0 +1,106 @@
+const User = require('../models/user');
+const Notification = require('../models/notification');
+const { checkThrottling } = require('../queries/throttling');
+const { isInQuietHours } = require('../queries/quietHours');
+const { checkDuplication } = require('../queries/deduplication');
+const { aggregateLowPriorityNotifications } = require('../queries/aggregation');
+const { getUrgentAlerts } = require('../queries/urgentAlerts');
+const { MongoClient } = require('mongodb');
+const mongoClient = new MongoClient('mongodb://localhost:27017');
+
+async function deliverNotifications() {
+  
+  const urgentAlerts = await getUrgentAlerts();
+  for (const alert of urgentAlerts) {
+    await processNotification(alert);
+  }
+
+  const pendingNotifications = await Notification.find({ status: 'pending', priority: { $ne: 'high' } });
+  for (const notification of pendingNotifications) {
+    await processNotification(notification);
+  }
+}
+
+async function processNotification(notification) {
+  await mongoClient.connect();
+  const user = await User.findOne({ userId: notification.userId });
+  if (!user) {
+    console.error(`User not found for notification: ${notification._id}`);
+    return;
+  }
+
+  if (!(await checkThrottling(user.userId))) {
+    await rescheduleNotification(notification);
+    return;
+  }
+
+  if (isInQuietHours(user, new Date())) {
+    await rescheduleNotification(notification);
+    return;
+  }
+
+  if (await checkDuplication(notification)) {
+    
+    const db = mongoClient.db('test');
+    await Notification.updateOne({ _id: notification._id }, { status: 'suppressed' });
+        // await db.collection('scheduledNotifications').deleteOne({ _id: notification._id });
+    return;
+  }
+
+  if (notification.priority === 'low') {
+    const aggregatedNotification = await aggregateLowPriorityNotifications(user.userId);
+    if (aggregatedNotification) {
+      await sendNotification(aggregatedNotification, user);
+      return;
+    }
+  }
+
+  await sendNotification(notification, user);
+}
+
+async function sendNotification(notification, user) {
+  for (const channel of user.preferences.channels) {
+    try {
+      switch (channel) {
+        case 'email':
+          await sendEmail(notification, user);
+          break;
+        case 'sms':
+          await sendSMS(notification, user);
+          break;
+        case 'push':
+          await sendPushNotification(notification, user);
+          break;
+      }
+      await Notification.updateOne({ _id: notification._id }, { status: 'sent' });
+      return;
+    } catch (error) {
+      console.error(`Failed to send notification via ${channel}:`, error);
+    }
+  }
+  
+  await Notification.updateOne({ _id: notification._id }, { status: 'failed' });
+}
+
+async function rescheduleNotification(notification) {
+  const newSendTime = new Date(Date.now() + 60 * 60 * 1000); // Reschedule for 1 hour later
+  const db = mongoClient.db('test');
+  await Notification.updateOne({ _id: notification._id }, { sendTime: newSendTime });
+  await db.collection('scheduledNotifications').updateOne({ _id: notification._id }, { sendTime: newSendTime });
+}
+
+// Mock delivery functions
+async function sendEmail(notification, user) {
+  console.log(`Sending email to ${user.userId}: ${notification.message}`);
+}
+
+async function sendSMS(notification, user) {
+  console.log(`Sending SMS to ${user.userId}: ${notification.message}`);
+}
+
+async function sendPushNotification(notification, user) {
+  console.log(`Sending push notification to ${user.userId}: ${notification.message}`);
+}
+
+module.exports = { deliverNotifications,sendNotification };
+

services/notificationIngestion.js

@@ -0,0 +1,50 @@
+const express = require('express');
+const { Kafka } = require('kafkajs');
+const router = express.Router();
+const Notification = require('../models/notification');
+const { MongoClient } = require('mongodb');
+const kafka = new Kafka({ clientId: 'notification-app', brokers: ['localhost:9092'] });
+const producer = kafka.producer();
+const mongoClient = new MongoClient('mongodb://localhost:27017');
+
+router.post('/notify', async (req, res) => {
+  try {
+    const { message, userId, priority, sendTime } = req.body;
+    
+    if (!message || !userId) {
+      return res.status(400).json({ error: 'Missing required fields' });
+    }
+
+    const notification = new Notification({
+      message,
+      userId,
+      priority: priority || 'normal',
+      sendTime: sendTime,
+      createdAt: new Date(),
+      status: 'pending'
+    });
+
+    await notification.save(); 
+    // await mongoClient.connect();
+    // const db = mongoClient.db('notifications');
+    // await db.collection('notification').insertOne(notification);
+  
+
+    // Publish to Kafka
+    await producer.connect();
+    await producer.send({
+      topic: 'notifications',
+      messages: [{ value: JSON.stringify(notification) }],
+    });
+
+    res.status(200).json({ message: 'Notification received' });
+  } catch (error) {
+    console.error('Error processing notification:', error);
+    res.status(500).json({ error: 'Internal server error' });
+  } finally {
+    await producer.disconnect();
+  }
+});
+
+module.exports = router;
+

services/notificationProcessing.js

@@ -0,0 +1,110 @@
+const { Kafka } = require('kafkajs');
+const { MongoClient } = require('mongodb');
+const User = require('../models/user');
+const Notification = require('../models/notification');
+const schedule = require('node-schedule');
+// const {sendNotification} = require('./notificationDelivery')
+const { checkDuplication } = require('../queries/deduplication');
+const kafka = new Kafka({ clientId: 'notification-processor', brokers: ['localhost:9092'] });
+const consumer = kafka.consumer({ groupId: 'notification-processor-group' });
+const mongoClient = new MongoClient('mongodb://localhost:27017');
+
+async function processNotifications() {
+  await consumer.connect();
+  await consumer.subscribe({ topic: 'notifications', fromBeginning: true });
+
+  await consumer.run({
+    eachMessage: async ({ message }) => {
+      const notification = JSON.parse(message.value.toString());
+      if (notification.priority == 'high' || !notification.sendTime || new Date(notification.sendTime) <= new Date() )
+{
+  //
+        // Process immediately
+        // if (await checkDuplication(notification)) {
+        //   await Notification.updateOne({ _id: notification._id }, { status: 'suppressed' });
+        //   return;
+        // }
+        // const user = await User.findOne({ userId: notification.userId });
+        await sendNotification(notification);
+    
+      } else {
+        // Schedule for later
+        
+        await storeNotification(notification);
+      }
+    },
+  });
+}
+
+async function storeNotification(notification) {
+  await mongoClient.connect();
+  const db = mongoClient.db('test');
+  await db.collection('scheduledNotifications').insertOne({
+    ...notification,
+    sendTime: new Date(notification.sendTime), 
+    createdAt:new Date(notification.createdAt)
+  });
+}
+
+async function checkScheduledNotifications() {
+  const db = mongoClient.db('test');
+  const now = new Date();
+  const notifications = await db.collection('scheduledNotifications')
+    .find({ sendTime: { $lte: now } })
+    .toArray();
+  console.log(notifications)
+  for (const notification of notifications) {
+    const user = await User.findOne({ userId: notification.userId });
+    await sendNotification(notification,user);
+    await db.collection('scheduledNotifications').deleteOne({ _id: notification._id });
+  }
+}
+
+async function sendNotification(notification) {
+  
+  console.log('Sending notification:', notification);
+}
+// async function sendNotification(notification, user) {
+//   for (const channel of user.preferences.channels) {
+//     try {
+//       switch (channel) {
+//         case 'email':
+//           await sendEmail(notification, user);
+//           break;
+//         case 'sms':
+//           await sendSMS(notification, user);
+//           break;
+//         case 'push':
+//           await sendPushNotification(notification, user);
+//           break;
+//       }
+//       await Notification.updateOne({ _id: notification._id }, { status: 'sent' });
+//       return;
+//     } catch (error) {
+//       console.error(`Failed to send notification via ${channel}:`, error);
+//     }
+//   }
+
+// async function sendEmail(notification, user) {
+//   console.log(`Sending email to ${user.userId}: ${notification.message}`);
+// }
+
+// async function sendSMS(notification, user) {
+//   console.log(`Sending SMS to ${user.userId}: ${notification.message}`);
+// }
+
+// async function sendPushNotification(notification, user) {
+//   console.log(`Sending push notification to ${user.userId}: ${notification.message}`);
+// }
+  
+//   // If all channels fail, mark as failed
+//   await Notification.updateOne({ _id: notification._id }, { status: 'failed' });
+// }
+// Run the processor
+processNotifications().catch(console.error);
+
+// Schedule periodic checks for pending notifications
+schedule.scheduleJob('*/1 * * * *', checkScheduledNotifications);
+
+module.exports = { processNotifications, checkScheduledNotifications };
+