feat(notification): handle deduping for tokens

Author: prathameshkurunkar7

raven_cloud/api/notification.py

@@ -8,7 +8,7 @@
 from frappe.utils.response import Response
 
 from raven_cloud.utils.fcm import get_app
-from raven_cloud.utils.notification import sanitize_fcm_data
+from raven_cloud.utils.notification import sanitize_fcm_data, get_background_job_id
 from raven_cloud.utils.rc_caching import get_push_tokens_for_user
 
 Message = dict[str, Any]
@@ -58,12 +58,16 @@ def send(messages: str, site_name: str):
     if isinstance(messages, str):
         messages = json.loads(messages)
 
+    job_id = get_background_job_id(messages, site_name)
+
     # Enqueue the job of sending notifications
     frappe.enqueue(
         _send,
         messages=messages,
         site_url=site_name,
         queue='short',
+        job_id=job_id,
+        deduplicate=True
     )
 
     # TODO: Return the response from api itself.
@@ -92,6 +96,12 @@ def _send(messages: Messages, site_url: str):
 
     try:
         for message in messages:
+            # remove duplicate tokens to avoid sending duplicate notifications
+            message_tokens = list(dict.fromkeys(message.get("tokens", [])))
+
+            if not message_tokens:
+                continue
+
             notification = None
             data = None
             webpush = None
@@ -138,7 +148,7 @@ def _send(messages: Messages, site_url: str):
             if message.get('data'):
                 data = sanitize_fcm_data(message['data'])
 
-            for token in message.get('tokens', []):
+            for token in message_tokens:
                 fcm_message = messaging.Message(
                     token=token,
                     notification=notification,
@@ -224,8 +234,17 @@ def send_to_users(messages: str, site_name: str):
     if isinstance(messages, str):
         messages = json.loads(messages)
 
+    job_id = get_background_job_id(messages, site_name)
+
     # enqueue the job of sending notifications
-    frappe.enqueue(_send_to_users, messages=messages, site_url=site_name, queue='short')
+    frappe.enqueue(
+        _send_to_users,
+        messages=messages,
+        site_url=site_name,
+        queue='short',
+        job_id=job_id,
+        deduplicate=True
+    )
 
     # TODO: Return the response from api itself.
     return
@@ -258,6 +277,9 @@ def _send_to_users(messages: Messages, site_url: str):
             for user in message.get("users", []):
                 message_tokens.extend(get_push_tokens_for_user(user, site_url))
 
+            # remove duplicate tokens to avoid sending duplicate notifications - not using set() because it doesn't preserve order
+            message_tokens = list(dict.fromkeys(message_tokens))
+
             if not message_tokens:
                 continue
 

raven_cloud/api/test_notification.py

@@ -0,0 +1,81 @@
+from types import SimpleNamespace
+from unittest.mock import patch
+
+import frappe
+from frappe.tests import IntegrationTestCase
+
+import raven_cloud.api.notification as notification_api
+
+
+class TestNotificationSendToUsers(IntegrationTestCase):
+    def setUp(self):
+        super().setUp()
+        self.site_name = f"rc-push-repro-{frappe.generate_hash(length=8)}.test"
+        frappe.get_doc({"doctype": "RC Site", "site": self.site_name}).insert(ignore_permissions=True)
+
+    def tearDown(self):
+        frappe.db.rollback()
+        super().tearDown()
+
+    def create_site_user_with_token(self, user_id: str, token: str):
+        site_user = frappe.get_doc(
+            {"doctype": "RC Site User", "site": self.site_name, "user_id": user_id}
+        ).insert(ignore_permissions=True)
+        frappe.get_doc(
+            {"doctype": "RC Site User Token", "user": site_user.name, "fcm_token": token}
+        ).insert(ignore_permissions=True)
+
+    def test_send_to_users_should_send_once_for_duplicate_users(self):
+        self.create_site_user_with_token("prathamesh@example.com", "tok-prathamesh")
+
+        sent_tokens = []
+
+        with patch.object(notification_api, "get_app", return_value=object()), patch.object(
+            notification_api.messaging,
+            "Message",
+            side_effect=lambda **kwargs: SimpleNamespace(token=kwargs["token"]),
+        ), patch.object(
+            notification_api.messaging,
+            "send_each",
+            side_effect=lambda fcm_messages, app=None: (
+                sent_tokens.extend(message.token for message in fcm_messages)
+                or SimpleNamespace(failure_count=0, responses=[])
+            ),
+        ):
+            notification_api._send_to_users(
+                [{"users": ["prathamesh@example.com", "prathamesh@example.com", "prathamesh@example.com"]}],
+                self.site_name,
+            )
+
+        log = frappe.get_last_doc("RC Push Notification Log", filters={"site": self.site_name})
+
+        self.assertEqual(["tok-prathamesh"], sent_tokens)
+        self.assertEqual(1, log.number_of_tokens)
+
+    def test_send_to_users_should_send_once_for_shared_token(self):
+        self.create_site_user_with_token("prathamesh@example.com", "tok-shared")
+        self.create_site_user_with_token("aditya@example.com", "tok-shared")
+
+        sent_tokens = []
+
+        with patch.object(notification_api, "get_app", return_value=object()), patch.object(
+            notification_api.messaging,
+            "Message",
+            side_effect=lambda **kwargs: SimpleNamespace(token=kwargs["token"]),
+        ), patch.object(
+            notification_api.messaging,
+            "send_each",
+            side_effect=lambda fcm_messages, app=None: (
+                sent_tokens.extend(message.token for message in fcm_messages)
+                or SimpleNamespace(failure_count=0, responses=[])
+            ),
+        ):
+            notification_api._send_to_users(
+                [{"users": ["prathamesh@example.com", "aditya@example.com"]}],
+                self.site_name,
+            )
+
+        log = frappe.get_last_doc("RC Push Notification Log", filters={"site": self.site_name})
+
+        self.assertEqual(["tok-shared"], sent_tokens)
+        self.assertEqual(1, log.number_of_tokens)

raven_cloud/utils/notification.py

@@ -1,4 +1,4 @@
-# import frappe
+import frappe
 
 def sanitize_fcm_data(data_dict):
     """
@@ -7,9 +7,19 @@ def sanitize_fcm_data(data_dict):
     """
     if not data_dict:
         return None
-    
+
     sanitized = {}
     for key, value in data_dict.items():
         # convert None to empty string, everything else to string
         sanitized[key] = str(value) if value is not None else ""
-    return sanitized
+    return sanitized
+
+
+def get_background_job_id(messages, site_name: str) -> str:
+    first = (messages or [{}])[0]
+    data = first.get("data") or {}
+
+    channel_id = data.get("channel_id") or first.get("channel_id") or "no-channel"
+    message_id = data.get("message_id") or first.get("message_id") or "no-message"
+
+    return f"rc_push_{frappe.scrub(site_name)}_{frappe.scrub(channel_id)}_{frappe.scrub(message_id)}"