"""Notification preferences and unsubscribe endpoints. Endpoints: GET /notifications/preferences — current user's preferences (auth required) PUT /notifications/preferences — update preferences (auth required) GET /notifications/unsubscribe — signed-token unsubscribe (no auth) """ from __future__ import annotations import logging from typing import Annotated from fastapi import APIRouter, Depends, HTTPException, Query, status from fastapi.responses import HTMLResponse from sqlalchemy import select, update from sqlalchemy.ext.asyncio import AsyncSession from auth import get_current_user from config import get_settings from database import get_session from models import User from schemas import NotificationPreferences, NotificationPreferencesUpdate from tasks.notifications import verify_unsubscribe_token logger = logging.getLogger(__name__) router = APIRouter(prefix="/notifications", tags=["notifications"]) @router.get("/preferences", response_model=NotificationPreferences) async def get_preferences( current_user: Annotated[User, Depends(get_current_user)], ): """Return the current user's notification preferences.""" prefs = current_user.notification_preferences or {} return NotificationPreferences( email_digests=prefs.get("email_digests", True), digest_frequency=prefs.get("digest_frequency", "daily"), ) @router.put("/preferences", response_model=NotificationPreferences) async def update_preferences( body: NotificationPreferencesUpdate, current_user: Annotated[User, Depends(get_current_user)], session: Annotated[AsyncSession, Depends(get_session)], ): """Update the current user's notification preferences. Only provided fields are updated; omitted fields keep their current value. """ # Validate digest_frequency if provided valid_frequencies = {"daily", "weekly"} if body.digest_frequency is not None and body.digest_frequency not in valid_frequencies: raise HTTPException( status_code=422, detail=f"digest_frequency must be one of: {', '.join(sorted(valid_frequencies))}", ) current_prefs = dict(current_user.notification_preferences or {}) if body.email_digests is not None: current_prefs["email_digests"] = body.email_digests if body.digest_frequency is not None: current_prefs["digest_frequency"] = body.digest_frequency await session.execute( update(User) .where(User.id == current_user.id) .values(notification_preferences=current_prefs) ) await session.commit() logger.info( "notification_preferences updated user_id=%s prefs=%s", current_user.id, current_prefs, ) return NotificationPreferences( email_digests=current_prefs.get("email_digests", True), digest_frequency=current_prefs.get("digest_frequency", "daily"), ) _UNSUB_SUCCESS_HTML = """ Unsubscribed

Unsubscribed

You've been unsubscribed from email digests. You can re-enable them in your settings.

""" _UNSUB_ERROR_HTML = """ Unsubscribe Error

Link Expired or Invalid

This unsubscribe link is no longer valid. Please log in to manage your notification preferences.

""" @router.get("/unsubscribe", response_class=HTMLResponse) async def unsubscribe( token: Annotated[str, Query(description="Signed unsubscribe token")], session: Annotated[AsyncSession, Depends(get_session)], ): """Unsubscribe a user from email digests via signed token. No authentication required — the token itself proves identity. """ settings = get_settings() user_id = verify_unsubscribe_token(token, settings.app_secret_key) if user_id is None: logger.warning("unsubscribe: invalid or expired token") return HTMLResponse(content=_UNSUB_ERROR_HTML, status_code=400) # Update user preferences result = await session.execute(select(User).where(User.id == user_id)) user = result.scalar_one_or_none() if user is None: logger.warning("unsubscribe: user_id=%s not found", user_id) return HTMLResponse(content=_UNSUB_ERROR_HTML, status_code=400) prefs = dict(user.notification_preferences or {}) prefs["email_digests"] = False await session.execute( update(User) .where(User.id == user.id) .values(notification_preferences=prefs) ) await session.commit() logger.info("unsubscribe: user_id=%s unsubscribed via token", user_id) return HTMLResponse(content=_UNSUB_SUCCESS_HTML, status_code=200)