feat(auth): add modular signup endpoint with language enum and standa…

.github/workflows/ci.yml

@@ -31,4 +31,4 @@ jobs:
           DATABASE_URL: postgresql://postgres:postgres@localhost:5432/fluentmeet_test
           REDIS_URL: redis://localhost:6379/1
         run: |
-          pytest --cov=app --cov-fail-under=60 tests/
+          pytest --cov=app --cov-fail-under=80 tests/

alembic/versions/11781e907181_initial_migration.py

@@ -29,9 +29,9 @@ def upgrade() -> None:
         sa.Column("full_name", sa.String(length=255), nullable=True),
         sa.Column("is_active", sa.Boolean(), nullable=False),
         sa.Column("is_verified", sa.Boolean(), nullable=False),
-        sa.Column("created_at", sa.DateTime(), nullable=False),
-        sa.Column("updated_at", sa.DateTime(), nullable=False),
-        sa.Column("deleted_at", sa.DateTime(), nullable=True),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("deleted_at", sa.DateTime(timezone=True), nullable=True),
         sa.Column("speaking_language", sa.String(length=10), nullable=False),
         sa.Column("listening_language", sa.String(length=10), nullable=False),
         sa.PrimaryKeyConstraint("id"),

app/api/v1/__init__.py

@@ -0,0 +1,3 @@
+from app.api.v1.api import api_router
+
+__all__ = ["api_router"]

app/api/v1/api.py

@@ -0,0 +1,6 @@
+from fastapi import APIRouter
+
+from app.api.v1.endpoints.auth import router as auth_router
+
+api_router = APIRouter()
+api_router.include_router(auth_router)

app/api/v1/endpoints/auth.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter, Depends, status
+from sqlalchemy.orm import Session
+
+from app.crud.user import create_user
+from app.db.session import get_db
+from app.schemas.auth import SignupResponse
+from app.schemas.user import UserCreate
+
+router = APIRouter(prefix="/auth", tags=["auth"])
+DB_SESSION_DEPENDENCY = Depends(get_db)
+
+
+@router.post(
+    "/signup",
+    response_model=SignupResponse,
+    status_code=status.HTTP_201_CREATED,
+)
+def signup(user_in: UserCreate, db: Session = DB_SESSION_DEPENDENCY) -> SignupResponse:
+    user = create_user(db=db, user_in=user_in)
+    return SignupResponse.model_validate(user)

app/crud/__init__.py

@@ -0,0 +1,3 @@
+from app.crud.user import create_user, get_user_by_email
+
+__all__ = ["create_user", "get_user_by_email"]

app/crud/user.py

@@ -0,0 +1,49 @@
+from typing import cast
+
+import bcrypt
+from passlib.context import CryptContext
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core.exceptions import ConflictException
+from app.models.user import User
+from app.schemas.user import UserCreate
+
+pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+
+def hash_password(password: str) -> str:
+    try:
+        return cast(str, pwd_context.hash(password))
+    except ValueError:
+        # Passlib's bcrypt backend probing can fail with newer bcrypt builds.
+        salt = bcrypt.gensalt()
+        return bcrypt.hashpw(password.encode("utf-8"), salt).decode("utf-8")
+
+
+def get_user_by_email(db: Session, email: str) -> User | None:
+    statement = select(User).where(User.email == email.lower())
+    return db.execute(statement).scalar_one_or_none()
+
+
+def create_user(db: Session, user_in: UserCreate) -> User:
+    existing_user = get_user_by_email(db, user_in.email)
+    if existing_user:
+        raise ConflictException(
+            code="EMAIL_ALREADY_REGISTERED",
+            message="An account with this email already exists.",
+        )
+
+    db_user = User(
+        email=user_in.email.lower(),
+        hashed_password=hash_password(user_in.password),
+        full_name=user_in.full_name,
+        speaking_language=user_in.speaking_language.value,
+        listening_language=user_in.listening_language.value,
+        is_active=True,
+        is_verified=False,
+    )
+    db.add(db_user)
+    db.commit()
+    db.refresh(db_user)
+    return db_user

app/db/__init__.py

@@ -0,0 +1,3 @@
+from app.db.session import SessionLocal, engine, get_db
+
+__all__ = ["SessionLocal", "engine", "get_db"]

app/db/session.py

@@ -0,0 +1,19 @@
+from collections.abc import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+
+from app.core.config import settings
+
+DATABASE_URL = settings.DATABASE_URL or "sqlite:///./fluentmeet.db"
+
+engine = create_engine(DATABASE_URL, pool_pre_ping=True)
+SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)
+
+
+def get_db() -> Generator[Session, None, None]:
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/kafka/schemas.py

@@ -1,5 +1,5 @@
 import uuid
-from datetime import datetime
+from datetime import UTC, datetime
 from typing import Any, Generic, TypeVar
 
 from pydantic import BaseModel, Field
@@ -14,7 +14,7 @@ class BaseEvent(BaseModel, Generic[T]):
 
     event_id: uuid.UUID = Field(default_factory=uuid.uuid4)
     event_type: str
-    timestamp: datetime = Field(default_factory=datetime.utcnow)
+    timestamp: datetime = Field(default_factory=lambda: datetime.now(UTC))
     payload: T
 
 
@@ -28,7 +28,7 @@ class DLQEvent(BaseModel):
     original_topic: str
     original_event: dict[str, Any]
     error_message: str
-    failed_at: datetime = Field(default_factory=datetime.utcnow)
+    failed_at: datetime = Field(default_factory=lambda: datetime.now(UTC))
     retry_count: int
 
 

app/main.py

@@ -4,6 +4,7 @@
 from fastapi import FastAPI
 from fastapi.middleware.cors import CORSMiddleware
 
+from app.api.v1 import api_router
 from app.core.config import settings
 from app.core.exception_handlers import register_exception_handlers
 from app.kafka.manager import get_kafka_manager
@@ -36,6 +37,7 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
 )
 
 register_exception_handlers(app)
+app.include_router(api_router, prefix=settings.API_V1_STR)
 
 
 @app.get("/health", tags=["health"])

app/models/user.py

@@ -1,4 +1,4 @@
-from datetime import datetime
+from datetime import UTC, datetime
 
 from sqlalchemy import Boolean, DateTime, String
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
@@ -8,6 +8,10 @@ class Base(DeclarativeBase):
     pass
 
 
+def utc_now() -> datetime:
+    return datetime.now(UTC)
+
+
 class User(Base):
     __tablename__ = "users"
 
@@ -19,11 +23,15 @@ class User(Base):
     full_name: Mapped[str | None] = mapped_column(String(255), nullable=True)
     is_active: Mapped[bool] = mapped_column(Boolean, default=True)
     is_verified: Mapped[bool] = mapped_column(Boolean, default=False)
-    created_at: Mapped[datetime] = mapped_column(DateTime, default=datetime.utcnow)
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=utc_now
+    )
     updated_at: Mapped[datetime] = mapped_column(
-        DateTime, default=datetime.utcnow, onupdate=datetime.utcnow
+        DateTime(timezone=True), default=utc_now, onupdate=utc_now
+    )
+    deleted_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
     )
-    deleted_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True)
 
     # Language preferences
     speaking_language: Mapped[str] = mapped_column(String(10), default="en")

app/schemas/auth.py

@@ -0,0 +1,5 @@
+from app.schemas.user import UserResponse
+
+
+class SignupResponse(UserResponse):
+    """Public payload returned by the signup endpoint."""

app/schemas/user.py

@@ -1,34 +1,64 @@
 from datetime import datetime
+from enum import StrEnum
 
-from pydantic import BaseModel, EmailStr, Field
+from pydantic import BaseModel, ConfigDict, EmailStr, Field, field_validator
+
+
+class SupportedLanguage(StrEnum):
+    ENGLISH = "en"
+    FRENCH = "fr"
+    GERMAN = "de"
+    SPANISH = "es"
+    ITALIAN = "it"
+    PORTUGUESE = "pt"
 
 
 class UserBase(BaseModel):
     email: EmailStr
-    full_name: str | None = None
-    speaking_language: str = "en"
-    listening_language: str = "en"
+    full_name: str | None = Field(default=None, max_length=255)
+    speaking_language: SupportedLanguage = SupportedLanguage.ENGLISH
+    listening_language: SupportedLanguage = SupportedLanguage.ENGLISH
+
+    @field_validator("email", mode="before")
+    @classmethod
+    def normalize_email(cls, value: str) -> str:
+        return value.strip().lower()
+
+    @field_validator("full_name", mode="before")
+    @classmethod
+    def strip_full_name(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        stripped_value = value.strip()
+        return stripped_value or None
 
 
 class UserCreate(UserBase):
     password: str = Field(..., min_length=8)
 
 
 class UserUpdate(BaseModel):
-    full_name: str | None = None
-    speaking_language: str | None = None
-    listening_language: str | None = None
+    full_name: str | None = Field(default=None, max_length=255)
+    speaking_language: SupportedLanguage | None = None
+    listening_language: SupportedLanguage | None = None
     password: str | None = Field(None, min_length=8)
 
+    @field_validator("full_name", mode="before")
+    @classmethod
+    def strip_full_name(cls, value: str | None) -> str | None:
+        if value is None:
+            return None
+        stripped_value = value.strip()
+        return stripped_value or None
+
 
 class UserResponse(UserBase):
     id: int
     is_active: bool
     is_verified: bool
     created_at: datetime
 
-    class Config:
-        from_attributes = True
+    model_config = ConfigDict(from_attributes=True)
 
 
 class Token(BaseModel):

docs/email_service.md

@@ -0,0 +1,68 @@
+### Feature: Implement Mailgun Email Service via Kafka
+
+**Problem**
+The FluentMeet application needs to send transactional emails for user account verification, password reset, and other notifications. Currently, no email service is integrated. Directly calling a third-party email API synchronously inside request handlers would increase latency and couple the request lifecycle to an external service, creating a poor user experience if the provider is slow or unavailable.
+
+**Proposed Solution**
+Integrate Mailgun as the email provider and decouple email sending from the HTTP request lifecycle using Apache Kafka. When an email needs to be sent, the application will publish a structured message to a Kafka topic (`notifications.email`). A dedicated consumer worker will pick up the message and dispatch it via the Mailgun REST API. This approach makes email sending asynchronous, resilient, and independently scalable.
+
+**User Stories**
+*   **As a new user,** I want to receive a verification email immediately after signing up, so I can activate my account without experiencing delays in the registration response.
+*   **As a user,** I want to receive a password reset email when I request one, so I can regain access to my account.
+*   **As a developer,** I want a reusable, Kafka-backed email service so that any part of the system can trigger an email without being blocked by the Mailgun API call.
+*   **As a DevOps engineer,** I want email failures to be retried automatically and logged clearly, so transient Mailgun outages don't result in silently dropped emails.
+
+**Acceptance Criteria**
+1.  A `Mailgun` configuration block (API key, domain, sender address) is defined in `app/core/config.py` and sourced from environment variables — never hardcoded.
+2.  A Kafka topic `notifications.email` is created and documented in the infrastructure setup.
+3.  An `EmailProducerService` is implemented with a `send_email(to, subject, html_body, template_data)` method that publishes a structured JSON message to the `notifications.email` Kafka topic.
+4.  An `EmailConsumerWorker` is implemented to:
+    *   Consume messages from the `notifications.email` topic.
+    *   Call the Mailgun REST API (`/messages`) to deliver the email.
+    *   Log success and failure outcomes.
+    *   Handle retries on transient failures using Kafka consumer group offsets.
+5.  Email templates are implemented for:
+    *   **Account Verification**: Contains the verification link.
+    *   **Password Reset**: Contains the time-limited reset link.
+6.  The `EmailProducerService` is injected into and called from the user registration and password reset flows.
+7.  Unit tests verify that the producer publishes the correct message payload to the Kafka topic.
+8.  Integration tests verify the full flow: producer publishes → consumer dispatches → Mailgun API is called.
+
+**Proposed Technical Details**
+*   **Mailgun SDK**: Use the `mailgun2` library (already in `requirements.txt`) or direct `httpx` calls to the Mailgun `/messages` endpoint.
+*   **Kafka Topic**: `notifications.email` — messages follow a standard envelope:
+    ```json
+    {
+      "to": "user@example.com",
+      "subject": "Verify your FluentMeet account",
+      "template": "verification",
+      "data": { "verification_link": "https://..." }
+    }
+    ```
+*   **Producer**: `app/services/email_producer.py` — uses `aiokafka.AIOKafkaProducer` to publish messages asynchronously.
+*   **Consumer Worker**: `app/services/email_consumer.py` — long-running `aiokafka.AIOKafkaConsumer` in a background task, started via FastAPI `lifespan` events.
+*   **Templates**: Jinja2 HTML templates stored in `app/templates/email/` (e.g., `verification.html`, `password_reset.html`).
+*   **Config**: New fields in `app/core/config.py`:
+    ```python
+    MAILGUN_API_KEY: str
+    MAILGUN_DOMAIN: str
+    MAILGUN_FROM_ADDRESS: str = "no-reply@fluentmeet.com"
+    ```
+
+**Tasks**
+- [ ] Add `MAILGUN_API_KEY`, `MAILGUN_DOMAIN`, and `MAILGUN_FROM_ADDRESS` to `.env.example` and `app/core/config.py`.
+- [ ] Create the `notifications.email` Kafka topic and document it in `infra/`.
+- [ ] Implement `EmailProducerService` in `app/services/email_producer.py`.
+- [ ] Implement `EmailConsumerWorker` in `app/services/email_consumer.py`.
+- [ ] Register the consumer as a background task in the FastAPI `lifespan` context manager in `app/main.py`.
+- [ ] Create Jinja2 HTML templates for verification and password reset emails.
+- [ ] Integrate the email producer into the user registration endpoint (`POST /api/v1/auth/signup`).
+- [ ] Integrate the email producer into the password reset endpoint (`POST /api/v1/auth/forgot-password`).
+- [ ] Write unit tests for the `EmailProducerService` (mock the Kafka producer).
+- [ ] Write integration tests for the full consumer → Mailgun dispatch flow (mock the Mailgun API).
+
+**Open Questions/Considerations**
+*   What is the retry strategy for failed Mailgun deliveries — dead-letter queue or fixed retry count?
+*   Should email sending failures be surfaced to the user (e.g., "email failed to send, please try again") or handled silently with a background retry?
+*   What is the Kafka consumer group ID for the email worker, and how should it be managed across deployments?
+*   Should we implement a resend-verification endpoint for users whose verification tokens have expired?