diff --git a/admin/notifications/forms.py b/admin/notifications/forms.py index 94b571dbb65..ea9417a89c2 100644 --- a/admin/notifications/forms.py +++ b/admin/notifications/forms.py @@ -48,6 +48,21 @@ class NotificationCampaignCreateForm(forms.ModelForm): sendgrid_bulk = forms.BooleanField( required=False, initial=False, + help_text=( + 'SendGrid caps personalizations at about 1000 per request - keep Batch Size ≤ 1000 for bulk requests' + ), + ) + + max_queued_batches = forms.IntegerField( + min_value=1, + initial=settings.MAX_QUEUED_CAMPAIGN_BATCHES, + help_text='Maximum number of queued campaign batches allowed before new batches are rejected.', + ) + + dispatch_interval = forms.IntegerField( + min_value=1, + initial=settings.CAMPAIGN_DISPATCH_INTERVAL, + help_text='The time in seconds between dispatches of campaign batches.', ) class Meta: diff --git a/admin/notifications/urls.py b/admin/notifications/urls.py index c9c03343fd8..de7139bb491 100644 --- a/admin/notifications/urls.py +++ b/admin/notifications/urls.py @@ -17,4 +17,6 @@ re_path(r'notification_campaigns_recipients_preview/$', views.NotificationCampaignsRecipientsPreview.as_view(), name='notification_campaigns_recipients_preview'), re_path(r'notification_campaigns_recipients_list/$', views.NotificationCampaignsRecipientsView.as_view(), name='notification_campaigns_recipients_list'), re_path(r'notification_campaigns_start/(?P\d+)/$', views.StartNotificationCampaign.as_view(), name='notification_campaigns_start'), + re_path(r'notification_campaigns_create_recipients/(?P\d+)/$', views.CreateNotificationCampaignRecipients.as_view(), name='notification_campaigns_create_recipients'), + re_path(r'notification_campaigns_delete/(?P\d+)/$', views.DeleteNotificationCampaign.as_view(), name='notification_campaigns_delete'), ] diff --git a/admin/notifications/views.py b/admin/notifications/views.py index fc27b02417b..bef2ceed94d 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_query +from osf.email.notification_campaign import FILTER_PRESETS, counter_subquery, build_campaign_filter_query from website import settings from urllib.parse import urlencode @@ -431,6 +431,8 @@ def get_context_data(self, *args, **kwargs): if k not in {'filters', 'context', 'execution', 'template'} }, 'allow_restart_stuck': True if timezone.now() - notification_campaign.updated_at > timedelta(minutes=15) else False, + 'delete_allowed': notification_campaign.status != NotificationCampaignStatus.RUNNING, + 'start_allowed': notification_campaign.status == NotificationCampaignStatus.CREATED and metadata.get('recipients_creation_finished', False), } if notification_campaign.status != NotificationCampaignStatus.CREATED: @@ -581,8 +583,11 @@ def form_valid(self, form): 'max_retries': form.cleaned_data['max_retries'], 'activity_threshold': form.cleaned_data['activity_threshold'], 'time_window': form.cleaned_data['time_window'], + 'max_queued_batches': form.cleaned_data['max_queued_batches'], + 'dispatch_interval': form.cleaned_data['dispatch_interval'], }, 'sendgrid_bulk': form.cleaned_data.get('sendgrid_bulk', False), + 'recipients_creation_finished': False, } try: _render_email_html(form.instance.notification_type, form.cleaned_data['context']) @@ -641,10 +646,7 @@ def get_queryset(self): raw_filters = self.request.GET.get('filters', None) if raw_filters: json_filters = json.loads(raw_filters) - if predefined := json_filters.get('predefined'): - query = Q(**FILTER_PRESETS.get(predefined, {})) - else: - query = build_query(json_filters.get('manual')) + query = build_campaign_filter_query(json_filters) qs = OSFUser.objects.filter(query) qs = qs.annotate( @@ -748,3 +750,41 @@ def post(self, request, *args, **kwargs): 'notifications:notification_campaigns_detail', pk=notification_campaign.pk, ) + +class CreateNotificationCampaignRecipients(PermissionRequiredMixin, View): + permission_required = 'osf.change_notificationcampaign' + + def post(self, request, *args, **kwargs): + notification_campaign = get_object_or_404( + NotificationCampaign, + pk=kwargs['pk'], + ) + + if notification_campaign.status != NotificationCampaignStatus.CREATED: + messages.error(request, 'Recipients can only be created for campaigns in CREATED status.') + return redirect( + 'notifications:notification_campaigns_detail', + pk=notification_campaign.pk, + ) + + notification_campaign.create_recipients() + + return redirect( + 'notifications:notification_campaigns_detail', + pk=notification_campaign.pk, + ) + +class DeleteNotificationCampaign(PermissionRequiredMixin, View): + permission_required = 'osf.delete_notificationcampaign' + + def post(self, request, *args, **kwargs): + notification_campaign = get_object_or_404( + NotificationCampaign, + pk=kwargs['pk'], + ) + if notification_campaign.status == NotificationCampaignStatus.RUNNING: + messages.error(request, 'Cannot delete a running campaign.') + return redirect('notifications:notification_campaigns_detail', pk=notification_campaign.pk) + notification_campaign.delete() + messages.success(request, f'Notification campaign {notification_campaign.name} deleted successfully.') + return redirect('notifications:notification_campaigns_list') diff --git a/admin/templates/notifications/notification_campaigns_detail.html b/admin/templates/notifications/notification_campaigns_detail.html index 1c75d1f382a..883b6e3e482 100644 --- a/admin/templates/notifications/notification_campaigns_detail.html +++ b/admin/templates/notifications/notification_campaigns_detail.html @@ -158,7 +158,7 @@

Progress

{% if notification_campaign.developer_reminder_sent %}
Warning! - The campaign exceeded the expected timeframe ({{ metadata.execution.time_window }}s). A reminder was sent. + The high-activity phase exceeded the expected timeframe ({{ metadata.execution.time_window }}s). A reminder was sent.
{% endif %} {% endif %} @@ -169,10 +169,21 @@

Progress

style="display:inline;" > {% csrf_token %} - +
+ {% csrf_token %} + +
Progress Cancel Campaign
+
+ {% csrf_token %} + +
@@ -267,25 +290,19 @@

General

Recipient Filters

- {% if not "predefined" in metadata.filters %} - - {% if not "predefined" in metadata.filters %} - {% if metadata.filters.manual %} - {% include "notifications/campaign_filter_group.html" with group=metadata.filters.manual is_root=True %} - {% else %} -

No filters configured.

- {% endif %} - {% endif %} - - {% elif "predefined" in metadata.filters %} - + {% if "predefined" in metadata.filters %}
Predefined Filter {{ metadata.filters.predefined }}
+ {% endif %} + {% if metadata.filters.manual %} + {% include "notifications/campaign_filter_group.html" with group=metadata.filters.manual is_root=True %} + {% elif not "predefined" in metadata.filters %} +

No filters configured.

{% endif %} Additional Metadata }); } + const deleteForm = document.getElementById("delete-campaign-form"); + + if (deleteForm) { + deleteForm.addEventListener("submit", function (e) { + confirmCampaign(this, e); + }); + } + {% endblock %} diff --git a/admin/templates/notifications/notification_campaigns_list.html b/admin/templates/notifications/notification_campaigns_list.html index 16b609cefac..486fa2d5814 100644 --- a/admin/templates/notifications/notification_campaigns_list.html +++ b/admin/templates/notifications/notification_campaigns_list.html @@ -5,6 +5,16 @@ List of Notification Campaigns {% endblock title %} {% block content %} +
+ {% if messages %} +
    + {% for message in messages %} + {{ message }} + + {% endfor %} +
+ {% endif %} +

List of Notification Campaigns

{% if active_campaign %}

Active Campaign

diff --git a/admin/templates/notifications/notification_campaing_create.html b/admin/templates/notifications/notification_campaing_create.html index c4e09df6a9b..5ca5e46f03b 100644 --- a/admin/templates/notifications/notification_campaing_create.html +++ b/admin/templates/notifications/notification_campaing_create.html @@ -163,6 +163,21 @@

Recipient Filters

+
+ +

+ When enabled, users who have not confirmed their accounts + are excluded from the recipient list. +

+
+ Execution name="sendgrid_bulk" value="{{ form.sendgrid_bulk.initial }}" > +

+ {{ form.sendgrid_bulk.help_text }} +

+ + + + Max Queued Batches + + + + + + Dispatch Interval + + +

+ {{ form.dispatch_interval.help_text }} +

@@ -406,15 +449,37 @@

Execution

} function buildFilters() { + let filters; if (filterMode.value === "predefined") { - return { + filters = { "predefined": document.getElementById("filter-row").value }; + } else { + const root = builder.querySelector(":scope > .group"); + filters = {"manual": serializeGroup(root)}; + } + + if (document.getElementById("exclude-unconfirmed").checked) { + const confirmed = { + field: "date_confirmed", + lookup: "isnull", + value: false, + }; + if (filters.manual) { + filters.manual = { + operator: "AND", + children: [filters.manual, confirmed], + }; + } else { + filters.manual = { + operator: "AND", + children: [confirmed], + }; + } } - const root = builder.querySelector(":scope > .group"); - return {"manual": serializeGroup(root)} + return filters; } function updateFiltersInput() { diff --git a/admin_tests/notifications/test_campaigns.py b/admin_tests/notifications/test_campaigns.py index a213975163c..771081e5b24 100644 --- a/admin_tests/notifications/test_campaigns.py +++ b/admin_tests/notifications/test_campaigns.py @@ -16,6 +16,7 @@ NotificationCampaignDetail, NotificationCampaignsList, StartNotificationCampaign, + DeleteNotificationCampaign, ) from admin_tests.utilities import setup_form_view from osf.models import NotificationType @@ -60,6 +61,8 @@ def _valid_form_data(notification_type, **overrides): 'activity_threshold': settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD, 'sendgrid_bulk': False, 'time_window': 8 * 60 * 60, + 'max_queued_batches': settings.MAX_QUEUED_CAMPAIGN_BATCHES, + 'dispatch_interval': settings.CAMPAIGN_DISPATCH_INTERVAL, } data.update(overrides) return data @@ -260,7 +263,9 @@ def test_form_valid_persists_execution_metadata(self): max_retries=4, activity_threshold=77, sendgrid_bulk=True, - time_window=28800 + time_window=28800, + max_queued_batches=10, + dispatch_interval=60, ), ) request.user = self.user @@ -282,7 +287,9 @@ def test_form_valid_persists_execution_metadata(self): 'batch_size': 25, 'max_retries': 4, 'activity_threshold': 77, - 'time_window': 28800 + 'time_window': 28800, + 'max_queued_batches': 10, + 'dispatch_interval': 60, } assert campaign.metadata['sendgrid_bulk'] is True assert campaign.metadata['filters'] == {'predefined': 'active'} @@ -419,3 +426,33 @@ def test_start_rejects_when_another_campaign_is_running(self): mock_start.assert_not_called() self.campaign.refresh_from_db() assert self.campaign.status == NotificationCampaignStatus.CREATED + +class TestNotificationCampaignDeleteView(AdminTestCase): + + def setUp(self): + super().setUp() + self.user = AuthUserFactory() + self.notification_type, _ = NotificationType.objects.get_or_create( + name='blank', + defaults={'subject': 'Test', 'template': 'Hello'}, + ) + self.campaign = NotificationCampaign.objects.create( + name='Campaign', + notification_type=self.notification_type, + metadata={'filters': {}, 'context': {}, 'execution': {}}, + ) + + def test_delete_requires_change_notificationcampaign_permission(self): + request = RequestFactory().post( + reverse('notifications:notification_campaigns_delete', kwargs={'pk': self.campaign.pk}) + ) + request.user = self.user + patch_messages(request) + + with self.assertRaises(PermissionDenied): + DeleteNotificationCampaign.as_view()(request, pk=self.campaign.pk) + + grant_permission(self.user, 'delete_notificationcampaign') + response = DeleteNotificationCampaign.as_view()(request, pk=self.campaign.pk) + assert response.status_code == 302 + assert not NotificationCampaign.objects.filter(pk=self.campaign.pk).exists() diff --git a/api/base/urls.py b/api/base/urls.py index 63984efd287..8572f9ae6b0 100644 --- a/api/base/urls.py +++ b/api/base/urls.py @@ -21,6 +21,7 @@ re_path(r'^ia/', include('api.ia.urls', namespace='ia')), re_path(r'^banners/', include('api.banners.urls', namespace='banners')), re_path(r'^crossref/', include('api.crossref.urls', namespace='crossref')), + re_path(r'^sendgrid/', include('api.sendgrid.urls', namespace='sendgrid')), re_path(r'^chronos/', include('api.chronos.urls', namespace='chronos')), re_path(r'^cedar_metadata_templates/', include('api.cedar_metadata_templates.urls', namespace='cedar-metadata-templates')), re_path(r'^cedar_metadata_records/', include('api.cedar_metadata_records.urls', namespace='cedar-metadata-records')), diff --git a/api/sendgrid/__init__.py b/api/sendgrid/__init__.py new file mode 100644 index 00000000000..e69de29bb2d diff --git a/api/sendgrid/permissions.py b/api/sendgrid/permissions.py new file mode 100644 index 00000000000..067eaa3168a --- /dev/null +++ b/api/sendgrid/permissions.py @@ -0,0 +1,38 @@ +from rest_framework import permissions +from rest_framework import exceptions +from sendgrid.helpers.eventwebhook import EventWebhook, EventWebhookHeader + +from framework import sentry +from website import settings + + +class RequestComesFromSendGrid(permissions.BasePermission): + """Verify that the request comes from SendGrid via signed Event Webhook. + + Uses ECDSA signature verification against the raw request body as documented at: + https://www.twilio.com/docs/sendgrid/for-developers/tracking-events/getting-started-event-webhook-security-features + """ + + def has_permission(self, request, view): + if request.method != 'POST': + raise exceptions.MethodNotAllowed(method=request.method) + + public_key = settings.SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY + if not public_key: + sentry.log_message('SendGrid Event Webhook public key is not configured') + return False + + signature = request.headers.get(EventWebhookHeader.SIGNATURE) + timestamp = request.headers.get(EventWebhookHeader.TIMESTAMP) + if not signature or not timestamp: + error_message = 'SendGrid Event Webhook signature headers required' + sentry.log_message(error_message) + raise exceptions.ParseError(error_message) + + # Must verify the raw body; re-serializing parsed JSON breaks the signature. + payload = request.body.decode('utf-8') + event_webhook = EventWebhook(public_key) + if not event_webhook.verify_signature(payload, signature, timestamp): + raise exceptions.ParseError('Invalid SendGrid Event Webhook signature') + + return True diff --git a/api/sendgrid/urls.py b/api/sendgrid/urls.py new file mode 100644 index 00000000000..64ce9714bf3 --- /dev/null +++ b/api/sendgrid/urls.py @@ -0,0 +1,9 @@ +from django.urls import re_path + +from api.sendgrid import views + +app_name = 'osf' + +urlpatterns = [ + re_path(r'^events/$', views.SendGridEventWebhook.as_view(), name=views.SendGridEventWebhook.view_name), +] diff --git a/api/sendgrid/views.py b/api/sendgrid/views.py new file mode 100644 index 00000000000..45f14c235f1 --- /dev/null +++ b/api/sendgrid/views.py @@ -0,0 +1,87 @@ +import json +import logging + +from django.http import HttpResponse +from django.views.decorators.csrf import csrf_exempt +from rest_framework.views import APIView + +from api.sendgrid.permissions import RequestComesFromSendGrid +from osf.email.notification_campaign import ( + CAMPAIGN_CUSTOM_ARG_KEYS, + process_sendgrid_campaign_events, +) + +logger = logging.getLogger(__name__) + + +def _is_campaign_related_event(event): + """SendGrid flattens personalization ``custom_args`` onto each event object.""" + if not isinstance(event, dict): + return False + return all(event.get(key) for key in CAMPAIGN_CUSTOM_ARG_KEYS) + + +class SendGridEventWebhook(APIView): + """Receive SendGrid Event Webhook POSTs and enqueue campaign event handling. + + Mounted under the ``_/`` namespace (no user auth). Non-campaign events are + ignored; campaign-tagged events are processed asynchronously. + + Local testing + ------------- + Requests must be ECDSA-signed by SendGrid (see ``RequestComesFromSendGrid``), + so plain ``curl`` posts are rejected unless you forge a valid signature or + disable the signature check. Use a public tunnel and a real SendGrid Event Webhook + instead: + + 1. Run the API (``localhost:8000``) and a Celery worker (events are enqueued + to ``email.process_sendgrid_campaign_events``). + 2. Expose the API, e.g. ``ngrok http 8000``. Add ``*.ngrok-free.dev`` in + ``ALLOWED_HOSTS`` in the settings. See https://ngrok.com/docs/share-localhost/overview + for more details. + 3. In SendGrid → Mail Settings → Event Webhook: + - POST URL: ``https:///_/sendgrid/events/`` + - 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``. + 4. Send a notification campaign email (personalization ``custom_args`` must + include ``campaign_id``, ``campaign_recipient_id``, and ``run_id``). + SendGrid's "Test Your Integration" payload lacks those keys, so this view + accepts it with 200 but does not update campaign recipients. + """ + + view_name = 'sendgrid_event_webhook' + view_category = 'sendgrid' + + authentication_classes = () + permission_classes = ( + RequestComesFromSendGrid, + ) + + @csrf_exempt + def dispatch(self, request, *args, **kwargs): + return super().dispatch(request, *args, **kwargs) + + def get_serializer_class(self): + return None + + def post(self, request): + try: + events = json.loads(request.body) + except (TypeError, ValueError, json.JSONDecodeError): + return HttpResponse('Invalid JSON', status=400) + + if isinstance(events, dict): + events = [events] + if not isinstance(events, list): + return HttpResponse('Expected a JSON array of events', status=400) + + # Process only campaign-related events + # TODO: add handling for non-campaign events here if required + campaign_events = [event for event in events if _is_campaign_related_event(event)] + + if campaign_events: + process_sendgrid_campaign_events.delay(campaign_events) + logger.info(f'Enqueued {len(campaign_events)} campaign-related SendGrid event(s) of {len(events)} received') + + return HttpResponse('Events received', status=200) diff --git a/api_tests/notifications/test_campaign_recipient_cleanup.py b/api_tests/notifications/test_campaign_recipient_cleanup.py new file mode 100644 index 00000000000..963540d0ae4 --- /dev/null +++ b/api_tests/notifications/test_campaign_recipient_cleanup.py @@ -0,0 +1,115 @@ +from datetime import timedelta + +import pytest +from django.utils import timezone + +from website import settings +from osf.models import ( + NotificationCampaign, + NotificationCampaignRecipient, + NotificationTypeEnum, +) +from osf_tests.factories import AuthUserFactory +from notifications.tasks import delete_notification_campaign_recipients + + +@pytest.mark.django_db +class TestDeleteNotificationCampaignRecipients: + + @pytest.fixture(autouse=True) + def setup(self): + self.notification_type = NotificationTypeEnum.BLANK.instance + + def test_deletes_recipients_for_old_campaigns(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Old campaign', + notification_type=self.notification_type, + completed_at=cutoff - timedelta(seconds=1), + ) + + recipients = NotificationCampaignRecipient.objects.bulk_create([ + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ), + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ), + ]) + + delete_notification_campaign_recipients() + + assert not NotificationCampaignRecipient.objects.filter( + id__in=[recipient.id for recipient in recipients], + ).exists() + + def test_does_not_delete_recipients_for_recent_campaigns(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Recent campaign', + notification_type=self.notification_type, + completed_at=cutoff + timedelta(seconds=1), + ) + + recipient = NotificationCampaignRecipient.objects.create( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + + delete_notification_campaign_recipients() + + assert NotificationCampaignRecipient.objects.filter( + id=recipient.id, + ).exists() + + def test_does_not_delete_recipients_for_incomplete_campaigns(self): + campaign = NotificationCampaign.objects.create( + name='Incomplete campaign', + notification_type=self.notification_type, + completed_at=None, + ) + + recipient = NotificationCampaignRecipient.objects.create( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + + delete_notification_campaign_recipients() + + assert NotificationCampaignRecipient.objects.filter( + id=recipient.id, + ).exists() + + def test_deletes_recipients_in_batches(self): + now = timezone.now() + cutoff = now - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaign = NotificationCampaign.objects.create( + name='Large campaign', + notification_type=self.notification_type, + completed_at=cutoff - timedelta(seconds=1), + ) + + NotificationCampaignRecipient.objects.bulk_create([ + NotificationCampaignRecipient( + campaign=campaign, + user_id=AuthUserFactory().id, + ) + for _ in range(10) + ]) + + assert NotificationCampaignRecipient.objects.filter( + campaign=campaign, + ).count() == 10 + + delete_notification_campaign_recipients() + + assert not NotificationCampaignRecipient.objects.filter( + campaign=campaign, + ).exists() diff --git a/notifications/tasks.py b/notifications/tasks.py index 9775a44fc7a..3f58770ef79 100644 --- a/notifications/tasks.py +++ b/notifications/tasks.py @@ -11,7 +11,7 @@ from framework.postcommit_tasks.handlers import run_postcommit from osf.models import OSFUser, Notification, NotificationTypeEnum, EmailTask, RegistrationProvider, \ - CollectionProvider, AbstractProvider + CollectionProvider, AbstractProvider, NotificationCampaign, NotificationCampaignRecipient from framework.sentry import log_message from osf.registrations.utils import get_registration_provider_submissions_url from osf.utils.permissions import ADMIN @@ -544,3 +544,37 @@ def delete_batch( logger.info(f'Deleted {deleted} rows from {model_name}') delete_batch.delay(app_label, model_name, filters, order_field, batch_size) + + +@celery_app.task( + bind=True, + name='notifications.tasks.delete_notification_campaign_recipients', +) +def delete_notification_campaign_recipients(self): + """Delete recipients for old notification campaigns.""" + + cutoff = timezone.now() - settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE + + campaigns = NotificationCampaign.objects.filter( + completed_at__lt=cutoff, + ).values_list('id', flat=True) + + for campaign_id in campaigns.iterator(): + total_deleted = 0 + while True: + recipient_ids = list( + NotificationCampaignRecipient.objects + .filter(campaign_id=campaign_id) + .values_list('id', flat=True)[:settings.NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE] + ) + + if not recipient_ids: + break + + deleted, _ = NotificationCampaignRecipient.objects.filter( + id__in=recipient_ids, + ).delete() + + total_deleted += deleted + + logger.info(f'Deleted {total_deleted} recipients for campaign {campaign_id}') diff --git a/osf/email/__init__.py b/osf/email/__init__.py index 1cf39af809b..54be37c898d 100644 --- a/osf/email/__init__.py +++ b/osf/email/__init__.py @@ -199,7 +199,7 @@ def _safe_categories(cats): out.append(c) return out[:10] -def send_email_over_smtp(to_email, notification_type, context, email_context): +def send_email_over_smtp(to_email, notification_type, context, email_context, rendered_html=None): if waffle.switch_is_active(features.ENABLE_MAILHOG): host = settings.MAILHOG_HOST port = settings.MAILHOG_PORT @@ -212,7 +212,7 @@ def send_email_over_smtp(to_email, notification_type, context, email_context): raise NotImplementedError('MAIL_SERVER or MAIL_PORT is not set') subject = None if not notification_type.subject else notification_type.subject.format(**context) - body_html = _render_email_html(notification_type, context) + body_html = rendered_html or _render_email_html(notification_type, context) or '

(no content)

' email = EmailMessage( subject=subject, @@ -238,7 +238,54 @@ def send_email_over_smtp(to_email, notification_type, context, email_context): email.attach(attachment_name, attachment_content) email.send() -def send_email_with_send_grid(to_addr, notification_type, context, email_context=None): +def _email_objects(addrs): + if not addrs: + return None + if isinstance(addrs, str): + addrs = [addrs] + return [{'email': a} for a in addrs] + + +def _build_sendgrid_personalizations(to_list, email_context=None, is_multiple=False): + """Build SendGrid personalizations. + + When ``is_multiple`` is True, each address gets its own personalization (separate + delivery; recipients do not see each other). + + Optional ``email_context`` keys: + - ``custom_args``: dict applied to every personalization + - ``custom_args_list``: list of dicts parallel to ``to_list`` (used when + ``is_multiple`` is True; each entry is attached to that recipient only) + """ + email_context = email_context or {} + cc = _email_objects(email_context.get('cc_addr')) + bcc = _email_objects(email_context.get('bcc_addr')) + shared_custom_args = email_context.get('custom_args') + custom_args_list = email_context.get('custom_args_list') or [] + + def personalization(recipients, custom_args=None): + item = {'to': [{'email': a} for a in recipients]} + if cc: + item['cc'] = cc + if bcc: + item['bcc'] = bcc + if custom_args: + item['custom_args'] = {str(k): str(v) for k, v in custom_args.items()} + return item + + if not is_multiple: + return [personalization(to_list, shared_custom_args)] + + return [ + personalization( + [addr], + custom_args_list[i] if i < len(custom_args_list) else shared_custom_args, + ) + for i, addr in enumerate(to_list) + ] + + +def send_email_with_send_grid(to_addr, notification_type, context, email_context=None, *, is_multiple=False, rendered_html=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] @@ -251,23 +298,17 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context logging.error('SendGrid: missing SENDGRID_FROM_EMAIL/FROM_EMAIL') return False - html = _render_email_html(notification_type, context) or '

(no content)

' + html = rendered_html or _render_email_html(notification_type, context) or '

(no content)

' subject_tpl = getattr(notification_type, 'subject', None) subject = subject_tpl.format(**context) if subject_tpl else f'Notification: {getattr(notification_type, "name", "OSF")}' - personalization = {'to': [{'email': addr} for addr in to_list]} - cc_addr = email_context.get('cc_addr') - if cc_addr: - personalization['cc'] = [{'email': a} for a in ([cc_addr] if isinstance(cc_addr, str) else cc_addr)] - bcc_addr = email_context.get('bcc_addr') - if bcc_addr: - personalization['bcc'] = [{'email': a} for a in ([bcc_addr] if isinstance(bcc_addr, str) else bcc_addr)] - payload = { 'from': {'email': from_email}, 'subject': subject, - 'personalizations': [personalization], + 'personalizations': _build_sendgrid_personalizations( + to_list, email_context=email_context, is_multiple=is_multiple + ), 'content': [ {'type': 'text/html', 'value': html}, ], @@ -338,3 +379,31 @@ def send_email_with_send_grid(to_addr, notification_type, context, email_context else: 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): + """ + Send an email using either SMTP or SendGrid based on settings and feature flags. + """ + if waffle.switch_is_active(features.ENABLE_MAILHOG): + send_email_over_smtp( + recipient_address, + notification_type, + event_context, + email_context, + rendered_html=rendered_html, + ) + + if not settings.LOCAL_MODE: + send_email_with_send_grid( + recipient_address, + notification_type, + event_context, + email_context, + rendered_html=rendered_html, + ) + + if settings.LOCAL_MODE and not waffle.switch_is_active(features.ENABLE_MAILHOG): + logging.warning( + 'Both ENABLE_MAILHOG and LOCAL_MODE are disabled. Emails will not be sent to MailHog or real email addresses. ' + 'Turn on ENABLE_MAILHOG to send emails to MailHog for testing, or turn on LOCAL_MODE to send emails with SendGrid.' + ) diff --git a/osf/email/notification_campaign.py b/osf/email/notification_campaign.py index a6db5fc9ca0..3ce4b2bd1d0 100644 --- a/osf/email/notification_campaign.py +++ b/osf/email/notification_campaign.py @@ -1,16 +1,15 @@ import logging - +import uuid from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter, Email from osf.models.spam import SpamStatus from django.db import transaction -from django.db.models import OuterRef, Subquery, Case, When, CharField, Count, Q +from django.db.models import OuterRef, Subquery, Case, When, Value, CharField, Count, Q, BooleanField, TextField from django.db.models.functions import Coalesce from framework.celery_tasks import app as celery_app -from celery import group, chain 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 +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 @@ -24,6 +23,11 @@ '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')) @@ -38,6 +42,39 @@ ) +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. @@ -54,6 +91,9 @@ def build_query(node): 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 @@ -79,7 +119,26 @@ def build_query(node): return query -def create_campaign_recipients(filters, campaign_id): + +def build_campaign_filter_query(filters): + """AND together optional predefined and manual 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) + 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) @@ -89,6 +148,7 @@ def create_campaign_recipients(filters, campaign_id): 'activity_score', ) ) + processed_records = 0 for rows in batched(qs.iterator(chunk_size=BULK_CREATE_SIZE), BULK_CREATE_SIZE): NotificationCampaignRecipient.objects.bulk_create( @@ -100,72 +160,24 @@ def create_campaign_recipients(filters, campaign_id): ) for user_id, activity_score in rows ], - ignore_conflicts=True + 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() -def get_campaign_recipient_batches( - campaign_id, - batch_size, - restart_failed=False, - min_activity=None, - max_activity=None, - spam=None, -): - qs = NotificationCampaignRecipient.objects.filter( - campaign_id=campaign_id, - ) - - if restart_failed: - qs = qs.filter(status=NotificationCampaignRecipientStatus.FAILED) - else: - qs = qs.filter(status=NotificationCampaignRecipientStatus.PENDING) - - # Minimum and maximum activity are mutually exclusive and use the same threshold. - if min_activity is not None: - qs = qs.filter(activity_score__gte=min_activity) - - if max_activity is not None: - qs = qs.filter(activity_score__lt=max_activity) - - if spam is True: - qs = qs.filter(user__spam_status=SpamStatus.SPAM) - elif spam is False: - qs = qs.exclude(user__spam_status=SpamStatus.SPAM) - - yield from batched( - qs.values_list('id', flat=True).iterator(chunk_size=batch_size), - batch_size, - ) - -def build_campaign_group( - campaign_id, - batch_size, - restart_failed=False, - min_activity=None, - max_activity=None, - spam=None, - **send_kwargs, -): - tasks = [] - - for batch in get_campaign_recipient_batches( - campaign_id=campaign_id, - batch_size=batch_size, - restart_failed=restart_failed, - min_activity=min_activity, - max_activity=max_activity, - spam=spam, - ): - tasks.append( - send_campaign_batch.si( - recipients_ids=batch, - campaign_id=campaign_id, - **send_kwargs, - ) - ) - - return group(tasks) + 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): @@ -188,144 +200,342 @@ def get_campaign_recipient_stats(campaign_id): ), ) -@celery_app.task(name='email.process_campaign_retry') -def process_campaign_retry(*args, **kwargs): - campaign_id = kwargs.get('campaign_id') - campaign = NotificationCampaign.objects.get(id=campaign_id) - if kwargs.get('run_id') != campaign.run_id: + +@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 - final_status = NotificationCampaignStatus.COMPLETED - - if campaign.status != NotificationCampaignStatus.CANCELLED: - failed_recipients = NotificationCampaignRecipient.objects.filter(campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED) - max_retries = campaign.metadata.get('execution', {}).get('max_retries', settings.DEFAULT_CAMPAIGN_MAX_RETRIES) - batch_size = campaign.metadata.get('execution', {}).get('batch_size', settings.DEFAULT_CAMPAIGN_BATCH_SIZE) - failed_recipients_count = failed_recipients.count() - if failed_recipients_count: - if campaign.retries < max_retries: - message = (f'[Notification Campaign #{campaign_id}] WARNING: ' - f'Retrying {failed_recipients_count} failed recipients, ' - f'previous retry attempts: {campaign.retries}/{max_retries}') - logger.info(message) - sentry.log_message(message) - campaign.retries += 1 - campaign.save(update_fields=['retries']) - retry_group = build_campaign_group( - batch_size=batch_size, - campaign_id=campaign_id, - restart_failed=True, - notification_type_name=campaign.notification_type.name, - context=campaign.metadata.get('context', {}), - run_id=campaign.run_id, - ) - chain( - retry_group, - process_campaign_retry.si(campaign_id=campaign_id, run_id=campaign.run_id), - ).apply_async() - 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 - final_status = NotificationCampaignStatus.PARTIALLY_COMPLETED - else: + 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(), + ), + ) + + +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.process_campaign_retry') +def process_campaign_retry(self, campaign_id, run_id): + campaign = self.get_campaign(campaign_id, run_id, abort_if_cancelled=False) + if campaign is None: + return + + campaign.refresh_from_db() + execution = campaign.metadata.get('execution', {}) + + queued_qs = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + status=NotificationCampaignRecipientStatus.QUEUED, + ) + if queued_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}, + countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), + ) + return + + timed_out = queued_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'({delivery_timeout}s) for campaign {campaign.name}.' + ) + logger.warning(message) + sentry.log_message(message) + + # Do not retry timed-out deliveries; close the run as partially completed. + 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 - # Refresh in case the campaign was cancelled while we were running. - campaign.refresh_from_db(fields=['status', 'completed_at']) - - # Sync statistics regardless of status. - 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'] + failed_recipients_count = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + status=NotificationCampaignRecipientStatus.FAILED, + ).count() + max_retries = execution.get('max_retries', settings.DEFAULT_CAMPAIGN_MAX_RETRIES) - if campaign.completed_at is None: - campaign.completed_at = timezone.now() + if failed_recipients_count: + if campaign.retries < max_retries: + message = (f'[Notification Campaign #{campaign_id}] WARNING: ' + f'Retrying {failed_recipients_count} failed recipients, ' + f'previous retry attempts: {campaign.retries}/{max_retries}') + logger.info(message) + sentry.log_message(message) + campaign.retries += 1 + campaign.save() - # Don't overwrite CANCELLED. - if campaign.status != NotificationCampaignStatus.CANCELLED: - campaign.status = final_status + dispatch_campaign.apply_async( + args=[campaign_id, campaign.run_id], + kwargs={ + 'restart_failed': True, + }, + ) + return - campaign.save() + final_status = NotificationCampaignStatus.PARTIALLY_COMPLETED + else: + final_status = NotificationCampaignStatus.COMPLETED + self.finish_campaign(campaign, final_status) -@celery_app.task(name='email.start_notification_campaign') -def start_notification_campaign(campaign_id, restart_failed=False, restart_stuck=False): - campaign = NotificationCampaign.objects.get(id=campaign_id) - filters = campaign.metadata.get('filters', {}) - context = campaign.metadata.get('context', {}) +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.start_notification_campaign') +def start_notification_campaign(self, campaign_id, restart_failed=False, restart_stuck=False): + campaign = self.get_campaign(campaign_id) + if campaign is None: + return notification_type_name = campaign.notification_type.name if hasattr(NotificationTypeEnum, notification_type_name): del getattr(NotificationTypeEnum, notification_type_name).instance - if predefined_filter_name := filters.get('predefined'): - filters = Q(**FILTER_PRESETS.get(predefined_filter_name, {})) + if restart_stuck: + NotificationCampaignRecipient.objects.filter( + campaign_id=campaign_id, + status=NotificationCampaignRecipientStatus.QUEUED + ).update(status=NotificationCampaignRecipientStatus.PENDING, batch_id=None) + + dispatch_campaign.apply_async( + args=[campaign_id, campaign.run_id], + kwargs={ + 'restart_failed': restart_failed, + }, + ) + + +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 = build_query(filters.get('manual', [])) + filters &= Q( + status=NotificationCampaignRecipientStatus.PENDING, + batch_id__isnull=True, + ) - if not restart_failed and not restart_stuck: - recipients_creation_started_at = timezone.now() - create_campaign_recipients(filters=filters, campaign_id=campaign_id) - campaign.recipient_count = NotificationCampaignRecipient.objects.filter(campaign_id=campaign_id).count() - campaign.save() - recipients_creation_finished_at = timezone.now() - recipients_creation_run_time = (recipients_creation_finished_at - recipients_creation_started_at).total_seconds() - 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) + 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): + campaign = self.get_campaign(campaign_id, run_id) + if campaign is None: + return + + if campaign.status != NotificationCampaignStatus.RUNNING: + message = f'[Notification Campaign #{campaign_id}] ERROR: Campaign {campaign.name} is not in RUNNING status.' + logger.error(message) sentry.log_message(message) + return + + queued_batches_count = NotificationCampaignRecipient.objects.filter( + campaign=campaign, + batch_id__isnull=False, + status=NotificationCampaignRecipientStatus.QUEUED + ).order_by().values('batch_id').distinct().count() execution = campaign.metadata.get('execution', {}) batch_size = execution.get('batch_size', settings.DEFAULT_CAMPAIGN_BATCH_SIZE) activity_threshold = execution.get('activity_threshold', settings.DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD) - batch_task_kwargs = dict( - batch_size=batch_size, - campaign_id=campaign_id, - restart_failed=restart_failed, - notification_type_name=notification_type_name, - context=context, - run_id=campaign.run_id + max_queued_batches = execution.get('max_queued_batches', settings.MAX_QUEUED_CAMPAIGN_BATCHES) + notification_type_name = campaign.notification_type.name + to_queue = max_queued_batches - queued_batches_count + + priority_groups = ( + { + 'min_activity': activity_threshold, + 'spam': False, + 'developer_reminder': True, + }, + { + 'max_activity': activity_threshold, + 'spam': False, + 'developer_reminder': False, + }, + { + 'spam': True, + 'developer_reminder': False, + }, ) + total_new_queued_batches = 0 + new_queued_batches = 0 + no_more_recipients = False + for priority_group in priority_groups: + no_more_recipients = False + developer_reminder = priority_group.pop('developer_reminder', False) + for _ in range(to_queue): + batch_id = assign_batch_id_to_recipients( + campaign_id, + batch_size=batch_size, + restart_failed=restart_failed, + **priority_group, + ) - workflow = [] - high_activity_tasks = build_campaign_group( - min_activity=activity_threshold, - spam=False, - **batch_task_kwargs - ) - if high_activity_tasks: - workflow.append(high_activity_tasks) + if batch_id is None: + no_more_recipients = True + break - low_activity_tasks = build_campaign_group( - max_activity=activity_threshold, - spam=False, - **batch_task_kwargs - ) - if low_activity_tasks: - workflow.append(low_activity_tasks) + send_campaign_batch.delay( + context=campaign.metadata.get('context', {}), + batch_id=batch_id, + campaign_id=campaign_id, + notification_type_name=notification_type_name, + run_id=run_id, + developer_reminder=developer_reminder, + ) + new_queued_batches += 1 - spam_users_tasks = build_campaign_group( - spam=True, - **batch_task_kwargs - ) - if spam_users_tasks: - workflow.append(spam_users_tasks) + to_queue -= new_queued_batches + total_new_queued_batches += new_queued_batches + new_queued_batches = 0 + if to_queue <= 0: + break + + logger.info(f'[Notification Campaign #{campaign_id}] INFO: Dispatched {total_new_queued_batches} new batches for campaign {campaign.name}.') - chain(*workflow, process_campaign_retry.si(campaign_id=campaign_id, run_id=campaign.run_id)).apply_async() + if no_more_recipients: + process_campaign_retry.delay( + campaign_id=campaign_id, run_id=campaign.run_id + ) + else: + self.apply_async( + args=[campaign_id, run_id], + kwargs={ + 'restart_failed': restart_failed, + }, + countdown=execution.get('dispatch_interval', settings.CAMPAIGN_DISPATCH_INTERVAL), + ) -@celery_app.task(name='email.send_campaign_batch', ignore_result=False) -def send_campaign_batch(context, recipients_ids, notification_type_name='blank', campaign_id=None, run_id=None): - campaign = NotificationCampaign.objects.get(id=campaign_id) - if campaign.run_id != run_id: - return - if campaign.status == NotificationCampaignStatus.CANCELLED: - logger.warning(f"Campaign {campaign_id} was cancelled") +@celery_app.task(bind=True, base=NotificationCampaignTask, name='email.send_campaign_batch', ignore_result=False) +def send_campaign_batch( + self, + context, + batch_id=None, + notification_type_name='blank', + campaign_id=None, + run_id=None, + developer_reminder=False, +): + campaign = self.get_campaign(campaign_id, run_id) + if campaign is None: return + + recipients_qs = NotificationCampaignRecipient.objects.filter(batch_id=batch_id).select_related('user') + batch_started_at = timezone.now() if hasattr(NotificationTypeEnum, notification_type_name): notification_type = getattr(NotificationTypeEnum, notification_type_name).instance @@ -337,22 +547,31 @@ def send_campaign_batch(context, recipients_ids, notification_type_name='blank', if campaign.status != NotificationCampaignStatus.FAILED: campaign.status = NotificationCampaignStatus.FAILED campaign.save() + recipients_qs.update( + status=NotificationCampaignRecipientStatus.FAILED, + error_message='Notification type not found', + ) message = f'[Notification Campaign #{campaign_id}] ERROR: Batch failed due to none notification_type (template)' logger.error(message) sentry.log_message(message) return - execution_time_window = campaign.metadata.get('execution', {}).get('time_window', settings.DEFAULT_CAMPAIGN_WINDOW_TIME) - if campaign.started_at < timezone.now() - timedelta(seconds=execution_time_window): - if not campaign.developer_reminder_sent: - message = (f'[Notification Campaign #{campaign_id}] WARNING: ' - f'Exceeded execution time window ({execution_time_window}s): name={campaign.name}.') - logger.warning(message) - sentry.log_message(message) - campaign.developer_reminder_sent = True - campaign.save() + if developer_reminder: + execution_time_window = campaign.metadata.get('execution', {}).get('time_window', settings.DEFAULT_CAMPAIGN_WINDOW_TIME) + if campaign.started_at < timezone.now() - timedelta(seconds=execution_time_window): + # Atomic claim so concurrent high-activity batches only alert once + updated = NotificationCampaign.objects.filter( + pk=campaign_id, + developer_reminder_sent=False, + ).update(developer_reminder_sent=True) + if updated: + message = ( + f'[Notification Campaign #{campaign_id}] WARNING: Campaign {campaign.name} exceeded ' + f'its high-activity execution time window ({execution_time_window} seconds).' + ) + logger.warning(message) + sentry.log_message(message) - recipients_qs = NotificationCampaignRecipient.objects.filter(id__in=recipients_ids).select_related('user') recipient_records = [] recipients_qs_annotated = recipients_qs.annotate( recipient_address=Case( @@ -366,11 +585,27 @@ def send_campaign_batch(context, recipients_ids, notification_type_name='blank', invalid_emails_qs.update(status=NotificationCampaignRecipientStatus.SKIPPED, error_message='Invalid email address') if campaign.metadata.get('sendgrid_bulk', False): - # NOTE: sendgrid bulk send feature has not been fully implemented and tested - recipient_emails = list(valid_emails_qs.values_list('recipient_address', flat=True)) + recipients = list(valid_emails_qs) + recipient_emails = [] + custom_args_list = [] + 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), + } + ) try: - send_email_with_send_grid(to_addr=recipient_emails, notification_type=notification_type, context=context) - valid_emails_qs.update(status=NotificationCampaignRecipientStatus.SENT, error_message=None) + send_email_with_send_grid( + to_addr=recipient_emails, + notification_type=notification_type, + context=context, + email_context={'custom_args_list': custom_args_list}, + is_multiple=True, + ) + # 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)}') @@ -378,13 +613,22 @@ def send_campaign_batch(context, recipients_ids, notification_type_name='blank', sentry.log_message(message) valid_emails_qs.update(status=NotificationCampaignRecipientStatus.FAILED, error_message=str(exc)) else: + rendered_html = _render_email_html(notification_type, context) for recipient in valid_emails_qs: notification_started_at = timezone.now() try: - notification_type.emit( - user=recipient.user, + send_email( + recipient_address=recipient.recipient_address, + notification_type=notification_type, event_context=context, - save=False, # Too many write operations + email_context={ + 'custom_args': { + 'campaign_recipient_id': str(recipient.id), + 'campaign_id': str(campaign_id), + 'run_id': str(run_id), + }, + }, + rendered_html=rendered_html, ) recipient.status = NotificationCampaignRecipientStatus.SENT recipient.error_message = None @@ -407,15 +651,14 @@ def send_campaign_batch(context, recipients_ids, notification_type_name='blank', f'campaign_name={campaign.name}') logger.warning(message) sentry.log_message(message) - NotificationCampaignRecipient.objects.bulk_update(recipient_records, ['status', 'error_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) - stats = get_campaign_recipient_stats(campaign_id) - notification_campaign.sent_count = stats['sent_count'] - notification_campaign.failed_count = stats['failed_count'] - notification_campaign.save(update_fields=['sent_count', 'failed_count']) + self.sync_campaign_stats(notification_campaign) + notification_campaign.save(update_fields=['sent_count', 'failed_count', 'recipient_count', 'updated_at']) batch_finished_at = timezone.now() batch_run_time = (batch_finished_at - batch_started_at).total_seconds() diff --git a/osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py b/osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py new file mode 100644 index 00000000000..9cc303d5fcb --- /dev/null +++ b/osf/migrations/0055_notificationcampaignrecipient_batch_id_and_more.py @@ -0,0 +1,23 @@ +# Generated by Django 4.2.26 on 2026-09-11 17:03 + +from django.db import migrations, models + + +class Migration(migrations.Migration): + + dependencies = [ + ('osf', '0054_merge_20260909_reconcile'), + ] + + operations = [ + migrations.AddField( + model_name='notificationcampaignrecipient', + name='batch_id', + field=models.UUIDField(blank=True, db_index=True, null=True), + ), + migrations.AlterField( + model_name='notificationcampaignrecipient', + name='status', + field=models.CharField(choices=[('queued', 'Queued'), ('pending', 'Pending'), ('sent', 'Sent'), ('failed', 'Failed'), ('skipped', 'Skipped'), ('postponed', 'Postponed')], db_index=True, default='pending', max_length=20), + ), + ] diff --git a/osf/migrations/__init__.py b/osf/migrations/__init__.py index 9aca9319438..6118efdd0c6 100644 --- a/osf/migrations/__init__.py +++ b/osf/migrations/__init__.py @@ -121,6 +121,7 @@ def get_admin_write_permissions(): 'change_emailtask', 'delete_emailtask', 'change_notificationcampaign', + 'delete_notificationcampaign', ]) diff --git a/osf/models/notification.py b/osf/models/notification.py index d3071442af6..3f9a66b7daa 100644 --- a/osf/models/notification.py +++ b/osf/models/notification.py @@ -1,12 +1,10 @@ import logging -import waffle from django.db import models from django.utils import timezone from api.base import settings as api_settings -from website import settings as osf_settings -from osf import email, features +from osf import email class Notification(models.Model): @@ -38,27 +36,12 @@ def send( f"\nemail_context={email_context}" ) - if waffle.switch_is_active(features.ENABLE_MAILHOG): - email.send_email_over_smtp( - recipient_address, - self.subscription.notification_type, - self.event_context, - email_context - ) - - if not osf_settings.LOCAL_MODE: - email.send_email_with_send_grid( - recipient_address, - self.subscription.notification_type, - self.event_context, - email_context - ) - - if osf_settings.LOCAL_MODE and not waffle.switch_is_active(features.ENABLE_MAILHOG): - logging.warning( - 'Both ENABLE_MAILHOG and LOCAL_MODE are disabled. Emails will not be sent to MailHog or real email addresses. ' - 'Turn on ENABLE_MAILHOG to send emails to MailHog for testing, or turn on LOCAL_MODE to send emails with SendGrid.' - ) + email.send_email( + recipient_address=recipient_address, + notification_type=self.subscription.notification_type, + event_context=self.event_context, + email_context=email_context + ) if save: self.mark_sent() diff --git a/osf/models/notification_campaign.py b/osf/models/notification_campaign.py index 4b03337f25e..ef3441233fb 100644 --- a/osf/models/notification_campaign.py +++ b/osf/models/notification_campaign.py @@ -13,6 +13,7 @@ class NotificationCampaignStatus(models.TextChoices): ENDED = 'ended', 'Ended' class NotificationCampaignRecipientStatus(models.TextChoices): + QUEUED = 'queued', 'Queued' PENDING = 'pending', 'Pending' SENT = 'sent', 'Sent' FAILED = 'failed', 'Failed' @@ -83,7 +84,6 @@ def start(self, restart_failed=False, restart_stuck=False): self.started_at = timezone.now() self.run_id = uuid.uuid4() if not restart_failed and not restart_stuck: - self.recipient_count = 0 self.sent_count = 0 self.failed_count = 0 self.retries = 0 @@ -99,6 +99,16 @@ def start(self, restart_failed=False, restart_stuck=False): ) ) + def create_recipients(self): + from osf.email.notification_campaign import create_campaign_recipients + + self.metadata.update({'recipients_creation_finished': False}) + self.save() + + transaction.on_commit( + lambda: create_campaign_recipients.delay(campaign_id=self.id) + ) + class NotificationCampaignRecipient(models.Model): campaign = models.ForeignKey( @@ -120,6 +130,7 @@ class NotificationCampaignRecipient(models.Model): error_message = models.TextField(null=True, blank=True) activity_score = models.IntegerField(default=0) + batch_id = models.UUIDField(null=True, blank=True, db_index=True) class Meta: unique_together = ('campaign', 'user') diff --git a/osf_tests/test_notification_campaign.py b/osf_tests/test_notification_campaign.py index a54e1f16fed..04504c16c75 100644 --- a/osf_tests/test_notification_campaign.py +++ b/osf_tests/test_notification_campaign.py @@ -6,11 +6,14 @@ from django.utils import timezone from django.db.models import Q +from osf.email import _build_sendgrid_personalizations from osf.email.notification_campaign import ( + NotificationCampaignTask, + assign_batch_id_to_recipients, create_campaign_recipients, - get_campaign_recipient_batches, get_campaign_recipient_stats, process_campaign_retry, + process_sendgrid_campaign_events, send_campaign_batch, start_notification_campaign, build_query, @@ -29,6 +32,47 @@ pytestmark = pytest.mark.django_db +class TestNotificationCampaignTask: + + @pytest.fixture + def task(self): + return NotificationCampaignTask() + + def test_get_campaign_returns_campaign(self, task, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + loaded = task.get_campaign(campaign.id, campaign.run_id) + assert loaded.pk == campaign.pk + + def test_get_campaign_skips_stale_run_id(self, task, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + assert task.get_campaign(campaign.id, uuid.uuid4()) is None + + def test_get_campaign_skips_cancelled_by_default(self, task, campaign): + campaign.status = NotificationCampaignStatus.CANCELLED + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['status', 'run_id']) + assert task.get_campaign(campaign.id, campaign.run_id) is None + + def test_get_campaign_can_include_cancelled(self, task, campaign): + campaign.status = NotificationCampaignStatus.CANCELLED + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['status', 'run_id']) + loaded = task.get_campaign(campaign.id, campaign.run_id, abort_if_cancelled=False) + assert loaded.pk == campaign.pk + + 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 + ) + stats = task.sync_campaign_stats(campaign) + assert stats['sent_count'] == 1 + assert campaign.sent_count == 1 + + @pytest.fixture def notification_type(): notification_type, _ = NotificationType.objects.get_or_create(name='blank') @@ -58,26 +102,23 @@ def _set_activity(user, total): defaults={'total': total, 'action': {}, 'date': {}}, ) +def reset_batch_ids(campaign_id): + NotificationCampaignRecipient.objects.filter(campaign_id=campaign_id).update(batch_id=None) def _recipient_user_ids(campaign_id, **batch_kwargs): """Flatten all batches into an list of user ids and preserve order""" - user_ids = [] - for batch in get_campaign_recipient_batches(campaign_id=campaign_id, batch_size=1000, **batch_kwargs): - recipients = NotificationCampaignRecipient.objects.filter(id__in=batch) - by_id = {r.id: r.user_id for r in recipients} - user_ids.extend(by_id[recipient_id] for recipient_id in batch) + + batch_id = assign_batch_id_to_recipients(campaign_id=campaign_id, batch_size=1000, **batch_kwargs) + recipients = NotificationCampaignRecipient.objects.filter(batch_id=batch_id) + user_ids = [r.user_id for r in recipients] return user_ids def _recipient_scores(campaign_id, **batch_kwargs): """Flatten all batches into an list of activity scores and preserve order""" - scores = [] - for batch in get_campaign_recipient_batches(campaign_id=campaign_id, batch_size=1000, **batch_kwargs): - by_id = { - r.id: r.activity_score - for r in NotificationCampaignRecipient.objects.filter(id__in=batch) - } - scores.extend(by_id[recipient_id] for recipient_id in batch) + batch_id = assign_batch_id_to_recipients(campaign_id=campaign_id, batch_size=1000, **batch_kwargs) + recipients = NotificationCampaignRecipient.objects.filter(batch_id=batch_id) + scores = [r.activity_score for r in recipients] return scores @@ -135,6 +176,61 @@ def test_or_combines_usernames(self): assert plain_user.id in user_ids assert other_email.id not in user_ids + def test_isnull_false_excludes_unconfirmed_accounts(self): + confirmed = UserFactory() + unconfirmed = UserFactory(date_confirmed=None, is_registered=False) + + query = build_query({ + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }) + user_ids = set(OSFUser.objects.filter(query).values_list('id', flat=True)) + + assert user_ids == {confirmed.id} + + def test_build_campaign_filter_query_ands_predefined_and_manual(self): + from osf.email.notification_campaign import build_campaign_filter_query + + confirmed = UserFactory() + unconfirmed = UserFactory(date_confirmed=None, is_registered=False) + inactive = UserFactory() + inactive.is_disabled = True + inactive.save() + assert inactive.is_active is False + + query = build_campaign_filter_query({ + 'predefined': 'active', + 'manual': { + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id},{inactive.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }, + }) + user_ids = set(OSFUser.objects.filter(query).values_list('id', flat=True)) + + assert user_ids == {confirmed.id} + class TestCreateCampaignRecipients: @@ -197,6 +293,7 @@ def test_ordered_by_activity_score_descending(self, campaign): scores = _recipient_scores(campaign.id) assert scores == sorted(scores, reverse=True) + reset_batch_ids(campaign.id) user_ids = _recipient_user_ids(campaign.id) assert user_ids == [newer.id, mid.id, older.id] @@ -304,6 +401,7 @@ def test_high_activity_non_spam(self, campaign, users_and_recipients): min_activity=data['threshold'], spam=False, ) + assert set(user_ids) == {data['high'].id, data['flagged'].id} def test_low_activity_non_spam_includes_zero(self, campaign, users_and_recipients): @@ -340,8 +438,14 @@ def test_batches_respect_batch_size(self, campaign): Q(**{'id__in': [u.id for u in users]}), campaign_id=campaign.id, ) + batches = [] + while True: + batch_id = assign_batch_id_to_recipients(campaign_id=campaign.id, batch_size=2) + if not batch_id: + break + + batches.append(list(NotificationCampaignRecipient.objects.filter(batch_id=batch_id).values_list('user_id', flat=True))) - batches = list(get_campaign_recipient_batches(campaign_id=campaign.id, batch_size=2)) assert [len(batch) for batch in batches] == [2, 2, 1] flat = {recipient_id for batch in batches for recipient_id in batch} assert len(flat) == 5 @@ -352,10 +456,10 @@ def test_restart_failed_only_returns_failed(self, campaign, users_and_recipients failed.status = NotificationCampaignRecipientStatus.FAILED failed.save(update_fields=['status']) - pending_ids = _recipient_user_ids(campaign.id, spam=False, min_activity=data['threshold']) + pending_ids = _recipient_user_ids(campaign.id, min_activity=data['threshold'], spam=False) assert data['high'].id not in pending_ids - failed_ids = _recipient_user_ids(campaign.id, restart_failed=True) + failed_ids = _recipient_user_ids(campaign.id, restart_failed=True, min_activity=data['threshold'], spam=False) assert failed_ids == [data['high'].id] def test_ignore_conflicts_on_duplicate_create(self, campaign): @@ -412,8 +516,8 @@ def test_campaign_start_restart_stuck_counts(self, mock_delay, campaign): restart_stuck=True, ) - @mock.patch('osf.email.notification_campaign.chain') - def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign.apply_async') + def test_start_schedules_workflow(self, mock_dispatch_campaign, campaign): high = UserFactory() low = UserFactory() spam = UserFactory() @@ -427,8 +531,7 @@ def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campa } campaign.run_id = uuid.uuid4() campaign.save() - - mock_chain.return_value.apply_async = mock.Mock() + create_campaign_recipients(campaign_id=campaign.id) start_notification_campaign(campaign.id) @@ -439,11 +542,49 @@ def test_start_creates_recipients_and_schedules_workflow(self, mock_chain, campa } assert recipients == {high.id: 250, low.id: 10, spam.id: 0} assert campaign.recipient_count == 3 - mock_chain.assert_called_once() - mock_chain.return_value.apply_async.assert_called_once() + mock_dispatch_campaign.assert_called_once() + + @mock.patch('osf.email.notification_campaign.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) + _set_activity(confirmed, 100) + _set_activity(unconfirmed, 100) + + campaign.metadata['filters'] = { + 'predefined': 'active', + 'manual': { + 'operator': 'AND', + 'children': [ + { + 'field': 'id', + 'lookup': 'in', + 'value': f'{confirmed.id},{unconfirmed.id}', + }, + { + 'field': 'date_confirmed', + 'lookup': 'isnull', + 'value': False, + }, + ], + }, + } + campaign.run_id = uuid.uuid4() + campaign.save() + create_campaign_recipients(campaign_id=campaign.id) + mock_dispatch_campaign.return_value.apply_async = mock.Mock() + + start_notification_campaign(campaign.id) - @mock.patch('osf.email.notification_campaign.chain') - def test_start_restart_failed_does_not_recreate_recipients(self, mock_chain, campaign): + recipient_ids = set( + NotificationCampaignRecipient.objects.filter(campaign=campaign).values_list( + 'user_id', flat=True + ) + ) + assert recipient_ids == {confirmed.id} + + @mock.patch('osf.email.notification_campaign.dispatch_campaign') + def test_start_restart_failed_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() _set_activity(user, 50) create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) @@ -456,21 +597,21 @@ def test_start_restart_failed_does_not_recreate_recipients(self, mock_chain, cam 'manual': {'operator': 'AND', 'children': [{'field': 'id', 'lookup': 'in', 'value': str(user.id)}]}, } campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id, restart_failed=True) assert NotificationCampaignRecipient.objects.filter(campaign=campaign).count() == 1 assert NotificationCampaignRecipient.objects.get(pk=recipient.pk).status == NotificationCampaignRecipientStatus.FAILED - @mock.patch('osf.email.notification_campaign.chain') - def test_start_restart_stuck_does_not_recreate_recipients(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.dispatch_campaign') + def test_start_restart_stuck_does_not_recreate_recipients(self, mock_dispatch_campaign, campaign): user = UserFactory() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) campaign.run_id = uuid.uuid4() campaign.recipient_count = 1 campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() start_notification_campaign(campaign.id, restart_stuck=True) @@ -489,15 +630,16 @@ def running_campaign(self, campaign): campaign.save() return campaign - @mock.patch.object(NotificationType, 'emit') - def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_campaign): + @mock.patch('osf.email.notification_campaign.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) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -507,18 +649,19 @@ def test_send_campaign_batch_marks_recipients_sent(self, mock_emit, running_camp running_campaign.refresh_from_db() assert recipient.status == NotificationCampaignRecipientStatus.SENT assert running_campaign.sent_count == 1 - mock_emit.assert_called_once() + mock_send_email.assert_called_once() - @mock.patch.object(NotificationType, 'emit', side_effect=Exception('send failed')) + @mock.patch('osf.email.notification_campaign.send_email', side_effect=Exception('send failed')) @mock.patch('osf.email.notification_campaign.sentry.log_exception') - def test_send_campaign_batch_marks_recipients_failed(self, mock_sentry, mock_emit, running_campaign): + 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) + batch_id = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -537,11 +680,12 @@ def test_send_campaign_batch_skips_invalid_email_addresses(self, running_campaig user.emails.all().delete() 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -560,11 +704,12 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -573,10 +718,61 @@ def test_send_campaign_batch_sendgrid_bulk_success(self, mock_sendgrid, running_ recipient.refresh_from_db() running_campaign.refresh_from_db() mock_sendgrid.assert_called_once() - assert recipient.status == NotificationCampaignRecipientStatus.SENT - assert running_campaign.sent_count == 1 + 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 + def test_build_sendgrid_personalizations_shared_to_list(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + is_multiple=False, + ) + assert personalizations == [{ + 'to': [ + {'email': 'a@example.com'}, + {'email': 'b@example.com'}, + ], + }] + + def test_build_sendgrid_personalizations_one_per_recipient(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + is_multiple=True, + ) + assert personalizations == [ + {'to': [{'email': 'a@example.com'}]}, + {'to': [{'email': 'b@example.com'}]}, + ] + + def test_build_sendgrid_personalizations_attaches_custom_args(self): + personalizations = _build_sendgrid_personalizations( + ['a@example.com', 'b@example.com'], + email_context={ + 'custom_args_list': [ + {'campaign_recipient_id': '1', 'campaign_id': '9'}, + {'campaign_recipient_id': '2', 'campaign_id': '9'}, + ], + }, + is_multiple=True, + ) + assert personalizations == [ + { + 'to': [{'email': 'a@example.com'}], + 'custom_args': {'campaign_recipient_id': '1', 'campaign_id': '9'}, + }, + { + 'to': [{'email': 'b@example.com'}], + 'custom_args': {'campaign_recipient_id': '2', 'campaign_id': '9'}, + }, + ] + @mock.patch( 'osf.email.notification_campaign.send_email_with_send_grid', side_effect=Exception('bulk failed'), @@ -587,11 +783,12 @@ def test_send_campaign_batch_sendgrid_bulk_failure(self, mock_sentry, mock_sendg 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -610,15 +807,79 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r running_campaign.metadata['execution']['time_window'] = 8 running_campaign.save() create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=running_campaign.id) - recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) + 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'): + send_campaign_batch( + context={}, + batch_id=batch_id, + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + developer_reminder=True, + ) + + running_campaign.refresh_from_db() + assert running_campaign.developer_reminder_sent is True + mock_sentry.assert_called_once() - with mock.patch.object(NotificationType, 'emit'): + @mock.patch('osf.email.notification_campaign.sentry.log_message') + def test_send_campaign_batch_skips_reminder_without_developer_reminder_flag( + self, mock_sentry, running_campaign + ): + user = UserFactory() + running_campaign.started_at = timezone.now() - timedelta(seconds=9) + running_campaign.metadata['execution']['time_window'] = 8 + running_campaign.save() + 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) + mock_sentry.reset_mock() + + with mock.patch('osf.email.notification_campaign.send_email'): + send_campaign_batch( + context={}, + batch_id=batch_id, + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + ) + + running_campaign.refresh_from_db() + assert running_campaign.developer_reminder_sent is False + mock_sentry.assert_not_called() + + @mock.patch('osf.email.notification_campaign.sentry.log_message') + def test_send_campaign_batch_reminder_claimed_only_once(self, mock_sentry, running_campaign): + user_a = UserFactory() + user_b = UserFactory() + running_campaign.started_at = timezone.now() - timedelta(seconds=9) + running_campaign.metadata['execution']['time_window'] = 8 + running_campaign.save() + create_campaign_recipients( + Q(**{'id__in': [user_a.id, user_b.id]}), + campaign_id=running_campaign.id, + ) + batch_id_1 = assign_batch_id_to_recipients(campaign_id=running_campaign.id, batch_size=1) + 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'): + send_campaign_batch( + context={}, + batch_id=batch_id_1, + notification_type_name='blank', + campaign_id=running_campaign.id, + run_id=running_campaign.run_id, + developer_reminder=True, + ) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id_2, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, + developer_reminder=True, ) running_campaign.refresh_from_db() @@ -628,11 +889,12 @@ def test_send_campaign_batch_logs_when_time_window_exceeded(self, mock_sentry, r def test_send_campaign_batch_marks_failed_when_notification_type_missing(self, running_campaign): user = UserFactory() 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='does-not-exist', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -641,44 +903,46 @@ def test_send_campaign_batch_marks_failed_when_notification_type_missing(self, r running_campaign.refresh_from_db() recipient.refresh_from_db() assert running_campaign.status == NotificationCampaignStatus.FAILED - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.FAILED def test_send_campaign_batch_skips_stale_run_id(self, running_campaign): user = UserFactory() 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=uuid.uuid4(), ) recipient.refresh_from_db() - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED def test_send_campaign_batch_skips_cancelled_campaign(self, running_campaign): user = UserFactory() running_campaign.status = NotificationCampaignStatus.CANCELLED running_campaign.save(update_fields=['status']) 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, ) recipient.refresh_from_db() - assert recipient.status == NotificationCampaignRecipientStatus.PENDING + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED - @mock.patch.object(NotificationType, 'emit') - def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, mock_emit, running_campaign): + @mock.patch('osf.email.notification_campaign.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' user.save(update_fields=['username']) @@ -686,11 +950,12 @@ def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, m user.emails.create(address='fallback@example.com') 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) recipient = NotificationCampaignRecipient.objects.get(campaign=running_campaign, user=user) send_campaign_batch( context={}, - recipients_ids=[recipient.id], + batch_id=batch_id, notification_type_name='blank', campaign_id=running_campaign.id, run_id=running_campaign.run_id, @@ -701,7 +966,7 @@ def test_send_campaign_batch_uses_fallback_email_when_username_has_no_at(self, m assert recipient.status == NotificationCampaignRecipientStatus.SENT assert running_campaign.sent_count == 1 assert running_campaign.failed_count == 0 - mock_emit.assert_called_once() + mock_send_email.assert_called_once() class TestNotificationCampaignCancel: @@ -788,6 +1053,7 @@ def test_process_campaign_retry_keeps_cancelled_status_and_syncs_stats(self, moc campaign.run_id = uuid.uuid4() campaign.status = NotificationCampaignStatus.CANCELLED campaign.save() + mock_sentry.reset_mock() process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) @@ -799,8 +1065,8 @@ 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.chain') - def test_process_campaign_retry_retries_failed_recipients(self, mock_chain, campaign): + @mock.patch('osf.email.notification_campaign.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) recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) @@ -810,14 +1076,14 @@ def test_process_campaign_retry_retries_failed_recipients(self, mock_chain, camp campaign.retries = 0 campaign.status = NotificationCampaignStatus.RUNNING campaign.save() - mock_chain.return_value.apply_async = mock.Mock() + mock_dispatch_campaign.return_value.apply_async = mock.Mock() process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) campaign.refresh_from_db() assert campaign.retries == 1 assert campaign.status == NotificationCampaignStatus.RUNNING - mock_chain.assert_called_once() + mock_dispatch_campaign.assert_called_once() def test_process_campaign_retry_marks_partially_completed_after_max_retries(self, campaign): user = UserFactory() @@ -836,3 +1102,163 @@ def test_process_campaign_retry_marks_partially_completed_after_max_retries(self assert campaign.failed_count == 1 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): + user = UserFactory() + create_campaign_recipients(Q(**{'id__in': [user.id]}), campaign_id=campaign.id) + NotificationCampaignRecipient.objects.filter(campaign=campaign).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + campaign.run_id = uuid.uuid4() + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() + campaign.metadata['execution']['dispatch_interval'] = 60 + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.RUNNING + assert campaign.completed_at is None + assert campaign.recipient_count == 1 + 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}, + countdown=60, + ) + + def test_process_campaign_retry_times_out_queued_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 + ) + campaign.run_id = uuid.uuid4() + campaign.retries = 0 + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() - timedelta(seconds=10) + campaign.metadata['execution']['delivery_timeout'] = 1 + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + recipient = NotificationCampaignRecipient.objects.get(campaign=campaign, user=user) + assert recipient.status == NotificationCampaignRecipientStatus.FAILED + assert recipient.error_message == 'SendGrid delivery timeout' + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.PARTIALLY_COMPLETED + assert campaign.retries == 0 + 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( + self, mock_dispatch_campaign, mock_apply_async, campaign + ): + queued_user = UserFactory() + failed_user = UserFactory() + create_campaign_recipients( + Q(**{'id__in': [queued_user.id, failed_user.id]}), + campaign_id=campaign.id, + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=queued_user).update( + status=NotificationCampaignRecipientStatus.QUEUED + ) + NotificationCampaignRecipient.objects.filter(campaign=campaign, user=failed_user).update( + status=NotificationCampaignRecipientStatus.FAILED + ) + campaign.run_id = uuid.uuid4() + campaign.retries = 0 + campaign.status = NotificationCampaignStatus.RUNNING + campaign.started_at = timezone.now() + campaign.save() + + process_campaign_retry(campaign_id=campaign.id, run_id=campaign.run_id) + + campaign.refresh_from_db() + assert campaign.status == NotificationCampaignStatus.RUNNING + assert campaign.retries == 0 + assert campaign.completed_at is None + mock_apply_async.assert_called_once() + mock_dispatch_campaign.assert_not_called() + + +class TestProcessSendgridCampaignEvents: + + def _event(self, campaign, recipient, event, **extra): + payload = { + 'event': event, + 'campaign_id': str(campaign.id), + 'campaign_recipient_id': str(recipient.id), + 'run_id': str(campaign.run_id), + } + payload.update(extra) + return payload + + def _queued_recipient(self, campaign, user=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']) + return recipient + + def test_delivered_marks_queued_sent(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + + 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_failure_marks_queued_failed(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_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.error_message == 'mailbox full' + + def test_delivered_overrides_failed(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + recipient.status = NotificationCampaignRecipientStatus.FAILED + recipient.error_message = 'bounce' + 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_ignores_stale_run_id(self, campaign): + campaign.run_id = uuid.uuid4() + campaign.save(update_fields=['run_id']) + recipient = self._queued_recipient(campaign) + + process_sendgrid_campaign_events([{ + 'event': 'delivered', + 'campaign_id': str(campaign.id), + 'campaign_recipient_id': str(recipient.id), + 'run_id': str(uuid.uuid4()), + }]) + + recipient.refresh_from_db() + assert recipient.status == NotificationCampaignRecipientStatus.QUEUED diff --git a/tests/utils.py b/tests/utils.py index ba208573f37..54616ad86d8 100644 --- a/tests/utils.py +++ b/tests/utils.py @@ -320,7 +320,7 @@ def _wrapped_emit(self, *emit_args, **emit_kwargs): _real_send_over_smtp = _osf_email.send_email_over_smtp _real_send_with_sendgrid = _osf_email.send_email_with_send_grid - def _fake_send_over_smtp(to_email, notification_type, context=None, email_context=None): + def _fake_send_over_smtp(to_email, notification_type, context=None, email_context=None, rendered_html=None): captured['emails'].append({ 'protocol': 'smtp', 'to': to_email, @@ -329,9 +329,9 @@ def _fake_send_over_smtp(to_email, notification_type, context=None, email_contex 'email_context': email_context.copy() if isinstance(email_context, dict) else email_context, }) if passthrough: - return _real_send_over_smtp(to_email, notification_type, context, email_context) + 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): + def _fake_send_with_sendgrid(user, notification_type, context=None, email_context=None, rendered_html=None): captured['emails'].append({ 'protocol': 'sendgrid', 'to': user, @@ -340,7 +340,7 @@ 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) + return _real_send_with_sendgrid(user, notification_type, context, email_context, rendered_html) 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 d1e077aea54..e42992339d6 100644 --- a/website/settings/defaults.py +++ b/website/settings/defaults.py @@ -181,6 +181,8 @@ def parent_dir(path): # OR, if using Sendgrid's API # WARNING: If `SENDGRID_WHITELIST_MODE` is True, SENDGRID_API_KEY = None +# Public verification key from SendGrid Event Webhook (Mail Settings -> Event Webhook -> Signed Event Webhook) +SENDGRID_EVENT_WEBHOOK_PUBLIC_KEY = None # Mailchimp MAILCHIMP_API_KEY = None @@ -200,12 +202,17 @@ def parent_dir(path): NO_LOGIN_OSF4M_WAIT_TIME = timedelta(weeks=52) # 1 year for "We miss you at OSF" email to users created from OSF4M NOTIFICATIONS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notifications and email tasks NOTIFICATIONS_CLEANUP_BATCH_SIZE = 10000 # Batch size for notifications and email tasks cleanup +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notification campaign recipients +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE = 5000 # Batch size for notification campaign recipients cleanup # Notification campaign execution defaults (overridable per campaign in admin metadata) DEFAULT_CAMPAIGN_ACTIVITY_THRESHOLD = 3 # Users at/above this activity total are scheduled in the high-activity phase 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. +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) @@ -458,7 +465,9 @@ class CeleryConfig: 'website.identifiers.tasks.task__update_verified_links' } - external_low_modules = {} + external_low_modules = { + 'email.process_sendgrid_campaign_events', + } account_status_changes_modules = {} @@ -521,6 +530,7 @@ class CeleryConfig: 'scripts.check_manual_restart_approval', 'scripts.enhanced_stuck_registration_audit', 'email.start_notification_campaign', + 'email.dispatch_campaign', } background_migration_modules = { @@ -725,6 +735,10 @@ class CeleryConfig: 'schedule': crontab(minute=0, hour=7), # Daily 2 a.m 'kwargs': {'dry_run': False}, }, + 'delete_notification_campaign_recipients': { + 'task': 'notifications.tasks.delete_notification_campaign_recipients', + 'schedule': crontab(minute=0, hour=3, day_of_month=1), + }, 'clear_expired_sessions': { 'task': 'osf.management.commands.clear_expired_sessions', 'schedule': crontab(minute=0, hour=5), # Daily 12 a.m diff --git a/website/settings/local-ci.py b/website/settings/local-ci.py index eec0e7070e2..1c3636405fb 100644 --- a/website/settings/local-ci.py +++ b/website/settings/local-ci.py @@ -67,6 +67,8 @@ class CeleryConfig(defaults.CeleryConfig): NO_ADDON_WAIT_TIME = timedelta(weeks=8) NO_LOGIN_WAIT_TIME = timedelta(weeks=4) NO_LOGIN_OSF4M_WAIT_TIME = timedelta(weeks=6) +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_AGE = timedelta(weeks=12) # 3 months to clean up old notification campaign recipients +NOTIFICATION_CAMPAIGN_RECIPIENTS_CLEANUP_BATCH_SIZE = 10 # Batch size for notification campaign recipients cleanup # Configuration for "We miss you at OSF" email (`NotificationTypeEnum.USER_NO_LOGIN`) MAX_DAILY_NO_LOGIN_EMAILS = None