From a6d582623af50782791d29750995355867387907 Mon Sep 17 00:00:00 2001 From: sublime247 Date: Sat, 25 Jul 2026 14:22:34 +0100 Subject: [PATCH 1/2] feat(notifications): wire producers, batched fan-out, redis pubsub, pagination limit & schema index fix --- app.js | 2 + src/controllers/books/bookController.js | 6 + src/controllers/courses/courseController.js | 6 + src/controllers/notificationController.js | 346 ++++++++++++++------ src/controllers/userController.js | 11 + src/models/Notification.js | 11 +- test/app.test.js | 21 +- test/auth.test.js | 4 + test/notification.test.js | 282 ++++++++++++++++ test/payout.test.js | 4 + 10 files changed, 582 insertions(+), 111 deletions(-) create mode 100644 test/notification.test.js diff --git a/app.js b/app.js index 6dd44dc..24bbe51 100644 --- a/app.js +++ b/app.js @@ -43,6 +43,7 @@ import stellarWalletRoutes from "./src/routes/stellar/walletRoutes.js"; import stellarPaymentRoutes from "./src/routes/stellar/paymentRoutes.js"; import stellarDonationRoutes from "./src/routes/stellar/donationRoutes.js"; import payoutRoutes from "./src/routes/payoutRoutes.js"; +import notificationRoutes from "./src/routes/notificationRoutes.js"; handleUncaughtException(); validateEnv(); @@ -179,6 +180,7 @@ app.use("/api/stellar/wallet", stellarWalletRoutes); app.use("/api/stellar/payment", stellarPaymentRoutes); app.use("/api/stellar/donation", stellarDonationRoutes); app.use("/api/payouts", payoutRoutes); +app.use("/api/notifications", notificationRoutes); // ====================== // ERROR HANDLING diff --git a/src/controllers/books/bookController.js b/src/controllers/books/bookController.js index 08b44ed..c020eef 100644 --- a/src/controllers/books/bookController.js +++ b/src/controllers/books/bookController.js @@ -3,6 +3,7 @@ import Book from "../../models/Book.js"; import User from "../../models/User.js"; import cloudinary from "../../utils/cloudinary.js"; import logger from "../../config/logger.js"; +import { createNewBookNotification } from "../notificationController.js"; //cretae a book export const createBook = async (req, res) => { @@ -64,6 +65,11 @@ export const createBook = async (req, res) => { fileUrl: fileUpload.secure_url, }); + // Emit new book notification asynchronously to followers + createNewBookNotification(book._id, req.user._id, book.title).catch((err) => + logger.error("Error creating book notification:", err) + ); + res.status(201).json({ success: true, book }); } catch (err) { logger.error("Book creation error:", err); diff --git a/src/controllers/courses/courseController.js b/src/controllers/courses/courseController.js index d783da6..a31b5ac 100644 --- a/src/controllers/courses/courseController.js +++ b/src/controllers/courses/courseController.js @@ -2,6 +2,7 @@ import Course from "../../models/Course.js"; import mongoose from "mongoose"; import logger from "../../config/logger.js"; import { catchAsync, APIError } from "../../middlewares/errorHandler.js"; +import { createNewCourseNotification } from "../notificationController.js"; /** * Create a new course @@ -33,6 +34,11 @@ export const createCourse = catchAsync(async (req, res, next) => { logger.info(`✅ Course created successfully: ${course._id} - ${title}`); + // Emit new course notification asynchronously to followers + createNewCourseNotification(course._id, req.user._id, course.title).catch((err) => + logger.error("Error creating course notification:", err) + ); + res.status(201).json({ success: true, message: "Course created successfully", diff --git a/src/controllers/notificationController.js b/src/controllers/notificationController.js index 326f87d..811bfac 100644 --- a/src/controllers/notificationController.js +++ b/src/controllers/notificationController.js @@ -1,42 +1,151 @@ import Notification from "../models/Notification.js"; import logger from "../config/logger.js"; import User from "../models/User.js"; +import { getRedisClient, isRedisReady } from "../config/redis.js"; -// Store active SSE connections +/** + * SSE Connections Map + * Maps userId (string) -> Set of active Express Response streams. + * Allows multiple concurrent browser tabs/sessions per user and proper auth-scoped cleanup on disconnect. + */ const sseConnections = new Map(); +/** + * Multi-Instance SSE Delivery via Redis Pub/Sub: + * When horizontally scaled (e.g. on Render), an instance producing a notification + * publishes the event to the Redis channel 'notifications:sse'. All backend instances + * listen on this channel and deliver the payload to any locally connected client SSE stream. + * If Redis is unavailable, the system falls back seamlessly to in-memory local delivery. + */ +let isSubscriberInit = false; +export const initRedisSubscriber = async () => { + if (isSubscriberInit) return; + if (!isRedisReady()) return; + + try { + const mainClient = getRedisClient(); + if (!mainClient) return; + + const subClient = mainClient.duplicate(); + await subClient.connect(); + + await subClient.subscribe("notifications:sse", (message) => { + try { + const { recipientId, notification } = JSON.parse(message); + deliverLocalSSENotification(recipientId, notification); + } catch (err) { + logger.error("Error handling Redis SSE pub/sub message:", err); + } + }); + + isSubscriberInit = true; + logger.info("✅ Redis SSE Pub/Sub subscriber initialized"); + } catch (err) { + logger.error("Failed to initialize Redis SSE subscriber:", err); + } +}; + +/** + * Deliver SSE payload to active local connections for a given user + */ +export const deliverLocalSSENotification = (recipientId, notification) => { + const userConns = sseConnections.get(recipientId.toString()); + if (userConns && userConns.size > 0) { + const payload = `data: ${JSON.stringify({ + type: "new_notification", + notification, + })}\n\n`; + + userConns.forEach((res) => { + try { + res.write(payload); + } catch (err) { + logger.error(`Error writing SSE payload to user ${recipientId}:`, err); + } + }); + } +}; + +/** + * Dispatch SSE notification across multi-instance deployment via Redis or local Map + */ +export const dispatchSSENotification = async (recipientId, notification) => { + if (isRedisReady()) { + try { + await initRedisSubscriber(); + const client = getRedisClient(); + if (client) { + await client.publish( + "notifications:sse", + JSON.stringify({ recipientId: recipientId.toString(), notification }) + ); + return; + } + } catch (err) { + logger.error("Error publishing SSE notification to Redis:", err); + } + } + + // Fallback to local delivery if Redis is not active or fails + deliverLocalSSENotification(recipientId, notification); +}; + // SSE endpoint for real-time notifications export const sseNotifications = async (req, res) => { const userId = req.user._id.toString(); - + // Set SSE headers res.writeHead(200, { - 'Content-Type': 'text/event-stream', - 'Cache-Control': 'no-cache', - 'Connection': 'keep-alive', - 'Access-Control-Allow-Origin': '*', - 'Access-Control-Allow-Headers': 'Cache-Control' + "Content-Type": "text/event-stream", + "Cache-Control": "no-cache", + Connection: "keep-alive", + "Access-Control-Allow-Origin": "*", + "Access-Control-Allow-Headers": "Cache-Control", }); + // Send initial connection message - res.write(`data: ${JSON.stringify({ type: 'connection', message: 'Connected to notifications' })}\n\n`); + res.write( + `data: ${JSON.stringify({ + type: "connection", + message: "Connected to notifications", + })}\n\n` + ); + + // Store connection in user's connection set + if (!sseConnections.has(userId)) { + sseConnections.set(userId, new Set()); + } + const userConns = sseConnections.get(userId); + userConns.add(res); - // Store connection - sseConnections.set(userId, res); + // Initialize Redis subscriber if available + initRedisSubscriber().catch((err) => + logger.error("Redis subscriber init error:", err) + ); // Handle client disconnect - req.on('close', () => { - sseConnections.delete(userId); + req.on("close", () => { + const conns = sseConnections.get(userId); + if (conns) { + conns.delete(res); + if (conns.size === 0) { + sseConnections.delete(userId); + } + } logger.info(`SSE connection closed for user: ${userId}`); }); // Keep connection alive const keepAlive = setInterval(() => { - if (sseConnections.has(userId)) { - res.write(`data: ${JSON.stringify({ type: 'ping', timestamp: Date.now() })}\n\n`); + const conns = sseConnections.get(userId); + if (conns && conns.has(res)) { + res.write( + `data: ${JSON.stringify({ type: "ping", timestamp: Date.now() })}\n\n` + ); } else { clearInterval(keepAlive); } - }, 30000); // Send ping every 30 seconds + }, 30000); }; // Send notification to specific user (for real-time updates) @@ -44,138 +153,177 @@ export const sendNotificationToUser = async (userId, notificationData) => { try { const notification = await Notification.create({ recipient: userId, - ...notificationData + ...notificationData, }); // Populate sender info - await notification.populate('sender', 'name avatar'); - - // Send real-time notification via SSE - const connection = sseConnections.get(userId.toString()); - if (connection) { - connection.write(`data: ${JSON.stringify({ - type: 'new_notification', - notification: notification - })}\n\n`); - } + await notification.populate("sender", "name avatar"); + + // Dispatch via SSE + await dispatchSSENotification(userId, notification); return notification; } catch (error) { - logger.error('Error sending notification:', error); + logger.error("Error sending notification:", error); throw error; } }; // Create follow notification export const createFollowNotification = async (followerId, followedId) => { - const follower = await User.findById(followerId).select('name avatar'); - - await sendNotificationToUser(followedId, { + const follower = await User.findById(followerId).select("name avatar"); + if (!follower) return; + + return sendNotificationToUser(followedId, { sender: followerId, - type: 'follow', - title: 'New Follower', + type: "follow", + title: "New Follower", message: `${follower.name} started following you`, - priority: 'medium' + priority: "medium", }); }; // Create unfollow notification export const createUnfollowNotification = async (unfollowerId, unfollowedId) => { - const unfollower = await User.findById(unfollowerId).select('name avatar'); - - await sendNotificationToUser(unfollowedId, { + const unfollower = await User.findById(unfollowerId).select("name avatar"); + if (!unfollower) return; + + return sendNotificationToUser(unfollowedId, { sender: unfollowerId, - type: 'unfollow', - title: 'User Unfollowed', + type: "unfollow", + title: "User Unfollowed", message: `${unfollower.name} unfollowed you`, - priority: 'low' + priority: "low", }); }; -// Create new course notification for followers +// Create new course notification for followers (batched fan-out) export const createNewCourseNotification = async (courseId, creatorId, courseTitle) => { - const creator = await User.findById(creatorId); - const followers = creator.followers; + try { + const creator = await User.findById(creatorId).select("name avatar followers"); + if (!creator || !creator.followers || creator.followers.length === 0) return []; - for (const followerId of followers) { - await sendNotificationToUser(followerId, { + const docs = creator.followers.map((followerId) => ({ + recipient: followerId, sender: creatorId, - type: 'new_course', - title: 'New Course Available', + type: "new_course", + title: "New Course Available", message: `${creator.name} created a new course: ${courseTitle}`, data: { courseId }, - priority: 'medium' - }); + priority: "medium", + })); + + // Batched DB insert in one write operation + const createdNotifications = await Notification.insertMany(docs); + + // Broadcast SSE notifications with pre-populated sender info + const senderPayload = { _id: creatorId, name: creator.name, avatar: creator.avatar }; + for (const notification of createdNotifications) { + const plainNotif = notification.toObject ? notification.toObject() : { ...notification }; + plainNotif.sender = senderPayload; + dispatchSSENotification(plainNotif.recipient, plainNotif); + } + + return createdNotifications; + } catch (error) { + logger.error("Error creating new course notifications:", error); + throw error; } }; -// Create new book notification for followers +// Create new book notification for followers (batched fan-out) export const createNewBookNotification = async (bookId, authorId, bookTitle) => { - const author = await User.findById(authorId); - const followers = author.followers; + try { + const author = await User.findById(authorId).select("name avatar followers"); + if (!author || !author.followers || author.followers.length === 0) return []; - for (const followerId of followers) { - await sendNotificationToUser(followerId, { + const docs = author.followers.map((followerId) => ({ + recipient: followerId, sender: authorId, - type: 'new_book', - title: 'New Book Available', + type: "new_book", + title: "New Book Available", message: `${author.name} published a new book: ${bookTitle}`, data: { bookId }, - priority: 'medium' - }); + priority: "medium", + })); + + // Batched DB insert in one write operation + const createdNotifications = await Notification.insertMany(docs); + + // Broadcast SSE notifications with pre-populated author info + const senderPayload = { _id: authorId, name: author.name, avatar: author.avatar }; + for (const notification of createdNotifications) { + const plainNotif = notification.toObject ? notification.toObject() : { ...notification }; + plainNotif.sender = senderPayload; + dispatchSSENotification(plainNotif.recipient, plainNotif); + } + + return createdNotifications; + } catch (error) { + logger.error("Error creating new book notifications:", error); + throw error; } }; -// Get user notifications +// Get user notifications (bounded pagination) export const getUserNotifications = async (req, res) => { try { const userId = req.user._id; - const { page = 1, limit = 20, unreadOnly = false } = req.query; + + let page = parseInt(req.query.page, 10); + if (isNaN(page) || page < 1) page = 1; + + let limit = parseInt(req.query.limit, 10); + if (isNaN(limit) || limit < 1) limit = 20; + if (limit > 100) limit = 100; // Enforce pagination cap of 100 + + const unreadOnly = req.query.unreadOnly === "true"; const query = { recipient: userId, - isDeleted: false + isDeleted: false, }; - if (unreadOnly === 'true') { + if (unreadOnly) { query.isRead = false; } + const skip = (page - 1) * limit; + const notifications = await Notification.find(query) - .populate('sender', 'name avatar') - .populate('data.courseId', 'title thumbnail') - .populate('data.bookId', 'title image') + .populate("sender", "name avatar") + .populate("data.courseId", "title thumbnail") + .populate("data.bookId", "title image") .sort({ createdAt: -1 }) - .limit(limit * 1) - .skip((page - 1) * limit) + .skip(skip) + .limit(limit) .lean(); const total = await Notification.countDocuments(query); const unreadCount = await Notification.countDocuments({ recipient: userId, isRead: false, - isDeleted: false + isDeleted: false, }); res.status(200).json({ success: true, notifications, pagination: { - currentPage: parseInt(page), - totalPages: Math.ceil(total / limit), + currentPage: page, + totalPages: Math.ceil(total / limit) || 1, totalNotifications: total, hasNextPage: page * limit < total, - hasPrevPage: page > 1 + hasPrevPage: page > 1, }, - unreadCount + unreadCount, }); - } catch (error) { - logger.error('Get notifications error:', error); + logger.error("Get notifications error:", error); res.status(500).json({ success: false, - message: 'Failed to fetch notifications', - error: error.message + message: "Failed to fetch notifications", + error: error.message, }); } }; @@ -195,21 +343,20 @@ export const markNotificationAsRead = async (req, res) => { if (!notification) { return res.status(404).json({ success: false, - message: 'Notification not found' + message: "Notification not found", }); } res.status(200).json({ success: true, - notification + notification, }); - } catch (error) { - logger.error('Mark notification read error:', error); + logger.error("Mark notification read error:", error); res.status(500).json({ success: false, - message: 'Failed to mark notification as read', - error: error.message + message: "Failed to mark notification as read", + error: error.message, }); } }; @@ -226,15 +373,14 @@ export const markAllNotificationsAsRead = async (req, res) => { res.status(200).json({ success: true, - message: 'All notifications marked as read' + message: "All notifications marked as read", }); - } catch (error) { - logger.error('Mark all notifications read error:', error); + logger.error("Mark all notifications read error:", error); res.status(500).json({ success: false, - message: 'Failed to mark notifications as read', - error: error.message + message: "Failed to mark notifications as read", + error: error.message, }); } }; @@ -254,21 +400,20 @@ export const deleteNotification = async (req, res) => { if (!notification) { return res.status(404).json({ success: false, - message: 'Notification not found' + message: "Notification not found", }); } res.status(200).json({ success: true, - message: 'Notification deleted' + message: "Notification deleted", }); - } catch (error) { - logger.error('Delete notification error:', error); + logger.error("Delete notification error:", error); res.status(500).json({ success: false, - message: 'Failed to delete notification', - error: error.message + message: "Failed to delete notification", + error: error.message, }); } }; @@ -277,25 +422,24 @@ export const deleteNotification = async (req, res) => { export const getNotificationSettings = async (req, res) => { try { const userId = req.user._id; - const user = await User.findById(userId).select('notificationSettings'); + const user = await User.findById(userId).select("notificationSettings"); res.status(200).json({ success: true, - settings: user.notificationSettings || { + settings: user?.notificationSettings || { follow: true, newContent: true, likes: true, comments: true, - system: true - } + system: true, + }, }); - } catch (error) { - logger.error('Get notification settings error:', error); + logger.error("Get notification settings error:", error); res.status(500).json({ success: false, - message: 'Failed to fetch notification settings', - error: error.message + message: "Failed to fetch notification settings", + error: error.message, }); } -}; \ No newline at end of file +}; \ No newline at end of file diff --git a/src/controllers/userController.js b/src/controllers/userController.js index b2175a2..be044fa 100644 --- a/src/controllers/userController.js +++ b/src/controllers/userController.js @@ -3,6 +3,7 @@ import cloudinary from "../utils/cloudinary.js"; import Course from "../models/Course.js"; import Book from "../models/Book.js"; import logger from "../config/logger.js"; +import { createFollowNotification, createUnfollowNotification } from "./notificationController.js"; // Update user profile (including avatar upload to Cloudinary) export const updateUser = async (req, res) => { @@ -177,6 +178,11 @@ export const followUser = async (req, res) => { $push: { followers: currentUserId }, }); + // Emit follow notification asynchronously + createFollowNotification(currentUserId, userId).catch((err) => + logger.error("Error creating follow notification:", err) + ); + res.status(200).json({ success: true, message: "Successfully followed user", @@ -232,6 +238,11 @@ export const unfollowUser = async (req, res) => { $pull: { followers: currentUserId }, }); + // Emit unfollow notification asynchronously + createUnfollowNotification(currentUserId, userId).catch((err) => + logger.error("Error creating unfollow notification:", err) + ); + res.status(200).json({ success: true, message: "Successfully unfollowed user", diff --git a/src/models/Notification.js b/src/models/Notification.js index 452af9a..e76cd33 100644 --- a/src/models/Notification.js +++ b/src/models/Notification.js @@ -74,13 +74,12 @@ const notificationSchema = new mongoose.Schema( }, { timestamps: true, - // Index for efficient queries - indexes: [ - { recipient: 1, createdAt: -1 }, - { recipient: 1, isRead: 1 }, - { recipient: 1, isDeleted: 1 }, - ], } ); +// Indexes for efficient querying +notificationSchema.index({ recipient: 1, createdAt: -1 }); +notificationSchema.index({ recipient: 1, isRead: 1 }); +notificationSchema.index({ recipient: 1, isDeleted: 1 }); + export default mongoose.model("Notification", notificationSchema); diff --git a/test/app.test.js b/test/app.test.js index 41bc297..acc40e2 100644 --- a/test/app.test.js +++ b/test/app.test.js @@ -1,5 +1,6 @@ import request from "supertest"; import mongoose from "mongoose"; +import { MongoMemoryServer } from "mongodb-memory-server"; import app from "../app.js"; import { calculateFeeSplit, @@ -13,14 +14,26 @@ import { paymentsFailed, } from "../src/config/metrics.js"; -// app.js skips connectDB() under NODE_ENV=test; routes that touch the DB need -// a live connection, so this suite manages its own. +let mongoServer; + beforeAll(async () => { - await mongoose.connect(process.env.MONGO_URI); -}); + if (process.env.MONGO_URI && !process.env.MONGO_URI.includes("localhost")) { + try { + await mongoose.connect(process.env.MONGO_URI); + return; + } catch (_err) { + // Fallback to MongoMemoryServer + } + } + mongoServer = await MongoMemoryServer.create(); + await mongoose.connect(mongoServer.getUri()); +}, 30000); afterAll(async () => { await mongoose.disconnect(); + if (mongoServer) { + await mongoServer.stop(); + } }); // Read a labeled counter value via prom-client's public API (internals like diff --git a/test/auth.test.js b/test/auth.test.js index 740d381..1cb5dbc 100644 --- a/test/auth.test.js +++ b/test/auth.test.js @@ -366,4 +366,8 @@ describe("Authentication & Session Management", () => { expect(activeSessions.length).toBe(0); }); }); + + afterAll(() => { + jest.restoreAllMocks(); + }); }); diff --git a/test/notification.test.js b/test/notification.test.js new file mode 100644 index 0000000..405ac7a --- /dev/null +++ b/test/notification.test.js @@ -0,0 +1,282 @@ +import { jest } from "@jest/globals"; +import express from "express"; +import mongoose from "mongoose"; +import { MongoMemoryServer } from "mongodb-memory-server"; +import Notification from "../src/models/Notification.js"; +import User from "../src/models/User.js"; +import Course from "../src/models/Course.js"; +import Book from "../src/models/Book.js"; +import { + sseNotifications, + getUserNotifications, + createFollowNotification, + createUnfollowNotification, + createNewCourseNotification, + createNewBookNotification, +} from "../src/controllers/notificationController.js"; + +describe("Notification System & Event Wiring", () => { + let mongoServer; + + beforeAll(async () => { + mongoServer = await MongoMemoryServer.create(); + const mongoUri = mongoServer.getUri(); + await mongoose.connect(mongoUri); + }, 30000); + + afterAll(async () => { + await mongoose.disconnect(); + if (mongoServer) { + await mongoServer.stop(); + } + }); + + beforeEach(async () => { + await Notification.deleteMany({}); + await User.deleteMany({}); + await Course.deleteMany({}); + await Book.deleteMany({}); + }); + + describe("Mongoose Schema Compound Indexes", () => { + it("has expected compound indexes declared on Notification schema", () => { + const indexes = Notification.schema.indexes(); + const indexFields = indexes.map(([spec]) => Object.keys(spec).join(",")); + + expect(indexFields).toContain("recipient,createdAt"); + expect(indexFields).toContain("recipient,isRead"); + expect(indexFields).toContain("recipient,isDeleted"); + }); + }); + + describe("Event Producer Notifications", () => { + it("creates a follow notification for the followed user", async () => { + const follower = await User.create({ + name: "Follower User", + email: "follower@example.com", + password: "password123", + }); + + const followed = await User.create({ + name: "Followed User", + email: "followed@example.com", + password: "password123", + }); + + await createFollowNotification(follower._id, followed._id); + + const notifs = await Notification.find({ recipient: followed._id }); + expect(notifs).toHaveLength(1); + expect(notifs[0].type).toBe("follow"); + expect(notifs[0].sender.toString()).toBe(follower._id.toString()); + expect(notifs[0].message).toContain("Follower User started following you"); + }); + + it("creates an unfollow notification for the unfollowed user", async () => { + const unfollower = await User.create({ + name: "Unfollower User", + email: "unfollower@example.com", + password: "password123", + }); + + const unfollowed = await User.create({ + name: "Unfollowed User", + email: "unfollowed@example.com", + password: "password123", + }); + + await createUnfollowNotification(unfollower._id, unfollowed._id); + + const notifs = await Notification.find({ recipient: unfollowed._id }); + expect(notifs).toHaveLength(1); + expect(notifs[0].type).toBe("unfollow"); + expect(notifs[0].sender.toString()).toBe(unfollower._id.toString()); + expect(notifs[0].message).toContain("Unfollower User unfollowed you"); + }); + + it("creates batched new_course notifications for all followers", async () => { + const follower1 = await User.create({ + name: "Follower 1", + email: "f1@example.com", + password: "password123", + }); + const follower2 = await User.create({ + name: "Follower 2", + email: "f2@example.com", + password: "password123", + }); + + const creator = await User.create({ + name: "Course Creator", + email: "creator@example.com", + password: "password123", + followers: [follower1._id, follower2._id], + }); + + const course = await Course.create({ + title: "Mastering Node.js", + description: "Comprehensive Node.js course", + category: "Programming", + createdBy: creator._id, + }); + + const result = await createNewCourseNotification( + course._id, + creator._id, + course.title + ); + + expect(result).toHaveLength(2); + + const notifs = await Notification.find({ sender: creator._id }); + expect(notifs).toHaveLength(2); + const recipientIds = notifs.map((n) => n.recipient.toString()); + expect(recipientIds).toContain(follower1._id.toString()); + expect(recipientIds).toContain(follower2._id.toString()); + expect(notifs[0].type).toBe("new_course"); + expect(notifs[0].message).toContain("Course Creator created a new course: Mastering Node.js"); + }); + + it("creates batched new_book notifications for all followers", async () => { + const follower1 = await User.create({ + name: "Book Reader 1", + email: "r1@example.com", + password: "password123", + }); + const follower2 = await User.create({ + name: "Book Reader 2", + email: "r2@example.com", + password: "password123", + }); + + const author = await User.create({ + name: "Book Author", + email: "author@example.com", + password: "password123", + followers: [follower1._id, follower2._id], + }); + + const book = await Book.create({ + title: "Clean Architecture in JS", + description: "Guide to scalable code architecture", + category: "Tech", + price: 15, + author: author._id, + thumbnail: "https://example.com/thumb.jpg", + image: "https://example.com/thumb.jpg", + fileUrl: "https://example.com/book.pdf", + }); + + const result = await createNewBookNotification( + book._id, + author._id, + book.title + ); + + expect(result).toHaveLength(2); + + const notifs = await Notification.find({ sender: author._id }); + expect(notifs).toHaveLength(2); + const recipientIds = notifs.map((n) => n.recipient.toString()); + expect(recipientIds).toContain(follower1._id.toString()); + expect(recipientIds).toContain(follower2._id.toString()); + expect(notifs[0].type).toBe("new_book"); + expect(notifs[0].message).toContain("Book Author published a new book: Clean Architecture in JS"); + }); + }); + + describe("getUserNotifications Pagination & Bounding", () => { + it("caps limit to 100 and validates invalid page/limit params", async () => { + const recipient = await User.create({ + name: "Recipient User", + email: "recipient@example.com", + password: "password123", + }); + + const sender = await User.create({ + name: "Sender User", + email: "sender@example.com", + password: "password123", + }); + + // Insert 120 notifications for recipient + const docs = Array.from({ length: 120 }, (_, i) => ({ + recipient: recipient._id, + sender: sender._id, + type: "system", + title: `Notification ${i + 1}`, + message: `Message ${i + 1}`, + })); + + await Notification.insertMany(docs); + + // Request with limit=500 -> should be capped at 100 + const reqLarge = { + user: { _id: recipient._id }, + query: { limit: "500", page: "1" }, + }; + + const resLarge = { + status: jest.fn().mockReturnThis(), + json: jest.fn(), + }; + + await getUserNotifications(reqLarge, resLarge); + + expect(resLarge.status).toHaveBeenCalledWith(200); + const dataLarge = resLarge.json.mock.calls[0][0]; + expect(dataLarge.notifications).toHaveLength(100); + expect(dataLarge.pagination.currentPage).toBe(1); + + // Request with negative page and non-numeric limit -> should default to page=1, limit=20 + const reqInvalid = { + user: { _id: recipient._id }, + query: { limit: "invalid", page: "-5" }, + }; + + const resInvalid = { + status: jest.fn().mockReturnThis(), + json: jest.fn(), + }; + + await getUserNotifications(reqInvalid, resInvalid); + + const dataInvalid = resInvalid.json.mock.calls[0][0]; + expect(dataInvalid.notifications).toHaveLength(20); + expect(dataInvalid.pagination.currentPage).toBe(1); + }); + }); + + describe("SSE Connection & Delivery", () => { + it("establishes SSE connection stream and handles payload delivery", async () => { + const user = await User.create({ + name: "SSE User", + email: "sse@example.com", + password: "password123", + }); + + const writeHeadMock = jest.fn(); + const writeMock = jest.fn(); + const reqMock = { + user: { _id: user._id }, + on: jest.fn(), + }; + const resMock = { + writeHead: writeHeadMock, + write: writeMock, + }; + + await sseNotifications(reqMock, resMock); + + expect(writeHeadMock).toHaveBeenCalledWith( + 200, + expect.objectContaining({ + "Content-Type": "text/event-stream", + }) + ); + expect(writeMock).toHaveBeenCalledWith( + expect.stringContaining("Connected to notifications") + ); + }); + }); +}); diff --git a/test/payout.test.js b/test/payout.test.js index c251808..1bba6df 100644 --- a/test/payout.test.js +++ b/test/payout.test.js @@ -496,4 +496,8 @@ describe("Payout Service & Earnings Ledger", () => { expect(isPayoutAdmin({ _id: "admin_user_1" })).toBe(false); }); }); + + afterAll(() => { + jest.restoreAllMocks(); + }); }); From 2827f8a46a3c4469bed33e6f80beedc6e3f5485b Mon Sep 17 00:00:00 2001 From: sublime247 Date: Sat, 25 Jul 2026 14:27:57 +0100 Subject: [PATCH 2/2] fix(notifications): restore notification controller imports in book, course, and user controllers --- src/controllers/books/bookController.js | 1 + src/controllers/courses/courseController.js | 1 + src/controllers/userController.js | 1 + 3 files changed, 3 insertions(+) diff --git a/src/controllers/books/bookController.js b/src/controllers/books/bookController.js index ebc0149..947525a 100644 --- a/src/controllers/books/bookController.js +++ b/src/controllers/books/bookController.js @@ -4,6 +4,7 @@ import User from "../../models/User.js"; import cloudinary from "../../utils/cloudinary.js"; import logger from "../../config/logger.js"; import { validateMagicBytes } from "../../utils/fileValidation.js"; +import { createNewBookNotification } from "../notificationController.js"; //cretae a book export const createBook = async (req, res) => { diff --git a/src/controllers/courses/courseController.js b/src/controllers/courses/courseController.js index 5751d0d..8b3d414 100644 --- a/src/controllers/courses/courseController.js +++ b/src/controllers/courses/courseController.js @@ -3,6 +3,7 @@ import mongoose from "mongoose"; import logger from "../../config/logger.js"; import { catchAsync, APIError } from "../../middlewares/errorHandler.js"; import { getCacheOrSet, CACHE_TTL, CACHE_KEYS } from "../../utils/cache.js"; +import { createNewCourseNotification } from "../notificationController.js"; /** * Create a new course diff --git a/src/controllers/userController.js b/src/controllers/userController.js index b6a452c..cbb7e99 100644 --- a/src/controllers/userController.js +++ b/src/controllers/userController.js @@ -4,6 +4,7 @@ import Course from "../models/Course.js"; import Book from "../models/Book.js"; import logger from "../config/logger.js"; import { validateMagicBytes } from "../utils/fileValidation.js"; +import { createFollowNotification, createUnfollowNotification } from "./notificationController.js"; // Update user profile (including avatar upload to Cloudinary) export const updateUser = async (req, res) => {