"""User settings + per-sender image allowlist (SPEC §9.1 Settings, §9.4).""" from __future__ import annotations from typing import Annotated from fastapi import APIRouter, Depends from pydantic import BaseModel from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.db import get_session from app.identity.deps import CurrentUser, require_csrf from app.models import Message, User router = APIRouter(prefix="/settings", tags=["settings"]) DbSession = Annotated[AsyncSession, Depends(get_session)] _ALLOWED_ORDER = {"asc", "desc"} _ALLOWED_TIME_FORMAT = {"12", "24"} _ALLOWED_REMINDER_METHOD = {"alert", "email"} _ALLOWED_INVITE_POLICY = {"add_update", "update_only", "none"} _ALLOWED_INVITE_FROM = {"contacts", "anyone"} class ReminderPref(BaseModel): method: str = "alert" # alert = in-app notification, email = email me minutes: int = 0 # minutes before the event start (0 = at time of event) class SettingsOut(BaseModel): thread_order: str = "asc" tz: str = "UTC" time_format: str = "24" # 12 = AM/PM, 24 = 24-hour (default); applies app-wide jitsi_server: str = "" # blank => https://meet.jit.si/ default_calendar: str = "" # calendar id new events default to (blank => first) default_reminders_timed: list[ReminderPref] = [] # defaults for timed events default_reminders_allday: list[ReminderPref] = [] # defaults for all-day events invitation_policy: str = "add_update" # how emailed events update the calendar invitation_from: str = "contacts" # whose emailed events auto-add (contacts|anyone) invitation_calendar: str = "" # calendar id emailed events land in (blank => default) invitation_file: bool = False # also mark the invite message read + filed allowed_image_senders: list[str] = [] class SettingsPatch(BaseModel): thread_order: str | None = None tz: str | None = None time_format: str | None = None jitsi_server: str | None = None default_calendar: str | None = None default_reminders_timed: list[ReminderPref] | None = None default_reminders_allday: list[ReminderPref] | None = None invitation_policy: str | None = None invitation_from: str | None = None invitation_calendar: str | None = None invitation_file: bool | None = None class QuotaOut(BaseModel): used_bytes: int quota_bytes: int def allowed_image_senders(user: User) -> set[str]: value = user.settings.get("allowed_image_senders", []) if isinstance(value, list): return {str(e).lower() for e in value} return set() def add_allowed_image_sender(user: User, email: str) -> None: current = sorted(allowed_image_senders(user)) if email.lower() not in current: current.append(email.lower()) # Reassign the whole dict so SQLAlchemy detects the JSONB change. user.settings = {**user.settings, "allowed_image_senders": current} def _reminder_list(value: object) -> list[ReminderPref]: """Normalize stored reminders. Accepts the legacy ``list[int]`` (minutes only, method defaults to alert) as well as the ``{method, minutes}`` object form.""" if not isinstance(value, list): return [] out: list[ReminderPref] = [] for x in value: if isinstance(x, dict): method = str(x.get("method", "alert")) out.append( ReminderPref( method=method if method in _ALLOWED_REMINDER_METHOD else "alert", minutes=int(x.get("minutes", 0)), ) ) else: out.append(ReminderPref(method="alert", minutes=int(x))) return out def _settings_out(user: User) -> SettingsOut: s = user.settings return SettingsOut( thread_order=str(s.get("thread_order", "asc")), tz=str(s.get("tz", "UTC")), time_format=str(s.get("time_format", "24")), jitsi_server=str(s.get("jitsi_server", "")), default_calendar=str(s.get("default_calendar", "")), default_reminders_timed=_reminder_list(s.get("default_reminders_timed")), default_reminders_allday=_reminder_list(s.get("default_reminders_allday")), invitation_policy=str(s.get("invitation_policy", "add_update")), invitation_from=str(s.get("invitation_from", "contacts")), invitation_calendar=str(s.get("invitation_calendar", "")), invitation_file=bool(s.get("invitation_file", False)), allowed_image_senders=sorted(allowed_image_senders(user)), ) @router.get("", response_model=SettingsOut) async def get_settings_endpoint(user: CurrentUser) -> SettingsOut: return _settings_out(user) @router.get("/quota", response_model=QuotaOut) async def get_quota(user: CurrentUser, db: DbSession) -> QuotaOut: """Storage quota and current usage (sum of stored message sizes).""" used = await db.scalar( select(func.coalesce(func.sum(Message.size_bytes), 0)).where(Message.user_id == user.id) ) return QuotaOut(used_bytes=int(used or 0), quota_bytes=user.quota_bytes) @router.patch("", response_model=SettingsOut, dependencies=[Depends(require_csrf)]) async def patch_settings(body: SettingsPatch, user: CurrentUser, db: DbSession) -> SettingsOut: updates = dict(user.settings) if body.thread_order is not None and body.thread_order in _ALLOWED_ORDER: updates["thread_order"] = body.thread_order if body.tz is not None: updates["tz"] = body.tz if body.time_format is not None and body.time_format in _ALLOWED_TIME_FORMAT: updates["time_format"] = body.time_format if body.jitsi_server is not None: updates["jitsi_server"] = body.jitsi_server.strip() if body.default_calendar is not None: updates["default_calendar"] = body.default_calendar if body.default_reminders_timed is not None: updates["default_reminders_timed"] = [ r.model_dump() for r in body.default_reminders_timed ] if body.default_reminders_allday is not None: updates["default_reminders_allday"] = [ r.model_dump() for r in body.default_reminders_allday ] if body.invitation_policy is not None and body.invitation_policy in _ALLOWED_INVITE_POLICY: updates["invitation_policy"] = body.invitation_policy if body.invitation_from is not None and body.invitation_from in _ALLOWED_INVITE_FROM: updates["invitation_from"] = body.invitation_from if body.invitation_calendar is not None: updates["invitation_calendar"] = body.invitation_calendar if body.invitation_file is not None: updates["invitation_file"] = body.invitation_file user.settings = updates await db.commit() return _settings_out(user)