feat(notifications): wire producers, batched fan-out, redis pubsub, p… - #76
Conversation
…agination limit & schema index fix
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Path: .coderabbit.yaml Review profile: CHILL Plan: Pro Plus Run ID: 📒 Files selected for processing (3)
🚧 Files skipped from review as they are similar to previous changes (3)
WalkthroughNotification producers now run after user, course, and book events. Delivery supports Redis Pub/Sub and multiple SSE connections, fan-out uses batched inserts, notification pagination is bounded, indexes are declared explicitly, and integration tests cover the updated behavior. ChangesNotification system
Estimated code review effort: 4 (Complex) | ~45 minutes Sequence Diagram(s)sequenceDiagram
participant EventController
participant NotificationController
participant MongoDB
participant Redis
participant SSEClient
EventController->>NotificationController: invoke notification producer
NotificationController->>MongoDB: insert notification records
NotificationController->>Redis: publish SSE event
Redis->>NotificationController: deliver subscribed event
NotificationController->>SSEClient: write notification event
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
…course, and user controllers
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
src/models/Notification.js (1)
80-84: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winCorrect fix — this addresses a real, previously silent bug.
Mongoose doesn't support an
indexesschema-constructor option for compound indexes; onlyschema.index()(or per-fieldindex: true) actually registers them. Moving to explicit.index()calls means these compound indexes are now genuinely created, matching the PR's stated goal of fixing the notification index declarations.One optional follow-up:
getUserNotificationsfilters by{recipient, isDeleted}(plus optionalisRead) and sorts bycreatedAt. None of the three indexes here covers all three fields together, so that query still needs an extra in-memory filter/sort step beyond the{recipient, createdAt}index. Consider addingnotificationSchema.index({ recipient: 1, isDeleted: 1, createdAt: -1 })to fully serve that hot path in one index.⚡ Optional additional index
notificationSchema.index({ recipient: 1, createdAt: -1 }); notificationSchema.index({ recipient: 1, isRead: 1 }); notificationSchema.index({ recipient: 1, isDeleted: 1 }); +notificationSchema.index({ recipient: 1, isDeleted: 1, createdAt: -1 });🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/models/Notification.js` around lines 80 - 84, Optionally add a compound index declaration alongside the existing notificationSchema.index calls covering recipient, isDeleted, and createdAt in descending order, so getUserNotifications can efficiently apply its filters and sort without extra in-memory processing.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/controllers/notificationController.js`:
- Around line 20-46: Replace the check-then-act boolean guard in
initRedisSubscriber with a shared in-flight initialization promise so concurrent
callers await the same connect/subscribe operation. Set the promise before any
await, reuse it while initialization is pending or complete, and clear it on
failure so later calls can retry; ensure only one duplicated Redis subscriber is
created and preserve the existing success and error logging.
- Around line 25-31: Add an error listener to the duplicated client created in
the notification controller’s subscriber setup before calling
subClient.connect(), handling subscriber connection errors through the existing
application logging or error-reporting mechanism. Keep the mainClient flow and
connection behavior unchanged.
In `@test/notification.test.js`:
- Around line 251-280: Update the “establishes SSE connection stream and handles
payload delivery” test to capture the callback registered through reqMock.on,
then invoke that close-handler after assertions so the interval created by
sseNotifications is cleared. Keep the existing connection and payload assertions
unchanged.
---
Nitpick comments:
In `@src/models/Notification.js`:
- Around line 80-84: Optionally add a compound index declaration alongside the
existing notificationSchema.index calls covering recipient, isDeleted, and
createdAt in descending order, so getUserNotifications can efficiently apply its
filters and sort without extra in-memory processing.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro Plus
Run ID: 69885440-e3b7-4ac6-9a22-fd78c2b2a527
📒 Files selected for processing (9)
src/controllers/books/bookController.jssrc/controllers/courses/courseController.jssrc/controllers/notificationController.jssrc/controllers/userController.jssrc/models/Notification.jstest/app.test.jstest/auth.test.jstest/notification.test.jstest/payout.test.js
| 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); | ||
| } | ||
| }; |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟠 Major | ⚡ Quick win
initRedisSubscriber has a check-then-act race that can spawn duplicate, never-closed subscribers.
isSubscriberInit is only flipped to true after the awaits for connect()/subscribe() resolve, but the function is called from every sseNotifications connection (line 122) and every dispatchSSENotification (line 75) without any in-flight guard. If several SSE connections open concurrently on the same instance (e.g. a reconnect storm after a deploy), each call passes the if (isSubscriberInit) return; check before the first completes, so multiple duplicate Redis subscriber clients get created — each is never closed and each independently re-delivers every published message locally, so users end up receiving duplicate notifications and the process accumulates leaked Redis connections over time.
🔒 Proposed fix: guard with an in-flight promise instead of a boolean
let isSubscriberInit = false;
+let subscriberInitPromise = null;
export const initRedisSubscriber = async () => {
if (isSubscriberInit) return;
if (!isRedisReady()) return;
+ if (subscriberInitPromise) return subscriberInitPromise;
- 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);
- }
+ subscriberInitPromise = (async () => {
+ try {
+ const mainClient = getRedisClient();
+ if (!mainClient) return;
+
+ const subClient = mainClient.duplicate();
+ subClient.on("error", (err) => logger.error("Redis SSE subscriber client error:", err));
+ 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);
+ } finally {
+ subscriberInitPromise = null;
+ }
+ })();
+ return subscriberInitPromise;
};📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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); | |
| } | |
| }; | |
| let isSubscriberInit = false; | |
| let subscriberInitPromise = null; | |
| export const initRedisSubscriber = async () => { | |
| if (isSubscriberInit) return; | |
| if (!isRedisReady()) return; | |
| if (subscriberInitPromise) return subscriberInitPromise; | |
| subscriberInitPromise = (async () => { | |
| try { | |
| const mainClient = getRedisClient(); | |
| if (!mainClient) return; | |
| const subClient = mainClient.duplicate(); | |
| subClient.on("error", (err) => logger.error("Redis SSE subscriber client error:", err)); | |
| 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); | |
| } finally { | |
| subscriberInitPromise = null; | |
| } | |
| })(); | |
| return subscriberInitPromise; | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/controllers/notificationController.js` around lines 20 - 46, Replace the
check-then-act boolean guard in initRedisSubscriber with a shared in-flight
initialization promise so concurrent callers await the same connect/subscribe
operation. Set the promise before any await, reuse it while initialization is
pending or complete, and clear it on failure so later calls can retry; ensure
only one duplicated Redis subscriber is created and preserve the existing
success and error logging.
| try { | ||
| const mainClient = getRedisClient(); | ||
| if (!mainClient) return; | ||
|
|
||
| const subClient = mainClient.duplicate(); | ||
| await subClient.connect(); | ||
|
|
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🌐 Web query:
node-redis v4 duplicate() client error event listener required
💡 Result:
Yes, it is strictly required to register an error event listener for any client created using the duplicate method in node-redis v4 [1][2]. The node-redis client functions as a Node.js EventEmitter [2][3]. According to official documentation, you must listen for error events on every client instance, including those created via duplicate [4][2]. If a client instance does not have at least one error listener registered and an error occurs (such as a network disruption), the error will be thrown, which may cause your entire Node.js process to exit unexpectedly [4][2][3]. When using duplicate to create a dedicated connection—for example, for Pub/Sub operations—ensure you attach an error listener to the new instance before calling.connect [1]. Example implementation: const subscriber = client.duplicate; subscriber.on('error', err => console.error('Redis subscriber error:', err)); await subscriber.connect;
Citations:
- 1: https://github.com/redis/node-redis/blob/HEAD/docs/pub-sub.md
- 2: https://github.com/redis/node-redis
- 3: https://github.com/redis/node-redis/blob/master/README.md
- 4: https://redis.io/docs/latest/develop/clients/nodejs/error-handling/
Attach an error handler to the duplicated Redis subscriber client
mainClient.duplicate() creates a separate Redis client, so it does not inherit the main connection’s listeners. Without an error handler on subClient, a Redis hiccup on this pub/sub connection can emit an unhandled 'error' and crash the process.
Suggested fix
const subClient = mainClient.duplicate();
+ subClient.on("error", (err) => logger.error("Redis SSE subscriber client error:", err));
await subClient.connect();📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| try { | |
| const mainClient = getRedisClient(); | |
| if (!mainClient) return; | |
| const subClient = mainClient.duplicate(); | |
| await subClient.connect(); | |
| try { | |
| const mainClient = getRedisClient(); | |
| if (!mainClient) return; | |
| const subClient = mainClient.duplicate(); | |
| subClient.on("error", (err) => logger.error("Redis SSE subscriber client error:", err)); | |
| await subClient.connect(); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@src/controllers/notificationController.js` around lines 25 - 31, Add an error
listener to the duplicated client created in the notification controller’s
subscriber setup before calling subClient.connect(), handling subscriber
connection errors through the existing application logging or error-reporting
mechanism. Keep the mainClient flow and connection behavior unchanged.
| 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") | ||
| ); | ||
| }); |
There was a problem hiding this comment.
🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
Dangling setInterval from the real SSE handler isn't cleaned up in this test.
sseNotifications starts a real 30s setInterval keep-alive ping. reqMock.on is a no-op mock, so the close handler that would let the interval self-clear is never invoked — the timer keeps firing for the life of the Jest worker process after this test finishes, which is exactly the kind of open handle that causes Jest to hang or need --forceExit.
🧹 Proposed fix: capture and trigger the close handler
const reqMock = {
user: { _id: user._id },
- on: jest.fn(),
+ on: jest.fn((event, cb) => {
+ if (event === "close") closeHandler = cb;
+ }),
};
+ let closeHandler;
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")
);
+ closeHandler?.();📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| 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") | |
| ); | |
| }); | |
| 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((event, cb) => { | |
| if (event === "close") closeHandler = cb; | |
| }), | |
| }; | |
| let closeHandler; | |
| 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") | |
| ); | |
| closeHandler?.(); | |
| }); |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@test/notification.test.js` around lines 251 - 280, Update the “establishes
SSE connection stream and handles payload delivery” test to capture the callback
registered through reqMock.on, then invoke that close-handler after assertions
so the interval created by sseNotifications is cleared. Keep the existing
connection and payload assertions unchanged.
2 participants
closes #22
Summary by CodeRabbit
New Features
Bug Fixes
Tests