refactor(messaging): unificar envio de e-mails via EmailConsumer e remover BackgroundTasks redundantes

app/api/routers/auth.py

@@ -4,7 +4,6 @@
 
 from fastapi import (
     APIRouter,
-    BackgroundTasks,
     Depends,
     Request,
     Response,
@@ -187,13 +186,11 @@ async def logout(
 async def request_password_reset(
     data: PasswordResetRequest,
     request: Request,
-    background_tasks: BackgroundTasks,
+    db: SessionDep,
 ) -> dict[str, str]:
     client_ip = request.client.host if request.client else None
-    background_tasks.add_task(
-        auth_service.request_password_reset_bg,
-        data.email,
-        client_ip=client_ip,
+    await auth_service.request_password_reset(
+        db, data.email, client_ip=client_ip
     )
     return {
         'message': 'If the e-mail is registered, a password reset link has been sent.'
@@ -233,10 +230,12 @@ async def verify_email(
 )
 async def resend_verification(
     data: ResendVerificationRequest,
-    background_tasks: BackgroundTasks,
+    request: Request,
+    db: SessionDep,
 ) -> dict[str, str]:
-    background_tasks.add_task(
-        auth_service.resend_verification_email_bg, data.email
+    client_ip = request.client.host if request.client else None
+    await auth_service.resend_verification_email(
+        db, data.email, client_ip=client_ip
     )
     return {
         'message': 'If the e-mail is registered and unverified, a new link has been sent.'

app/messaging/consumers/email_consumer.py

@@ -10,6 +10,7 @@
 from app.messaging.events.user_events import UserEvents
 from app.schemas.notification import (
     EmailNotification,
+    EmailVerificationRequestedPayload,
     EmailVerifiedPayload,
     PasswordChangedPayload,
     PasswordResetCompletedPayload,
@@ -69,6 +70,10 @@ async def subscribe(self) -> None:
             AuthEvents.PASSWORD_RESET_COMPLETED,
             self._handle_password_reset_completed,
         )
+        await bus.subscribe(
+            AuthEvents.EMAIL_VERIFICATION_REQUESTED,
+            self._handle_email_verification_requested,
+        )
 
         self._subscribed = True
         logger.info('EmailConsumer subscribed to domain events')
@@ -102,6 +107,10 @@ async def unsubscribe(self) -> None:
             AuthEvents.PASSWORD_RESET_COMPLETED,
             self._handle_password_reset_completed,
         )
+        await bus.unsubscribe(
+            AuthEvents.EMAIL_VERIFICATION_REQUESTED,
+            self._handle_email_verification_requested,
+        )
 
         self._subscribed = False
         logger.info('EmailConsumer unsubscribed from domain events')
@@ -145,11 +154,20 @@ async def _handle_email_verified(self, event: Event) -> None:
         logger.info('Email verified for user %s', payload.user_id)
 
     async def _handle_password_reset_requested(self, event: Event) -> None:
+        from app.core.config import get_settings
+        from app.services import email_service
+
         payload = PasswordResetRequestedPayload(**event.payload)
         logger.info(
-            'Password reset requested event logged for user %s',
+            'Sending password reset email for user %s',
             payload.user_id,
         )
+        settings = get_settings()
+        base_url = getattr(settings, 'APP_BASE_URL', 'http://localhost:8000')
+        reset_link = f'{base_url}/auth/password-reset/confirm?token={payload.reset_token}'
+        await email_service.send_password_reset_email(
+            payload.email, reset_link
+        )
 
     async def _handle_password_reset_completed(self, event: Event) -> None:
         payload = PasswordResetCompletedPayload(**event.payload)
@@ -160,6 +178,22 @@ async def _handle_password_reset_completed(self, event: Event) -> None:
         email = create_password_reset_completed_email(payload)
         await self._send_email(email)
 
+    async def _handle_email_verification_requested(self, event: Event) -> None:
+        from app.core.config import get_settings
+        from app.services import email_service
+
+        payload = EmailVerificationRequestedPayload(**event.payload)
+        logger.info(
+            'Sending email verification link for user %s',
+            payload.user_id,
+        )
+        settings = get_settings()
+        base_url = getattr(settings, 'APP_BASE_URL', 'http://localhost:8000')
+        verify_link = (
+            f'{base_url}/auth/verify-email?token={payload.verify_token}'
+        )
+        await email_service.send_verification_email(payload.email, verify_link)
+
     async def _send_email(self, email: EmailNotification) -> None:
         from app.services.email_service import _send_via_smtp, render_template
 

app/messaging/events/__init__.py

@@ -2,6 +2,7 @@
 
 from app.messaging.events.auth_events import (
     AuthEvents,
+    EmailVerificationRequestedEvent,
     PasswordResetCompletedEvent,
     PasswordResetRequestedEvent,
 )
@@ -17,6 +18,7 @@
 
 __all__ = [
     'AuthEvents',
+    'EmailVerificationRequestedEvent',
     'EmailVerifiedEvent',
     'PasswordChangedEvent',
     'PasswordResetCompletedEvent',

app/messaging/events/auth_events.py

@@ -16,6 +16,7 @@ class AuthEvents:
     TOKEN_REFRESHED = 'auth.token_refreshed'
     PASSWORD_RESET_REQUESTED = 'auth.password_reset_requested'
     PASSWORD_RESET_COMPLETED = 'auth.password_reset_completed'
+    EMAIL_VERIFICATION_REQUESTED = 'auth.email_verification_requested'
     ACCOUNT_TEMPORARILY_LOCKED = 'auth.account_temporarily_locked'
 
 
@@ -31,3 +32,12 @@ class PasswordResetCompletedEvent(DomainEvent):
     """Event emitted when a password reset is completed successfully."""
 
     type: str = field(default=AuthEvents.PASSWORD_RESET_COMPLETED, init=False)
+
+
+@dataclass
+class EmailVerificationRequestedEvent(DomainEvent):
+    """Event emitted when an email verification link is requested."""
+
+    type: str = field(
+        default=AuthEvents.EMAIL_VERIFICATION_REQUESTED, init=False
+    )

app/schemas/notification.py

@@ -201,6 +201,18 @@ class PasswordResetRequestedPayload(BaseModel):
     client_ip: str | None = None
 
 
+class EmailVerificationRequestedPayload(BaseModel):
+    """Payload for auth.email_verification_requested event."""
+
+    model_config = ConfigDict(extra='forbid')
+
+    user_id: str
+    email: EmailStr
+    verify_token: str
+    expires_at: datetime
+    client_ip: str | None = None
+
+
 class PasswordResetCompletedPayload(BaseModel):
     """Payload for auth.password_reset_completed event."""
 

app/services/auth_service.py

@@ -36,7 +36,6 @@
     hash_password_async,
 )
 from app.core.security.security_logger import log_security_event
-from app.db.session import get_session_factory
 from app.messaging import Event
 from app.messaging.buses import get_event_bus
 from app.messaging.events import AuthEvents, UserEvents
@@ -368,11 +367,6 @@ async def request_password_reset(
         'PASSWORD_RESET_REQUESTED', user_id=user.id, ip=client_ip
     )
 
-    reset_link = (
-        f'{settings.APP_BASE_URL}/auth/password-reset/confirm?token={token}'
-    )
-    await email_service.send_password_reset_email(user.email, reset_link)
-
     # Publish auth.password_reset_requested event
     bus = get_event_bus()
 
@@ -382,22 +376,14 @@ async def request_password_reset(
             payload={
                 'user_id': user.id,
                 'email': user.email,
+                'reset_token': token,
                 'expires_at': expires_at.isoformat(),
                 'client_ip': client_ip,
             },
         )
     )
 
 
-async def request_password_reset_bg(
-    email: str, *, client_ip: str | None = None
-) -> None:
-    """Background task entry point for requesting password reset with an isolated DB session."""
-    session_factory = get_session_factory()
-    async with session_factory() as session:
-        await request_password_reset(session, email, client_ip=client_ip)
-
-
 async def reset_password(
     db: AsyncSession,
     token: str,
@@ -516,9 +502,9 @@ async def verify_email(
 
 
 async def send_verification_email_for_user(
-    db: AsyncSession, user: User
+    db: AsyncSession, user: User, *, client_ip: str | None = None
 ) -> None:
-    """Create a verification token and e-mail it to the user (if unverified)."""
+    """Create a verification token and publish email verification event."""
     settings = get_settings()
 
     if user.is_verified:
@@ -535,22 +521,29 @@ async def send_verification_email_for_user(
     )
     await db.commit()
 
-    verify_link = f'{settings.APP_BASE_URL}/auth/verify-email?token={token}'
-    await email_service.send_verification_email(user.email, verify_link)
+    # Publish auth.email_verification_requested event
+    bus = get_event_bus()
+    await bus.publish(
+        Event(
+            type=AuthEvents.EMAIL_VERIFICATION_REQUESTED,
+            payload={
+                'user_id': user.id,
+                'email': user.email,
+                'verify_token': token,
+                'expires_at': expires_at.isoformat(),
+                'client_ip': client_ip,
+            },
+        )
+    )
 
 
-async def resend_verification_email(db: AsyncSession, email: str) -> None:
+async def resend_verification_email(
+    db: AsyncSession, email: str, *, client_ip: str | None = None
+) -> None:
     """Send a fresh verification link to a registered, unverified user."""
     user_repository = UserRepository(db)
     user = await user_repository.get_by_email(email)
     if user is None or user.is_verified or not user.is_active:
         return
 
-    await send_verification_email_for_user(db, user)
-
-
-async def resend_verification_email_bg(email: str) -> None:
-    """Background task entry point for resending verification email with an isolated DB session."""
-    session_factory = get_session_factory()
-    async with session_factory() as session:
-        await resend_verification_email(session, email)
+    await send_verification_email_for_user(db, user, client_ip=client_ip)

app/services/notification_service.py

@@ -10,6 +10,7 @@
 from app.messaging.events import AuthEvents, UserEvents
 from app.schemas.notification import (
     EmailNotification,
+    EmailVerificationRequestedPayload,
     EmailVerifiedPayload,
     NotificationPriority,
     PasswordChangedPayload,
@@ -145,6 +146,30 @@ def create_password_reset_email(
     )
 
 
+def create_verification_requested_email(
+    verification: EmailVerificationRequestedPayload,
+) -> EmailNotification:
+    """Create email verification notification email."""
+    settings = get_settings()
+    base_url = getattr(settings, 'APP_BASE_URL', 'http://localhost:8000')
+    verify_link = (
+        f'{base_url}/auth/verify-email?token={verification.verify_token}'
+    )
+    return EmailNotification(
+        notification_id=f'email-verification-{verification.user_id}',
+        priority=NotificationPriority.HIGH,
+        recipient_id=verification.user_id,
+        to_email=verification.email,
+        subject='Verify your email',
+        template_id='account_created',
+        template_data={
+            'verify_link': verify_link,
+            'expires_at': verification.expires_at.isoformat(),
+            'client_ip': verification.client_ip,
+        },
+    )
+
+
 def create_password_reset_completed_email(
     reset: PasswordResetCompletedPayload,
 ) -> EmailNotification:

tests/conftest.py

@@ -105,15 +105,19 @@ async def _clear_database_session_cache() -> AsyncIterator[None]:
 
 @pytest_asyncio.fixture(autouse=True)
 async def _setup_test_notifications() -> AsyncIterator[None]:
+    from app.messaging.consumers import get_email_consumer
     from app.services.notification_service import (
         setup_notifications,
         teardown_notifications,
     )
 
+    email_consumer = get_email_consumer()
+    await email_consumer.subscribe()
     await setup_notifications()
     try:
         yield
     finally:
+        await email_consumer.unsubscribe()
         await teardown_notifications()
 
 
@@ -151,9 +155,6 @@ async def isolated_session_factory(
     )
 
     monkeypatch.setattr('app.db.session.get_session_factory', lambda: factory)
-    monkeypatch.setattr(
-        'app.services.auth_service.get_session_factory', lambda: factory
-    )
 
     try:
         yield factory

tests/test_events_resilience.py

@@ -103,3 +103,53 @@ async def _handler(event: dict) -> None:
     # DLQ should now be empty
     dlq_after = await bus.get_dlq_events()
     assert len(dlq_after) == 0
+
+
+@pytest.mark.asyncio
+async def test_email_consumer_handles_auth_events() -> None:
+    """Test EmailConsumer processes password reset and email verification requested events."""
+    from datetime import UTC, datetime
+    from unittest.mock import AsyncMock, patch
+
+    from app.messaging.consumers.email_consumer import EmailConsumer
+    from app.messaging.events import AuthEvents
+
+    consumer = EmailConsumer()
+
+    # 1. Password reset requested
+    reset_event = Event(
+        type=AuthEvents.PASSWORD_RESET_REQUESTED,
+        payload={
+            'user_id': 'usr-123',
+            'email': 'user@example.com',
+            'reset_token': 'test-token-xyz',
+            'expires_at': datetime.now(UTC).isoformat(),
+        },
+    )
+    with patch(
+        'app.services.email_service.send_password_reset_email',
+        new_callable=AsyncMock,
+    ) as mock_reset_email:
+        await consumer._handle_password_reset_requested(reset_event)
+        assert mock_reset_email.called
+        assert mock_reset_email.call_args[0][0] == 'user@example.com'
+        assert 'token=test-token-xyz' in mock_reset_email.call_args[0][1]
+
+    # 2. Email verification requested
+    verify_event = Event(
+        type=AuthEvents.EMAIL_VERIFICATION_REQUESTED,
+        payload={
+            'user_id': 'usr-123',
+            'email': 'user@example.com',
+            'verify_token': 'verify-token-abc',
+            'expires_at': datetime.now(UTC).isoformat(),
+        },
+    )
+    with patch(
+        'app.services.email_service.send_verification_email',
+        new_callable=AsyncMock,
+    ) as mock_verify_email:
+        await consumer._handle_email_verification_requested(verify_event)
+        assert mock_verify_email.called
+        assert mock_verify_email.call_args[0][0] == 'user@example.com'
+        assert 'token=verify-token-abc' in mock_verify_email.call_args[0][1]