feat(auth): add email verification and resend verification flow

.env.example

@@ -7,6 +7,7 @@ API_V1_STR="/api/v1"
 SECRET_KEY="your-super-secret-key-here"
 ACCESS_TOKEN_EXPIRE_MINUTES=15
 REFRESH_TOKEN_EXPIRE_DAYS=7
+VERIFICATION_TOKEN_EXPIRE_HOURS=24
 
 # Database (PostgreSQL)
 POSTGRES_SERVER=localhost

alembic/versions/4b4b6b5d1c2a_add_verification_tokens_table.py

@@ -0,0 +1,51 @@
+"""add verification tokens table
+
+Revision ID: 4b4b6b5d1c2a
+Revises: 11781e907181
+Create Date: 2026-03-17 11:00:00.000000
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision: str = "4b4b6b5d1c2a"
+down_revision: str | Sequence[str] | None = "11781e907181"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "verification_tokens",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("user_id", sa.Integer(), nullable=False),
+        sa.Column("token", sa.String(length=36), nullable=False),
+        sa.Column("expires_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("created_at", sa.DateTime(timezone=True), nullable=False),
+        sa.ForeignKeyConstraint(["user_id"], ["users.id"]),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_verification_tokens_id"),
+        "verification_tokens",
+        ["id"],
+        unique=False,
+    )
+    op.create_index(
+        op.f("ix_verification_tokens_token"),
+        "verification_tokens",
+        ["token"],
+        unique=True,
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(
+        op.f("ix_verification_tokens_token"), table_name="verification_tokens"
+    )
+    op.drop_index(op.f("ix_verification_tokens_id"), table_name="verification_tokens")
+    op.drop_table("verification_tokens")

app/api/v1/endpoints/auth.py

@@ -1,25 +1,33 @@
 import logging
 from uuid import uuid4
 
-from fastapi import APIRouter, Depends, status
+from fastapi import APIRouter, Depends, Query, Request, status
 from sqlalchemy.orm import Session
 
 from app.core.config import settings
+from app.core.rate_limiter import limiter
 from app.crud.user.user import create_user, get_user_by_email
 from app.db.session import get_db
 from app.schemas.auth import (
     ActionAcknowledgement,
     ForgotPasswordRequest,
+    ResendVerificationRequest,
     SignupResponse,
+    VerifyEmailResponse,
 )
 from app.schemas.user import UserCreate
+from app.services.auth_verification import (
+    AuthVerificationService,
+    get_auth_verification_service,
+)
 from app.services.email_producer import EmailProducerService, get_email_producer_service
 
 logger = logging.getLogger(__name__)
 
 router = APIRouter(prefix="/auth", tags=["auth"])
 DB_SESSION_DEPENDENCY = Depends(get_db)
 EMAIL_PRODUCER_DEPENDENCY = Depends(get_email_producer_service)
+AUTH_VERIFICATION_SERVICE_DEPENDENCY = Depends(get_auth_verification_service)
 
 
 @router.post(
@@ -31,11 +39,18 @@
     user_in: UserCreate,
     db: Session = DB_SESSION_DEPENDENCY,
     email_producer: EmailProducerService = EMAIL_PRODUCER_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
 ) -> SignupResponse:
     user = create_user(db=db, user_in=user_in)
+    verification_token = auth_verification_service.create_verification_token(
+        db=db,
+        user_id=user.id,
+    )
 
     verification_link = (
-        f"{settings.FRONTEND_BASE_URL}/verify-email?user={user.id}&token={uuid4()}"
+        f"{settings.FRONTEND_BASE_URL}/verify-email?token={verification_token.token}"
     )
     try:
         await email_producer.send_email(
@@ -91,3 +106,104 @@
             "password reset instructions."
         )
     )
+
+
+@router.get(
+    "/verify-email",
+    response_model=VerifyEmailResponse,
+    status_code=status.HTTP_200_OK,
+    summary="Verify user email address",
+    description=(
+        "Validates an email verification token, activates the user account, "
+        "and invalidates the token."
+    ),
+    responses={
+        400: {
+            "description": "Missing, invalid, or expired token",
+            "content": {
+                "application/json": {
+                    "examples": {
+                        "missing": {
+                            "value": {
+                                "status": "error",
+                                "code": "MISSING_TOKEN",
+                                "message": "Verification token is required.",
+                                "details": [],
+                            }
+                        },
+                        "invalid": {
+                            "value": {
+                                "status": "error",
+                                "code": "INVALID_TOKEN",
+                                "message": "Verification token is invalid.",
+                                "details": [],
+                            }
+                        },
+                        "expired": {
+                            "value": {
+                                "status": "error",
+                                "code": "TOKEN_EXPIRED",
+                                "message": (
+                                    "Verification token has expired. "
+                                    "Please request a new one."
+                                ),
+                                "details": [],
+                            }
+                        },
+                    }
+                }
+            },
+        }
+    },
+)
+def verify_email(
+    token: str | None = Query(default=None),
+    db: Session = DB_SESSION_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
+) -> VerifyEmailResponse:
+    auth_verification_service.verify_email(db=db, token=token)
+    return VerifyEmailResponse(
+        message="Email successfully verified. You can now log in.",
+    )
+
+
+@router.post(
+    "/resend-verification",
+    response_model=ActionAcknowledgement,
+    status_code=status.HTTP_200_OK,
+    summary="Resend email verification link",
+    description=(
+        "Queues a new verification email when the account exists and is not "
+        "verified. Always returns a generic response to prevent user enumeration."
+    ),
+)
+@limiter.limit("3/minute")
+async def resend_verification(
+    request: Request,
+    payload: ResendVerificationRequest,
+    db: Session = DB_SESSION_DEPENDENCY,
+    email_producer: EmailProducerService = EMAIL_PRODUCER_DEPENDENCY,
+    auth_verification_service: AuthVerificationService = (
+        AUTH_VERIFICATION_SERVICE_DEPENDENCY
+    ),
+) -> ActionAcknowledgement:
+    del request
+    try:
+        await auth_verification_service.resend_verification_email(
+            db=db,
+            email=str(payload.email),
+            email_producer=email_producer,
+        )
+    except Exception as exc:
+        logger.warning(
+            "Failed to enqueue verification resend for %s: %s", payload.email, exc
+        )
+
+    return ActionAcknowledgement(
+        message=(
+            "If an account with that email exists, we have sent a verification "
+            "email."
+        )
+    )

app/core/config.py

@@ -22,6 +22,7 @@ class Settings(BaseSettings):
     SECRET_KEY: str = "placeholder_secret_key"
     ACCESS_TOKEN_EXPIRE_MINUTES: int = 15
     REFRESH_TOKEN_EXPIRE_DAYS: int = 7
+    VERIFICATION_TOKEN_EXPIRE_HOURS: int = 24
 
     # Database
     POSTGRES_SERVER: str = "localhost"

app/core/rate_limiter.py

@@ -0,0 +1,20 @@
+from fastapi import Request
+from fastapi.responses import JSONResponse
+from slowapi import Limiter
+from slowapi.errors import RateLimitExceeded
+from slowapi.util import get_remote_address
+
+from app.core.error_responses import create_error_response
+
+limiter = Limiter(key_func=get_remote_address)
+
+
+async def rate_limit_exception_handler(
+    _request: Request,
+    _exc: RateLimitExceeded,
+) -> JSONResponse:
+    return create_error_response(
+        status_code=429,
+        code="RATE_LIMIT_EXCEEDED",
+        message="Too many requests. Please try again later.",
+    )

app/crud/auth_verification_api.md

@@ -0,0 +1,83 @@
+# Email Verification API
+
+This document describes the email verification endpoints in `app/api/v1/endpoints/auth.py`.
+
+## 1) Verify email
+
+- **Method**: `GET`
+- **Path**: `/api/v1/auth/verify-email`
+- **Auth required**: No
+- **Query params**:
+  - `token` (string UUID)
+
+### Success response
+
+```json
+{
+  "status": "ok",
+  "message": "Email successfully verified. You can now log in."
+}
+```
+
+### Error responses
+
+- Missing token (`400`)
+
+```json
+{
+  "status": "error",
+  "code": "MISSING_TOKEN",
+  "message": "Verification token is required.",
+  "details": []
+}
+```
+
+- Invalid token (`400`)
+
+```json
+{
+  "status": "error",
+  "code": "INVALID_TOKEN",
+  "message": "Verification token is invalid.",
+  "details": []
+}
+```
+
+- Expired token (`400`)
+
+```json
+{
+  "status": "error",
+  "code": "TOKEN_EXPIRED",
+  "message": "Verification token has expired. Please request a new one.",
+  "details": []
+}
+```
+
+## 2) Resend verification
+
+- **Method**: `POST`
+- **Path**: `/api/v1/auth/resend-verification`
+- **Auth required**: No
+- **Rate limit**: `3/minute` per IP
+- **Request body**:
+
+```json
+{
+  "email": "user@example.com"
+}
+```
+
+### Response (`200` for both existing and non-existing emails)
+
+```json
+{
+  "message": "If an account with that email exists, we have sent a verification email."
+}
+```
+
+## Notes
+
+- Signup creates a verification token and queues the verification email through Kafka topic `notifications.email`.
+- Verification tokens are single-use: successful verification deletes the token.
+- Already verified users are handled idempotently by `GET /verify-email` and receive `200`.

app/crud/verification_token.py

@@ -0,0 +1,57 @@
+from datetime import UTC, datetime, timedelta
+
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from app.core.config import settings
+from app.models.verification_token import VerificationToken
+
+
+class VerificationTokenRepository:
+    def get_token(self, db: Session, token: str) -> VerificationToken | None:
+        statement = select(VerificationToken).where(VerificationToken.token == token)
+        return db.execute(statement).scalar_one_or_none()
+
+    def create_token(self, db: Session, user_id: int) -> VerificationToken:
+        expires_at = datetime.now(UTC) + timedelta(
+            hours=settings.VERIFICATION_TOKEN_EXPIRE_HOURS
+        )
+        verification_token = VerificationToken(user_id=user_id, expires_at=expires_at)
+        db.add(verification_token)
+        db.commit()
+        db.refresh(verification_token)
+        return verification_token
+
+    def delete_token(self, db: Session, token_id: int) -> None:
+        token = db.get(VerificationToken, token_id)
+        if token is None:
+            return
+        db.delete(token)
+        db.commit()
+
+    def delete_unexpired_tokens_for_user(self, db: Session, user_id: int) -> None:
+        now = datetime.now(UTC)
+        statement = select(VerificationToken).where(
+            VerificationToken.user_id == user_id,
+            VerificationToken.expires_at >= now,
+        )
+        tokens = db.execute(statement).scalars().all()
+        for token in tokens:
+            db.delete(token)
+        if tokens:
+            db.commit()
+
+
+verification_token_repository = VerificationTokenRepository()
+
+
+def get_token(db: Session, token: str) -> VerificationToken | None:
+    return verification_token_repository.get_token(db=db, token=token)
+
+
+def create_token(db: Session, user_id: int) -> VerificationToken:
+    return verification_token_repository.create_token(db=db, user_id=user_id)
+
+
+def delete_token(db: Session, token_id: int) -> None:
+    verification_token_repository.delete_token(db=db, token_id=token_id)