fix: remove send_email flag and set up email queue in DB

biocommons/bundles.py

@@ -37,15 +37,23 @@ def _add_platform_memberships(self, user: BiocommonsUser, session: Session, auth
             )
             session.add(platform_membership)
 
-    def create_memberships(self, user: BiocommonsUser, auth0_client: Auth0Client, db_session: Session):
+    def create_memberships(
+        self,
+        user: BiocommonsUser,
+        auth0_client: Auth0Client,
+        db_session: Session,
+        commit: bool = False,
+    ):
         """
         Create group and platform memberships for the bundle user
         """
         # Create group membership
         self._add_group_membership(user=user, session=db_session)
         # Add extra platform memberships based on bundle configuration
         self._add_platform_memberships(user=user, session=db_session, auth0_client=auth0_client)
-        db_session.commit()
+        db_session.flush()
+        if commit:
+            db_session.commit()
         return user
 
 

config.py

@@ -16,7 +16,6 @@ class Settings(BaseSettings):
     jwt_secret_key: str
     auth0_algorithms: list[str] = ["RS256"]
     admin_roles: list[str] = []
-    send_email: bool = False
     # Note we process this separately in app startup as it needs
     #   to be available before the app starts
     cors_allowed_origins: str

db/models.py

@@ -1,19 +1,20 @@
 import uuid
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from logging import getLogger
 from typing import Optional, Self
 
 from pydantic import AwareDatetime
-from sqlalchemy import Column, String, UniqueConstraint
+from sqlalchemy import Column, String, Text, UniqueConstraint
 from sqlmodel import DateTime, Field, Relationship, Session, select
 from sqlmodel import Enum as DbEnum
 from starlette.exceptions import HTTPException
 
 import schemas
 from auth0.client import Auth0Client
-from db.core import SoftDeleteModel
+from db.core import BaseModel, SoftDeleteModel
 from db.types import (
     ApprovalStatusEnum,
+    EmailStatusEnum,
     GroupMembershipData,
     PlatformEnum,
     PlatformMembershipData,
@@ -877,10 +878,61 @@ def get_for_admin_roles(cls, role_names: list[str], session: Session) -> list[Se
         ).all()
 
 
+class EmailNotification(BaseModel, table=True):
+    """
+    Stores pending outbound emails for asynchronous processing.
+    """
+    __tablename__ = "emailnotification"
+
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    to_address: str = Field(index=True)
+    subject: str = Field()
+    body_html: str = Field(sa_column=Column(Text(), nullable=False))
+    status: EmailStatusEnum = Field(
+        default=EmailStatusEnum.PENDING,
+        sa_type=DbEnum(EmailStatusEnum, name="EmailStatusEnum"),
+        nullable=False,
+        index=True,
+    )
+    attempts: int = Field(default=0, nullable=False)
+    last_error: str | None = Field(default=None, sa_column=Column(String(1024), nullable=True))
+    send_after: AwareDatetime | None = Field(default=None, sa_type=DateTime(timezone=True))
+    sent_at: AwareDatetime | None = Field(default=None, sa_type=DateTime(timezone=True))
+    created_at: AwareDatetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_type=DateTime(timezone=True),
+    )
+    updated_at: AwareDatetime = Field(
+        default_factory=lambda: datetime.now(timezone.utc),
+        sa_type=DateTime(timezone=True),
+    )
+
+    def mark_sending(self) -> None:
+        self.status = EmailStatusEnum.SENDING
+        self.attempts += 1
+        self.updated_at = datetime.now(timezone.utc)
+
+    def mark_sent(self) -> None:
+        now = datetime.now(timezone.utc)
+        self.status = EmailStatusEnum.SENT
+        self.sent_at = now
+        self.updated_at = now
+        self.last_error = None
+
+    def mark_failed(self, error: str, retry_delay_seconds: int | None = None) -> None:
+        now = datetime.now(timezone.utc)
+        self.status = EmailStatusEnum.FAILED
+        self.last_error = error[:1024]
+        self.updated_at = now
+        if retry_delay_seconds:
+            self.send_after = now + timedelta(seconds=retry_delay_seconds)
+
+
 # Update all model references
 BiocommonsUser.model_rebuild()
 Platform.model_rebuild()
 PlatformMembership.model_rebuild()
 PlatformMembershipHistory.model_rebuild()
 GroupMembership.model_rebuild()
 GroupMembershipHistory.model_rebuild()
+EmailNotification.model_rebuild()

db/types.py

@@ -16,6 +16,13 @@ class ApprovalStatusEnum(str, Enum):
     REVOKED = "revoked"
 
 
+class EmailStatusEnum(str, Enum):
+    PENDING = "pending"
+    SENDING = "sending"
+    SENT = "sent"
+    FAILED = "failed"
+
+
 class PlatformEnum(str, Enum):
     GALAXY = "galaxy"
     BPA_DATA_PORTAL = "bpa_data_portal"

migrations/versions/d64e9ebd0253_new_email_queue.py

@@ -0,0 +1,48 @@
+"""new-email-queue
+
+Revision ID: d64e9ebd0253
+Revises: 4594b458279c
+Create Date: 2025-11-20 14:30:18.888335
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'd64e9ebd0253'
+down_revision: Union[str, None] = '4594b458279c'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('emailnotification',
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('to_address', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('subject', sqlmodel.sql.sqltypes.AutoString(), nullable=False),
+    sa.Column('body_html', sa.Text(), nullable=False),
+    sa.Column('status', sa.Enum('PENDING', 'SENDING', 'SENT', 'FAILED', name='EmailStatusEnum'), nullable=False),
+    sa.Column('attempts', sa.Integer(), nullable=False),
+    sa.Column('last_error', sa.String(length=1024), nullable=True),
+    sa.Column('send_after', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('sent_at', sa.DateTime(timezone=True), nullable=True),
+    sa.Column('created_at', sa.DateTime(timezone=True), nullable=False),
+    sa.Column('updated_at', sa.DateTime(timezone=True), nullable=False),
+    sa.PrimaryKeyConstraint('id', name=op.f('pk_emailnotification'))
+    )
+    op.create_index(op.f('ix_emailnotification_status'), 'emailnotification', ['status'], unique=False)
+    op.create_index(op.f('ix_emailnotification_to_address'), 'emailnotification', ['to_address'], unique=False)
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_index(op.f('ix_emailnotification_to_address'), table_name='emailnotification')
+    op.drop_index(op.f('ix_emailnotification_status'), table_name='emailnotification')
+    op.drop_table('emailnotification')
+    # ### end Alembic commands ###

routers/admin.py

@@ -2,14 +2,13 @@
 from datetime import datetime, timezone
 from typing import Annotated, Any
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 from fastapi.params import Query
 from pydantic import BaseModel, Field, ValidationError, field_validator
 from sqlalchemy import func, or_
 from sqlmodel import Session, select
 from sqlmodel.sql._expression_select_cls import SelectOfScalar
 
-from auth.ses import EmailService, get_email_service
 from auth.user_permissions import (
     get_db_user,
     get_session_user,
@@ -37,9 +36,10 @@
     GroupMembershipData,
     PlatformMembershipData,
 )
-from routers.biocommons_groups import send_group_membership_approved_email
+from routers.biocommons_groups import compose_group_membership_approved_email
 from schemas.biocommons import Auth0UserDataWithMemberships, ServiceIdParam, UserIdParam
 from schemas.user import SessionUser
+from services.email_queue import enqueue_email
 
 logger = logging.getLogger('uvicorn.error')
 
@@ -200,7 +200,7 @@ def _approve_group_membership(
     membership.updated_at = datetime.now(timezone.utc)
     membership.updated_by = admin_record
     membership.grant_auth0_role(auth0_client=client)
-    membership.save(session=db_session, commit=True)
+    membership.save(session=db_session, commit=False)
     db_session.refresh(membership)
     if membership.user is None:
         db_session.refresh(membership, attribute_names=["user"])
@@ -627,9 +627,7 @@ def approve_group_membership(user_id: Annotated[str, UserIdParam],
                              client: Annotated[Auth0Client, Depends(get_auth0_client)],
                              admin_record: Annotated[BiocommonsUser, Depends(get_db_user)],
                              db_session: Annotated[Session, Depends(get_db_session)],
-                             background_tasks: BackgroundTasks,
-                             settings: Annotated[Settings, Depends(get_settings)],
-                             email_service: Annotated[EmailService, Depends(get_email_service)]):
+                             settings: Annotated[Settings, Depends(get_settings)]):
     group_record = BiocommonsGroup.get_by_id_or_404(group_id, db_session)
     membership, status_changed = _approve_group_membership(
         user_id=user_id,
@@ -638,15 +636,19 @@ def approve_group_membership(user_id: Annotated[str, UserIdParam],
         client=client,
         db_session=db_session,
     )
-    if status_changed and settings.send_email and membership.user and membership.user.email:
-        background_tasks.add_task(
-            send_group_membership_approved_email,
-            membership.user.email,
-            group_record.name,
-            group_record.short_name,
-            settings,
-            email_service,
+    if status_changed and membership.user and membership.user.email:
+        subject, body_html = compose_group_membership_approved_email(
+            group_name=group_record.name,
+            group_short_name=group_record.short_name,
+            settings=settings,
         )
+        enqueue_email(
+            db_session,
+            to_address=membership.user.email,
+            subject=subject,
+            body_html=body_html,
+        )
+    db_session.commit()
     return _membership_response()
 
 

routers/biocommons_groups.py

@@ -2,11 +2,10 @@
 from http import HTTPStatus
 from typing import Annotated
 
-from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 from sqlmodel import Session
 
-from auth.ses import EmailService, get_email_service
 from auth.user_permissions import get_session_user
 from auth0.client import Auth0Client, get_auth0_client
 from biocommons.groups import (
@@ -21,6 +20,7 @@
 )
 from db.setup import get_db_session
 from schemas.user import SessionUser
+from services.email_queue import enqueue_email
 
 logger = logging.getLogger('uvicorn.error')
 
@@ -39,8 +39,6 @@ def request_group_access(
         auth0_client: Annotated[Auth0Client, Depends(get_auth0_client)],
         db_session: Annotated[Session, Depends(get_db_session)],
         settings: Annotated[Settings, Depends(get_settings)],
-        email_service: Annotated[EmailService, Depends(get_email_service)],
-        background_tasks: BackgroundTasks
     ):
     """
     Request access to a group. Assumes the user does not already have a
@@ -69,13 +67,18 @@ def request_group_access(
         approval_status=ApprovalStatusEnum.PENDING,
         updated_by=None
     )
-    membership.save(session=db_session, commit=True)
-    if settings.send_email:
-        logger.info("Sending emails to group admins for approval")
-        admin_emails = membership.group.get_admins(auth0_client=auth0_client)
-        for email in admin_emails:
-            background_tasks.add_task(send_group_approval_email,
-                                      approver_email=email, request=membership, email_service=email_service, settings=settings)
+    membership.save(session=db_session, commit=False)
+    logger.info("Queueing emails to group admins for approval")
+    admin_emails = membership.group.get_admins(auth0_client=auth0_client)
+    for email in admin_emails:
+        subject, body_html = compose_group_approval_email(request=membership, settings=settings)
+        enqueue_email(
+            db_session,
+            to_address=email,
+            subject=subject,
+            body_html=body_html,
+        )
+    db_session.commit()
     return {"message": f"Group membership for {group_id} requested successfully."}
 
 
@@ -90,9 +93,7 @@ def approve_group_access(
     approving_user: Annotated[SessionUser, Depends(get_session_user)],
     db_session: Annotated[Session, Depends(get_db_session)],
     auth0_client: Annotated[Auth0Client, Depends(get_auth0_client)],
-    background_tasks: BackgroundTasks,
     settings: Annotated[Settings, Depends(get_settings)],
-    email_service: Annotated[EmailService, Depends(get_email_service)],
 ):
     group = BiocommonsGroup.get_by_id(data.group_id, db_session)
     is_admin = group.user_is_admin(approving_user)
@@ -115,47 +116,43 @@ def approve_group_access(
     membership.approval_status = ApprovalStatusEnum.APPROVED
     membership.updated_by = approving_user_record
     membership.grant_auth0_role(auth0_client=auth0_client)
-    membership.save(session=db_session, commit=True)
+    membership.save(session=db_session, commit=False)
     if membership.user is None:
         db_session.refresh(membership, attribute_names=["user"])
-    if settings.send_email and membership.user and membership.user.email:
-        background_tasks.add_task(
-            send_group_membership_approved_email,
-            membership.user.email,
-            group.name,
-            group.short_name,
-            settings,
-            email_service,
+    if membership.user and membership.user.email:
+        subject, body_html = compose_group_membership_approved_email(
+            group_name=group.name,
+            group_short_name=group.short_name,
+            settings=settings,
+        )
+        enqueue_email(
+            db_session,
+            to_address=membership.user.email,
+            subject=subject,
+            body_html=body_html,
         )
+    db_session.commit()
     return {"message": f"Group membership for {group.name} approved successfully."}
 
 
-def send_group_approval_email(approver_email: str, request: GroupMembership, email_service: EmailService, settings: Settings):
+def compose_group_approval_email(request: GroupMembership, settings: Settings) -> tuple[str, str]:
     subject = f"New request to join {request.group.name}"
-
     body_html = f"""
         <p>A new user has requested access to the {request.group.name} group.</p>
         <p><strong>User:</strong> {request.user.email}</p>
         <p>Please <a href='{settings.aai_portal_url}/requests'>log into the BioCommons account dashboard</a> to review and approve access.</p>
     """
+    return subject, body_html
 
-    email_service.send(approver_email, subject, body_html)
 
-
-def send_group_membership_approved_email(
-    recipient_email: str,
+def compose_group_membership_approved_email(
     group_name: str,
     group_short_name: str,
     settings: Settings,
-    email_service: EmailService,
-):
+) -> tuple[str, str]:
     """
     Notify a user that their group/bundle access was approved.
     """
-    if not recipient_email:
-        logger.warning("Skipping group approval email due to missing recipient email")
-        return
-
     short_name = group_short_name or group_name
     portal_url = settings.aai_portal_url.rstrip("/")
     subject = f"Access approved for {short_name}"
@@ -165,4 +162,4 @@ def send_group_membership_approved_email(
         <p>You now have access to all services included with this bundle. Sign in to the <a href="{portal_url}">AAI Portal</a> to review the bundle details and launch its platforms.</p>
         <p>If you have any questions, please reply to this email.</p>
     """
-    email_service.send(recipient_email, subject, body_html)
+    return subject, body_html