M9 notifications

src/services/comment.service.js

@@ -122,7 +122,7 @@ async function notifyTrackCommentIfNeeded({ actorUserId, trackId, commentId }) {
     userId: trackOwnerId,
     actionUserId: actorUserId,
     type: 'comment',
-    referenceId: commentId,
-    referenceType: 'comment',
+    referenceId: trackId,
+    referenceType: 'track',
   });
 }

src/services/messages.service.js

@@ -6,6 +6,36 @@ const pushNotificationsService = require('./push-notifications.service');
 
 const ALLOWED_EMBED_TYPES = ['track', 'playlist'];
 
+const logNotificationFailure = (channel, err) => {
+  console.error(`[Messages] ${channel} notification failed:`, err?.message || err);
+};
+
+const sendDirectMessageNotifications = ({
+  conversationId,
+  senderId,
+  recipientId,
+  messageBody,
+  embedType,
+}) => {
+  emailNotificationsService
+    .sendDirectMessageEmailIfEligible({
+      conversationId,
+      senderId,
+      recipientId,
+    })
+    .catch((err) => logNotificationFailure('Email', err));
+
+  pushNotificationsService
+    .sendDirectMessagePushIfEligible({
+      conversationId,
+      senderId,
+      recipientId,
+      messageBody,
+      embedType,
+    })
+    .catch((err) => logNotificationFailure('Push', err));
+};
+
 const validateSenderId = (senderId) => {
   if (!senderId) {
     throw new AppError('Authenticated sender is required.', 401, 'UNAUTHORIZED');
@@ -174,13 +204,7 @@ exports.startConversation = async ({ senderId, recipientId, body, resource }) =>
     embedId: payload.resource?.id ?? null,
   });
 
-  await emailNotificationsService.sendDirectMessageEmailIfEligible({
-    conversationId: conversation.id,
-    senderId,
-    recipientId,
-  });
-
-  await pushNotificationsService.sendDirectMessagePushIfEligible({
+  sendDirectMessageNotifications({
     conversationId: conversation.id,
     senderId,
     recipientId,
@@ -273,6 +297,8 @@ exports.listConversations = async ({ userId, page, limit }) => {
 // ------------------------------------------------------------
 
 exports.getConversation = async ({ conversationId, userId, page, limit, offset: rawOffset }) => {
+  const conversationActivity = require('../utils/conversation-activity');
+
   // 1. Find conversation
   const conversation = await messageModel.findConversationById(conversationId);
   if (!conversation) {
@@ -293,6 +319,9 @@ exports.getConversation = async ({ conversationId, userId, page, limit, offset:
     throw new AppError('Conversation not found.', 404, 'CONVERSATION_NOT_FOUND');
   }
 
+  // Mark as active for push-notification suppression (best-effort)
+  conversationActivity.markActive({ userId, conversationId });
+
   // 4. Sanitize pagination inputs
   const safePage = Math.max(1, parseInt(page) || 1);
   const safeLimit = Math.min(100, Math.max(1, parseInt(limit) || 50));
@@ -406,13 +435,7 @@ exports.sendMessage = async ({ conversationId, senderId, body, resource }) => {
     embedId: payload.resource?.id ?? null,
   });
 
-  await emailNotificationsService.sendDirectMessageEmailIfEligible({
-    conversationId,
-    senderId,
-    recipientId,
-  });
-
-  await pushNotificationsService.sendDirectMessagePushIfEligible({
+  sendDirectMessageNotifications({
     conversationId,
     senderId,
     recipientId,
@@ -439,6 +462,8 @@ exports.getUnreadCount = async ({ userId }) => {
 // ------------------------------------------------------------
 
 exports.markMessageReadState = async ({ conversationId, messageId, userId, isRead }) => {
+  const conversationActivity = require('../utils/conversation-activity');
+
   // 1. Find conversation
   const conversation = await messageModel.findConversationById(conversationId);
   if (!conversation) {
@@ -462,6 +487,9 @@ exports.markMessageReadState = async ({ conversationId, messageId, userId, isRea
     throw new AppError('You cannot change the read state of your own message.', 403, 'FORBIDDEN');
   }
 
+  // Mark as active for push-notification suppression (best-effort)
+  conversationActivity.markActive({ userId, conversationId });
+
   // 5. Check if already in the requested state — 409 conflict
   if (message.is_read === isRead) {
     throw new AppError(

src/services/push-notifications.service.js

@@ -104,12 +104,21 @@ exports.sendDirectMessagePushIfEligible = async ({
 }) => {
   try {
     const notificationModel = require('../models/notification.model');
+    const conversationActivity = require('../utils/conversation-activity');
+    const dmPushThrottle = require('../utils/dm-push-throttle');
 
     // Check if user has push notifications enabled for new messages
     const recipientPrefs = await pushTokenModel.getPushPreferencesByUserId(recipientId);
     const shouldSendPush = recipientPrefs?.new_message_push !== false;
     if (!shouldSendPush) return;
 
+    // HTTP-only: if recipient fetched/updated this conversation very recently, skip push.
+    if (conversationActivity.isRecentlyActive({ userId: recipientId, conversationId })) return;
+
+    // Anti-spam: throttle DM pushes per conversation per recipient.
+    // First message sends a push; subsequent messages within the cooldown window do not.
+    if (dmPushThrottle.isThrottled({ recipientId, conversationId })) return;
+
     // Get sender info for title
     const sender = await notificationModel.getUserEmailIdentity(senderId);
     if (!sender) return;
@@ -136,6 +145,8 @@ exports.sendDirectMessagePushIfEligible = async ({
         data: { type: 'new_message', conversationId },
       })
       .catch((err) => console.error('[Push] Direct message push failed:', err?.message));
+
+    dmPushThrottle.markSent({ recipientId, conversationId });
   } catch (err) {
     console.error('Direct message push notification skipped:', err?.message || err);
   }

src/utils/conversation-activity.js

@@ -0,0 +1,45 @@
+// backend-only  for "user is currently in a conversation" without relying on Socket.IO. to prevent edits for cross/front teams
+// We mark a user as active when they hit conversation-related HTTP endpoints.
+//
+//  This is per-node-process state (not shared across multiple server instances).
+
+const lastActiveByUserAndConversation = new Map(); // to avoid unbounded memory growth, we only store keys for active conversations and rely on expiration (see isRecentlyActive)
+
+const normalizeKey = (value) => {
+  if (typeof value !== 'string') return null;
+  const trimmed = value.trim();
+  return trimmed ? trimmed : null;
+};
+
+const makeKey = (userId, conversationId) => {
+  const uid = normalizeKey(userId);
+  const cid = normalizeKey(conversationId);
+  if (!uid || !cid) return null;
+  return `${uid}:${cid}`;
+};
+
+exports.markActive = ({ userId, conversationId, now = Date.now() } = {}) => {
+  const key = makeKey(userId, conversationId);
+  if (!key) return;
+  lastActiveByUserAndConversation.set(key, now);
+};
+
+exports.isRecentlyActive = ({
+  userId,
+  conversationId,
+  withinMs = 30_000,
+  now = Date.now(),
+} = {}) => {
+  const key = makeKey(userId, conversationId);
+  if (!key) return false;
+
+  const last = lastActiveByUserAndConversation.get(key);
+  if (!last) return false;
+
+  return now - last <= withinMs;
+};
+
+// For unit tests only
+exports._resetForTests = () => {
+  lastActiveByUserAndConversation.clear();
+};

src/utils/dm-push-throttle.js

@@ -0,0 +1,43 @@
+// Simple in-memory throttle to avoid spamming DM push notifications
+// For each recipient+conversation, we allow one push within the cooldown window (e.g. 60s)
+
+const lastPushByRecipientAndConversation = new Map(); // `${recipientId}:${conversationId}` -> timestampMs
+
+const normalizeKey = (value) => {
+  if (typeof value !== 'string') return null;
+  const trimmed = value.trim();
+  return trimmed ? trimmed : null;
+};
+
+const makeKey = (recipientId, conversationId) => {
+  const rid = normalizeKey(recipientId);
+  const cid = normalizeKey(conversationId);
+  if (!rid || !cid) return null;
+  return `${rid}:${cid}`;
+};
+
+exports.isThrottled = ({
+  recipientId,
+  conversationId,
+  withinMs = 60_000,
+  now = Date.now(),
+} = {}) => {
+  const key = makeKey(recipientId, conversationId);
+  if (!key) return false;
+
+  const last = lastPushByRecipientAndConversation.get(key);
+  if (!last) return false;
+
+  return now - last <= withinMs;
+};
+
+exports.markSent = ({ recipientId, conversationId, now = Date.now() } = {}) => {
+  const key = makeKey(recipientId, conversationId);
+  if (!key) return;
+  lastPushByRecipientAndConversation.set(key, now);
+};
+
+// For unit tests only
+exports._resetForTests = () => {
+  lastPushByRecipientAndConversation.clear();
+};

tests/dm-push-throttle.unit.test.js

@@ -0,0 +1,63 @@
+const throttle = require('../src/utils/dm-push-throttle');
+
+describe('dm-push-throttle', () => {
+  beforeEach(() => throttle._resetForTests());
+
+  it('is not throttled before any push is sent', () => {
+    expect(
+      throttle.isThrottled({ recipientId: 'u1', conversationId: 'c1', withinMs: 60_000, now: 1000 })
+    ).toBe(false);
+  });
+
+  it('throttles within window after markSent', () => {
+    throttle.markSent({ recipientId: 'u1', conversationId: 'c1', now: 1000 });
+
+    expect(
+      throttle.isThrottled({ recipientId: 'u1', conversationId: 'c1', withinMs: 60_000, now: 2000 })
+    ).toBe(true);
+  });
+
+  it('does not throttle after window passes', () => {
+    throttle.markSent({ recipientId: 'u1', conversationId: 'c1', now: 1000 });
+
+    expect(
+      throttle.isThrottled({ recipientId: 'u1', conversationId: 'c1', withinMs: 60_000, now: 62_000 })
+    ).toBe(false);
+  });
+
+  it('does not throttle when ids are missing/invalid', () => {
+    expect(throttle.isThrottled()).toBe(false);
+    expect(throttle.isThrottled({ recipientId: null, conversationId: 'c1' })).toBe(false);
+    expect(throttle.isThrottled({ recipientId: 'u1', conversationId: null })).toBe(false);
+    expect(throttle.isThrottled({ recipientId: 123, conversationId: 'c1' })).toBe(false);
+    expect(throttle.isThrottled({ recipientId: 'u1', conversationId: {} })).toBe(false);
+    expect(throttle.isThrottled({ recipientId: '   ', conversationId: 'c1' })).toBe(false);
+    expect(throttle.isThrottled({ recipientId: 'u1', conversationId: '   ' })).toBe(false);
+  });
+
+  it('ignores markSent when ids are invalid (no throw)', () => {
+    throttle.markSent();
+    throttle.markSent({ recipientId: null, conversationId: 'c1' });
+    throttle.markSent({ recipientId: 'u1', conversationId: null });
+    throttle.markSent({ recipientId: '   ', conversationId: 'c1' });
+    throttle.markSent({ recipientId: 'u1', conversationId: '   ' });
+
+    expect(throttle.isThrottled({ recipientId: 'u1', conversationId: 'c1' })).toBe(false);
+  });
+
+  it('trims ids so whitespace does not bypass throttling', () => {
+    throttle.markSent({ recipientId: ' u1 ', conversationId: ' c1 ', now: 1000 });
+
+    expect(
+      throttle.isThrottled({ recipientId: 'u1', conversationId: 'c1', withinMs: 60_000, now: 2000 })
+    ).toBe(true);
+  });
+
+  it('throttles per recipient+conversation key (different conversation not throttled)', () => {
+    throttle.markSent({ recipientId: 'u1', conversationId: 'c1', now: 1000 });
+
+    expect(
+      throttle.isThrottled({ recipientId: 'u1', conversationId: 'c2', withinMs: 60_000, now: 2000 })
+    ).toBe(false);
+  });
+});