Send CfP emails via a queue

apps/admin/email.py

@@ -1,7 +1,7 @@
-from collections.abc import Sequence
 from typing import Literal
 
 from flask import flash, redirect, render_template, url_for
+from flask.typing import ResponseReturnValue
 from sqlalchemy import select
 from wtforms import SelectField, StringField, SubmitField
 from wtforms.validators import DataRequired
@@ -15,8 +15,9 @@
 from models.village import VillageMember
 
 from ..common.email import (
-    enqueue_trusted_emails,
+    enqueue_emails,
     format_trusted_html_email,
+    format_trusted_plaintext_email,
     preview_trusted_email,
 )
 from ..common.forms import Form
@@ -43,7 +44,7 @@ class EmailComposeForm(Form):
     send = SubmitField("Send Email")
 
 
-def get_users(dest: Literal["ticket", "cfp", "purchasers", "villages"]) -> Sequence[User]:
+def get_users(dest: Literal["ticket", "cfp", "purchasers", "villages"]) -> list[User]:
     query = select(User)
     if dest == "ticket":
         query = (
@@ -81,13 +82,13 @@ def get_email_reason(dest: str) -> str:
 
 
 @admin.route("/email", methods=["GET", "POST"])
-def email():
+def email() -> ResponseReturnValue:
+    # This function is almost identical to apps.villages.admin.admin_email_owners, consider updating there too
     form = EmailComposeForm()
     if form.validate_on_submit():
+        users: list[User]
         if form.destination.data == "ticket_and_cfp":
-            users = set()
-            users.update(get_users("ticket"))
-            users.update(get_users("cfp"))
+            users = list(set(get_users("ticket")) | set(get_users("cfp")))
         else:
             users = get_users(form.destination.data)
 
@@ -117,12 +118,18 @@ def email():
             )
 
         if form.send.data is True:
-            enqueue_trusted_emails(
-                users,
-                form.subject.data,
-                form.text.data,
-                reason=reason,
+            assert form.text.data  # DataRequired()
+            assert form.subject.data  # DataRequired()
+            body: str = form.text.data
+            subject: str = form.subject.data
+            enqueue_emails(
+                "bulk_contact",
+                users=users,
+                subject=subject,
+                text_body=format_trusted_plaintext_email(body),
+                html_body=format_trusted_html_email(body, subject, reason=reason),
             )
+            db.session.commit()
             flash(f"Email queued for sending to {len(users)} users")
             return redirect(url_for(".email"))
 

apps/base/scheduled_tasks.py

@@ -1,61 +1,115 @@
+import logging
+from datetime import timedelta
+from itertools import groupby
+from typing import Any
+
 from flask import current_app as app
+from prometheus_client import Counter
+from sqlalchemy import case, select
+from sqlalchemy.orm import joinedload
 
 from main import db, mail
-from models.email import EmailJobRecipient
+from models import naive_utcnow
+from models.email import EmailJob, EmailJobRecipient, EmailJobType
 from models.scheduled_task import scheduled_task
-from models.volunteer.notify import VolunteerNotifyRecipient
 
 from ..config import config
 
+logger = logging.getLogger(__name__)
+
+EMAIL_PRIORITIES: dict[EmailJobType, int] = {
+    "notify_volunteer": 0,
+    "cfp": 1,
+    "bulk_contact": 2,
+}
+EMAIL_YIELD_INTERVAL = timedelta(seconds=30)
+
+email_yields = Counter("emf_email_yields", "Queued email yields")
+
 
 @scheduled_task(minutes=1)
-def send_emails():
-    """Send queued emails, allowing for failure"""
-    count = 0
+def send_emails() -> int:
+    """
+    Send queued emails in priority order, allowing for failure.
 
-    with mail.get_connection(app.config.get("BULK_MAIL_BACKEND")) as conn:
-        for rec in EmailJobRecipient.query.filter(EmailJobRecipient.sent == False):
-            count += send_email(conn, rec)
-    return count
+    The job only runs once a minute, so this isn't really suitable for time-sensitive emails.
 
+    We yield if the process takes too long (e.g. mail server is struggling),
+    as only one scheduled task can run at once.
+    """
 
-def send_email(conn, rec):
-    sent_count = mail.send_mail(
-        subject=rec.job.subject,
-        message=rec.job.text_body,
-        from_email=config.from_email("CONTACT_EMAIL"),
-        recipient_list=[rec.user.email],
-        fail_silently=True,
-        connection=conn,
-        html_message=rec.job.html_body,
-    )
-    if sent_count > 0:
-        rec.sent = True
-        db.session.commit()
-    return sent_count
+    start = naive_utcnow()
 
+    recs: list[EmailJobRecipient] = list(
+        db.session.scalars(
+            select(EmailJobRecipient)
+            .join(EmailJobRecipient.job)
+            .options(
+                joinedload(EmailJobRecipient.job),
+            )
+            .where(
+                EmailJobRecipient.sent == False,
+                EmailJobRecipient.sent_at.is_(None),
+            )
+            .order_by(
+                case(EMAIL_PRIORITIES, value=EmailJob.type),
+                EmailJobRecipient.id,
+            )
+        )
+    )
 
-@scheduled_task(minutes=1)
-def send_volunteer_emails():
-    """Send queued volunteer notifications"""
     count = 0
-    with mail.get_connection() as conn:
-        for rec in VolunteerNotifyRecipient.query.filter(VolunteerNotifyRecipient.sent == False):
-            count += send_volunteer_email(conn, rec)
+    job_type: EmailJobType
+    for job_type, grouped_recs in groupby(recs, lambda r: r.job.type):
+        if job_type == "bulk_contact":
+            # Use config beginning BULK_MAIL_ on production-like systems (see apps/common/backends/bulk.py)
+            backend = app.config.get("BULK_MAIL_BACKEND")
+        else:
+            backend = None
+
+        with mail.get_connection(backend=backend) as conn:
+            for rec in grouped_recs:
+                count += send_email(conn, rec)
+
+                if naive_utcnow() - start > EMAIL_YIELD_INTERVAL:
+                    logger.warning("Email sending is taking too long, yielding")
+                    email_yields.inc()
+                    return count
+
     return count
 
 
-def send_volunteer_email(conn, rec):
-    sent_count = mail.send_mail(
+def send_email(conn: Any, rec: EmailJobRecipient) -> int:
+    match rec.job.type:
+        case "notify_volunteer":
+            from_email = config.from_email("VOLUNTEER_EMAIL")
+            assert rec.user.volunteer
+            recipient = rec.user.volunteer.volunteer_email
+
+        case "cfp":
+            from_email = config.from_email("CONTENT_EMAIL")
+            recipient = rec.user
+
+        case "cfp_speakers":
+            from_email = config.from_email("SPEAKERS_EMAIL")
+            recipient = rec.user
+
+        case "bulk_contact":
+            from_email = config.from_email("CONTACT_EMAIL")
+            recipient = rec.user
+
+    sent_count: int = mail.send_mail(
         subject=rec.job.subject,
         message=rec.job.text_body,
-        from_email=config.from_email("VOLUNTEER_EMAIL"),
-        recipient_list=[rec.volunteer.volunteer_email],
+        html_message=rec.job.html_body,
+        from_email=from_email,
+        recipient_list=[recipient],
         fail_silently=True,
         connection=conn,
-        html_message=rec.job.html_body,
     )
     if sent_count > 0:
         rec.sent = True
+        rec.sent_at = naive_utcnow()
         db.session.commit()
+
     return sent_count

apps/cfp_review/email.py

@@ -6,11 +6,10 @@
     current_app as app,
 )
 from flask import render_template
-from flask_mailman import EmailMessage
 
+from apps.common.email import enqueue_emails
 from models.content import Proposal
-
-from ..config import config
+from models.email import EmailJobType
 
 ProposalEmailReason = Literal[
     "accepted",
@@ -25,8 +24,8 @@
 
 
 def send_email_for_proposal(proposal: Proposal, reason: ProposalEmailReason) -> None:
-    from_email_ = config.from_email("CONTENT_EMAIL")
     title = (proposal.schedule_item and proposal.schedule_item.title) or proposal.title
+    email_type: EmailJobType = "cfp"
 
     if reason == "accepted":
         subject = f'''Your EMF {proposal.human_type} "{title}" has been accepted!'''
@@ -44,43 +43,47 @@ def send_email_for_proposal(proposal: Proposal, reason: ProposalEmailReason) ->
     elif reason == "reserve-list":
         subject = f'''Your EMF {proposal.human_type} "{title}", and EMF tickets'''
         template = "emails/cfp-reserve-list.txt"
-        from_email_ = config.from_email("SPEAKERS_EMAIL")
+        email_type = "cfp_speakers"
 
     elif reason == "please-finalise":
         # Can be sent before or after scheduling. We want them
         # to update for the line-up, so the earlier the better.
         subject = f'''We need information about your EMF {proposal.human_type} "{title}"'''
         template = "emails/cfp-please-finalise.txt"
-        from_email_ = config.from_email("SPEAKERS_EMAIL")
+        email_type = "cfp_speakers"
 
     elif reason == "check-scheduled-duration":
         # This email is basically the same as "please-finalise" but less urgent
         subject = f'''Your EMF {proposal.human_type} "{title}" is ready to schedule, please check your slot'''
         template = "emails/cfp-check-scheduled-duration.txt"
-        from_email_ = config.from_email("SPEAKERS_EMAIL")
+        email_type = "cfp_speakers"
 
     elif reason == "slot-scheduled":
         subject = f'''Your EMF {proposal.human_type} "{title}" has been scheduled'''
         template = "emails/cfp-slot-scheduled.txt"
-        from_email_ = config.from_email("SPEAKERS_EMAIL")
+        email_type = "cfp_speakers"
 
     elif reason == "slot-moved":
         # TODO: might be nice to highlight which slot has moved
         subject = f'''Your EMF {proposal.human_type} slot has been moved ("{title}")'''
         template = "emails/cfp-slot-moved.txt"
-        from_email_ = config.from_email("SPEAKERS_EMAIL")
+        email_type = "cfp_speakers"
 
     else:
         raise Exception(f"Invalid proposal email type {reason}")
 
     app.logger.info("Sending %s email for proposal %s", reason, proposal.id)
 
-    msg = EmailMessage(subject, from_email=from_email_, to=[proposal.user.email])
-    msg.body = render_template(
+    text_body = render_template(
         template,
         user=proposal.user,
         proposal=proposal,
         reserve_ticket_link=app.config["RESERVE_LIST_TICKET_LINK"],
     )
 
-    msg.send()
+    enqueue_emails(
+        type=email_type,
+        users=[proposal.user],
+        subject=subject,
+        text_body=text_body,
+    )

apps/common/email.py

@@ -6,7 +6,8 @@
 from markupsafe import Markup
 
 from main import db, mail
-from models.email import EmailJob, EmailJobRecipient
+from models.email import EmailJob, EmailJobRecipient, EmailJobType
+from models.user import User
 
 from ..config import config as app_config
 
@@ -123,27 +124,28 @@ def format_trusted_plaintext_email(markdown_text, **kwargs):
 def preview_trusted_email(preview_address, subject, body):
     subject = "[PREVIEW] " + subject
     formatted_plaintext = format_trusted_plaintext_email(body)
+    # FIXME: add reason?
     formatted_html = format_trusted_html_email(body, subject)
 
     mail.send_mail(
         subject=subject,
         message=formatted_plaintext,
+        html_message=formatted_html,
         from_email=app_config.from_email("CONTACT_EMAIL"),
         recipient_list=[preview_address],
-        html_message=formatted_html,
     )
 
 
-def enqueue_trusted_emails(users, subject, body, **kwargs):
+def enqueue_emails(
+    type: EmailJobType, users: list[User], subject: str, text_body: str, html_body: Markup | None = None
+) -> None:
     """Queue an email for sending by the background email worker."""
     job = EmailJob(
-        subject,
-        format_trusted_plaintext_email(body, **kwargs),
-        format_trusted_html_email(body, subject, **kwargs),
+        type=type,
+        subject=subject,
+        text_body=text_body,
+        html_body=html_body,
     )
     db.session.add(job)
-
     for user in users:
-        db.session.add(EmailJobRecipient(job, user))
-
-    db.session.commit()
+        db.session.add(EmailJobRecipient(job=job, user=user))