[ENG-11605][ENG-11695][ENG-11696][ENG-11697][ENG-11605][ENG-11698] ENTER - Part 1 (#11805)

* Implement notification campaign models and email handling
  - Add NotificationCampaign and NotificationCampaignRecipient models.
  - Introduce notification campaign status management.
  - Update OSFUser model to track received notification campaigns.
  - Enhance email templates for notification campaigns.
  - Create migration for new models and relationships.
* Remove notification settings table from blank email template
* Code clean-up and fixes

---------

Co-authored-by: Longze Chen <cslzchen@gmail.com>

Author: Ostap-Zherebetskyi

admin/notifications/views.py

@@ -11,6 +11,7 @@
 from mako.lexer import Lexer
 from mako.parsetree import ControlLine
 import re
+from string import Formatter
 
 def delete_selected_notifications(selected_ids):
     NotificationSubscription.objects.filter(id__in=selected_ids).delete()
@@ -82,7 +83,7 @@ def generate_mock_json(structure, list_name=None):
     return result
 
 
-def build_safe_context(template: str) -> dict:
+def build_safe_context(template: str, subject: str) -> dict:
     templatenode = Lexer(text=template).parse()
     identifiers_location = []
     for node in templatenode.get_children():
@@ -103,6 +104,9 @@ def build_safe_context(template: str) -> dict:
     mock_json = generate_mock_json(identifier_structure)
     context = {identifier: f'mock_{identifier}' for identifier in flatten_identifiers if identifier not in TEMPLATE_IDENTIFIER_BLACKLIST}
     context.update(mock_json)
+
+    # subject
+    context.update({key: key for _, key, _, _ in Formatter().parse(subject) if key})
     return context
 
 class NotificationsList(PermissionRequiredMixin, ListView):
@@ -282,12 +286,12 @@ def get_context_data(self, *args, **kwargs):
                 return kwargs
         else:
             if notification_type.is_digest_type:
-                inner_context = build_safe_context(notification_type.template)
+                inner_context = build_safe_context(notification_type.template, notification_type.subject)
                 inner_template = _render_email_html(notification_type, ctx=inner_context, return_original_error=True)
                 safe_context = {'notifications': [inner_template]}
                 return_context = inner_context
             else:
-                safe_context = build_safe_context(notification_type.template)
+                safe_context = build_safe_context(notification_type.template, notification_type.subject)
                 return_context = safe_context
 
         if notification_type.is_digest_type:
@@ -300,6 +304,7 @@ def get_context_data(self, *args, **kwargs):
         except Exception as e:
             kwargs['rendered_template'] = f"Error rendering template: {str(e)}"
 
+        kwargs['rendered_subject'] = notification_type.subject.format(**safe_context)
         kwargs['context'] = json.dumps(return_context, indent=4)
 
         return kwargs

admin/templates/notifications/notification_type_preview.html

@@ -1,5 +1,6 @@
 <h2>Notification Template Preview</h2>
 <h3>Rendered Template</h3>
+<h3>Subject: {{ rendered_subject|safe }}</h3>
 
 
 

osf/email/notification_campaign.py

@@ -0,0 +1,221 @@
+import logging
+from osf.models import NotificationType, NotificationTypeEnum, OSFUser, UserActivityCounter
+from django.db.models import Q
+from django.db.models import OuterRef, Subquery, Exists, F
+from django.db.models.functions import Coalesce
+from framework.celery_tasks import app as celery_app
+from celery import chord
+from django.utils import timezone
+from osf.models.notification_campaign import NotificationCampaign, NotificationCampaignRecipient, NotificationCampaignStatus, NotificationCampaignRecipientStatus
+
+logger = logging.getLogger(__name__)
+
+
+counter_subquery = (
+    UserActivityCounter.objects
+    .filter(_id=OuterRef('guids___id'))
+    .values('total')[:1]
+)
+
+
+def get_filtered_batches(filters, batch_size=1000, campaign_id=None):
+    already_sent_subquery = NotificationCampaignRecipient.objects.filter(
+        campaign_id=campaign_id,
+        user_id=OuterRef('pk'),
+    )
+
+    qs = (
+        OSFUser.objects
+        .annotate(already_sent=Exists(already_sent_subquery))
+        .filter(already_sent=False)
+        .filter(**filters)
+        .annotate(activity_total=Coalesce(Subquery(counter_subquery), 0))
+        .order_by('-activity_total', '-date_registered', '-id')
+    )
+
+    last_total = None
+    last_date = None
+    last_id = None
+
+    while True:
+        batch_qs = qs
+
+        if last_total is not None:
+            batch_qs = batch_qs.filter(
+                Q(activity_total__lt=last_total) |
+                Q(activity_total=last_total, date_registered__lt=last_date) |
+                Q(activity_total=last_total, date_registered=last_date, id__lt=last_id)
+            )
+
+        batch = batch_qs[:batch_size]
+
+        if not batch:
+            break
+
+        rows = list(batch.values_list('id', 'activity_total', 'date_registered'))
+
+        if not rows:
+            break
+
+        batch_ids = [r[0] for r in rows]
+
+        last_id, last_total, last_date = (
+            rows[-1][0],
+            rows[-1][1],
+            rows[-1][2],
+        )
+
+        yield batch_ids
+
+
+FILTER_PRESETS = {
+    'all': {},
+    'active': {'is_active': True},
+    'internal': {'is_active': True, 'is_staff': True, 'username__endswith': '@cos.io'},
+}
+
+@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)
+    failed_recipients = NotificationCampaignRecipient.objects.filter(campaign=campaign, status=NotificationCampaignRecipientStatus.FAILED)
+    max_retries = campaign.metadata.get('execution', {}).get('max_retries', 3)
+    batch_size = campaign.metadata.get('execution', {}).get('batch_size', 1000)
+    failed_recipients_count = failed_recipients.count()
+    if not failed_recipients_count:
+        campaign.status = NotificationCampaignStatus.COMPLETED
+        campaign.completed_at = timezone.now()
+        campaign.failed_count = 0
+        campaign.save(update_fields=['status', 'completed_at', 'failed_count'])
+        return
+
+    if campaign.retries < max_retries:
+        logger.info(f'Retrying {failed_recipients_count} failed recipients for campaign {campaign_id}')
+        filters = {'id__in': failed_recipients.values_list('user_id', flat=True)}
+        tasks = []
+        for batch in get_filtered_batches(filters=filters, batch_size=batch_size, campaign_id=campaign_id):
+            tasks.append(
+                send_campaign_batch.s(
+                    notification_type_name=campaign.notification_type.name,
+                    recipients_ids=batch,
+                    context=campaign.metadata.get('context', {}),
+                    campaign_id=campaign_id,
+                )
+            )
+        chord(tasks)(
+            process_campaign_retry.s(campaign_id=campaign_id)
+        )
+        campaign.retries += 1
+        campaign.save(update_fields=['retries'])
+    else:
+        campaign.failed_count = failed_recipients_count
+        campaign.status = NotificationCampaignStatus.PARTIALLY_COMPLETED
+        campaign.completed_at = timezone.now()
+        campaign.save(update_fields=['status', 'completed_at', 'failed_count'])
+
+
+@celery_app.task(name='email.start_notification_campaign')
+def start_notification_campaign(campaign_id):
+    campaign = NotificationCampaign.objects.get(id=campaign_id)
+    filters = campaign.metadata.get('filters', {})
+    context = campaign.metadata.get('context', {})
+    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 = FILTER_PRESETS.get(predefined_filter_name, {})
+
+    tasks = []
+    total_recipients = 0
+    batch_size = campaign.metadata.get('execution', {}).get('batch_size', 1000)
+    for batch in get_filtered_batches(filters=filters, batch_size=batch_size):
+        tasks.append(
+            send_campaign_batch.s(
+                notification_type_name=notification_type_name,
+                recipients_ids=batch,
+                context=context,
+                campaign_id=campaign_id,
+            )
+        )
+        total_recipients += len(batch)
+
+    campaign.recipient_count = total_recipients
+    campaign.save(update_fields=['recipient_count'])
+
+    chord(tasks)(
+        process_campaign_retry.s(campaign_id=campaign_id)
+    )
+
+
+@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):
+    campaign = NotificationCampaign.objects.get(id=campaign_id)
+    if campaign.status == NotificationCampaignStatus.CANCELLED:
+        logger.warning(f"Campaign {campaign_id} was cancelled")
+        return
+    if hasattr(NotificationTypeEnum, notification_type_name):
+        notification_type = getattr(NotificationTypeEnum, notification_type_name).instance
+    else:
+        notification_type = NotificationType.objects.filter(
+            name=notification_type_name
+        ).first()  # TODO cache
+
+        if notification_type is None:
+            return
+
+    recipients_qs = OSFUser.objects.filter(id__in=recipients_ids)
+    recipient_records = {
+        'to_create': [],
+        'to_update': [],
+    }
+    existing = {
+        r.user_id: r
+        for r in NotificationCampaignRecipient.objects.filter(
+            campaign_id=campaign_id,
+            user_id__in=recipients_ids,
+        )
+    }
+    success_count = 0
+    failure_count = 0
+
+    for recipient in recipients_qs:
+        recipient_record = existing.get(recipient.id)
+
+        if recipient_record is None:
+            recipient_record = NotificationCampaignRecipient(
+                campaign_id=campaign_id,
+                user=recipient,
+            )
+            operation = 'to_create'
+        else:
+            operation = 'to_update'
+
+        try:
+            notification_type.emit(
+                user=recipient,
+                event_context=context,
+                save=False,  # Too many write operations
+            )
+
+            recipient_record.status = NotificationCampaignRecipientStatus.SENT
+            recipient_record.error_message = None
+            recipient_records[operation].append(recipient_record)
+            success_count += 1
+
+        except Exception as exc:
+            logger.error(exc)  # TODO update error
+            recipient_record.status = NotificationCampaignRecipientStatus.FAILED
+            recipient_record.error_message = str(exc)
+            recipient_records[operation].append(recipient_record)
+
+            failure_count += 1
+            pass
+
+    NotificationCampaignRecipient.objects.bulk_create(recipient_records['to_create'])
+    NotificationCampaignRecipient.objects.bulk_update(recipient_records['to_update'], ['status', 'error_message'])
+
+    NotificationCampaign.objects.filter(pk=campaign_id).update(sent_count=F('sent_count') + success_count, failed_count=F('failed_count') + failure_count)
+    logger.info('Batch finished')  # TODO: add/update logs

osf/migrations/0045_notificationcampaign_notificationcampaignrecipient_and_more.py

@@ -0,0 +1,53 @@
+# Generated by Django 4.2.26 on 2026-07-09 11:38
+
+from django.conf import settings
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('osf', '0044_notification_scheduled'),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name='NotificationCampaign',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('created_at', models.DateTimeField(auto_now_add=True)),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('name', models.CharField(max_length=255)),
+                ('status', models.CharField(choices=[('created', 'Created'), ('running', 'Running'), ('completed', 'Completed'), ('partially_completed', 'Partially Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled'), ('ended', 'Ended')], default='created', max_length=20)),
+                ('started_at', models.DateTimeField(blank=True, null=True)),
+                ('completed_at', models.DateTimeField(blank=True, null=True)),
+                ('metadata', models.JSONField(blank=True, default=dict)),
+                ('recipient_count', models.PositiveIntegerField(default=0)),
+                ('sent_count', models.PositiveIntegerField(default=0)),
+                ('failed_count', models.PositiveIntegerField(default=0)),
+                ('retries', models.PositiveIntegerField(default=0)),
+                ('created_by', models.ForeignKey(null=True, on_delete=django.db.models.deletion.SET_NULL, to=settings.AUTH_USER_MODEL)),
+                ('notification_type', models.ForeignKey(on_delete=django.db.models.deletion.PROTECT, to='osf.notificationtype')),
+            ],
+        ),
+        migrations.CreateModel(
+            name='NotificationCampaignRecipient',
+            fields=[
+                ('id', models.AutoField(auto_created=True, primary_key=True, serialize=False, verbose_name='ID')),
+                ('updated_at', models.DateTimeField(auto_now=True)),
+                ('status', models.CharField(choices=[('pending', 'Pending'), ('sent', 'Sent'), ('failed', 'Failed')], db_index=True, default='pending', max_length=20)),
+                ('error_message', models.TextField(blank=True, null=True)),
+                ('campaign', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name='recipients', to='osf.notificationcampaign')),
+                ('user', models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, to=settings.AUTH_USER_MODEL)),
+            ],
+            options={
+                'unique_together': {('campaign', 'user')},
+            },
+        ),
+        migrations.AddField(
+            model_name='osfuser',
+            name='received_notification_campaigns',
+            field=models.ManyToManyField(through='osf.NotificationCampaignRecipient', to='osf.notificationcampaign'),
+        ),
+    ]