fix: improve email queue and send email to group admins

Author: amandazhuyilan

routers/biocommons_register.py

@@ -10,7 +10,7 @@
 from biocommons.bundles import BUNDLES, BiocommonsBundle
 from biocommons.default import DEFAULT_PLATFORMS
 from config import Settings, get_settings
-from db.models import BiocommonsUser
+from db.models import BiocommonsGroup, BiocommonsUser
 from db.setup import get_db_session
 from routers.errors import RegistrationRoute
 from schemas.biocommons import Auth0UserData, BiocommonsRegisterData
@@ -105,15 +105,26 @@ async def register_biocommons_user(
             session=db_session,
             auth0_client=auth0_client
         )
-        # Queue approval email for manual bundles
+        # Queue approval email(s) for manual bundles
         if bundle is not None and not bundle.group_auto_approve:
+            admin_emails: set[str] = set()
+            group_record = BiocommonsGroup.get_by_id(bundle.group_id.value, db_session)
+            if group_record is not None:
+                admin_emails = group_record.get_admins(auth0_client=auth0_client)
+            if not admin_emails:
+                logger.warning(
+                    "No admins found for bundle %s, notifying ops inbox",
+                    bundle.group_id.value,
+                )
+                admin_emails.add("aai-dev@biocommons.org.au")
             subject, body_html = compose_biocommons_registration_email(registration, settings)
-            enqueue_email(
-                db_session,
-                to_address="aai-dev@biocommons.org.au",
-                subject=subject,
-                body_html=body_html,
-            )
+            for email in admin_emails:
+                enqueue_email(
+                    db_session,
+                    to_address=email,
+                    subject=subject,
+                    body_html=body_html,
+                )
 
         db_session.commit()
 

scheduled_tasks/tasks.py

@@ -1,5 +1,6 @@
 import re
 from datetime import datetime, timezone
+from uuid import UUID
 
 from loguru import logger
 from pydantic import BaseModel
@@ -31,6 +32,7 @@
 
 EMAIL_QUEUE_BATCH_SIZE = 25
 EMAIL_RETRY_DELAY_SECONDS = 300
+EMAIL_JOB_ID_PREFIX = "email_notification_"
 
 
 def _ensure_user_from_auth0(session: Session, user_data: Auth0UserData) -> tuple[BiocommonsUser, bool, bool]:
@@ -72,7 +74,7 @@ async def process_email_queue(
     retry_delay_seconds: int = EMAIL_RETRY_DELAY_SECONDS,
 ) -> int:
     """
-    Send pending email notifications stored in the database.
+    Schedule pending email notifications for delivery.
     """
     logger.info("Processing email notification queue")
     session = next(get_db_session())
@@ -96,34 +98,60 @@ async def process_email_queue(
         if not notifications:
             logger.info("No email notifications ready for delivery")
             return 0
+        scheduled = 0
         for notification in notifications:
             notification.mark_sending()
             session.add(notification)
+            session.flush()
+            job_id = f"{EMAIL_JOB_ID_PREFIX}{notification.id}"
+            SCHEDULER.add_job(
+                send_email_notification,
+                args=[notification.id],
+                id=job_id,
+                replace_existing=True,
+                kwargs={"retry_delay_seconds": retry_delay_seconds},
+            )
+            scheduled += 1
         session.commit()
+        logger.info("Queued %d email notifications for delivery", scheduled)
+        return scheduled
+    finally:
+        session.close()
+
+
+async def send_email_notification(
+    notification_id: UUID,
+    retry_delay_seconds: int = EMAIL_RETRY_DELAY_SECONDS,
+) -> bool:
+    """
+    Deliver a single queued email notification.
+    """
+    session = next(get_db_session())
+    try:
+        notification = session.get(EmailNotification, notification_id)
+        if notification is None:
+            logger.warning("Email notification %s not found", notification_id)
+            return False
         email_service = get_email_service()
-        sent_count = 0
-        for notification in notifications:
-            try:
-                email_service.send(
-                    notification.to_address,
-                    notification.subject,
-                    notification.body_html,
-                )
-            except Exception as exc:  # noqa: BLE001
-                logger.warning(
-                    "Failed to send email %s: %s", notification.id, exc
-                )
-                session.refresh(notification)
-                notification.mark_failed(str(exc), retry_delay_seconds)
-                session.add(notification)
-                session.commit()
-            else:
-                session.refresh(notification)
-                notification.mark_sent()
-                session.add(notification)
-                session.commit()
-                sent_count += 1
-        return sent_count
+        try:
+            email_service.send(
+                notification.to_address,
+                notification.subject,
+                notification.body_html,
+            )
+        except Exception as exc:  # noqa: BLE001
+            logger.warning(
+                "Failed to send email %s: %s", notification.id, exc
+            )
+            notification.mark_failed(str(exc), retry_delay_seconds)
+            session.add(notification)
+            session.commit()
+            return False
+        else:
+            notification.mark_sent()
+            session.add(notification)
+            session.commit()
+            return True
     finally:
         session.close()
 

services/email_queue.py

@@ -15,6 +15,8 @@ def enqueue_email(
 ) -> EmailNotification:
     """
     Persist an outbound email so the scheduler can deliver it later.
+
+    Note: the caller is responsible for committing the session after enqueueing.
     """
     notification = EmailNotification(
         to_address=to_address,

tests/scheduled_tasks/test_tasks.py

@@ -21,6 +21,7 @@
     _get_group_membership_including_deleted,
     populate_db_groups,
     process_email_queue,
+    send_email_notification,
     sync_auth0_roles,
     sync_auth0_users,
     sync_group_user_roles,
@@ -94,7 +95,7 @@ async def test_sync_auth0_users_creates_and_soft_deletes(mocker, test_db_session
     mocker.patch.object(test_db_session, "close", return_value=None)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, object()),
     )
 
     await sync_auth0_users()
@@ -127,7 +128,7 @@ async def test_update_auth0_user_updates_existing(test_db_session, mocker, persi
     mocker.patch.object(test_db_session, "close", return_value=None)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, object()),
     )
 
     await update_auth0_user(user_data=user_data)
@@ -145,7 +146,7 @@ async def test_update_auth0_user_creates_when_missing(test_db_session, mocker):
     mocker.patch.object(test_db_session, "close", return_value=None)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, object()),
     )
 
     result = await update_auth0_user(user_data=user_data)
@@ -263,7 +264,7 @@ async def test_sync_auth0_roles_updates_and_soft_deletes(mocker, test_db_session
     mocker.patch.object(test_db_session, "close", return_value=None)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, object()),
     )
 
     await sync_auth0_roles()
@@ -356,7 +357,7 @@ async def test_sync_auth0_group_roles_syncs_assignments(mocker, test_db_session,
     mocker.patch.object(test_db_session, "close", return_value=None)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, None),
     )
 
     await sync_group_user_roles()
@@ -413,7 +414,7 @@ class TestGroups(Enum):
     mocker.patch("scheduled_tasks.tasks.get_settings", return_value=mock_settings)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, None),
     )
 
     BiocommonsGroupFactory.create_sync(group_id=TestGroups.TSI.value)
@@ -560,12 +561,15 @@ async def test_process_email_queue_sends_notifications(test_db_session, mocker):
     mocker.patch("scheduled_tasks.tasks.get_email_service", return_value=mock_service)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, None),
     )
+    mock_scheduler = mocker.patch("scheduled_tasks.tasks.SCHEDULER.add_job")
 
-    sent = await process_email_queue()
+    scheduled = await process_email_queue()
 
-    assert sent == 1
+    assert scheduled == 1
+    mock_scheduler.assert_called_once()
+    await send_email_notification(notification_id)
     updated = test_db_session.get(EmailNotification, notification_id)
     assert updated.status == EmailStatusEnum.SENT
     assert mock_service.send.called
@@ -587,12 +591,15 @@ async def test_process_email_queue_retries_failed_notifications(test_db_session,
     mocker.patch("scheduled_tasks.tasks.get_email_service", return_value=mock_service)
     mocker.patch(
         "scheduled_tasks.tasks.get_db_session",
-        return_value=(test_db_session for _ in range(1)),
+        return_value=iter(lambda: test_db_session, None),
     )
+    mock_scheduler = mocker.patch("scheduled_tasks.tasks.SCHEDULER.add_job")
 
-    sent = await process_email_queue()
+    scheduled = await process_email_queue()
 
-    assert sent == 0
+    assert scheduled == 1
+    mock_scheduler.assert_called_once()
+    await send_email_notification(notification_id)
     updated = test_db_session.get(EmailNotification, notification_id)
     assert updated.status == EmailStatusEnum.FAILED
     assert updated.last_error is not None

tests/test_biocommons_register.py

@@ -12,6 +12,7 @@
 from tests.datagen import (
     Auth0UserDataFactory,
     BiocommonsRegistrationRequestFactory,
+    RoleUserDataFactory,
 )
 from tests.db.datagen import Auth0RoleFactory, BiocommonsGroupFactory, PlatformFactory
 
@@ -258,6 +259,12 @@ def test_successful_biocommons_registration_endpoint(
     """Test successful biocommons registration via HTTP endpoint"""
     auth0_data = Auth0UserDataFactory.build()
     mock_auth0_client.create_user.return_value = auth0_data
+    admin_stub = RoleUserDataFactory.build(email="tsi.admin@example.com")
+    mock_auth0_client.get_all_role_users.return_value = [admin_stub]
+    mock_auth0_client.get_user.return_value = Auth0UserDataFactory.build(
+        user_id=admin_stub.user_id,
+        email=admin_stub.email,
+    )
 
     registration_data = {
         "first_name": "Test",
@@ -291,7 +298,7 @@ def test_successful_biocommons_registration_endpoint(
     queued_emails = test_db_session.exec(select(EmailNotification)).all()
     assert len(queued_emails) == 1
     email = queued_emails[0]
-    assert email.to_address == "aai-dev@biocommons.org.au"
+    assert email.to_address == admin_stub.email
     assert email.status == EmailStatusEnum.PENDING