siftlode/backend/app/routes/me.py

from typing import Any

from fastapi import APIRouter, BackgroundTasks, Depends, Request
from pydantic import BaseModel, RootModel
from sqlalchemy import func, select
from sqlalchemy.orm import Session

from app.auth import (
    count_admins,
    count_pending_activations,
    current_user,
    drop_from_wallet,
    google_enabled,
    has_read_scope,
    has_write_scope,
    is_allowed,
    optional_current_user,
    purge_user,
)
from app import debugsession, diagnostics
from app.issues import gitea
from app import email as email_mod
from app import logs
from app import sysconfig
from app.db import get_db
from app.models import Invite, User
from app.utils import iso
from app.refusal import refused
from app.routes._common import not_found

router = APIRouter(prefix="/api/me", tags=["me"])


@router.get("/accounts")
def my_accounts(
    request: Request,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> list[dict]:
    """The accounts that have authenticated in this browser session (switchable without a
    new Google sign-in). The active one is flagged."""
    ids = request.session.get("account_ids") or [user.id]
    rows = {u.id: u for u in db.execute(select(User).where(User.id.in_(ids))).scalars()}
    out = []
    for uid in ids:  # preserve recency order from the session
        u = rows.get(uid)
        if u is not None:
            out.append(
                {
                    "id": u.id,
                    "email": u.email,
                    "display_name": u.display_name,
                    "avatar_url": u.avatar_url,
                    "active": u.id == user.id,
                }
            )
    return out


class SwitchAccountIn(BaseModel):
    user_id: int


@router.post("/switch")
def switch_account(
    body: SwitchAccountIn,
    request: Request,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Switch the active account to another one already authenticated in this browser. No
    Google round-trip — but only accounts present in the session list (proof they signed in
    here) are allowed, and access is re-checked in case the invite was revoked since."""
    target = body.user_id
    if target not in (request.session.get("account_ids") or []):
        raise refused(403, "not_your_account")
    # An account `_session_user` would refuse — gone, deactivated, suspended, or its session revoked — is
    # refused here too, and dropped from the switch list the way `_session_user` drops it. Made the session
    # default instead, the next request's current_user would refuse it as the DEFAULT and clear the whole
    # wallet session, logging out every account here. (B-core-9)
    u = db.get(User, target)
    if u is None:
        drop_from_wallet(request.session, target)
        raise not_found("account")
    if not is_allowed(db, u.email):
        raise refused(403, "account_no_access")
    if not u.is_active:
        drop_from_wallet(request.session, target)
        raise refused(403, "account_no_access")
    if u.is_suspended:
        drop_from_wallet(request.session, target)
        raise refused(403, "account_suspended")
    # Revoked: a password reset/change or "log out everywhere" bumped the account's epoch.
    epochs = request.session.get("epochs") or {}
    if int(epochs.get(str(target), 0)) != (u.session_epoch or 0):
        drop_from_wallet(request.session, target)
        raise refused(403, "account_signed_out")
    request.session["user_id"] = target
    return {"ok": True, "user_id": target}


@router.get("")
def get_me(
    user: User | None = Depends(optional_current_user), db: Session = Depends(get_db)
) -> dict:
    # The app's bootstrap probe: return 200 with `authenticated: False` when logged out (rather
    # than 401) so the public landing never logs a failed /api/me to the browser console. Other
    # protected endpoints still 401, so mid-session expiry is still caught by the global handler.
    if user is None:
        return {"authenticated": False}
    pending_invites = 0
    pending_activations = 0
    # Admin-only, because it drives an operator warning: email is fail-soft, so a half-filled SMTP
    # config costs verification links, password resets and access-request notices with nothing
    # visible anywhere. Surfacing it here puts the warning on the screens where onboarding happens.
    email_configured = True
    if user.role == "admin":
        pending_invites = (
            db.scalar(select(func.count()).select_from(Invite).where(Invite.status == "pending"))
            or 0
        )
        # Counted separately from pending_invites rather than folded into it: the two are different
        # facts (a request awaiting a decision vs a decided address whose sign-up never got let in),
        # and quietly widening one number would make an existing field mean something new.
        pending_activations = count_pending_activations(db)
        email_configured = email_mod.email_enabled(db)  # this request's session; no second connect
    return {
        "authenticated": True,
        "id": user.id,
        "email": user.email,
        "display_name": user.display_name,
        "avatar_url": user.avatar_url,
        "role": user.role,
        "has_google": user.google_sub is not None,
        # The address Google answers with, when it DIFFERS from the account's own. Same string for
        # most people, and then this is None so no surface has to decide whether to show it — the
        # comparison lives here, once, instead of in every consumer.
        "google_email": (
            user.google_email if user.google_email and user.google_email != user.email else None
        ),
        "has_password": user.password_hash is not None,
        # The account's latest sign-in on any device — shown in Settings → Account, so a sign-in the
        # user does not recognise is something they can see (null: none since 0.88).
        "last_login_at": iso(user.last_login_at),
        # Instance-wide: whether Google sign-in / YouTube connect is available at all (the UI hides
        # YouTube-access affordances and the onboarding nudge when it isn't configured).
        "google_enabled": google_enabled(db),
        # Whether the optional Plex module is enabled (UI shows the Plex feed source when so).
        "plex_enabled": sysconfig.get_bool(db, "plex_enabled"),
        # Which live-search source runs, so the search button can say what a search costs: "scrape"
        # spends no API quota, "api" spends 100 units a page.
        "search_source": sysconfig.get_str(db, "search_source"),
        "can_read": has_read_scope(user),
        "can_write": has_write_scope(user),
        "pending_invites": pending_invites,
        # Sign-ups an admin still has to let in (roadmap UX-25): approved BEFORE they registered, so
        # nothing ever reached the Access-requests queue and the badge used to read zero.
        "pending_activations": pending_activations,
        "email_configured": email_configured,
        "preferences": user.preferences or {},
        # The consented diagnostics level in force and when it lapses. The client applies
        # it to its own log and streams that log while it is raised.
        "diagnostics": diagnostics.state(db, user),
        # The live debug session, if one is recording — a reload resumes it.
        "debug_session": _live_session(db, user),
        # Where a bug report goes: the tracker's repository, or None — download only.
        "issues": gitea.target(db),
    }


def _live_session(db: Session, user: User) -> dict | None:
    session = debugsession.open_session(db, user)
    return None if session is None else debugsession.view(db, session)


@router.delete("/account")
def delete_my_account(
    request: Request,
    background: BackgroundTasks,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """GDPR self-service erasure: permanently delete the signed-in account and ALL its personal
    data — OAuth tokens, subscriptions, tags, watch/save/hide states, playlists, notifications —
    which all cascade on the users row. Quota-audit events are kept but anonymised (FK SET NULL).
    The last remaining admin can't delete itself (that would lock everyone out of the admin
    surfaces)."""
    if user.role == "admin" and count_admins(db) <= 1:
        raise refused(400, "only_admin_delete")
    user_id = user.id
    # Full erasure (cascades) + access-request cleanup + Google-grant revocation; shared with the
    # admin delete path. Capture the id first — `user` is detached once purge_user deletes the row.
    purge_user(db, user, background)
    # Drop this account from the browser session; switch to another signed-in account if any.
    remaining = [a for a in (request.session.get("account_ids") or []) if a != user_id]
    if remaining:
        request.session["account_ids"] = remaining
        request.session["user_id"] = remaining[-1]
    else:
        request.session.clear()
    return {"deleted": True}


class PreferencesIn(RootModel[dict[str, Any]]):
    """Free-form UI preferences — a JSON object of any keys (theme, filters, …), merged into the
    stored ones. A model rather than a bare `dict` parameter so a non-object body is a 422 at the
    boundary like every other request body."""


@router.put("/preferences")
def update_preferences(
    preferences: PreferencesIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    # Merge so partial updates don't wipe other keys.
    merged = dict(user.preferences or {})
    merged.update(preferences.root)
    user.preferences = merged
    db.add(user)
    db.commit()
    return {"ok": True, "preferences": merged}


class DiagnosticsIn(BaseModel):
    level: diagnostics.DiagnosticsLevel


@router.put("/diagnostics")
def update_diagnostics(
    body: DiagnosticsIn,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
) -> dict:
    """Raise the diagnostics level (the user consented in Settings → Diagnosztika) or return to the
    base. The expiry is the server's: a raised level always lapses after
    `diagnostics.ELEVATION_HOURS`, whatever the client asks."""
    diagnostics.set_level(user, body.level)
    db.commit()
    # The rest of THIS request, and the process's elevation registry, follow the new level now.
    logs.set_user(user.id, diagnostics.effective_level(user), user.diagnostics_until)
    return {"ok": True, "diagnostics": diagnostics.state(db, user)}