webmail-client/backend/app/api/settings.py

"""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)