diff --git a/admin/notifications/forms.py b/admin/notifications/forms.py index ea9417a89c2..06f04a799d8 100644 --- a/admin/notifications/forms.py +++ b/admin/notifications/forms.py @@ -65,6 +65,12 @@ class NotificationCampaignCreateForm(forms.ModelForm): help_text='The time in seconds between dispatches of campaign batches.', ) + log_email_recipients = forms.CharField( + required=False, + initial='', + help_text='List of email addresses to receive campaign log emails, separated by commas.', + ) + class Meta: model = NotificationCampaign fields = ( diff --git a/admin/notifications/views.py b/admin/notifications/views.py index bef2ceed94d..01c65dbb99a 100644 --- a/admin/notifications/views.py +++ b/admin/notifications/views.py @@ -19,7 +19,7 @@ from mako.parsetree import ControlLine from string import Formatter from osf.email import _render_email_html -from osf.email.notification_campaign import FILTER_PRESETS, counter_subquery, build_campaign_filter_query +from osf.email.notification_campaign import FILTER_PRESETS, counter_subquery, build_campaign_filter_query, get_campaign_recipient_stats from website import settings from urllib.parse import urlencode @@ -392,6 +392,19 @@ def get_object(self, queryset=None): def get_context_data(self, *args, **kwargs): notification_campaign = self.get_object() metadata = notification_campaign.metadata or {} + stats = get_campaign_recipient_stats(notification_campaign.id) + queued = stats['queued_count'] + awaiting = stats['awaiting_count'] + # Recipient rows are the source of truth while they exist; the stored counters + # are a snapshot for campaigns whose recipients were purged. + if stats['recipient_count']: + recipient_count = stats['recipient_count'] + sent_count = stats['sent_count'] + failed_count = stats['failed_count'] + else: + recipient_count = notification_campaign.recipient_count + sent_count = notification_campaign.sent_count + failed_count = notification_campaign.failed_count context = { 'notification_campaign': notification_campaign, @@ -401,9 +414,11 @@ def get_context_data(self, *args, **kwargs): ('Created By', notification_campaign.created_by), ('Developer reminder ', 'Sent' if notification_campaign.developer_reminder_sent else ''), ('Status', notification_campaign.get_status_display()), - ('Recipients', notification_campaign.recipient_count), - ('Sent', notification_campaign.sent_count), - ('Failed', notification_campaign.failed_count), + ('Recipients', recipient_count), + ('Sent', sent_count), + ('Queued', queued), + ('Awaiting Delivery', awaiting), + ('Failed', failed_count), ('Retries', notification_campaign.retries), ('Created', notification_campaign.created_at), ('Started', notification_campaign.started_at), @@ -436,19 +451,39 @@ def get_context_data(self, *args, **kwargs): } if notification_campaign.status != NotificationCampaignStatus.CREATED: - processed = notification_campaign.sent_count + notification_campaign.failed_count - pending = max(notification_campaign.recipient_count - processed, 0) + processed = sent_count + failed_count + pending = max( + recipient_count - processed - queued - awaiting, + 0, + ) + awaiting_delivery = ( + notification_campaign.status == NotificationCampaignStatus.RUNNING + and pending == 0 + and queued == 0 + and awaiting > 0 + ) sent_percent = ( - notification_campaign.sent_count / notification_campaign.recipient_count * 100 - if notification_campaign.recipient_count else 0 + sent_count / recipient_count * 100 + if recipient_count else 0 ) failed_percent = ( - notification_campaign.failed_count / notification_campaign.recipient_count * 100 - if notification_campaign.recipient_count else 0 + failed_count / recipient_count * 100 + if recipient_count else 0 + ) + queued_percent = ( + queued / recipient_count * 100 + if recipient_count else 0 + ) + awaiting_percent = ( + awaiting / recipient_count * 100 + if recipient_count else 0 ) - pending_percent = max(100 - sent_percent - failed_percent, 0) + pending_percent = max( + 100 - sent_percent - failed_percent - queued_percent - awaiting_percent, + 0, + ) elapsed = None speed = None eta = None @@ -462,22 +497,31 @@ def get_context_data(self, *args, **kwargs): elapsed_seconds = elapsed.total_seconds() if processed > 0 and elapsed_seconds > 0: speed = processed / elapsed_seconds - if pending: - eta = timedelta(seconds=int(pending / speed)) + remaining = pending + queued + if remaining: + eta = timedelta(seconds=int(remaining / speed)) estimated_finish = timezone.now() + eta if notification_campaign.updated_at: last_activity_ago = timezone.now() - notification_campaign.updated_at if processed: - failure_rate = notification_campaign.failed_count / processed * 100 + failure_rate = failed_count / processed * 100 else: failure_rate = 0 context.update({ + 'recipient_count': recipient_count, + 'sent_count': sent_count, + 'failed_count': failed_count, 'processed': processed, 'pending': pending, + 'queued': queued, + 'awaiting': awaiting, + 'awaiting_delivery': awaiting_delivery, 'sent_percent': sent_percent, 'failed_percent': failed_percent, + 'queued_percent': queued_percent, + 'awaiting_percent': awaiting_percent, 'pending_percent': pending_percent, 'elapsed': elapsed, 'speed': speed, @@ -585,6 +629,7 @@ def form_valid(self, form): 'time_window': form.cleaned_data['time_window'], 'max_queued_batches': form.cleaned_data['max_queued_batches'], 'dispatch_interval': form.cleaned_data['dispatch_interval'], + 'log_email_recipients': form.cleaned_data['log_email_recipients'], }, 'sendgrid_bulk': form.cleaned_data.get('sendgrid_bulk', False), 'recipients_creation_finished': False, diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index 618bb570d56..91486541212 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -62,16 +62,33 @@

{{ notification_campaign.name }}

Last activity: {{ last_activity_ago }} ago {% endif %} + {% if awaiting_delivery %} +
+ Awaiting delivery: + All recipients have been submitted to SendGrid. + Waiting for delivery confirmation ({{ awaiting }} awaiting). +
+ {% endif %}
- {{ notification_campaign.sent_count }} + {{ sent_count }} +
+ +
+ {{ awaiting }} +
+ +
+ {{ queued }}
- {{ notification_campaign.failed_count }} + {{ failed_count }}
Progress - + + + + + + + + + + + diff --git a/admin/templates/notifications/notification_campaing_create.html b/admin/templates/notifications/notification_campaing_create.html index 2f1a6d2f525..fa416ad9a80 100644 --- a/admin/templates/notifications/notification_campaing_create.html +++ b/admin/templates/notifications/notification_campaing_create.html @@ -324,6 +324,20 @@

Execution

+ + + +
Processed{{ processed }} / {{ notification_campaign.recipient_count }}{{ processed }} / {{ recipient_count }}
Queued{{ queued }}
Awaiting Delivery{{ awaiting }}
Log Email Recipients + +

+ {{ form.log_email_recipients.help_text }} +

+
diff --git a/admin_tests/notifications/test_campaigns.py b/admin_tests/notifications/test_campaigns.py index 8d548f89ba9..c3d79a0a40f 100644 --- a/admin_tests/notifications/test_campaigns.py +++ b/admin_tests/notifications/test_campaigns.py @@ -23,8 +23,11 @@ from osf.models import NotificationType from osf.models.notification_campaign import ( NotificationCampaign, + NotificationCampaignRecipient, + NotificationCampaignRecipientStatus, + NotificationCampaignStatus, ) -from osf_tests.factories import AuthUserFactory, ProjectFactory +from osf_tests.factories import AuthUserFactory, ProjectFactory, UserFactory from tests.base import AdminTestCase from website import settings @@ -64,6 +67,7 @@ def _valid_form_data(notification_type, **overrides): 'time_window': 8 * 60 * 60, 'max_queued_batches': settings.MAX_QUEUED_CAMPAIGN_BATCHES, 'dispatch_interval': settings.CAMPAIGN_DISPATCH_INTERVAL, + 'log_email_recipients': '', } data.update(overrides) return data @@ -81,6 +85,7 @@ def test_valid_form_parses_context_and_filters(self, notification_type): assert form.cleaned_data['activity_threshold'] == settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD assert form.cleaned_data['time_window'] == 8 * 60 * 60 assert form.cleaned_data['sendgrid_bulk'] is False + assert form.cleaned_data['log_email_recipients'] == '' def test_defaults_come_from_settings(self): form = NotificationCampaignCreateForm() @@ -288,6 +293,7 @@ def test_form_valid_persists_execution_metadata(self): 'time_window': 28800, 'max_queued_batches': 10, 'dispatch_interval': 60, + 'log_email_recipients': '' } assert campaign.metadata['sendgrid_bulk'] is True assert campaign.metadata['filters'] == {'predefined': 'active'} @@ -511,3 +517,75 @@ def test_delete_requires_change_notificationcampaign_permission(self): response = DeleteNotificationCampaign.as_view()(request, pk=self.campaign.pk) assert response.status_code == 302 assert not NotificationCampaign.objects.filter(pk=self.campaign.pk).exists() + + +class TestNotificationCampaignDetailCounters(AdminTestCase): + + def setUp(self): + super().setUp() + self.user = AuthUserFactory() + grant_permission(self.user, 'change_notificationcampaign') + notification_type, _ = NotificationType.objects.get_or_create( + name='blank', + defaults={'subject': 'Test', 'template': 'Hello'}, + ) + # Stored counters deliberately differ from recipient rows. + self.campaign = NotificationCampaign.objects.create( + name='Campaign', + notification_type=notification_type, + status=NotificationCampaignStatus.COMPLETED, + recipient_count=10, + sent_count=7, + failed_count=3, + metadata={'filters': {}, 'context': {}, 'execution': {}}, + ) + + def _context(self): + request = RequestFactory().get( + reverse('notifications:notification_campaigns_detail', kwargs={'pk': self.campaign.pk}) + ) + request.user = self.user + response = NotificationCampaignDetail.as_view()(request, pk=self.campaign.pk) + assert response.status_code == 200 + return response.context_data + + def _add_recipient(self, status): + NotificationCampaignRecipient.objects.create( + campaign=self.campaign, + user=UserFactory(), + status=status, + ) + + def test_uses_live_recipient_stats_when_recipients_exist(self): + self._add_recipient(NotificationCampaignRecipientStatus.SENT) + self._add_recipient(NotificationCampaignRecipientStatus.SKIPPED) + self._add_recipient(NotificationCampaignRecipientStatus.AWAITING_DELIVERY) + + context = self._context() + + assert (context['recipient_count'], context['sent_count'], context['failed_count']) == (3, 1, 1) + assert context['awaiting'] == 1 + assert context['processed'] == 2 + assert context['pending'] == 0 + + def test_falls_back_to_stored_counters_when_recipients_purged(self): + context = self._context() + + assert (context['recipient_count'], context['sent_count'], context['failed_count']) == (10, 7, 3) + assert context['processed'] == 10 + assert context['pending'] == 0 + + @mock.patch('osf.email.notification_campaign.start_notification_campaign.delay') + def test_restart_does_not_reset_displayed_counters(self, mock_delay): + self._add_recipient(NotificationCampaignRecipientStatus.SENT) + self._add_recipient(NotificationCampaignRecipientStatus.FAILED) + self._add_recipient(NotificationCampaignRecipientStatus.SKIPPED) + + for flags in ({'restart_failed': True}, {'restart_stuck': True}): + self.campaign.start(**flags) # resets stored failed_count to 0 + self.campaign.refresh_from_db() + assert self.campaign.failed_count == 0 + + context = self._context() + + assert (context['recipient_count'], context['sent_count'], context['failed_count']) == (3, 1, 2) diff --git a/api/sendgrid/views.py b/api/sendgrid/views.py index 45f14c235f1..36c9041fe6f 100644 --- a/api/sendgrid/views.py +++ b/api/sendgrid/views.py @@ -44,8 +44,10 @@ class SendGridEventWebhook(APIView): - Enable Signed Event Webhook; copy the verification key into ``SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY`` (``website/settings/local.py``). - Subscribe at least to ``delivered``, ``bounce``, and ``dropped``. + (``deferred`` is safe to enable but ignored for campaigns - SendGrid + retries those itself.) 4. Send a notification campaign email (personalization ``custom_args`` must - include ``campaign_id``, ``campaign_recipient_id``, and ``run_id``). + include ``campaign_id``, ``campaign_recipient_id``, and ``sent_at``). SendGrid's "Test Your Integration" payload lacks those keys, so this view accepts it with 200 but does not update campaign recipients. """ diff --git a/notifications.yaml b/notifications.yaml index c14324df931..872d68ad28a 100644 --- a/notifications.yaml +++ b/notifications.yaml @@ -828,3 +828,10 @@ notification_types: object_content_type_model_name: osfuser template: 'website/templates/blank.html.mako' tests: [] + + - name: notification_campaign_log + subject: 'Notification campaign "{campaign_name}" — {status}' + __docs__: ... + object_content_type_model_name: osfuser + template: 'website/templates/notification_campaign_log.html.mako' + tests: [] diff --git a/osf/email/__init__.py b/osf/email/__init__.py index 54be37c898d..3c5c785a813 100644 --- a/osf/email/__init__.py +++ b/osf/email/__init__.py @@ -4,6 +4,7 @@ import logging import importlib import sys +from functools import cache from html import unescape from typing import Optional from mako.template import Template as MakoTemplate @@ -285,7 +286,22 @@ def personalization(recipients, custom_args=None): ] -def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False, rendered_html=None): +@cache +def _sendgrid_client(api_key): + return SendGridAPIClient(api_key) + + +def get_sendgrid_client(use_webhook=False): + """Return a per-process cached SendGrid client. + + When ``use_webhook`` is True, the client authenticates as the sub-user with Event Webhook enabled + (``SENDGRID_API_KEY_WITH_WEBHOOK``), otherwise as the default user (``SENDGRID_API_KEY``). + """ + api_key = settings.SENDGRID_API_KEY_WITH_WEBHOOK if use_webhook else None + return _sendgrid_client(api_key or settings.SENDGRID_API_KEY) + + +def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False, rendered_html=None, client=None): email_context = email_context or {} to_list = [to_addr] if isinstance(to_addr, str) else [a for a in (to_addr or []) if a] @@ -337,7 +353,7 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context payload['attachments'] = [item] try: - sg = SendGridAPIClient(settings.SENDGRID_API_KEY) + sg = client or SendGridAPIClient(settings.SENDGRID_API_KEY) resp = sg.client.mail.send.post(request_body=payload) if resp.status_code not in (200, 201, 202): logging.error( @@ -380,9 +396,10 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context logging.error('SendGrid hit a blocked socket error: %r | payload=%s', exc, payload) raise -def send_email(recipient_address, notification_type, event_context=None, email_context=None, rendered_html=None): +def send_email(recipient_address, notification_type, event_context=None, email_context=None, rendered_html=None, client=None): """ Send an email using either SMTP or SendGrid based on settings and feature flags. + ``client`` is an optional SendGrid client to reuse; a default one is created per call otherwise. """ if waffle.switch_is_active(features.ENABLE_MAILHOG): send_email_over_smtp( @@ -400,6 +417,7 @@ def send_email(recipient_address, notification_type, event_context=None, email_c event_context, email_context, rendered_html=rendered_html, + client=client, ) if settings.LOCAL_MODE and not waffle.switch_is_active(features.ENABLE_MAILHOG): diff --git a/osf/email/notification_campaign/__init__.py b/osf/email/notification_campaign/__init__.py new file mode 100644 index 00000000000..aa1eddb3c0b --- /dev/null +++ b/osf/email/notification_campaign/__init__.py @@ -0,0 +1,55 @@ +from .base import NotificationCampaignTask +from .execution import ( + dispatch_campaign, + process_campaign_retry, + send_campaign_batch, + start_notification_campaign, +) +from .recipients import ( + BULK_CREATE_SIZE, + FILTER_PRESETS, + assign_batch_id_to_recipients, + build_campaign_filter_query, + build_query, + counter_subquery, + create_campaign_recipients, + first_email_subquery, + get_campaign_recipient_stats, +) +from .webhook import ( + CAMPAIGN_CUSTOM_ARG_KEYS, + SENDGRID_HARD_BOUNCE_TYPE, + SENDGRID_SKIP_EVENTS, + SENDGRID_SOFT_BOUNCE_TYPE, + SENDGRID_SUCCESS_EVENTS, + _classify_sendgrid_events, + _classify_sendgrid_failure, + _sendgrid_event_error_message, + process_sendgrid_campaign_events, +) + +__all__ = [ + 'BULK_CREATE_SIZE', + 'CAMPAIGN_CUSTOM_ARG_KEYS', + 'FILTER_PRESETS', + 'SENDGRID_HARD_BOUNCE_TYPE', + 'SENDGRID_SKIP_EVENTS', + 'SENDGRID_SOFT_BOUNCE_TYPE', + 'SENDGRID_SUCCESS_EVENTS', + 'NotificationCampaignTask', + '_classify_sendgrid_events', + '_classify_sendgrid_failure', + '_sendgrid_event_error_message', + 'assign_batch_id_to_recipients', + 'build_campaign_filter_query', + 'build_query', + 'counter_subquery', + 'create_campaign_recipients', + 'dispatch_campaign', + 'first_email_subquery', + 'get_campaign_recipient_stats', + 'process_campaign_retry', + 'process_sendgrid_campaign_events', + 'send_campaign_batch', + 'start_notification_campaign', +] diff --git a/osf/email/notification_campaign/base.py b/osf/email/notification_campaign/base.py new file mode 100644 index 00000000000..478bb587d31 --- /dev/null +++ b/osf/email/notification_campaign/base.py @@ -0,0 +1,70 @@ +import logging + +from django.utils import timezone + +from framework import sentry +from framework.celery_tasks import app as celery_app +from osf.email import send_email +from osf.models import NotificationTypeEnum +from osf.models.notification_campaign import ( + NotificationCampaign, + NotificationCampaignStatus, +) + +from .recipients import get_campaign_recipient_stats + +logger = logging.getLogger(__name__) + + +class NotificationCampaignTask(celery_app.Task): + """Shared guards for notification campaign Celery tasks.""" + + abstract = True + + def get_campaign(self, campaign_id, run_id=None, *, abort_if_cancelled=True): + """Load a campaign, or return None if the task should no-op. + """ + campaign = NotificationCampaign.objects.get(id=campaign_id) + if run_id is not None and campaign.run_id != run_id: + return None + if abort_if_cancelled and campaign.status == NotificationCampaignStatus.CANCELLED: + logger.warning(f"Campaign {campaign_id} was cancelled") + return None + return campaign + + def sync_campaign_stats(self, campaign, *, save=True): + stats = get_campaign_recipient_stats(campaign.id) + campaign.recipient_count = stats['recipient_count'] + campaign.sent_count = stats['sent_count'] + campaign.failed_count = stats['failed_count'] + if save: + campaign.save(update_fields=[ + 'recipient_count', + 'sent_count', + 'failed_count', + 'updated_at', + ]) + return stats + + def finish_campaign(self, campaign, status=None): + """Sync recipient counters, set completed_at once, optionally update status, and save.""" + self.sync_campaign_stats(campaign, save=False) + if campaign.completed_at is None: + campaign.completed_at = timezone.now() + if status is not None: + campaign.status = status + campaign.save() + self.send_campaign_log_email(campaign, status or campaign.status, f"Campaign finished with status {status or campaign.status}") + + def send_campaign_log_email(self, campaign, status, message): + """Send a notification campaign status email.""" + recipients = [recipient.strip() for recipient in campaign.metadata['execution'].get('log_email_recipients', '').split(',') if recipient] + for recipient in recipients: + try: + send_email( + recipient_address=recipient, + notification_type=NotificationTypeEnum.NOTIFICATION_CAMPAIGN_LOG.instance, + event_context={'campaign_name': campaign.name, 'status': status, 'message': message},) + except Exception as exc: + logger.error(f"Failed to send error logs email for campaign {campaign.id}: {exc}") + sentry.log_message(f"Failed to send error logs email for campaign {campaign.id}: {exc}") diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign/execution.py similarity index 52% rename from osf/email/notification_campaign.py rename to osf/email/notification_campaign/execution.py index 8fe0d98278a..7ab783a390c 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign/execution.py @@ -1,304 +1,29 @@ import logging -import uuid -from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter, Email, Contributor -from osf.models.spam import SpamStatus -from django.db import transaction -from django.db.models import OuterRef, Subquery, Case, When, Value, CharField, Count, Q, BooleanField, TextField, Exists -from django.db.models.functions import Coalesce -from framework.celery_tasks import app as celery_app -from django.utils import timezone from datetime import timedelta -from osf.models.notification_campaign import NotificationCampaign, NotificationCampaignRecipient, NotificationCampaignStatus, NotificationCampaignRecipientStatus -from osf.email import send_email_with_send_grid, _render_email_html, send_email -from framework import sentry -from website import settings -from itertools import batched - -logger = logging.getLogger(__name__) - -BULK_CREATE_SIZE = 5000 -FILTER_PRESETS = { - 'all': {}, - 'active': {'is_active': True}, - 'internal': {'is_active': True, 'is_staff': True, 'username__endswith': '@cos.io'}, -} - -# Flattened onto SendGrid Event Webhook payloads via personalization custom_args. -CAMPAIGN_CUSTOM_ARG_KEYS = ('campaign_id', 'campaign_recipient_id', 'run_id') -SENDGRID_SUCCESS_EVENTS = frozenset({'delivered'}) -SENDGRID_FAILURE_EVENTS = frozenset({'bounce', 'dropped'}) - -first_email_subquery = ( - Email.objects - .filter(user=OuterRef('user_id')) - .values('address')[:1] -) +from django.db.models import Case, CharField, Subquery, When +from django.utils import timezone -counter_subquery = ( - UserActivityCounter.objects - .filter(_id=OuterRef('guids___id')) - .values('total')[:1] +from framework import sentry +from framework.celery_tasks import app as celery_app +from osf.email import _render_email_html, get_sendgrid_client, send_email, send_email_with_send_grid +from osf.models import NotificationType, NotificationTypeEnum +from osf.models.notification_campaign import ( + NotificationCampaign, + NotificationCampaignRecipient, + NotificationCampaignRecipientStatus, + NotificationCampaignStatus, ) +from website import settings +from .base import NotificationCampaignTask +from .recipients import assign_batch_id_to_recipients, first_email_subquery -# Matches users who contribute to at least one project or component -# Counts: projects and components at any nesting depth - all share the osf.node type -# Does not count: registrations, draft nodes, and preprints, whose contributors live in a -# separate table entirely -# not filtered on: contributor permissions, whether the project is public, and -# whether the node is deleted or spam-flagged -project_contributor_subquery = Contributor.objects.filter(user_id=OuterRef('pk'), node__type='osf.node') - - -class NotificationCampaignTask(celery_app.Task): - """Shared guards for notification campaign Celery tasks.""" - - abstract = True - - def get_campaign(self, campaign_id, run_id=None, *, abort_if_cancelled=True): - """Load a campaign, or return None if the task should no-op. - """ - campaign = NotificationCampaign.objects.get(id=campaign_id) - if run_id is not None and campaign.run_id != run_id: - return None - if abort_if_cancelled and campaign.status == NotificationCampaignStatus.CANCELLED: - logger.warning(f"Campaign {campaign_id} was cancelled") - return None - return campaign - - def sync_campaign_stats(self, campaign): - stats = get_campaign_recipient_stats(campaign.id) - campaign.recipient_count = stats['recipient_count'] - campaign.sent_count = stats['sent_count'] - campaign.failed_count = stats['failed_count'] - return stats - - def finish_campaign(self, campaign, status=None): - """Sync recipient counters, set completed_at once, optionally update status, and save.""" - self.sync_campaign_stats(campaign) - if campaign.completed_at is None: - campaign.completed_at = timezone.now() - if status is not None: - campaign.status = status - campaign.save() - - -def build_query(node): - """ - Convert a filter tree into a Django Q object. - """ - - if 'field' in node: - value = node['value'] - lookup = node['lookup'] - negated_lookups = { # not native Django field lookups - 'not_contains': 'contains', - 'not_icontains': 'icontains', - } - - if lookup == 'in': - value = [v.strip() for v in value.split(',')] - - if lookup == 'isnull': - value = BooleanField().to_python(value) - - if lookup in negated_lookups: - return ~Q(**{ - f'{node["field"]}__{negated_lookups[lookup]}': value - }) - - return Q(**{ - f'{node["field"]}__{lookup}': value - }) - - operator = node.get('operator', 'AND') - children = node.get('children', []) - - if not children: - return Q() - - query = build_query(children[0]) - - for child in children[1:]: - if operator == 'AND': - query &= build_query(child) - else: - query |= build_query(child) - - return query - - -def build_campaign_filter_query(filters): - """AND together optional predefined, manual and contributor filter clauses.""" - filters = filters or {} - query = Q() - if predefined := filters.get('predefined'): - query &= Q(**FILTER_PRESETS.get(predefined, {})) - if manual := filters.get('manual'): - query &= build_query(manual) - if filters.get('exclude_non_contributors'): - query &= Q(Exists(project_contributor_subquery)) - return query - -@celery_app.task(name='email.create_campaign_recipients') -def create_campaign_recipients(filters=None, campaign_id=None): - recipients_creation_started_at = timezone.now() - campaign = NotificationCampaign.objects.get(id=campaign_id) - if not filters: - - raw_filters = campaign.metadata.get('filters', {}) - filters = build_campaign_filter_query(raw_filters) - - qs = ( - OSFUser.objects - .filter(filters) - .annotate(activity_score=Coalesce(Subquery(counter_subquery), 0)) - .values_list( - 'id', - 'activity_score', - ) - ) - processed_records = 0 - - for rows in batched(qs.iterator(chunk_size=BULK_CREATE_SIZE), BULK_CREATE_SIZE): - NotificationCampaignRecipient.objects.bulk_create( - [ - NotificationCampaignRecipient( - campaign_id=campaign_id, - user_id=user_id, - activity_score=activity_score, - ) - for user_id, activity_score in rows - ], - update_conflicts=True, - update_fields=['activity_score'], - unique_fields=['campaign', 'user'], - ) - processed_records += len(rows) - - campaign.recipient_count = processed_records - campaign.metadata['recipients_creation_finished'] = True - campaign.save() - - recipients_creation_finished_at = timezone.now() - recipients_creation_run_time = (recipients_creation_finished_at - recipients_creation_started_at) - message = (f'[Notification Campaign #{campaign_id}] INFO: ' - f'Recipients creation finished in {recipients_creation_run_time} seconds ' - f'(start={recipients_creation_started_at}, finish={recipients_creation_finished_at}) ' - f'for Campaign {campaign.name} (start={campaign.started_at}).') - logger.info(message) - sentry.log_message(message) - - -def get_campaign_recipient_stats(campaign_id): - return NotificationCampaignRecipient.objects.filter( - campaign_id=campaign_id - ).aggregate( - recipient_count=Count('id'), - sent_count=Count( - 'id', - filter=Q(status=NotificationCampaignRecipientStatus.SENT), - ), - failed_count=Count( - 'id', - filter=Q( - status__in=[ - NotificationCampaignRecipientStatus.FAILED, - NotificationCampaignRecipientStatus.SKIPPED, - ] - ), - ), - ) - - -@celery_app.task(name='email.process_sendgrid_campaign_events') -def process_sendgrid_campaign_events(events): - """Update campaign recipients from filtered SendGrid Event Webhook events. - - Expects events that already include campaign ``custom_args`` - (``campaign_id``, ``campaign_recipient_id``, ``run_id``). Only ``QUEUED`` - or ``FAILED`` recipients whose event ``run_id`` matches the campaign's - current run are updated; delayed webhooks from a prior run are ignored. - - A ``delivered`` event wins over failure events for the same recipient - (including a prior ``FAILED`` from an earlier webhook) so a confirmed - delivery is never left failed (and retried). - """ - campaign_ids = { - event.get('campaign_id') - for event in events - if event.get('campaign_id', False) - } - if not campaign_ids: - return - - current_run_ids = { - str(campaign.id): str(campaign.run_id) - for campaign in NotificationCampaign.objects.filter( - id__in=campaign_ids, - run_id__isnull=False, - ) - } - if not current_run_ids: - return - - success_ids = set() - failed = dict() - - for event in events: - campaign_id = event.get('campaign_id', '') - current_run_id = current_run_ids.get(str(campaign_id), None) - if current_run_id is None or event.get('run_id') != current_run_id: - continue - - event_type = event.get('event') - recipient_id = event.get('campaign_recipient_id') - if not recipient_id: - continue - try: - recipient_pk = int(recipient_id) - except (TypeError, ValueError): - continue - - if event_type in SENDGRID_SUCCESS_EVENTS: - success_ids.add(recipient_pk) - failed.pop(recipient_pk, None) - elif event_type in SENDGRID_FAILURE_EVENTS: - if recipient_pk in success_ids: - continue - error_message = event.get('reason') or event.get('type') or event_type - failed[recipient_pk] = error_message - - if success_ids: - NotificationCampaignRecipient.objects.filter( - id__in=success_ids, - status__in=[ - NotificationCampaignRecipientStatus.QUEUED, - NotificationCampaignRecipientStatus.FAILED, - ], - ).update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) - - if failed: - failed_errors = [ - When(id=recipient_pk, then=Value(error_message)) - for recipient_pk, error_message in failed.items() - ] - NotificationCampaignRecipient.objects.filter( - id__in=failed.keys(), - status=NotificationCampaignRecipientStatus.QUEUED, - ).update( - status=NotificationCampaignRecipientStatus.FAILED, - error_message=Case( - *failed_errors, - default=Value('SendGrid delivery failed'), - output_field=TextField(), - ), - ) +logger = logging.getLogger(__name__) @celery_app.task(bind=True, base=NotificationCampaignTask, name='email.process_campaign_retry') -def process_campaign_retry(self, campaign_id, run_id): +def process_campaign_retry(self, campaign_id, run_id, use_sendgrid_webhook=False): campaign = self.get_campaign(campaign_id, run_id, abort_if_cancelled=False) if campaign is None: return @@ -306,30 +31,40 @@ def process_campaign_retry(self, campaign_id, run_id): campaign.refresh_from_db() execution = campaign.metadata.get('execution', {}) - queued_qs = NotificationCampaignRecipient.objects.filter( + if campaign.status == NotificationCampaignStatus.CANCELLED: + message = f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} was cancelled.' + logger.info(message) + sentry.log_message(message) + self.finish_campaign(campaign) + return + + awaiting_qs = NotificationCampaignRecipient.objects.filter( campaign=campaign, - status=NotificationCampaignRecipientStatus.QUEUED, + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY, ) - if queued_qs.exists(): + if awaiting_qs.exists(): delivery_timeout = execution.get('delivery_timeout', settings.DEFAULT_CAMPAIGN_DELIVERY_TIMEOUT) reference_time = campaign.started_at or campaign.created_at if timezone.now() - reference_time < timedelta(seconds=delivery_timeout): # Still waiting for in-flight sends / SendGrid delivery webhooks. self.sync_campaign_stats(campaign) - campaign.save() process_campaign_retry.apply_async( - kwargs={'campaign_id': campaign_id, 'run_id': campaign.run_id}, + kwargs={ + 'campaign_id': campaign_id, + 'run_id': campaign.run_id, + 'use_sendgrid_webhook': use_sendgrid_webhook, + }, countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), ) return - timed_out = queued_qs.update( + timed_out = awaiting_qs.update( status=NotificationCampaignRecipientStatus.FAILED, error_message='SendGrid delivery timeout', ) message = ( f'[Notification Campaign #{campaign_id}] WARNING: ' - f'Marked {timed_out} queued recipients as FAILED after delivery timeout ' + f'Marked {timed_out} awaiting-delivery recipients as FAILED after delivery timeout ' f'({delivery_timeout}s) for campaign {campaign.name}.' ) logger.warning(message) @@ -339,13 +74,6 @@ def process_campaign_retry(self, campaign_id, run_id): self.finish_campaign(campaign, NotificationCampaignStatus.PARTIALLY_COMPLETED) return - if campaign.status == NotificationCampaignStatus.CANCELLED: - message = f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} was cancelled.' - logger.info(message) - sentry.log_message(message) - self.finish_campaign(campaign) - return - failed_recipients_count = NotificationCampaignRecipient.objects.filter( campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED, @@ -366,6 +94,7 @@ def process_campaign_retry(self, campaign_id, run_id): args=[campaign_id, campaign.run_id], kwargs={ 'restart_failed': True, + 'use_sendgrid_webhook': use_sendgrid_webhook, }, ) return @@ -387,61 +116,29 @@ def start_notification_campaign(self, campaign_id, restart_failed=False, restart del getattr(NotificationTypeEnum, notification_type_name).instance if restart_stuck: + # Only reclaim Celery-in-flight batches. Leave AWAITING_DELIVERY alone so + # SendGrid webhooks for already-accepted mail still match recipient.sent_at. NotificationCampaignRecipient.objects.filter( campaign_id=campaign_id, - status=NotificationCampaignRecipientStatus.QUEUED + status=NotificationCampaignRecipientStatus.QUEUED, ).update(status=NotificationCampaignRecipientStatus.PENDING, batch_id=None) + # Decide once per run which SendGrid user sends the campaign: delivery tracked via + # webhook requires the sub-user with Event Webhook enabled. + use_sendgrid_webhook = bool( + campaign.metadata.get('sendgrid_bulk', False) or settings.CAMPAIGN_SINGLE_SEND_WEBHOOK + ) dispatch_campaign.apply_async( args=[campaign_id, campaign.run_id], kwargs={ 'restart_failed': restart_failed, + 'use_sendgrid_webhook': use_sendgrid_webhook, }, ) -def assign_batch_id_to_recipients( - campaign_id, - batch_size, - restart_failed=False, - min_activity=None, - max_activity=None, - spam=None, -): - batch_id = uuid.uuid4() - filters = Q(campaign_id=campaign_id) - - if restart_failed: - filters &= Q( - status=NotificationCampaignRecipientStatus.FAILED, - ) - else: - filters &= Q( - status=NotificationCampaignRecipientStatus.PENDING, - batch_id__isnull=True, - ) - - if min_activity is not None: - filters &= Q(activity_score__gte=min_activity) - if max_activity is not None: - filters &= Q(activity_score__lt=max_activity) - - if spam is not None: - filters &= Q(user__spam_status=SpamStatus.SPAM) if spam else ~Q(user__spam_status=SpamStatus.SPAM) - - recipient_ids = NotificationCampaignRecipient.objects.filter( - filters - ).values_list('id', flat=True)[:batch_size] - - updated = NotificationCampaignRecipient.objects.filter(id__in=recipient_ids).update(batch_id=batch_id, status=NotificationCampaignRecipientStatus.QUEUED) - - if updated == 0: - return None - - return batch_id - @celery_app.task(bind=True, base=NotificationCampaignTask, name='email.dispatch_campaign') -def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_stuck=False): +def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_stuck=False, use_sendgrid_webhook=False): campaign = self.get_campaign(campaign_id, run_id) if campaign is None: return @@ -452,6 +149,10 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_s sentry.log_message(message) return + # Refresh counters once per dispatch interval (covers SENT without webhook and + # FAILED/SKIPPED set by batches) instead of re-aggregating after every batch. + self.sync_campaign_stats(campaign) + queued_batches_count = NotificationCampaignRecipient.objects.filter( campaign=campaign, batch_id__isnull=False, @@ -506,6 +207,7 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_s notification_type_name=notification_type_name, run_id=run_id, developer_reminder=developer_reminder, + use_sendgrid_webhook=use_sendgrid_webhook, ) new_queued_batches += 1 @@ -519,13 +221,14 @@ def dispatch_campaign(self, campaign_id, run_id, restart_failed=False, restart_s if no_more_recipients: process_campaign_retry.delay( - campaign_id=campaign_id, run_id=campaign.run_id + campaign_id=campaign_id, run_id=campaign.run_id, use_sendgrid_webhook=use_sendgrid_webhook ) else: self.apply_async( args=[campaign_id, run_id], kwargs={ 'restart_failed': restart_failed, + 'use_sendgrid_webhook': use_sendgrid_webhook, }, countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), ) @@ -540,6 +243,7 @@ def send_campaign_batch( campaign_id=None, run_id=None, developer_reminder=False, + use_sendgrid_webhook=False, ): campaign = self.get_campaign(campaign_id, run_id) if campaign is None: @@ -582,6 +286,7 @@ def send_campaign_batch( ) logger.warning(message) sentry.log_message(message) + self.send_campaign_log_email(campaign, message=message, status='WARNING') recipient_records = [] recipients_qs_annotated = recipients_qs.annotate( @@ -595,17 +300,20 @@ def send_campaign_batch( invalid_emails_qs = recipients_qs_annotated.filter(recipient_address__isnull=True) invalid_emails_qs.update(status=NotificationCampaignRecipientStatus.SKIPPED, error_message='Invalid email address') + sendgrid_client = get_sendgrid_client(use_webhook=use_sendgrid_webhook) + if campaign.metadata.get('sendgrid_bulk', False): recipients = list(valid_emails_qs) recipient_emails = [] custom_args_list = [] + sent_at = int(timezone.now().timestamp()) for recipient in recipients: recipient_emails.append(recipient.recipient_address) custom_args_list.append( { 'campaign_recipient_id': str(recipient.id), 'campaign_id': str(campaign_id), - 'run_id': str(run_id), + 'sent_at': str(sent_at), } ) try: @@ -615,8 +323,12 @@ def send_campaign_batch( context=context, email_context={'custom_args_list': custom_args_list}, is_multiple=True, + client=sendgrid_client, + ) + valid_emails_qs.update( + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY, + sent_at=sent_at, ) - # Leave QUEUED until SendGrid Event Webhook confirms delivery. except Exception as exc: message = (f'[Notification Campaign #{campaign_id}] ERROR: ' f'Campaign {campaign.name} sendgrid bulk request failed, error={str(exc)}') @@ -627,6 +339,7 @@ def send_campaign_batch( rendered_html = _render_email_html(notification_type, context) for recipient in valid_emails_qs: notification_started_at = timezone.now() + sent_at = int(notification_started_at.timestamp()) try: send_email( recipient_address=recipient.recipient_address, @@ -636,13 +349,18 @@ def send_campaign_batch( 'custom_args': { 'campaign_recipient_id': str(recipient.id), 'campaign_id': str(campaign_id), - 'run_id': str(run_id), + 'sent_at': str(sent_at), }, }, rendered_html=rendered_html, + client=sendgrid_client, ) - recipient.status = NotificationCampaignRecipientStatus.SENT + if use_sendgrid_webhook: + recipient.status = NotificationCampaignRecipientStatus.AWAITING_DELIVERY + else: + recipient.status = NotificationCampaignRecipientStatus.SENT recipient.error_message = None + recipient.sent_at = sent_at recipient_records.append(recipient) except Exception as exc: message = (f'[Notification Campaign #{campaign_id}] ERROR:' @@ -663,13 +381,7 @@ def send_campaign_batch( logger.warning(message) sentry.log_message(message) if recipient_records: - NotificationCampaignRecipient.objects.bulk_update(recipient_records, ['status', 'error_message']) - - # Lock the campaign row so concurrent batches cannot overwrite counters with a stale aggregate snapshot - with transaction.atomic(): - notification_campaign = NotificationCampaign.objects.select_for_update().get(pk=campaign_id) - self.sync_campaign_stats(notification_campaign) - notification_campaign.save(update_fields=['sent_count', 'failed_count', 'recipient_count', 'updated_at']) + NotificationCampaignRecipient.objects.bulk_update(recipient_records, ['status', 'error_message', 'sent_at']) batch_finished_at = timezone.now() batch_run_time = (batch_finished_at - batch_started_at).total_seconds() diff --git a/osf/email/notification_campaign/recipients.py b/osf/email/notification_campaign/recipients.py new file mode 100644 index 00000000000..b468364a157 --- /dev/null +++ b/osf/email/notification_campaign/recipients.py @@ -0,0 +1,226 @@ +import logging +import uuid +from itertools import batched + +from django.db.models import BooleanField, Count, OuterRef, Q, Subquery, Exists +from django.db.models.functions import Coalesce +from django.utils import timezone + +from framework import sentry +from framework.celery_tasks import app as celery_app +from osf.models import Email, OSFUser, UserActivityCounter, Contributor +from osf.models.notification_campaign import ( + NotificationCampaign, + NotificationCampaignRecipient, + NotificationCampaignRecipientStatus, +) +from osf.models.spam import SpamStatus + +logger = logging.getLogger(__name__) + +BULK_CREATE_SIZE = 5000 +FILTER_PRESETS = { + 'all': {}, + 'active': {'is_active': True}, + 'internal': {'is_active': True, 'is_staff': True, 'username__endswith': '@cos.io'}, +} + +first_email_subquery = ( + Email.objects + .filter(user=OuterRef('user_id')) + .values('address')[:1] +) + + +counter_subquery = ( + UserActivityCounter.objects + .filter(_id=OuterRef('guids___id')) + .values('total')[:1] +) + + +# Matches users who contribute to at least one project or component +# Counts: projects and components at any nesting depth - all share the osf.node type +# Does not count: registrations, draft nodes, and preprints, whose contributors live in a +# separate table entirely +# not filtered on: contributor permissions, whether the project is public, and +# whether the node is deleted or spam-flagged +project_contributor_subquery = Contributor.objects.filter(user_id=OuterRef('pk'), node__type='osf.node') + + +def build_query(node): + """ + Convert a filter tree into a Django Q object. + """ + + if 'field' in node: + value = node['value'] + lookup = node['lookup'] + negated_lookups = { # not native Django field lookups + 'not_contains': 'contains', + 'not_icontains': 'icontains', + } + + if lookup == 'in': + value = [v.strip() for v in value.split(',')] + + if lookup == 'isnull': + value = BooleanField().to_python(value) + + if lookup in negated_lookups: + return ~Q(**{ + f'{node["field"]}__{negated_lookups[lookup]}': value + }) + + return Q(**{ + f'{node["field"]}__{lookup}': value + }) + + operator = node.get('operator', 'AND') + children = node.get('children', []) + + if not children: + return Q() + + query = build_query(children[0]) + + for child in children[1:]: + if operator == 'AND': + query &= build_query(child) + else: + query |= build_query(child) + + return query + + +def build_campaign_filter_query(filters): + """AND together optional predefined, manual and contributor filter clauses.""" + filters = filters or {} + query = Q() + if predefined := filters.get('predefined'): + query &= Q(**FILTER_PRESETS.get(predefined, {})) + if manual := filters.get('manual'): + query &= build_query(manual) + if filters.get('exclude_non_contributors'): + query &= Q(Exists(project_contributor_subquery)) + return query + + +@celery_app.task(name='email.create_campaign_recipients') +def create_campaign_recipients(filters=None, campaign_id=None): + recipients_creation_started_at = timezone.now() + campaign = NotificationCampaign.objects.get(id=campaign_id) + if not filters: + + raw_filters = campaign.metadata.get('filters', {}) + filters = build_campaign_filter_query(raw_filters) + + qs = ( + OSFUser.objects + .filter(filters) + .annotate(activity_score=Coalesce(Subquery(counter_subquery), 0)) + .values_list( + 'id', + 'activity_score', + ) + ) + processed_records = 0 + + for rows in batched(qs.iterator(chunk_size=BULK_CREATE_SIZE), BULK_CREATE_SIZE): + NotificationCampaignRecipient.objects.bulk_create( + [ + NotificationCampaignRecipient( + campaign_id=campaign_id, + user_id=user_id, + activity_score=activity_score, + ) + for user_id, activity_score in rows + ], + update_conflicts=True, + update_fields=['activity_score'], + unique_fields=['campaign', 'user'], + ) + processed_records += len(rows) + + campaign.recipient_count = processed_records + campaign.metadata['recipients_creation_finished'] = True + campaign.save() + + recipients_creation_finished_at = timezone.now() + recipients_creation_run_time = (recipients_creation_finished_at - recipients_creation_started_at) + message = (f'[Notification Campaign #{campaign_id}] INFO: ' + f'Recipients creation finished in {recipients_creation_run_time} seconds ' + f'(start={recipients_creation_started_at}, finish={recipients_creation_finished_at}) ' + f'for Campaign {campaign.name} (start={campaign.started_at}).') + logger.info(message) + sentry.log_message(message) + + +def get_campaign_recipient_stats(campaign_id): + return NotificationCampaignRecipient.objects.filter( + campaign_id=campaign_id + ).aggregate( + recipient_count=Count('id'), + sent_count=Count( + 'id', + filter=Q(status=NotificationCampaignRecipientStatus.SENT), + ), + failed_count=Count( + 'id', + filter=Q( + status__in=[ + NotificationCampaignRecipientStatus.FAILED, + NotificationCampaignRecipientStatus.SKIPPED, + ] + ), + ), + queued_count=Count( + 'id', + filter=Q(status=NotificationCampaignRecipientStatus.QUEUED), + ), + awaiting_count=Count( + 'id', + filter=Q(status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY), + ), + ) + + +def assign_batch_id_to_recipients( + campaign_id, + batch_size, + restart_failed=False, + min_activity=None, + max_activity=None, + spam=None, +): + batch_id = uuid.uuid4() + filters = Q(campaign_id=campaign_id) + + if restart_failed: + filters &= Q( + status=NotificationCampaignRecipientStatus.FAILED, + ) + else: + filters &= Q( + status=NotificationCampaignRecipientStatus.PENDING, + batch_id__isnull=True, + ) + + if min_activity is not None: + filters &= Q(activity_score__gte=min_activity) + if max_activity is not None: + filters &= Q(activity_score__lt=max_activity) + + if spam is not None: + filters &= Q(user__spam_status=SpamStatus.SPAM) if spam else ~Q(user__spam_status=SpamStatus.SPAM) + + recipient_ids = NotificationCampaignRecipient.objects.filter( + filters + ).values_list('id', flat=True)[:batch_size] + + updated = NotificationCampaignRecipient.objects.filter(id__in=recipient_ids).update(batch_id=batch_id, status=NotificationCampaignRecipientStatus.QUEUED) + + if updated == 0: + return None + + return batch_id diff --git a/osf/email/notification_campaign/webhook.py b/osf/email/notification_campaign/webhook.py new file mode 100644 index 00000000000..ba2940ad85d --- /dev/null +++ b/osf/email/notification_campaign/webhook.py @@ -0,0 +1,181 @@ +from django.db import InterfaceError, OperationalError, transaction +from django.db.models import Case, TextField, Value, When + +from framework.celery_tasks import app as celery_app +from osf.models.notification_campaign import ( + NotificationCampaignRecipient, + NotificationCampaignRecipientStatus, +) + +from .base import NotificationCampaignTask + +# Flattened onto SendGrid Event Webhook payloads via personalization custom_args. +CAMPAIGN_CUSTOM_ARG_KEYS = ('campaign_id', 'campaign_recipient_id', 'sent_at') +SENDGRID_SUCCESS_EVENTS = frozenset({'delivered'}) +# Permanent / non-retryable delivery failures → SKIPPED (excluded from restart_failed). +SENDGRID_SKIP_EVENTS = frozenset({'dropped'}) +# Soft / retryable failures → FAILED (eligible for campaign retry). +SENDGRID_SOFT_BOUNCE_TYPE = 'blocked' +SENDGRID_HARD_BOUNCE_TYPE = 'bounce' + + +def _sendgrid_event_error_message(event): + return event.get('reason') or event.get('type') or event.get('event') or 'SendGrid delivery failed' + + +def _classify_sendgrid_failure(event): + """Return SKIPPED/FAILED status for a failure event, or None to ignore. + + - ``dropped`` / hard ``bounce`` (type ``bounce`` or missing): permanent → SKIPPED + - soft ``bounce`` (type ``blocked``): transient → FAILED (campaign may retry) + - ``deferred`` and other events: ignored (SendGrid keeps retrying deferred) + """ + event_type = event.get('event') + if event_type in SENDGRID_SKIP_EVENTS: + return NotificationCampaignRecipientStatus.SKIPPED + if event_type == 'bounce': + bounce_type = event.get('type') or SENDGRID_HARD_BOUNCE_TYPE + if bounce_type == SENDGRID_SOFT_BOUNCE_TYPE: + return NotificationCampaignRecipientStatus.FAILED + return NotificationCampaignRecipientStatus.SKIPPED + return None + + +def _classify_sendgrid_events(events, recipients): + """Classify SendGrid events into success/failure/skipped categories.""" + successed = set() + failed = dict() + skipped = dict() + + for event in events: + recipient_pk = event.get('campaign_recipient_id') + if recipient_pk is None: + continue + + recipient = recipients.get(int(recipient_pk)) + if recipient is None: + continue + + if ( + recipient.sent_at is None + or str(recipient.campaign_id) != str(event.get('campaign_id', '')) + or str(recipient.sent_at) != str(event.get('sent_at', '')) + ): + continue + + event_type = event.get('event') + if event_type in SENDGRID_SUCCESS_EVENTS: + successed.add(recipient_pk) + failed.pop(recipient_pk, None) + skipped.pop(recipient_pk, None) + continue + + failure_status = _classify_sendgrid_failure(event) + if failure_status is None or recipient_pk in successed: + continue + + error_message = _sendgrid_event_error_message(event) + if failure_status == NotificationCampaignRecipientStatus.SKIPPED: + skipped[recipient_pk] = error_message + failed.pop(recipient_pk, None) + else: + failed[recipient_pk] = error_message + skipped.pop(recipient_pk, None) + + return successed, failed, skipped + + +# Safe to retry / redeliver: updates are status-guarded and atomic, so replaying the +# same events is a no-op. Retry only transient DB errors (connection drops, deadlocks). +@celery_app.task( + bind=True, + base=NotificationCampaignTask, + name='email.process_sendgrid_campaign_events', + autoretry_for=(OperationalError, InterfaceError), + retry_backoff=True, + retry_backoff_max=60, + retry_jitter=True, + max_retries=5, + acks_late=True, + reject_on_worker_lost=True, +) +def process_sendgrid_campaign_events(self, events): + """Update campaign recipients from filtered SendGrid Event Webhook events. + + Expects events that already include campaign ``custom_args`` + (``campaign_id``, ``campaign_recipient_id``, ``sent_at``). Only + ``AWAITING_DELIVERY`` or ``FAILED`` recipients whose event ``sent_at`` + matches the recipient's stored send stamp are updated; delayed webhooks + from a prior send (e.g. after ``restart_failed``) are ignored. + + Permanent failures (``dropped``, hard bounce) become ``SKIPPED`` so + ``restart_failed`` will not resend them. Soft bounces (``type=blocked``) + become ``FAILED`` and remain retryable. ``deferred`` events are ignored + (SendGrid retries those itself). + + A ``delivered`` event wins over failure events for the same recipient + (including a prior ``FAILED``/``SKIPPED`` from an earlier webhook) so a + confirmed delivery is never left failed (and retried). + """ + recipient_ids = { + int(recipient_id) + for event in events + if (recipient_id := event.get('campaign_recipient_id')) is not None + } + recipients_by_id = { + recipient.id: recipient + for recipient in NotificationCampaignRecipient.objects.filter( + id__in=recipient_ids, + ).only('id', 'campaign_id', 'sent_at') + } + if not recipients_by_id: + return + + success_ids, failed, skipped = _classify_sendgrid_events(events, recipients_by_id) + if not success_ids and not failed and not skipped: + return + + with transaction.atomic(): + if success_ids: + NotificationCampaignRecipient.objects.filter( + id__in=success_ids, + status__in=[ + NotificationCampaignRecipientStatus.AWAITING_DELIVERY, + NotificationCampaignRecipientStatus.FAILED, + NotificationCampaignRecipientStatus.SKIPPED, + ], + ).update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) + + if failed: + failed_errors = [ + When(id=recipient_pk, then=Value(error_message)) + for recipient_pk, error_message in failed.items() + ] + NotificationCampaignRecipient.objects.filter( + id__in=failed.keys(), + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY, + ).update( + status=NotificationCampaignRecipientStatus.FAILED, + error_message=Case( + *failed_errors, + default=Value('SendGrid delivery failed'), + output_field=TextField(), + ), + ) + + if skipped: + skipped_errors = [ + When(id=recipient_pk, then=Value(error_message)) + for recipient_pk, error_message in skipped.items() + ] + NotificationCampaignRecipient.objects.filter( + id__in=skipped.keys(), + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY, + ).update( + status=NotificationCampaignRecipientStatus.SKIPPED, + error_message=Case( + *skipped_errors, + default=Value('SendGrid delivery skipped'), + output_field=TextField(), + ), + ) diff --git a/osf/migrations/0057_post_enter_p2.py b/osf/migrations/0057_post_enter_p2.py new file mode 100644 index 00000000000..7b6c1722b59 --- /dev/null +++ b/osf/migrations/0057_post_enter_p2.py @@ -0,0 +1,34 @@ +# Generated by Django 4.2.26 on 2026-10-02 14:16 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0056_merge_20260918'), + ] + + operations = [ + migrations.AddField( + model_name='notificationcampaignrecipient', + name='sent_at', + field=models.PositiveBigIntegerField(blank=True, db_column='sent_at_big_int', null=True), + ), + migrations.AlterField( + model_name='notificationcampaignrecipient', + name='status', + field=models.CharField( + choices=[ + ('queued', 'Queued'), + ('pending', 'Pending'), + ('awaiting_delivery', 'Awaiting Delivery'), + ('sent', 'Sent'), ('failed', 'Failed'), + ('skipped', 'Skipped'), + ('postponed', 'Postponed') + ], + db_index=True, + default='pending', + max_length=20), + ), + ] diff --git a/osf/migrations/0058_merge_20261028_2014.py b/osf/migrations/0058_merge_20261028_2014.py new file mode 100644 index 00000000000..3df1e06be11 --- /dev/null +++ b/osf/migrations/0058_merge_20261028_2014.py @@ -0,0 +1,14 @@ +# Generated by Django 4.2.30 on 2026-10-28 20:14 + +from django.db import migrations + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0057_alter_adminprofile_options'), + ('osf', '0057_post_enter_p2'), + ] + + operations = [ + ] diff --git a/osf/models/notification_campaign.py b/osf/models/notification_campaign.py index ef3441233fb..d36da346743 100644 --- a/osf/models/notification_campaign.py +++ b/osf/models/notification_campaign.py @@ -13,8 +13,11 @@ class NotificationCampaignStatus(models.TextChoices): ENDED = 'ended', 'Ended' class NotificationCampaignRecipientStatus(models.TextChoices): + # Batch assigned; Celery send task in flight (counts toward max_queued_batches). QUEUED = 'queued', 'Queued' PENDING = 'pending', 'Pending' + # SendGrid accepted the message; waiting on Event Webhook delivery/failure. + AWAITING_DELIVERY = 'awaiting_delivery', 'Awaiting Delivery' SENT = 'sent', 'Sent' FAILED = 'failed', 'Failed' SKIPPED = 'skipped', 'Skipped' @@ -131,6 +134,14 @@ class NotificationCampaignRecipient(models.Model): activity_score = models.IntegerField(default=0) batch_id = models.UUIDField(null=True, blank=True, db_index=True) + # Unix seconds when mail was handed to SendGrid for this send attempt. Mirrored + # into SendGrid custom_args as ``sent_at`` and matched on inbound webhooks so + # delayed events from a prior attempt (e.g. after restart_failed) are ignored. + sent_at = models.PositiveBigIntegerField( + null=True, + blank=True, + db_column='sent_at_big_int', + ) class Meta: unique_together = ('campaign', 'user') diff --git a/osf/models/notification_type.py b/osf/models/notification_type.py index ad8603cf5cd..5a042e63edc 100644 --- a/osf/models/notification_type.py +++ b/osf/models/notification_type.py @@ -15,6 +15,8 @@ def get_default_frequency_choices(): class NotificationTypeEnum(str, Enum): EMPTY = 'empty' BLANK = 'blank' + NOTIFICATION_CAMPAIGN_LOG = 'notification_campaign_log' + # Desk notifications REVIEWS_SUBMISSION_STATUS = 'reviews_submission_status' ADDONS_BOA_JOB_FAILURE = 'addon_boa_job_failure' diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index 4e135889c12..3fb139bb202 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -6,12 +6,13 @@ from django.utils import timezone from django.db.models import Q -from osf.email import _build_sendgrid_personalizations +from osf.email import _build_sendgrid_personalizations, _sendgrid_client, get_sendgrid_client from osf.email.notification_campaign import ( NotificationCampaignTask, assign_batch_id_to_recipients, build_campaign_filter_query, create_campaign_recipients, + dispatch_campaign, get_campaign_recipient_stats, process_campaign_retry, process_sendgrid_campaign_events, @@ -74,11 +75,15 @@ def test_sync_campaign_stats(self, task, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) NotificationCampaignRecipient.objects.filter(campaign=campaign).update( - status=NotificationCampaignRecipientStatus.SENT + status=NotificationCampaignRecipientStatus.QUEUED ) - stats = task.sync_campaign_stats(campaign) - assert stats['sent_count'] == 1 - assert campaign.sent_count == 1 + stats = task.sync_campaign_stats(campaign, save=True) + assert stats['queued_count'] == 1 + assert stats['awaiting_count'] == 0 + campaign.refresh_from_db() + assert campaign.recipient_count == 1 + assert campaign.sent_count == 0 + assert campaign.failed_count == 0 @pytest.fixture @@ -467,6 +472,8 @@ def test_empty_campaign(self, campaign): 'recipient_count': 0, 'sent_count': 0, 'failed_count': 0, + 'queued_count': 0, + 'awaiting_count': 0, } def test_counts_sent_failed_and_skipped(self, campaign, notification_type): @@ -474,8 +481,9 @@ def test_counts_sent_failed_and_skipped(self, campaign, notification_type): failed = UserFactory() skipped = UserFactory() pending = UserFactory() + queued = UserFactory() create_campaign_recipients( - Q(**{'id__in': [sent.id, failed.id, skipped.id, pending.id]}), + Q(**{'id__in': [sent.id, failed.id, skipped.id, pending.id, queued.id]}), campaign_id=campaign.id, ) NotificationCampaignRecipient.objects.filter(campaign=campaign, user=sent).update( @@ -487,11 +495,24 @@ def test_counts_sent_failed_and_skipped(self, campaign, notification_type): NotificationCampaignRecipient.objects.filter(campaign=campaign, user=skipped).update( status=NotificationCampaignRecipientStatus.SKIPPED ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=queued).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + awaiting = UserFactory() + create_campaign_recipients( + Q(**{'id__in': [awaiting.id]}), + campaign_id=campaign.id, + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=awaiting).update( + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY + ) assert get_campaign_recipient_stats(campaign.id) == { - 'recipient_count': 4, + 'recipient_count': 6, 'sent_count': 1, 'failed_count': 2, # FAILED + SKIPPED + 'queued_count': 1, + 'awaiting_count': 1, } def test_scopes_to_requested_campaign(self, campaign, notification_type): @@ -515,11 +536,15 @@ def test_scopes_to_requested_campaign(self, campaign, notification_type): 'recipient_count': 1, 'sent_count': 1, 'failed_count': 0, + 'queued_count': 0, + 'awaiting_count': 0, } assert get_campaign_recipient_stats(other.id) == { 'recipient_count': 1, 'sent_count': 0, 'failed_count': 1, + 'queued_count': 0, + 'awaiting_count': 0, } @@ -678,7 +703,7 @@ def test_campaign_start_restart_stuck_counts(self, mock_delay, campaign): restart_stuck=True, ) - @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign.apply_async') def test_start_schedules_workflow(self, mock_dispatch_campaign, campaign): high = UserFactory() low = UserFactory() @@ -706,7 +731,26 @@ def test_start_schedules_workflow(self, mock_dispatch_campaign, campaign): assert campaign.recipient_count == 3 mock_dispatch_campaign.assert_called_once() - @mock.patch('osf.email.notification_campaign.dispatch_campaign') + @pytest.mark.parametrize('sendgrid_bulk, single_send_webhook, expected', [ + (False, False, False), + (True, False, True), + (False, True, True), + ]) + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign.apply_async') + def test_start_decides_sendgrid_webhook_client( + self, mock_dispatch_campaign, campaign, sendgrid_bulk, single_send_webhook, expected + ): + campaign.metadata['sendgrid_bulk'] = sendgrid_bulk + campaign.run_id = uuid.uuid4() + campaign.save() + + with mock.patch('website.settings.CAMPAIGN_SINGLE_SEND_WEBHOOK', single_send_webhook): + start_notification_campaign(campaign.id) + + mock_dispatch_campaign.assert_called_once() + assert mock_dispatch_campaign.call_args.kwargs['kwargs']['use_sendgrid_webhook'] is expected + + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign') def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_dispatch_campaign, campaign): confirmed = UserFactory() unconfirmed = UserFactory(date_confirmed=None, is_registered=False) @@ -745,7 +789,7 @@ def test_start_excludes_unconfirmed_accounts_when_enabled(self, mock_dispatch_ca ) assert recipient_ids == {confirmed.id} - @mock.patch('osf.email.notification_campaign.dispatch_campaign') + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign') def test_start_restart_failed_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() _set_activity(user, 50) @@ -766,20 +810,59 @@ def test_start_restart_failed_does_not_recreate_recipients(self, mock_dispatch_c assert NotificationCampaignRecipient.objects.filter(campaign=campaign).count() == 1 assert NotificationCampaignRecipient.objects.get(pk=recipient.pk).status == NotificationCampaignRecipientStatus.FAILED - @mock.patch('osf.email.notification_campaign.dispatch_campaign') - def test_start_restart_stuck_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign') + def test_start_restart_stuck_resets_only_queued(self, mock_dispatch_campaign, campaign): user = UserFactory() - create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + other = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id, other.id]}), campaign_id=campaign.id) + awaiting_sent_at = int(timezone.now().timestamp()) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=user).update( + status=NotificationCampaignRecipientStatus.QUEUED, + batch_id=uuid.uuid4(), + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=other).update( + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY, + batch_id=uuid.uuid4(), + sent_at=awaiting_sent_at, + ) campaign.run_id = uuid.uuid4() - campaign.recipient_count = 1 + campaign.recipient_count = 2 campaign.save() mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id, restart_stuck=True) - assert NotificationCampaignRecipient.objects.filter(campaign=campaign).count() == 1 + queued = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) + awaiting = NotificationCampaignRecipient.objects.get(campaign=campaign, user=other) + assert queued.status == NotificationCampaignRecipientStatus.PENDING + assert queued.batch_id is None + assert awaiting.status == NotificationCampaignRecipientStatus.AWAITING_DELIVERY + assert awaiting.batch_id is not None + assert awaiting.sent_at == awaiting_sent_at campaign.refresh_from_db() - assert campaign.recipient_count == 1 + assert campaign.recipient_count == 2 + + +class TestDispatchCampaign: + + @mock.patch('osf.email.notification_campaign.execution.process_campaign_retry.delay') + @mock.patch('osf.email.notification_campaign.execution.send_campaign_batch.delay') + def test_dispatch_passes_sendgrid_webhook_flag_down(self, mock_send_batch, mock_retry, campaign): + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + campaign.run_id = uuid.uuid4() + campaign.status = NotificationCampaignStatus.RUNNING + campaign.save() + + dispatch_campaign(campaign.id, campaign.run_id, use_sendgrid_webhook=True) + + mock_send_batch.assert_called_once() + assert mock_send_batch.call_args.kwargs['use_sendgrid_webhook'] is True + mock_retry.assert_called_once_with( + campaign_id=campaign.id, + run_id=campaign.run_id, + use_sendgrid_webhook=True, + ) class TestSendCampaignBatch: @@ -792,7 +875,7 @@ def running_campaign(self, campaign): campaign.save() return campaign - @mock.patch('osf.email.notification_campaign.send_email') + @mock.patch('osf.email.notification_campaign.execution.send_email') def test_send_campaign_batch_marks_recipients_sent(self, mock_send_email, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) @@ -810,11 +893,11 @@ def test_send_campaign_batch_marks_recipients_sent(self, mock_send_email, runnin recipient.refresh_from_db() running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SENT - assert running_campaign.sent_count == 1 + assert get_campaign_recipient_stats(running_campaign.id)['sent_count'] == 1 mock_send_email.assert_called_once() - @mock.patch('osf.email.notification_campaign.send_email', side_effect=Exception('send failed')) - @mock.patch('osf.email.notification_campaign.sentry.log_exception') + @mock.patch('osf.email.notification_campaign.execution.send_email', side_effect=Exception('send failed')) + @mock.patch('osf.email.notification_campaign.execution.sentry.log_exception') def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_send_email, running_campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) @@ -833,7 +916,7 @@ def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_sen running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.FAILED assert 'send failed' in recipient.error_message - assert running_campaign.failed_count == 1 + assert get_campaign_recipient_stats(running_campaign.id)['failed_count'] == 1 def test_send_campaign_batch_skips_invalid_email_addresses(self, running_campaign): user = UserFactory() @@ -857,10 +940,10 @@ def test_send_campaign_batch_skips_invalid_email_addresses(self, running_campaig running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SKIPPED assert recipient.error_message == 'Invalid email address' - assert running_campaign.failed_count == 1 - assert running_campaign.sent_count == 0 + assert get_campaign_recipient_stats(running_campaign.id)['failed_count'] == 1 + assert get_campaign_recipient_stats(running_campaign.id)['sent_count'] == 0 - @mock.patch('osf.email.notification_campaign.send_email_with_send_grid') + @mock.patch('osf.email.notification_campaign.execution.send_email_with_send_grid') def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_campaign): user = UserFactory() running_campaign.metadata['sendgrid_bulk'] = True @@ -882,14 +965,58 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ mock_sendgrid.assert_called_once() assert mock_sendgrid.call_args.kwargs.get('is_multiple') is True email_context = mock_sendgrid.call_args.kwargs.get('email_context') or {} - assert email_context['custom_args_list'] == [{ - 'campaign_recipient_id': str(recipient.id), - 'campaign_id': str(running_campaign.id), - 'run_id': str(running_campaign.run_id), - }] - assert recipient.status == NotificationCampaignRecipientStatus.QUEUED - assert running_campaign.sent_count == 0 - assert running_campaign.failed_count == 0 + custom_args = email_context['custom_args_list'][0] + assert custom_args['campaign_recipient_id'] == str(recipient.id) + assert custom_args['campaign_id'] == str(running_campaign.id) + assert custom_args['sent_at'] == str(recipient.sent_at) + assert recipient.status == NotificationCampaignRecipientStatus.AWAITING_DELIVERY + assert recipient.sent_at is not None + assert get_campaign_recipient_stats(running_campaign.id)['sent_count'] == 0 + assert get_campaign_recipient_stats(running_campaign.id)['failed_count'] == 0 + assert get_campaign_recipient_stats(running_campaign.id)['awaiting_count'] == 1 + assert get_campaign_recipient_stats(running_campaign.id)['queued_count'] == 0 + + @mock.patch('osf.email.notification_campaign.execution.get_sendgrid_client') + @mock.patch('osf.email.notification_campaign.execution.send_email_with_send_grid') + def test_send_campaign_batch_sendgrid_bulk_uses_webhook_client(self, mock_sendgrid, mock_get_client, running_campaign): + user = UserFactory() + running_campaign.metadata['sendgrid_bulk'] = True + running_campaign.save(update_fields=['metadata']) + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) + + send_campaign_batch( + context={}, + batch_id=batch_id, + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + use_sendgrid_webhook=True, + ) + + mock_get_client.assert_called_once_with(use_webhook=True) + assert mock_sendgrid.call_args.kwargs['client'] is mock_get_client.return_value + + @mock.patch('osf.email.notification_campaign.execution.get_sendgrid_client') + @mock.patch('osf.email.notification_campaign.execution.send_email') + def test_send_campaign_batch_single_send_reuses_client_for_batch(self, mock_send_email, mock_get_client, running_campaign): + users = [UserFactory(), UserFactory()] + create_campaign_recipients(Q(**{'id__in': [u.id for u in users]}), campaign_id=running_campaign.id) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=2) + + send_campaign_batch( + context={}, + batch_id=batch_id, + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + use_sendgrid_webhook=True, + ) + + mock_get_client.assert_called_once_with(use_webhook=True) + assert mock_send_email.call_count == 2 + assert all(c.kwargs['client'] is mock_get_client.return_value for c in mock_send_email.call_args_list) + assert get_campaign_recipient_stats(running_campaign.id)['awaiting_count'] == 2 def test_build_sendgrid_personalizations_shared_to_list(self): personalizations = _build_sendgrid_personalizations( @@ -936,10 +1063,10 @@ def test_build_sendgrid_personalizations_attaches_custom_args(self): ] @mock.patch( - 'osf.email.notification_campaign.send_email_with_send_grid', + 'osf.email.notification_campaign.execution.send_email_with_send_grid', side_effect=Exception('bulk failed'), ) - @mock.patch('osf.email.notification_campaign.sentry.log_exception') + @mock.patch('osf.email.notification_campaign.execution.sentry.log_exception') def test_send_campaign_batch_sendgrid_bulk_failure(self, mock_sentry, mock_sendgrid, running_campaign): user = UserFactory() running_campaign.metadata['sendgrid_bulk'] = True @@ -959,10 +1086,10 @@ def test_send_campaign_batch_sendgrid_bulk_failure(self, mock_sentry, mock_sendg recipient.refresh_from_db() running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.FAILED - assert running_campaign.failed_count == 1 - assert running_campaign.sent_count == 0 + assert get_campaign_recipient_stats(running_campaign.id)['failed_count'] == 1 + assert get_campaign_recipient_stats(running_campaign.id)['sent_count'] == 0 - @mock.patch('osf.email.notification_campaign.sentry.log_message') + @mock.patch('osf.email.notification_campaign.execution.sentry.log_message') def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, running_campaign): user = UserFactory() running_campaign.started_at = timezone.now() - timedelta(seconds=9) @@ -972,7 +1099,7 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch('osf.email.notification_campaign.send_email'): + with mock.patch('osf.email.notification_campaign.execution.send_email'): send_campaign_batch( context={}, batch_id=batch_id, @@ -986,7 +1113,7 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r assert running_campaign.developer_reminder_sent is True mock_sentry.assert_called_once() - @mock.patch('osf.email.notification_campaign.sentry.log_message') + @mock.patch('osf.email.notification_campaign.execution.sentry.log_message') def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( self, mock_sentry, running_campaign ): @@ -998,7 +1125,7 @@ def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch('osf.email.notification_campaign.send_email'): + with mock.patch('osf.email.notification_campaign.execution.send_email'): send_campaign_batch( context={}, batch_id=batch_id, @@ -1011,7 +1138,7 @@ def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( assert running_campaign.developer_reminder_sent is False mock_sentry.assert_not_called() - @mock.patch('osf.email.notification_campaign.sentry.log_message') + @mock.patch('osf.email.notification_campaign.execution.sentry.log_message') def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, running_campaign): user_a = UserFactory() user_b = UserFactory() @@ -1026,7 +1153,7 @@ def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, runni batch_id_2 = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) mock_sentry.reset_mock() - with mock.patch('osf.email.notification_campaign.send_email'): + with mock.patch('osf.email.notification_campaign.execution.send_email'): send_campaign_batch( context={}, batch_id=batch_id_1, @@ -1103,7 +1230,7 @@ def test_send_campaign_batch_skips_cancelled_campaign(self, running_campaign): recipient.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.QUEUED - @mock.patch('osf.email.notification_campaign.send_email') + @mock.patch('osf.email.notification_campaign.execution.send_email') def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, mock_send_email, running_campaign): user = UserFactory() user.username = 'invalid' @@ -1126,10 +1253,39 @@ def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, m recipient.refresh_from_db() running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SENT - assert running_campaign.sent_count == 1 - assert running_campaign.failed_count == 0 + assert get_campaign_recipient_stats(running_campaign.id)['sent_count'] == 1 + assert get_campaign_recipient_stats(running_campaign.id)['failed_count'] == 0 mock_send_email.assert_called_once() +class TestGetSendgridClient: + + @pytest.fixture(autouse=True) + def clear_client_cache(self): + _sendgrid_client.cache_clear() + yield + _sendgrid_client.cache_clear() + + @mock.patch('website.settings.SENDGRID_API_KEY_WITH_WEBHOOK', 'webhook-key') + @mock.patch('website.settings.SENDGRID_API_KEY', 'default-key') + def test_selects_api_key_by_webhook_flag(self): + with mock.patch('osf.email.SendGridAPIClient') as mock_client_cls: + get_sendgrid_client(use_webhook=True) + get_sendgrid_client(use_webhook=False) + + assert mock_client_cls.call_args_list == [mock.call('webhook-key'), mock.call('default-key')] + + @mock.patch('website.settings.SENDGRID_API_KEY_WITH_WEBHOOK', None) + @mock.patch('website.settings.SENDGRID_API_KEY', 'default-key') + def test_webhook_falls_back_to_default_api_key(self): + with mock.patch('osf.email.SendGridAPIClient') as mock_client_cls: + get_sendgrid_client(use_webhook=True) + + mock_client_cls.assert_called_once_with('default-key') + + @mock.patch('website.settings.SENDGRID_API_KEY', 'default-key') + def test_client_is_cached_per_api_key(self): + assert get_sendgrid_client() is get_sendgrid_client() + class TestNotificationCampaignCancel: @@ -1198,7 +1354,7 @@ def test_process_campaign_retry_skips_stale_run_id(self, campaign): assert campaign.completed_at is None assert campaign.sent_count == 0 - @mock.patch('osf.email.notification_campaign.sentry.log_message') + @mock.patch('osf.email.notification_campaign.execution.sentry.log_message') def test_process_campaign_retry_keeps_cancelled_status_and_syncs_stats(self, mock_sentry, campaign): sent_user = UserFactory() failed_user = UserFactory() @@ -1227,7 +1383,7 @@ def test_process_campaign_retry_keeps_cancelled_status_and_syncs_stats(self, moc assert campaign.completed_at is not None mock_sentry.assert_called_once() - @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign.apply_async') def test_process_campaign_retry_retries_failed_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) @@ -1265,12 +1421,12 @@ def test_process_campaign_retry_marks_partially_completed_after_max_retries(self assert campaign.recipient_count == 1 assert campaign.completed_at is not None - @mock.patch('osf.email.notification_campaign.process_campaign_retry.apply_async') - def test_process_campaign_retry_waits_for_queued_recipients(self, mock_apply_async, campaign): + @mock.patch('osf.email.notification_campaign.execution.process_campaign_retry.apply_async') + def test_process_campaign_retry_waits_for_awaiting_delivery_recipients(self, mock_apply_async, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) NotificationCampaignRecipient.objects.filter(campaign=campaign).update( - status=NotificationCampaignRecipientStatus.QUEUED + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY ) campaign.run_id = uuid.uuid4() campaign.status = NotificationCampaignStatus.RUNNING @@ -1287,15 +1443,15 @@ def test_process_campaign_retry_waits_for_queued_recipients(self, mock_apply_asy assert campaign.sent_count == 0 assert campaign.failed_count == 0 mock_apply_async.assert_called_once_with( - kwargs={'campaign_id': campaign.id, 'run_id': campaign.run_id}, + kwargs={'campaign_id': campaign.id, 'run_id': campaign.run_id, 'use_sendgrid_webhook': False}, countdown=60, ) - def test_process_campaign_retry_times_out_queued_marks_partial_without_retry(self, campaign): + def test_process_campaign_retry_times_out_awaiting_delivery_marks_partial_without_retry(self, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) NotificationCampaignRecipient.objects.filter(campaign=campaign).update( - status=NotificationCampaignRecipientStatus.QUEUED + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY ) campaign.run_id = uuid.uuid4() campaign.retries = 0 @@ -1315,9 +1471,9 @@ def test_process_campaign_retry_times_out_queued_marks_partial_without_retry(sel assert campaign.failed_count == 1 assert campaign.completed_at is not None - @mock.patch('osf.email.notification_campaign.process_campaign_retry.apply_async') - @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') - def test_process_campaign_retry_waits_for_queued_before_retrying_failed( + @mock.patch('osf.email.notification_campaign.execution.process_campaign_retry.apply_async') + @mock.patch('osf.email.notification_campaign.execution.dispatch_campaign.apply_async') + def test_process_campaign_retry_waits_for_awaiting_delivery_before_retrying_failed( self, mock_dispatch_campaign, mock_apply_async, campaign ): queued_user = UserFactory() @@ -1327,7 +1483,7 @@ def test_process_campaign_retry_waits_for_queued_before_retrying_failed( campaign_id=campaign.id, ) NotificationCampaignRecipient.objects.filter(campaign=campaign, user=queued_user).update( - status=NotificationCampaignRecipientStatus.QUEUED + status=NotificationCampaignRecipientStatus.AWAITING_DELIVERY ) NotificationCampaignRecipient.objects.filter(campaign=campaign, user=failed_user).update( status=NotificationCampaignRecipientStatus.FAILED @@ -1355,23 +1511,24 @@ def _event(self, campaign, recipient, event, **extra): 'event': event, 'campaign_id': str(campaign.id), 'campaign_recipient_id': str(recipient.id), - 'run_id': str(campaign.run_id), + 'sent_at': str(recipient.sent_at), } payload.update(extra) return payload - def _queued_recipient(self, campaign, user=None): + def _awaiting_recipient(self, campaign, user=None, sent_at=None): user = user or UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) - recipient.status = NotificationCampaignRecipientStatus.QUEUED - recipient.save(update_fields=['status']) + recipient.status = NotificationCampaignRecipientStatus.AWAITING_DELIVERY + recipient.sent_at = sent_at if sent_at is not None else int(timezone.now().timestamp()) + recipient.save(update_fields=['status', 'sent_at']) return recipient - def test_delivered_marks_queued_sent(self, campaign): + def test_delivered_marks_awaiting_sent(self, campaign): campaign.run_id = uuid.uuid4() campaign.save(update_fields=['run_id']) - recipient = self._queued_recipient(campaign) + recipient = self._awaiting_recipient(campaign) process_sendgrid_campaign_events([ self._event(campaign, recipient, 'delivered'), @@ -1380,24 +1537,91 @@ def test_delivered_marks_queued_sent(self, campaign): recipient.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SENT assert recipient.error_message is None + assert get_campaign_recipient_stats(campaign.id)['sent_count'] == 1 + assert get_campaign_recipient_stats(campaign.id)['awaiting_count'] == 0 + assert get_campaign_recipient_stats(campaign.id)['queued_count'] == 0 - def test_failure_marks_queued_failed(self, campaign): + def test_hard_bounce_marks_awaiting_skipped(self, campaign): campaign.run_id = uuid.uuid4() campaign.save(update_fields=['run_id']) - recipient = self._queued_recipient(campaign) + recipient = self._awaiting_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event( + campaign, recipient, 'bounce', + type='bounce', reason='550 user unknown', + ), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SKIPPED + assert recipient.error_message == '550 user unknown' + assert get_campaign_recipient_stats(campaign.id)['failed_count'] == 1 # SKIPPED counts in failed_count + assert get_campaign_recipient_stats(campaign.id)['awaiting_count'] == 0 + assert get_campaign_recipient_stats(campaign.id)['queued_count'] == 0 + + def test_bounce_without_type_treated_as_hard_skipped(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) process_sendgrid_campaign_events([ self._event(campaign, recipient, 'bounce', reason='mailbox full'), ]) recipient.refresh_from_db() - assert recipient.status == NotificationCampaignRecipientStatus.FAILED + assert recipient.status == NotificationCampaignRecipientStatus.SKIPPED assert recipient.error_message == 'mailbox full' + def test_soft_bounce_blocked_marks_awaiting_failed(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event( + campaign, recipient, 'bounce', + type='blocked', reason='mailbox temporarily unavailable', + ), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.FAILED + assert recipient.error_message == 'mailbox temporarily unavailable' + assert get_campaign_recipient_stats(campaign.id)['failed_count'] == 1 + assert get_campaign_recipient_stats(campaign.id)['awaiting_count'] == 0 + assert get_campaign_recipient_stats(campaign.id)['queued_count'] == 0 + + def test_dropped_marks_awaiting_skipped(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'dropped', reason='Bounced Address'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SKIPPED + assert recipient.error_message == 'Bounced Address' + + def test_deferred_is_ignored(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'deferred', reason='try again later'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.AWAITING_DELIVERY + assert recipient.error_message in (None, '') + def test_delivered_overrides_failed(self, campaign): campaign.run_id = uuid.uuid4() campaign.save(update_fields=['run_id']) - recipient = self._queued_recipient(campaign) + recipient = self._awaiting_recipient(campaign) recipient.status = NotificationCampaignRecipientStatus.FAILED recipient.error_message = 'bounce' recipient.save(update_fields=['status', 'error_message']) @@ -1410,17 +1634,132 @@ def test_delivered_overrides_failed(self, campaign): assert recipient.status == NotificationCampaignRecipientStatus.SENT assert recipient.error_message is None - def test_ignores_stale_run_id(self, campaign): + def test_delivered_overrides_skipped(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) + recipient.status = NotificationCampaignRecipientStatus.SKIPPED + recipient.error_message = 'dropped' + recipient.save(update_fields=['status', 'error_message']) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'delivered'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SENT + assert recipient.error_message is None + + def test_delivered_in_same_batch_wins_over_hard_bounce(self, campaign): campaign.run_id = uuid.uuid4() campaign.save(update_fields=['run_id']) - recipient = self._queued_recipient(campaign) + recipient = self._awaiting_recipient(campaign) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'bounce', type='bounce', reason='hard'), + self._event(campaign, recipient, 'delivered'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SENT + assert recipient.error_message is None + + def test_restart_failed_excludes_skipped_hard_bounces(self, campaign): + soft = UserFactory() + hard = UserFactory() + create_campaign_recipients( + Q(**{'id__in': [soft.id, hard.id]}), + campaign_id=campaign.id, + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=soft).update( + status=NotificationCampaignRecipientStatus.FAILED, + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=hard).update( + status=NotificationCampaignRecipientStatus.SKIPPED, + ) + + batch_id = assign_batch_id_to_recipients( + campaign_id=campaign.id, + batch_size=10, + restart_failed=True, + ) + assert batch_id is not None + retried = list( + NotificationCampaignRecipient.objects.filter(batch_id=batch_id).values_list('user_id', flat=True) + ) + assert retried == [soft.id] + hard_recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=hard) + assert hard_recipient.status == NotificationCampaignRecipientStatus.SKIPPED + assert hard_recipient.batch_id is None + + def test_ignores_stale_sent_at(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) process_sendgrid_campaign_events([{ 'event': 'delivered', 'campaign_id': str(campaign.id), 'campaign_recipient_id': str(recipient.id), - 'run_id': str(uuid.uuid4()), + 'sent_at': str(recipient.sent_at - 60), }]) + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.AWAITING_DELIVERY + + def test_accepts_webhook_after_run_id_rotated(self, campaign): + """restart-stuck rotates run_id; awaiting webhooks must still apply via sent_at.""" + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._awaiting_recipient(campaign) + original_sent_at = recipient.sent_at + + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'delivered'), + ]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.SENT + assert recipient.sent_at == original_sent_at + + def test_ignores_prior_send_after_resent(self, campaign): + """After restart_failed resend, old sent_at webhooks must not apply.""" + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + old_sent_at = int(timezone.now().timestamp()) - 300 + recipient = self._awaiting_recipient(campaign, sent_at=old_sent_at) + new_sent_at = int(timezone.now().timestamp()) + recipient.sent_at = new_sent_at + recipient.save(update_fields=['sent_at']) + + process_sendgrid_campaign_events([{ + 'event': 'bounce', + 'campaign_id': str(campaign.id), + 'campaign_recipient_id': str(recipient.id), + 'sent_at': str(old_sent_at), + 'reason': 'stale', + }]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.AWAITING_DELIVERY + assert recipient.sent_at == new_sent_at + + def test_ignores_queued_recipients_not_yet_submitted(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) + recipient.status = NotificationCampaignRecipientStatus.QUEUED + recipient.sent_at = int(timezone.now().timestamp()) + recipient.save(update_fields=['status', 'sent_at']) + + process_sendgrid_campaign_events([ + self._event(campaign, recipient, 'delivered'), + ]) + recipient.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.QUEUED diff --git a/tests/utils.py b/tests/utils.py index 54616ad86d8..cb76319e466 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -331,7 +331,7 @@ def _fake_send_over_smtp(to_email, notification_type, context=None, email_contex if passthrough: return _real_send_over_smtp(to_email, notification_type, context, email_context, rendered_html) - def _fake_send_with_sendgrid(user, notification_type, context=None, email_context=None, rendered_html=None): + def _fake_send_with_sendgrid(user, notification_type, context=None, email_context=None, rendered_html=None, client=None): captured['emails'].append({ 'protocol': 'sendgrid', 'to': user, @@ -340,7 +340,9 @@ def _fake_send_with_sendgrid(user, notification_type, context=None, email_contex 'email_context': email_context.copy() if isinstance(email_context, dict) else email_context, }) if passthrough: - return _real_send_with_sendgrid(user, notification_type, context, email_context, rendered_html) + return _real_send_with_sendgrid( + user, notification_type, context, email_context, rendered_html=rendered_html, client=client + ) patches.extend([ mock.patch('osf.email.send_email_over_smtp', new=_fake_send_over_smtp), diff --git a/website/settings/defaults.py b/website/settings/defaults.py index 2a33333cfbd..647ff63681e 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -186,7 +186,10 @@ def parent_dir(path): # OR, if using Sendgrid's API # WARNING: If `SENDGRID_WHITELIST_MODE` is True, +# API key of either the primary/default SendGrid user or a different sub-suer, both without Event Webhook enabled SENDGRID_API_KEY = None +# API key of the SendGrid sub-user with Event Webhook enabled +SENDGRID_API_KEY_WITH_WEBHOOK = None # Public verification key from SendGrid Event Webhook (Mail Settings -> Event Webhook -> Signed Event Webhook) SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY = None @@ -216,12 +219,13 @@ def parent_dir(path): DEFAULT_CAMPAIGN_BATCH_SIZE = 1000 DEFAULT_CAMPAIGN_WINDOW_TIME = 28800 # 8 hours DEFAULT_CAMPAIGN_MAX_RETRIES = 3 -DEFAULT_CAMPAIGN_DELIVERY_TIMEOUT = 86400 # 24 hours; mark remaining QUEUED as FAILED after this from started_at -MAX_QUEUED_CAMPAIGN_BATCHES = 100 # Maximum number of queued campaign batches allowed before new batches are rejected. This is to prevent runaway campaigns from overwhelming the system. +DEFAULT_CAMPAIGN_DELIVERY_TIMEOUT = 86400 # 24 hours; mark remaining AWAITING_DELIVERY as FAILED after this from started_at +MAX_QUEUED_CAMPAIGN_BATCHES = 100 # Max QUEUED (Celery send in flight) batches before new batches are held back. AWAITING_DELIVERY (webhook pending) does not count. CAMPAIGN_DISPATCH_INTERVAL = 300 # 5 min (300 sec), minimum time before checking and dispatching new campaign batches. # The following are rough estimates so we can log to sentry those batches and sendgrid quests which run longer than normal ESTIMATED_PER_REQUEST_THRESHOLD = 0.3 # On production, sending one email via SendGrid takes 0.20 ~ 0.50 seconds, set default alert threshold at 0.30s ESTIMATED_BATCH_RUN_TIME_THRESHOLD = 300 # On production, with batch size 1000, we expect each batch to finish within 300s (5m) +CAMPAIGN_SINGLE_SEND_WEBHOOK = False # If True, recipient will be marked as SENT/FAILED/SKIPPED based on SendGrid webhook event, instead of when the request is completed. # Configuration for "We miss you at OSF" email (`NotificationTypeEnum.USER_NO_LOGIN`) # Note: 1) we can gradually increase `MAX_DAILY_NO_LOGIN_EMAILS` to 10000, 100000, etc. or set it to `None` after we diff --git a/website/templates/notification_campaign_log.html.mako b/website/templates/notification_campaign_log.html.mako new file mode 100644 index 00000000000..ceecf81c916 --- /dev/null +++ b/website/templates/notification_campaign_log.html.mako @@ -0,0 +1,22 @@ +<%inherit file="notify_base.mako" /> +<%def name="content()"> + + +

+ Notification campaign [${campaign_name}] +

+ + + + + Notification campaign [${campaign_name}] has reached the following status:
+
+ ${status}
+
+ ${message}
+ Sincerely,
+
+ The OSF Team
+ + + \ No newline at end of file