feat: added buffer to immediate email notification (#37741)

cms/envs/devstack.py

@@ -292,6 +292,9 @@ CREDENTIALS_PUBLIC_SERVICE_URL = 'http://localhost:18150'
 ########################## ORA MFE APP ##############################
 ORA_MICROFRONTEND_URL = 'http://localhost:1992'
 
+########################## LEARNER HOME APP ##############################
+LEARNER_HOME_MICROFRONTEND_URL = 'http://localhost:1996'
+
 ############################ AI_TRANSLATIONS ##################################
 AI_TRANSLATIONS_API_URL = 'http://localhost:18760/api/v1'
 

lms/envs/common.py

@@ -3482,10 +3482,6 @@ ENTERPRISE_MANUAL_REPORTING_CUSTOMER_UUIDS = []
 
 AVAILABLE_DISCUSSION_TOURS = []
 
-############## NOTIFICATIONS ##############
-NOTIFICATION_TYPE_ICONS = {}
-DEFAULT_NOTIFICATION_ICON_URL = ""
-
 ############## SELF PACED EMAIL ##############
 SELF_PACED_BANNER_URL = ""
 SELF_PACED_CLOUD_URL = ""

openedx/core/djangoapps/notifications/email/tasks.py

@@ -1,14 +1,20 @@
 """
 Celery tasks for sending email notifications
 """
+from datetime import datetime, timedelta
+
 from bs4 import BeautifulSoup
 from celery import shared_task
 from celery.utils.log import get_task_logger
+from django.conf import settings
 from django.contrib.auth import get_user_model
+from django.db import transaction
 from django.utils.translation import gettext as _, override as translation_override
-from edx_ace import ace
+from edx_ace import ace, presentation
+from edx_ace.channel.django_email import DjangoEmailChannel
 from edx_ace.recipient import Recipient
 from edx_django_utils.monitoring import set_code_owner_attribute
+from opaque_keys.edx.keys import CourseKey
 
 from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
 from openedx.core.djangoapps.notifications.models import (
@@ -29,7 +35,8 @@ from .utils import (
     get_text_for_notification_type,
     is_email_notification_flag_enabled,
 )
-
+from ..base_notification import COURSE_NOTIFICATION_APPS
+from ..config.waffle import ENABLE_EMAIL_NOTIFICATIONS
 
 User = get_user_model()
 logger = get_task_logger(__name__)
@@ -51,14 +58,27 @@ def get_audience_for_cadence_email(cadence_type):
     return users
 
 
-def send_digest_email_to_user(user, cadence_type, start_date, end_date, user_language='en', courses_data=None):
+def get_buffer_minutes() -> int:
+    """Get configured buffer period in minutes."""
+    return getattr(settings, 'NOTIFICATION_IMMEDIATE_EMAIL_BUFFER_MINUTES', 0)
+
+
+def send_digest_email_to_user(
+    user: User,
+    cadence_type: str,
+    start_date: datetime,
+    end_date: datetime,
+    user_language: str = 'en',
+    courses_data: dict = None
+):
     """
     Send [cadence_type] email to user.
     Cadence Type can be EmailCadence.DAILY or EmailCadence.WEEKLY
     start_date: Datetime object
     end_date: Datetime object
     """
-    if cadence_type not in [EmailCadence.DAILY, EmailCadence.WEEKLY]:
+
+    if cadence_type not in [EmailCadence.IMMEDIATELY, EmailCadence.DAILY, EmailCadence.WEEKLY]:
         raise ValueError('Invalid cadence_type')
     logger.info(f'<Email Cadence> Sending email to user {user.username} ==Temp Log==')
     if not user.has_usable_password():
@@ -75,13 +95,17 @@ def send_digest_email_to_user(user, cadence_type, start_date, end_date, user_lan
 
     with translation_override(user_language):
         preferences = NotificationPreference.objects.filter(user=user)
-        notifications = filter_email_enabled_notifications(notifications, preferences, user,
-                                                           cadence_type=cadence_type)
-
-        if not notifications:
+        notifications_list = filter_email_enabled_notifications(
+            notifications,
+            preferences,
+            user,
+            cadence_type=cadence_type
+        )
+        if not notifications_list:
             logger.info(f'<Email Cadence> No filtered notification for {user.username} ==Temp Log==')
             return
-        apps_dict = create_app_notifications_dict(notifications)
+
+        apps_dict = create_app_notifications_dict(notifications_list)
         message_context = create_email_digest_context(apps_dict, user.username, start_date, end_date,
                                                       cadence_type, courses_data=courses_data)
         recipient = Recipient(user.id, user.email)
@@ -91,7 +115,8 @@ def send_digest_email_to_user(user, cadence_type, start_date, end_date, user_lan
         message = add_headers_to_email_message(message, message_context)
         message.options['skip_disable_user_policy'] = True
         ace.send(message)
-        send_user_email_digest_sent_event(user, cadence_type, notifications, message_context)
+        notifications.update(email_sent_on=datetime.now())
+        send_user_email_digest_sent_event(user, cadence_type, notifications_list, message_context)
         logger.info(f'<Email Cadence> Email sent to {user.username} ==Temp Log==')
 
 
@@ -119,14 +144,17 @@ def send_immediate_cadence_email(email_notification_mapping, course_key):
     Parameters:
         email_notification_mapping: Dictionary of user_id and Notification object
         course_key: Course key for which the email is sent
+    1. First notification → Send immediately
+    2. Second notification → Schedule buffer job (15 min)
+    3. Third+ notifications → Just mark as scheduled (no new job)
     """
     if not email_notification_mapping:
         return
     user_list = email_notification_mapping.keys()
-    users = User.objects.filter(id__in=user_list)
+    users = list(User.objects.filter(id__in=user_list))
     language_prefs = get_language_preference_for_users(user_list)
     course_name = get_course_info(course_key).get("name", course_key)
-    for user in users.iterator(chunk_size=100):
+    for user in users:
         if not user.has_usable_password():
             logger.info(f'<Immediate Email> User is disabled {user.username}')
             continue
@@ -137,26 +165,342 @@ def send_immediate_cadence_email(email_notification_mapping, course_key):
         if not notification:
             logger.info(f'<Immediate Email> No notification for {user.username}')
             continue
+        # THE CORE DECISION LOGIC
+        decision = decide_email_action(user, course_key, notification)
+        user_language = language_prefs.get(user.id, 'en')
 
-        language = language_prefs.get(user.id, 'en')
-        with translation_override(language):
-            soup = BeautifulSoup(notification.content, "html.parser")
-            title = _("New Course Update") if notification.notification_type == "course_updates" else soup.get_text()
-            message_context = create_email_template_context(user.username)
-            message_context.update({
-                "course_id": course_key,
-                "course_name": course_name,
-                "content_url": notification.content_url,
-                "content_title": title,
-                "footer_email_reason": _(
-                    "You are receiving this email because you are enrolled in the edX course "
-                ) + str(course_name),
-                "content": notification.content_context.get("email_content", notification.content),
-                "view_text": get_text_for_notification_type(notification.notification_type),
-            })
+        if decision == 'send_immediate':
+            # CASE 1: First notification - send immediately
+            logger.info(
+                f"Email Buffered Digest: Sending immediate email for notification IDs: {notification.id}",
+            )
+            send_immediate_email(
+                user=user,
+                notification=notification,
+                course_key=course_key,
+                course_name=course_name,
+                user_language=user_language
+            )
+
+        elif decision == 'schedule_buffer':
+            # CASE 2: Second notification - schedule buffer job
+            logger.info(
+                f"Email Buffered Digest: Scheduling buffer for notification IDs: {notification.id}",
+            )
+            schedule_digest_buffer(
+                user=user,
+                notification=notification,
+                course_key=course_key,
+                user_language=user_language
+            )
+
+        elif decision == 'add_to_buffer':
+            logger.info(
+                f"Email Buffered Digest: "
+                f"Email Buffered Digest:Adding to existing buffer for notification IDs: {notification.id}\n",
+            )
+            # CASE 3: Third+ notification - just mark as scheduled
+            add_to_existing_buffer(notification)
+
+
+@transaction.atomic
+def decide_email_action(user: User, course_key: str, notification: Notification) -> str:
+    """
+    Decide what to do with this notification.
+
+    Logic:
+    - No recent email? → send_immediate (1st)
+    - Recent email + no buffer? → schedule_buffer (2nd)
+    - Recent email + buffer exists? → add_to_buffer (3rd+)
+
+    Returns:
+        'send_immediate', 'schedule_buffer', or 'add_to_buffer'
+    """
+    buffer_minutes = get_buffer_minutes()
+    buffer_threshold = datetime.now() - timedelta(minutes=buffer_minutes)
+
+    # Use select_for_update to prevent race conditions
+    recent_notifications = Notification.objects.select_for_update().filter(
+        user=user,
+        course_id=course_key,
+        created__gte=buffer_threshold
+    )
+
+    # Check if any email was sent recently
+    has_recent_email = recent_notifications.filter(
+        email_sent_on__isnull=False,
+        email_sent_on__gte=buffer_threshold
+    ).exists()
+
+    if not has_recent_email:
+        # CASE 1: No recent email → First notification
+        logger.info(f'[{user.username}] CASE 1: First notification, sending immediately')
+        return 'send_immediate'
+
+    # Check if buffer job already exists
+    # Buffer exists if there are notifications marked as scheduled
+    has_scheduled_buffer = recent_notifications.filter(
+        email_scheduled=True
+    ).exists()
+
+    if not has_scheduled_buffer:
+        # CASE 2: Recent email but no buffer → Second notification
+        logger.info(f'[{user.username}] CASE 2: Second notification, scheduling buffer')
+        return 'schedule_buffer'
+
+    # CASE 3: Buffer already exists → Third+ notification
+    logger.info(f'[{user.username}] CASE 3: Third+ notification, adding to buffer')
+    return 'add_to_buffer'
+
+
+def send_immediate_email(
+    user: User,
+    notification: Notification,
+    course_key: str,
+    course_name: str,
+    user_language: str
+) -> None:
+    """Send immediate email for the first notification."""
+    with translation_override(user_language):
+        soup = BeautifulSoup(notification.content, "html.parser")
+        title = (
+            _("New Course Update")
+            if notification.notification_type == "course_updates"
+            else soup.get_text()
+        )
+
+        message_context = create_email_template_context(user.username)
+        message_context.update({
+            "course_id": course_key,
+            "course_name": course_name,
+            "content_url": notification.content_url,
+            "content_title": title,
+            "footer_email_reason": _(
+                "You are receiving this email because you are enrolled in "
+                "the edX course "
+            ) + str(course_name),
+            "content": notification.content_context.get(
+                "email_content",
+                notification.content
+            ),
+            "view_text": get_text_for_notification_type(
+                notification.notification_type
+            ),
+        })
+
+        message = EmailNotificationMessageType(
+            app_label="notifications",
+            name="immediate_email"
+        ).personalize(
+            Recipient(user.id, user.email),
+            user_language,
+            message_context
+        )
+
+        message = add_headers_to_email_message(message, message_context)
+        ace.send(message)
+
+        # Mark as sent - this starts the buffer period
+        notification.email_sent_on = datetime.now()
+        notification.save(update_fields=["email_sent_on"])
+
+        logger.info(f'Email Buffered Digest: ✓ Sent immediate email to {user.username}')
+
+        send_immediate_email_digest_sent_event(
+            user,
+            EmailCadence.IMMEDIATELY,
+            notification
+        )
+
+
+def schedule_digest_buffer(
+    user: User,
+    notification: Notification,
+    course_key: str,
+    user_language: str
+) -> None:
+    """
+    Schedule a buffer job for digest email.
+    Called for the SECOND notification only.
+    """
+    buffer_minutes = get_buffer_minutes()
+
+    # Find when we last sent an email
+    last_sent = Notification.objects.filter(
+        user=user,
+        course_id=course_key,
+        email_sent_on__isnull=False
+    ).order_by('-email_sent_on').first()
+
+    if not last_sent:
+        logger.error(f'No last_sent found for {user.username}')
+        return
+
+    start_date = last_sent.email_sent_on
+    scheduled_time = datetime.now() + timedelta(minutes=buffer_minutes)
+
+    # Mark this notification as scheduled FIRST
+    notification.email_scheduled = True
+    notification.save(update_fields=['email_scheduled'])
+
+    # Then schedule the digest task
+    send_buffered_digest.apply_async(
+        kwargs={
+            'user_id': user.id,
+            'course_key': str(course_key),
+            'start_date': start_date,
+            'user_language': user_language,
+        },
+        eta=scheduled_time
+    )
+
+    logger.info(
+        f'Email Buffered Digest: ✓ Scheduled digest for {user.username} at {scheduled_time}, '
+        f'marked notification {notification.id} as scheduled'
+    )
+
+
+def add_to_existing_buffer(notification: Notification) -> None:
+    """
+    Add notification to existing buffer.
+    Just mark as scheduled - the existing job will find it!
+
+    Called for THIRD+ notifications.
+    """
+    notification.email_scheduled = True
+    notification.save(update_fields=['email_scheduled'])
+
+    logger.info(
+        f'✓ Marked notification {notification.id} as scheduled '
+        f'(will be picked up by existing buffer job)'
+    )
+
+
+@shared_task(bind=True, max_retries=3, default_retry_delay=60)
+@set_code_owner_attribute
+def send_buffered_digest(
+    self,
+    user_id: int,
+    course_key: str,
+    start_date: datetime,
+    user_language: str
+) -> None:
+    """
+    Send digest email with all buffered notifications.
+
+    This collects ALL notifications where email_scheduled=True
+    for this user+course within the buffer period.
+
+    Simple! No task ID tracking needed.
+    """
+    try:
+        # Re-check feature flags
+        if not ENABLE_EMAIL_NOTIFICATIONS.is_enabled():
+            logger.info('Email notifications disabled, cancelling digest')
+            return
+
+        user = User.objects.get(id=user_id)
+
+        if not user.has_usable_password():
+            logger.info(f'User {user.username} disabled')
+            return
+
+        end_date = datetime.now()
+
+        # Get ALL scheduled notifications
+        # Simple query: just find where email_scheduled=True
+        scheduled_notifications = Notification.objects.filter(
+            user=user,
+            course_id=course_key,
+            email_scheduled=True,  # This is all we need!
+            created__gte=start_date,
+            created__lte=end_date,
+            app_name__in=COURSE_NOTIFICATION_APPS
+        )
+
+        if not scheduled_notifications.exists():
+            logger.info(f'Email Buffered Digest: No scheduled notifications for {user.username}')
+            return
+        logger.info(
+            "Email Buffered Digest: "
+            f'Found {scheduled_notifications.count()} scheduled '
+            f'notifications for {user.username}'
+        )
+        with translation_override(user_language):
+            # Filter based on preferences
+            preferences = NotificationPreference.objects.filter(user=user)
+            notifications_list = filter_email_enabled_notifications(
+                scheduled_notifications,
+                preferences,
+                user,
+                cadence_type=EmailCadence.IMMEDIATELY
+            )
+
+            if not notifications_list:
+                logger.info(f'No email-enabled notifications for {user.username}')
+                # Reset flags even if we don't send
+                scheduled_notifications.update(email_scheduled=False)
+                return
+
+            # Build digest email
+            apps_dict = create_app_notifications_dict(notifications_list)
+            course_key = CourseKey.from_string(course_key)
+            course_name = get_course_info(course_key).get("name", course_key)
+
+            message_context = create_email_digest_context(
+                apps_dict,
+                user.username,
+                start_date,
+                end_date,
+                EmailCadence.IMMEDIATELY,
+                courses_data={course_key: {'name': course_name}}
+            )
+
+            # Send digest
+            recipient = Recipient(user.id, user.email)
             message = EmailNotificationMessageType(
-                app_label="notifications", name="immediate_email"
-            ).personalize(Recipient(user.id, user.email), language, message_context)
+                app_label="notifications",
+                name="email_digest"
+            ).personalize(recipient, user_language, message_context)
+
             message = add_headers_to_email_message(message, message_context)
+
+            render_msg = presentation.render(DjangoEmailChannel, message)
+
+            print(render_msg.body)  # For debugging purposes
+            print(render_msg.body_html)
             ace.send(message)
-            send_immediate_email_digest_sent_event(user, EmailCadence.IMMEDIATELY, notification)
+
+            # Mark ALL as sent and clear scheduled flag
+            notification_ids = [n.id for n in notifications_list]
+            logger.info(
+                f'Email Buffered Digest: Sent buffered digest to {user.username} for '""
+                f'notifications IDs: {notification_ids}'
+
+            )
+            updated_count = scheduled_notifications.filter(
+                id__in=notification_ids
+            ).update(
+                email_sent_on=datetime.now(),
+                email_scheduled=False  # Clear the flag
+            )
+
+            logger.info(
+                f'Email Buffered Digest: ✓ Sent buffered digest to {user.username} with '
+                f'{updated_count} notifications'
+            )
+
+            send_user_email_digest_sent_event(
+                user,
+                EmailCadence.IMMEDIATELY,
+                notifications_list,
+                message_context
+            )
+
+    except User.DoesNotExist:
+        logger.error(f'Email Buffered Digest: User {user_id} not found')
+
+    except Exception as exc:
+        logger.exception(f'Email Buffered Digest: Failed to send buffered digest: {exc}')
+        retry_countdown = 60 * (2 ** self.request.retries)
+        raise self.retry(exc=exc, countdown=retry_countdown)

openedx/core/djangoapps/notifications/email/tests/test_tasks.py

@@ -2,30 +2,43 @@
 Test cases for notifications/email/tasks.py
 """
 import datetime
+from datetime import datetime, timedelta
+from unittest.mock import Mock, patch
+
 import ddt
-
-from unittest.mock import patch
-
+from django.contrib.auth import get_user_model
+from django.test import override_settings
+from django.utils import timezone
 from edx_toggles.toggles.testutils import override_waffle_flag
+from freezegun import freeze_time
 
 from common.djangoapps.student.tests.factories import UserFactory
-from openedx.core.djangoapps.notifications.config.waffle import (
-    ENABLE_NOTIFICATIONS, ENABLE_EMAIL_NOTIFICATIONS
-)
-from openedx.core.djangoapps.notifications.tasks import send_notifications
-from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
+from openedx.core.djangoapps.notifications.config.waffle import ENABLE_EMAIL_NOTIFICATIONS, ENABLE_NOTIFICATIONS
 from openedx.core.djangoapps.notifications.email.tasks import (
+    add_to_existing_buffer,
+    decide_email_action,
     get_audience_for_cadence_email,
+    schedule_digest_buffer,
+    send_buffered_digest,
     send_digest_email_to_all_users,
-    send_digest_email_to_user
+    send_digest_email_to_user,
+    send_immediate_cadence_email,
+    send_immediate_email
 )
 from openedx.core.djangoapps.notifications.email.utils import get_start_end_date
-from openedx.core.djangoapps.notifications.models import NotificationPreference
+from openedx.core.djangoapps.notifications.email_notifications import EmailCadence
+from openedx.core.djangoapps.notifications.models import (
+    Notification,
+    NotificationPreference
+)
+from openedx.core.djangoapps.notifications.tasks import send_notifications
 from xmodule.modulestore.tests.django_utils import ModuleStoreTestCase
 from xmodule.modulestore.tests.factories import CourseFactory
 
 from .utils import create_notification
 
+User = get_user_model()
+
 
 @ddt.ddt
 class TestEmailDigestForUser(ModuleStoreTestCase):
@@ -57,7 +70,7 @@ class TestEmailDigestForUser(ModuleStoreTestCase):
         """
         Tests email is sent iff waffle flag is enabled
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, flag_value):
@@ -70,7 +83,7 @@ class TestEmailDigestForUser(ModuleStoreTestCase):
         Tests email is not sent if notification is created on next day
         """
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
-        create_notification(self.user, self.course.id, created=end_date + datetime.timedelta(minutes=2))
+        create_notification(self.user, self.course.id, created=end_date + timedelta(minutes=2))
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
         assert not mock_func.called
@@ -81,7 +94,7 @@ class TestEmailDigestForUser(ModuleStoreTestCase):
         """
         Tests email is not sent to disabled user
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
         if value:
@@ -98,21 +111,21 @@ class TestEmailDigestForUser(ModuleStoreTestCase):
         Tests email is not sent if notification is created day before yesterday
         """
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1, minutes=18)
+        created_date = datetime.now() - timedelta(days=1, minutes=18)
         create_notification(self.user, self.course.id, created=created_date)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
         assert not mock_func.called
 
     @ddt.data(
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(days=1, minutes=30), False),
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(minutes=10), True),
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(days=1), True),
-        (EmailCadence.DAILY, datetime.datetime.now() + datetime.timedelta(minutes=20), False),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(days=7, minutes=30), False),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(days=7), True),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(minutes=20), True),
-        (EmailCadence.WEEKLY, datetime.datetime.now() + datetime.timedelta(minutes=20), False),
+        (EmailCadence.DAILY, datetime.now() - timedelta(days=1, minutes=30), False),
+        (EmailCadence.DAILY, datetime.now() - timedelta(minutes=10), True),
+        (EmailCadence.DAILY, datetime.now() - timedelta(days=1), True),
+        (EmailCadence.DAILY, datetime.now() + timedelta(minutes=20), False),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(days=7, minutes=30), False),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(days=7), True),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(minutes=20), True),
+        (EmailCadence.WEEKLY, datetime.now() + timedelta(minutes=20), False),
     )
     @ddt.unpack
     @patch('edx_ace.ace.send')
@@ -157,7 +170,7 @@ class TestEmailDigestForUserWithAccountPreferences(ModuleStoreTestCase):
         """
         Tests email is sent iff waffle flag is enabled
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, flag_value):
@@ -170,7 +183,7 @@ class TestEmailDigestForUserWithAccountPreferences(ModuleStoreTestCase):
         Tests email is not sent if notification is created on next day
         """
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
-        create_notification(self.user, self.course.id, created=end_date + datetime.timedelta(minutes=2))
+        create_notification(self.user, self.course.id, created=end_date + timedelta(minutes=2))
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
         assert not mock_func.called
@@ -181,7 +194,7 @@ class TestEmailDigestForUserWithAccountPreferences(ModuleStoreTestCase):
         """
         Tests email is not sent to disabled user
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
         if value:
@@ -198,21 +211,21 @@ class TestEmailDigestForUserWithAccountPreferences(ModuleStoreTestCase):
         Tests email is not sent if notification is created day before yesterday
         """
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1, minutes=18)
+        created_date = datetime.now() - timedelta(days=1, minutes=18)
         create_notification(self.user, self.course.id, created=created_date)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
         assert not mock_func.called
 
     @ddt.data(
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(days=1, minutes=30), False),
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(minutes=10), True),
-        (EmailCadence.DAILY, datetime.datetime.now() - datetime.timedelta(days=1), True),
-        (EmailCadence.DAILY, datetime.datetime.now() + datetime.timedelta(minutes=20), False),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(days=7, minutes=30), False),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(days=7), True),
-        (EmailCadence.WEEKLY, datetime.datetime.now() - datetime.timedelta(minutes=20), True),
-        (EmailCadence.WEEKLY, datetime.datetime.now() + datetime.timedelta(minutes=20), False),
+        (EmailCadence.DAILY, datetime.now() - timedelta(days=1, minutes=30), False),
+        (EmailCadence.DAILY, datetime.now() - timedelta(minutes=10), True),
+        (EmailCadence.DAILY, datetime.now() - timedelta(days=1), True),
+        (EmailCadence.DAILY, datetime.now() + timedelta(minutes=20), False),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(days=7, minutes=30), False),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(days=7), True),
+        (EmailCadence.WEEKLY, datetime.now() - timedelta(minutes=20), True),
+        (EmailCadence.WEEKLY, datetime.now() + timedelta(minutes=20), False),
     )
     @ddt.unpack
     @patch('edx_ace.ace.send')
@@ -255,7 +268,7 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
         """
         Tests email sending function is called if user has notification
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_all_users(EmailCadence.DAILY)
@@ -267,7 +280,7 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
         Tests email sending function is not called if user has notification
         which is not in duration
         """
-        created_date = datetime.datetime.now() - datetime.timedelta(days=10)
+        created_date = datetime.now() - timedelta(days=10)
         create_notification(self.user, self.course.id, created=created_date)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_all_users(EmailCadence.DAILY)
@@ -275,7 +288,7 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
 
     @patch('edx_ace.ace.send')
     def test_email_is_sent_to_user_when_task_is_called(self, mock_func):
-        created_date = datetime.datetime.now() - datetime.timedelta(days=1)
+        created_date = datetime.now() - timedelta(days=1)
         create_notification(self.user, self.course.id, created=created_date)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_all_users(EmailCadence.DAILY)
@@ -294,7 +307,7 @@ class TestEmailDigestAudience(ModuleStoreTestCase):
         Tests email is sent only when notifications with email=True exists
         """
         start_date, end_date = get_start_end_date(EmailCadence.DAILY)
-        created_date = datetime.datetime.now() - datetime.timedelta(hours=23, minutes=59)
+        created_date = datetime.now() - timedelta(hours=23, minutes=59)
         create_notification(self.user, self.course.id, created=created_date, email=email_value)
         with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
             send_digest_email_to_user(self.user, EmailCadence.DAILY, start_date, end_date)
@@ -316,7 +329,7 @@ class TestAccountPreferences(ModuleStoreTestCase):
         self.course = CourseFactory.create(display_name='test course', run="Testing_course")
         self.preference, _ = NotificationPreference.objects.get_or_create(user=self.user, app="discussion",
                                                                           type="new_discussion_post")
-        created_date = datetime.datetime.now() - datetime.timedelta(hours=23)
+        created_date = datetime.now() - timedelta(hours=23)
         create_notification(self.user, self.course.id, notification_type='new_discussion_post', created=created_date)
 
     @patch('edx_ace.ace.send')
@@ -360,57 +373,829 @@ class TestAccountPreferences(ModuleStoreTestCase):
         assert not mock_func.called
 
 
-class TestImmediateEmail(ModuleStoreTestCase):
+class TestImmediateEmailNotifications(ModuleStoreTestCase):
     """
-    Tests immediate email
+    Tests for immediate email notifications functionality.
+    Covers both high-level notification triggering and specific task execution logic.
     """
 
     def setUp(self):
         """
-        Setup
+        Shared setup for user, course, and default preferences.
         """
         super().setUp()
         self.user = UserFactory()
         self.course = CourseFactory.create(display_name='test course', run="Testing_course")
+
+        # Ensure a clean slate for this user
+        NotificationPreference.objects.filter(user=self.user).delete()
+
+        # Create a default preference object that can be modified by individual tests
         self.preference, _ = NotificationPreference.objects.get_or_create(
             user=self.user,
             type='new_discussion_post',
-            app='discussion'
+            app='discussion',
+            defaults={
+                'web': True,
+                'push': True,
+                'email': True,
+                'email_cadence': EmailCadence.IMMEDIATELY
+            }
         )
 
     @patch('edx_ace.ace.send')
-    def test_email_sent_when_cadence_is_immediate(self, mock_func):
+    def test_email_sent_when_cadence_is_immediate(self, mock_ace_send):
         """
-        Tests email is sent when cadence is immediate
+        Tests that an email is sent via send_notifications when cadence is set to IMMEDIATE.
         """
-
+        # Ensure preference matches test case
         self.preference.email = True
         self.preference.email_cadence = EmailCadence.IMMEDIATELY
         self.preference.save()
+
         context = {
             'username': 'User',
             'post_title': 'title'
         }
-        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
-            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
-                send_notifications([self.user.id], str(self.course.id), 'discussion',
-                                   'new_discussion_post', context, 'http://test.url')
-        assert mock_func.call_count == 1
+
+        with (
+            override_waffle_flag(ENABLE_NOTIFICATIONS, True),
+            override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True)
+        ):
+            send_notifications(
+                [self.user.id],
+                str(self.course.id),
+                'discussion',
+                'new_discussion_post',
+                context,
+                'http://test.url'
+            )
+
+        assert mock_ace_send.call_count == 1
 
     @patch('edx_ace.ace.send')
-    def test_email_not_sent_when_cadence_is_not_immediate(self, mock_func):
+    def test_email_not_sent_when_cadence_is_not_immediate(self, mock_ace_send):
         """
-        Tests email is not sent when cadence is not immediate
+        Tests that an email is NOT sent via send_notifications when cadence is DAILY.
         """
+        # Modify preference for this test case
         self.preference.email = True
         self.preference.email_cadence = EmailCadence.DAILY
         self.preference.save()
+
         context = {
             'replier_name': 'User',
             'post_title': 'title'
         }
+
+        with (
+            override_waffle_flag(ENABLE_NOTIFICATIONS, True),
+            override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True)
+        ):
+            send_notifications(
+                [self.user.id],
+                str(self.course.id),
+                'discussion',
+                'new_response',
+                context,
+                'http://test.url'
+            )
+
+        assert mock_ace_send.call_count == 0
+
+
+@ddt.ddt
+class TestDecideEmailAction(ModuleStoreTestCase):
+    """Test the core decision logic for email buffering."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create()
+        self.course_key = str(self.course.id)
+
+    def _create_notification(self, **kwargs):
+        """Helper to create notification with defaults."""
+        defaults = {
+            'user': self.user,
+            'course_id': self.course_key,
+            'app_name': 'discussion',
+            'notification_type': 'new_discussion_post',
+            'content_url': 'http://example.com',
+            'email': True,
+        }
+        defaults.update(kwargs)
+        return Notification.objects.create(**defaults)
+
+    @freeze_time("2025-12-15 10:00:00")
+    def test_first_notification_sends_immediate(self):
+        """Test that first notification triggers immediate send."""
+        notification = self._create_notification()
+
+        decision = decide_email_action(self.user, self.course_key, notification)
+
+        assert decision == 'send_immediate'
+
+    @freeze_time("2025-12-15 10:00:00")
+    def test_second_notification_schedules_buffer(self):
+        """Test that second notification within buffer schedules digest."""
+        # First notification - sent 5 minutes ago
+        self._create_notification(
+            email_sent_on=timezone.now() - timedelta(minutes=5)
+        )
+
+        # Second notification - should schedule buffer
+        notification = self._create_notification()
+
+        decision = decide_email_action(self.user, self.course_key, notification)
+
+        assert decision == 'schedule_buffer'
+
+    @freeze_time("2025-12-15 10:00:00")
+    def test_third_notification_adds_to_buffer(self):
+        """Test that third notification just marks as scheduled."""
+        # First notification - sent 5 minutes ago
+        self._create_notification(
+            email_sent_on=timezone.now() - timedelta(minutes=5)
+        )
+
+        # Second notification - scheduled
+        self._create_notification(email_scheduled=True)
+
+        # Third notification - should add to existing buffer
+        notification = self._create_notification()
+
+        decision = decide_email_action(self.user, self.course_key, notification)
+
+        assert decision == 'add_to_buffer'
+
+    @freeze_time("2025-12-15 10:00:00")
+    @override_settings(NOTIFICATION_EMAIL_BUFFER_MINUTES=15)
+    def test_old_email_triggers_new_immediate_send(self):
+        """Test that email sent outside buffer period triggers new immediate send."""
+        # Email sent 20 minutes ago (outside 15-minute buffer)
+        self._create_notification(
+            email_sent_on=timezone.now() - timedelta(minutes=20)
+        )
+
+        notification = self._create_notification()
+
+        decision = decide_email_action(self.user, self.course_key, notification)
+
+        assert decision == 'send_immediate'
+
+    @freeze_time("2025-12-15 10:00:00")
+    def test_different_course_doesnt_affect_decision(self):
+        """Test that notifications from different courses are independent."""
+        other_course = CourseFactory.create()
+
+        # Notification from different course
+        self._create_notification(
+            course_id=str(other_course.id),
+            email_sent_on=timezone.now() - timedelta(minutes=5)
+        )
+
+        # This course should still send immediate
+        notification = self._create_notification()
+
+        decision = decide_email_action(self.user, self.course_key, notification)
+
+        assert decision == 'send_immediate'
+
+    @freeze_time("2025-12-15 10:00:00")
+    def test_race_condition_protection(self):
+        """Test that select_for_update prevents race conditions."""
+        # Simulate concurrent notifications
+        notification1 = self._create_notification()
+        notification2 = self._create_notification()
+
+        # Both should see no recent email initially
+        with patch('openedx.core.djangoapps.notifications.email.tasks.logger') as mock_logger:
+            decision1 = decide_email_action(self.user, self.course_key, notification1)
+
+            # Mark first as sent to simulate race
+            notification1.email_sent_on = timezone.now()
+            notification1.save()
+
+            decision2 = decide_email_action(self.user, self.course_key, notification2)
+
+            assert decision1 == 'send_immediate'
+            assert decision2 == 'schedule_buffer'
+
+
+@ddt.ddt
+class TestSendImmediateEmail(ModuleStoreTestCase):
+    """Test immediate email sending logic."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create(display_name='Test Course')
+        self.course_key = str(self.course.id)
+
+        self.notification = Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context=get_new_post_notification_content_context(),
+            email=True,
+        )
+
+    @freeze_time("2025-12-15 10:00:00")
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    def test_immediate_email_sent_successfully(self, mock_ace_send):
+        """Test that immediate email is sent and notification marked."""
+        send_immediate_email(
+            user=self.user,
+            notification=self.notification,
+            course_key=self.course_key,
+            course_name='Test Course',
+            user_language='en'
+        )
+
+        # Verify email was sent
+        assert mock_ace_send.called
+
+        # Verify notification marked with sent time
+        self.notification.refresh_from_db()
+        assert self.notification.email_sent_on is not None
+        assert self.notification.email_sent_on == timezone.now()
+
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    def test_email_content_includes_notification_data(self, mock_ace_send):
+        """Test that email contains all required notification data."""
+        send_immediate_email(
+            user=self.user,
+            notification=self.notification,
+            course_key=self.course_key,
+            course_name='Test Course',
+            user_language='en'
+        )
+
+        # Get the message that was sent
+        call_args = mock_ace_send.call_args
+        message = call_args[0][0]
+
+        # Verify message context
+        assert 'Test Course' in str(message.context)
+        assert 'Email content' in str(message.context.get('content', ''))
+
+
+@ddt.ddt
+class TestScheduleDigestBuffer(ModuleStoreTestCase):
+    """Test digest buffer scheduling logic."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create()
+        self.course_key = str(self.course.id)
+
+    @freeze_time("2025-12-15 10:00:00", tz_offset=0)
+    @patch('openedx.core.djangoapps.notifications.email.tasks.send_buffered_digest.apply_async')
+    @override_settings(NOTIFICATION_EMAIL_BUFFER_MINUTES=15)
+    def test_buffer_scheduled_with_correct_delay(self, mock_apply_async):
+        """Test that buffer task is scheduled with correct countdown."""
+        # Create notification that was sent 5 minutes ago
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+            email_sent_on=timezone.now() - timedelta(minutes=5)
+        )
+
+        new_notification = Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+        )
+
+        schedule_digest_buffer(
+            user=self.user,
+            notification=new_notification,
+            course_key=self.course_key,
+            user_language='en'
+        )
+
+        # Verify task was scheduled
+        assert mock_apply_async.called
+
+        # Verify notification marked as scheduled
+        new_notification.refresh_from_db()
+        assert new_notification.email_scheduled is True
+
+        # Verify scheduled time (should be 15 minutes from now)
+        call_kwargs = mock_apply_async.call_args[1]
+        eta = call_kwargs['eta']
+        expected_eta = timezone.now() + timedelta(minutes=15)
+        if timezone.is_naive(eta) and timezone.is_aware(expected_eta):
+            expected_eta = timezone.make_naive(expected_eta)
+        elif timezone.is_aware(eta) and timezone.is_naive(expected_eta):
+            expected_eta = timezone.make_aware(expected_eta)
+        # --- FIX END ---
+        # Allow 1 second tolerance
+        assert abs((eta - expected_eta).total_seconds()) < 1
+
+    @patch('openedx.core.djangoapps.notifications.email.tasks.send_buffered_digest.apply_async')
+    def test_schedule_includes_start_date(self, mock_apply_async):
+        """Test that scheduled task includes correct start date."""
+        sent_time = timezone.now() - timedelta(minutes=10)
+
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+            email_sent_on=sent_time
+        )
+
+        new_notification = Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+        )
+
+        schedule_digest_buffer(
+            user=self.user,
+            notification=new_notification,
+            course_key=self.course_key,
+            user_language='en'
+        )
+
+        # Verify start_date in task kwargs
+        call_kwargs = mock_apply_async.call_args[1]['kwargs']
+        assert call_kwargs['start_date'] == sent_time
+
+
+class TestAddToExistingBuffer(ModuleStoreTestCase):
+    """Test adding notifications to existing buffer."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create()
+
+    def test_notification_marked_as_scheduled(self):
+        """Test that notification is marked as scheduled."""
+        notification = Notification.objects.create(
+            user=self.user,
+            course_id=str(self.course.id),
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+            email_scheduled=False
+        )
+
+        add_to_existing_buffer(notification)
+
+        notification.refresh_from_db()
+        assert notification.email_scheduled is True
+
+    def test_only_scheduled_field_updated(self):
+        """Test that only email_scheduled field is updated."""
+        notification = Notification.objects.create(
+            user=self.user,
+            course_id=str(self.course.id),
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+            content_context=get_new_post_notification_content_context()
+        )
+
+        add_to_existing_buffer(notification)
+
+        notification.refresh_from_db()
+        assert 'Hello world' in notification.content
+        assert notification.email_scheduled is True
+
+
+@ddt.ddt
+class TestSendBufferedDigest(ModuleStoreTestCase):
+    """Test buffered digest email sending."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create(display_name='Test Course')
+        self.course_key = str(self.course.id)
+
+        # Create preference
+        NotificationPreference.objects.all().delete()
+        NotificationPreference.objects.create(
+            user=self.user,
+            app='discussion',
+            type='new_discussion_post',
+            email=True,
+            email_cadence=EmailCadence.IMMEDIATELY
+        )
+
+    @freeze_time("2025-12-15 10:15:00")
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    def test_digest_collects_all_scheduled_notifications(self, mock_ace_send):
+        """Test that digest email includes all scheduled notifications."""
+        start_time = timezone.now() - timedelta(minutes=15)
+
+        # Create 3 scheduled notifications
+        for i in range(3):
+            Notification.objects.create(
+                user=self.user,
+                course_id=self.course_key,
+                app_name='discussion',
+                notification_type='new_discussion_post',
+                content_url='http://example.com',
+                content_context=get_new_post_notification_content_context(),
+                email=True,
+                email_scheduled=True,
+                created=start_time + timedelta(minutes=i * 5)
+            )
+
         with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
             with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
-                send_notifications([self.user.id], str(self.course.id), 'discussion',
-                                   'new_response', context, 'http://test.url')
-        assert mock_func.call_count == 0
+                send_buffered_digest(  # pylint: disable=no-value-for-parameter
+                    user_id=self.user.id,
+                    course_key=self.course_key,
+                    start_date=start_time,
+                    user_language='en'
+                )
+
+        # Verify email was sent
+        assert mock_ace_send.called
+
+        # Verify all notifications marked as sent and unscheduled
+        notifications = Notification.objects.filter(
+            user=self.user,
+            course_id=self.course_key
+        )
+
+        for notif in notifications:
+            assert notif.email_sent_on is not None
+            assert notif.email_scheduled is False
+
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    def test_digest_skips_non_scheduled_notifications(self, mock_ace_send):
+        """Test that digest only includes scheduled notifications."""
+        start_time = timezone.now() - timedelta(minutes=15)
+
+        # Scheduled notification
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context=get_new_post_notification_content_context(),
+            email=True,
+            email_scheduled=True,
+            created=start_time + timedelta(minutes=5)
+        )
+
+        # Non-scheduled notification (should be ignored)
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context=get_new_post_notification_content_context(),
+            email=True,
+            email_scheduled=False,
+            created=start_time + timedelta(minutes=10)
+        )
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                send_buffered_digest(  # pylint: disable=no-value-for-parameter
+                    user_id=self.user.id,
+                    course_key=self.course_key,
+                    start_date=start_time,
+                    user_language='en'
+                )
+
+        # Only 1 notification should be marked as sent
+        sent_count = Notification.objects.filter(
+            user=self.user,
+            email_sent_on__isnull=False
+        ).count()
+
+        assert sent_count == 1
+
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    def test_digest_respects_user_preferences(self, mock_ace_send):
+        """Test that digest filters based on user preferences."""
+        start_time = timezone.now() - timedelta(minutes=15)
+        NotificationPreference.objects.all().delete()
+
+        # Create notification for type that user has disabled
+        NotificationPreference.objects.create(
+            user=self.user,
+            app='discussion',
+            type='new_comment',
+            email=False,  # Disabled
+            email_cadence=EmailCadence.IMMEDIATELY
+        )
+
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_comment',
+            content_context=get_new_post_notification_content_context(),
+            content_url='http://example.com',
+            email=True,
+            email_scheduled=True,
+            created=start_time + timedelta(minutes=5)
+        )
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                send_buffered_digest(  # pylint: disable=no-value-for-parameter
+                    user_id=self.user.id,
+                    course_key=self.course_key,
+                    start_date=start_time,
+                    user_language='en'
+                )
+
+        # Email should not be sent
+        assert not mock_ace_send.called
+
+        # Notification should still be marked as scheduled=False
+        notif = Notification.objects.get(
+            user=self.user,
+            notification_type='new_comment'
+        )
+        assert notif.email_scheduled is False
+
+    def test_digest_handles_missing_user(self):
+        """Test that digest handles non-existent user gracefully."""
+        start_time = timezone.now() - timedelta(minutes=15)
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                # Should not raise exception
+                send_buffered_digest(  # pylint: disable=no-value-for-parameter
+                    user_id=99999,  # Non-existent
+                    course_key=self.course_key,
+                    start_date=start_time,
+                    user_language='en'
+                )
+
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send', side_effect=Exception('Email failed'))
+    def test_digest_retries_on_failure(self, mock_ace_send):
+        """Test that digest task retries on failure."""
+        start_time = timezone.now() - timedelta(minutes=15)
+
+        Notification.objects.create(
+            user=self.user,
+            course_id=self.course_key,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context={'email_content': '<p>Email</p>'},
+            email=True,
+            email_scheduled=True,
+            created=start_time + timedelta(minutes=5)
+        )
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                # Create a mock task instance
+                mock_task = Mock()
+                mock_task.request.retries = 0
+
+                with self.assertRaises(Exception):
+                    send_buffered_digest.bind(mock_task)(
+                        user_id=self.user.id,
+                        course_key=self.course_key,
+                        start_date=start_time,
+                        user_language='en'
+                    )
+
+
+@ddt.ddt
+class TestIntegrationScenarios(ModuleStoreTestCase):
+    """Integration tests for complete notification flow."""
+
+    def setUp(self):
+        super().setUp()
+        self.user = UserFactory()
+        self.course = CourseFactory.create(display_name='Test Course')
+        NotificationPreference.objects.all().delete()
+
+        NotificationPreference.objects.create(
+            user=self.user,
+            app='discussion',
+            type='new_discussion_post',
+            email=True,
+            email_cadence=EmailCadence.IMMEDIATELY
+        )
+
+    @freeze_time("2025-12-15 10:00:00")
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    @patch('openedx.core.djangoapps.notifications.email.tasks.send_buffered_digest.apply_async')
+    @override_settings(NOTIFICATION_EMAIL_BUFFER_MINUTES=15)
+    def test_complete_three_notification_flow(self, mock_digest_async, mock_ace_send):
+        """Test complete flow: immediate → buffer → add to buffer."""
+        email_mapping = {}
+
+        # FIRST NOTIFICATION - should send immediately
+        notif1 = Notification.objects.create(
+            user=self.user,
+            course_id=self.course.id,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context=get_new_post_notification_content_context(),
+            email=True,
+        )
+        email_mapping[self.user.id] = notif1
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                send_immediate_cadence_email(email_mapping, self.course.id)
+
+        # Verify immediate email sent
+        assert mock_ace_send.call_count == 1
+        assert mock_digest_async.call_count == 0
+
+        notif1.refresh_from_db()
+        assert notif1.email_sent_on is not None
+        assert notif1.email_scheduled is False
+
+        # SECOND NOTIFICATION - should schedule buffer (5 minutes later)
+        with freeze_time("2025-12-15 10:05:00"):
+            notif2 = Notification.objects.create(
+                user=self.user,
+                course_id=self.course.id,
+                app_name='discussion',
+                notification_type='new_discussion_post',
+                content_url='http://example.com',
+                content_context=get_new_post_notification_content_context(),
+                email=True,
+            )
+            email_mapping = {self.user.id: notif2}
+
+            with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+                with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                    send_immediate_cadence_email(email_mapping, self.course.id)
+
+            # Verify buffer scheduled
+            assert mock_ace_send.call_count == 1  # Still just 1 immediate email
+            assert mock_digest_async.call_count == 1  # Buffer scheduled
+
+            notif2.refresh_from_db()
+            assert notif2.email_sent_on is None
+            assert notif2.email_scheduled is True
+
+        # THIRD NOTIFICATION - should just mark as scheduled (10 minutes later)
+        with freeze_time("2025-12-15 10:10:00"):
+            notif3 = Notification.objects.create(
+                user=self.user,
+                course_id=self.course.id,
+                app_name='discussion',
+                notification_type='new_discussion_post',
+                content_url='http://example.com',
+                content_context=get_new_post_notification_content_context(),
+                email=True,
+            )
+            email_mapping = {self.user.id: notif3}
+
+            with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+                with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                    send_immediate_cadence_email(email_mapping, self.course.id)
+
+            # Verify no new tasks scheduled
+            assert mock_ace_send.call_count == 1
+            assert mock_digest_async.call_count == 1  # Still just 1 buffer task
+
+            notif3.refresh_from_db()
+            assert notif3.email_sent_on is None
+            assert notif3.email_scheduled is True
+
+        # BUFFER FIRES - should send digest with notif2 and notif3
+        with freeze_time("2025-12-15 10:15:00"):
+            with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+                with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                    send_buffered_digest(  # pylint: disable=no-value-for-parameter
+                        user_id=self.user.id,
+                        course_key=str(self.course.id),
+                        start_date=notif1.email_sent_on,
+                        user_language='en'
+                    )
+
+                    # Verify digest email sent
+                    assert mock_ace_send.call_count == 2  # 1 immediate + 1 digest
+
+                    # Verify both buffered notifications marked as sent
+                    notif2.refresh_from_db()
+                    notif3.refresh_from_db()
+
+                    assert notif2.email_sent_on is not None
+                    assert notif2.email_scheduled is False
+                    assert notif3.email_sent_on is not None
+                    assert notif3.email_scheduled is False
+
+    @freeze_time("2025-12-15 10:00:00")
+    @patch('openedx.core.djangoapps.notifications.email.tasks.ace.send')
+    @override_settings(NOTIFICATION_EMAIL_BUFFER_MINUTES=15)
+    def test_notification_after_buffer_expires_sends_immediate(self, mock_ace_send):
+        """Test that notification after buffer period sends immediately again."""
+        # First notification
+        notif1 = Notification.objects.create(
+            user=self.user,
+            course_id=self.course.id,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            content_context=get_new_post_notification_content_context(),
+            email=True,
+        )
+        email_mapping = {self.user.id: notif1}
+
+        with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+            with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                send_immediate_cadence_email(email_mapping, self.course.id)
+
+        assert mock_ace_send.call_count == 1
+
+        # New notification 20 minutes later (after 15-minute buffer)
+        with freeze_time("2025-12-15 10:20:00"):
+            notif2 = Notification.objects.create(
+                user=self.user,
+                course_id=self.course.id,
+                app_name='discussion',
+                notification_type='new_discussion_post',
+                content_url='http://example.com',
+                content_context=get_new_post_notification_content_context(),
+                email=True,
+            )
+            email_mapping = {self.user.id: notif2}
+
+            with override_waffle_flag(ENABLE_NOTIFICATIONS, True):
+                with override_waffle_flag(ENABLE_EMAIL_NOTIFICATIONS, True):
+                    send_immediate_cadence_email(email_mapping, self.course.id)
+
+            # Should send immediate again (buffer expired)
+            assert mock_ace_send.call_count == 2
+
+            notif2.refresh_from_db()
+            assert notif2.email_sent_on is not None
+            assert notif2.email_scheduled is False
+
+    def test_multiple_courses_independent_buffers(self):
+        """Test that different courses maintain independent buffers."""
+        course2 = CourseFactory.create()
+
+        # Notifications in course 1
+        notif1 = Notification.objects.create(
+            user=self.user,
+            course_id=self.course.id,
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+            email_sent_on=timezone.now() - timedelta(minutes=5)
+        )
+
+        # Notification in course 2 should be independent
+        notif2 = Notification.objects.create(
+            user=self.user,
+            course_id=str(course2.id),
+            app_name='discussion',
+            notification_type='new_discussion_post',
+            content_url='http://example.com',
+            email=True,
+        )
+
+        decision = decide_email_action(self.user, str(course2.id), notif2)
+        assert decision == 'send_immediate'
+
+
+def get_new_post_notification_content_context(**kwargs):
+    """Helper to generate notification content for a new post."""
+    return {
+        "topic_id": "i4x-edx-eiorguegnru-course-foobarbaz",
+        "username": "verified",
+        "thread_id": "693fbf23ee2b892eaed49239",
+        "comment_id": None,
+        "post_title": "Hello world",
+        "course_name": "Demonstration Course",
+        "response_id": None,
+        "replier_name": "verified",
+        "email_content": "<p style=\"margin: 0\">Email content</p>",
+        **kwargs
+    }

openedx/core/djangoapps/notifications/migrations/0011_notification_email_scheduled_and_more.py

@@ -0,0 +1,29 @@
+# Generated by Django 5.2.7 on 2026-01-01 09:01
+
+from django.conf import settings
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('notifications', '0010_delete_coursenotificationpreference'),
+        migrations.swappable_dependency(settings.AUTH_USER_MODEL),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='notification',
+            name='email_scheduled',
+            field=models.BooleanField(db_index=True, default=False, help_text='True if this notification is waiting in buffer for digest'),
+        ),
+        migrations.AddField(
+            model_name='notification',
+            name='email_sent_on',
+            field=models.DateTimeField(blank=True, null=True),
+        ),
+        migrations.AddIndex(
+            model_name='notification',
+            index=models.Index(fields=['user', 'course_id', 'email_sent_on', 'email_scheduled'], name='notif_email_buffer_idx'),
+        ),
+    ]

openedx/core/djangoapps/notifications/models.py

@@ -90,6 +90,20 @@ class Notification(TimeStampedModel):
     last_read = models.DateTimeField(null=True, blank=True)
     last_seen = models.DateTimeField(null=True, blank=True)
     group_by_id = models.CharField(max_length=255, db_index=True, null=False, default="")
+    email_sent_on = models.DateTimeField(null=True, blank=True)
+    email_scheduled = models.BooleanField(
+        default=False,
+        db_index=True,
+        help_text="True if this notification is waiting in buffer for digest"
+    )
+
+    class Meta:
+        indexes = [
+            models.Index(
+                fields=['user', 'course_id', 'email_sent_on', 'email_scheduled'],
+                name='notif_email_buffer_idx'
+            ),
+        ]
 
     def __str__(self):
         return f'{self.user.username} - {self.course_id} - {self.app_name} - {self.notification_type}'

openedx/core/djangoapps/notifications/tasks.py

@@ -1,6 +1,7 @@
 """
 This file contains celery tasks for notifications.
 """
+import uuid
 from datetime import datetime, timedelta
 from typing import List
 
@@ -122,7 +123,7 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
     email_notification_mapping = {}
     push_notification_audience = []
     is_push_notification_enabled = ENABLE_PUSH_NOTIFICATIONS.is_enabled(course_key)
-
+    task_id = str(uuid.uuid4())
     for batch_user_ids in get_list_in_batches(user_ids, batch_size):
         logger.debug(f'Sending notifications to {len(batch_user_ids)} users in {course_key}')
         batch_user_ids = NotificationFilter().apply_filters(batch_user_ids, course_key, notification_type)
@@ -151,6 +152,7 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
             continue
 
         notifications = []
+        email_notification_user_ids = []
         for preference in preferences:
             user_id = preference.user_id
 
@@ -166,16 +168,17 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
                     user_id=user_id,
                     app_name=app_name,
                     notification_type=notification_type,
-                    content_context=context,
+                    content_context={**context, 'uuid': task_id},
                     content_url=content_url,
                     course_id=course_key,
                     web='web' in notification_preferences,
                     email=email_enabled,
                     push=push_notification,
                     group_by_id=group_by_id,
+                    email_scheduled=False
                 )
                 if email_enabled and (email_cadence == EmailCadence.IMMEDIATELY):
-                    email_notification_mapping[user_id] = new_notification
+                    email_notification_user_ids.append(user_id)
 
                 if push_notification:
                     push_notification_audience.append(user_id)
@@ -193,7 +196,22 @@ def send_notifications(user_ids, course_key: str, app_name, notification_type, c
         # send notification to users but use bulk_create
         Notification.objects.bulk_create(notifications)
 
+        # Get fresh records with pk so it can be used in email sending because there is a need to
+        # update the records further down the line.
+        if email_notification_user_ids:
+            email_notification_mapping = {
+                notif.user_id: notif
+                for notif in Notification.objects.filter(
+                    user_id__in=email_notification_user_ids,
+                    content_context__uuid=task_id,
+                )
+            }
     if email_notification_mapping:
+        logger.info(
+            f"Email Buffered Digest: Sending immediate email notifications to "
+            f"users {list(email_notification_mapping.keys())} "
+            f"for notification {notification_type}",
+        )
         send_immediate_cadence_email(email_notification_mapping, course_key)
 
     if generated_notification:

openedx/core/djangoapps/notifications/tests/test_tasks_with_account_level_pref.py

@@ -69,6 +69,9 @@ class SendNotificationsTest(ModuleStoreTestCase):
 
         # Assert that `Notification` objects have been created for the users.
         notification = Notification.objects.filter(user_id=self.user.id).first()
+        # Removing uuid from content_context for assertion
+        if 'uuid' in notification.content_context:
+            notification.content_context.pop('uuid')
         # Assert that the `Notification` objects have the correct properties.
         self.assertEqual(notification.user_id, self.user.id)
         self.assertEqual(notification.app_name, app_name)

openedx/envs/common.py

@@ -1544,6 +1544,11 @@ DISCUSSIONS_MFE_FEEDBACK_URL = None
 # .. toggle_tickets: https://openedx.atlassian.net/browse/VAN-838
 ENABLE_DYNAMIC_REGISTRATION_FIELDS = False
 
+# .. setting_name: LEARNER_HOME_MICROFRONTEND_URL
+# .. setting_default: None
+# .. setting_description: Base URL of the micro-frontend-based learner home page.
+LEARNER_HOME_MICROFRONTEND_URL = None
+
 ################################## Swift ###################################
 
 SWIFT_USERNAME = None
@@ -2224,6 +2229,9 @@ EXPIRED_NOTIFICATIONS_DELETE_BATCH_SIZE = 10000
 NOTIFICATION_CREATION_BATCH_SIZE = 76
 NOTIFICATIONS_DEFAULT_FROM_EMAIL = "no-reply@example.com"
 NOTIFICATION_DIGEST_LOGO = DEFAULT_EMAIL_LOGO_URL
+NOTIFICATION_IMMEDIATE_EMAIL_BUFFER_MINUTES = 15  # in minutes
+NOTIFICATION_TYPE_ICONS = {}
+DEFAULT_NOTIFICATION_ICON_URL = ""
 
 # These settings are used to override the default notification preferences values for apps and types.
 # Here is complete documentation about how to use them: openedx/core/djangoapps/notifications/docs/settings.md