fix: email notification duplicates (#3719)

apiserver/plane/bgtasks/email_notification_task.py

@@ -1,9 +1,9 @@
 from datetime import datetime
 from bs4 import BeautifulSoup
 
-
 # Third party imports
 from celery import shared_task
+from sentry_sdk import capture_exception
 
 # Django imports
 from django.utils import timezone
@@ -16,6 +16,17 @@ from plane.db.models import EmailNotificationLog, User, Issue
 from plane.license.utils.instance_value import get_email_configuration
 from plane.settings.redis import redis_instance
 
+# acquire and delete redis lock
+def acquire_lock(lock_id, expire_time=300):
+    redis_client = redis_instance()
+    """Attempt to acquire a lock with a specified expiration time."""
+    return redis_client.set(lock_id, 'true', nx=True, ex=expire_time)
+
+def release_lock(lock_id):
+    """Release a lock."""
+    redis_client = redis_instance()
+    redis_client.delete(lock_id)
+
 @shared_task
 def stack_email_notification():
     # get all email notifications
@@ -142,11 +153,20 @@ def process_html_content(content):
         processed_content_list.append(processed_content)  
     return processed_content_list
 
+
 @shared_task
 def send_email_notification(
     issue_id, notification_data, receiver_id, email_notification_ids
 ):
+    # Convert UUIDs to a sorted, concatenated string
+    sorted_ids = sorted(email_notification_ids)
+    ids_str = "_".join(str(id) for id in sorted_ids)
+    lock_id = f"send_email_notif_{issue_id}_{receiver_id}_{ids_str}"
+
+    # acquire the lock for sending emails
     try:
+        if acquire_lock(lock_id=lock_id):
+            # get the redis instance
             ri = redis_instance()
             base_api = (ri.get(str(issue_id)).decode())
             data = create_payload(notification_data=notification_data)
@@ -268,9 +288,18 @@ def send_email_notification(
                 EmailNotificationLog.objects.filter(
                     pk__in=email_notification_ids
                 ).update(sent_at=timezone.now())
+
+                # release the lock
+                release_lock(lock_id=lock_id)
                 return
             except Exception as e:
-            print(e)
+                capture_exception(e)
+                # release the lock
+                release_lock(lock_id=lock_id)
                 return
-    except Issue.DoesNotExist:
+        else:
+            print("Duplicate task recived. Skipping...")
+            return
+    except (Issue.DoesNotExist, User.DoesNotExist) as e:
+        release_lock(lock_id=lock_id)
         return