siftlode/backend/app/auth.py

import logging
import secrets
from datetime import datetime, timedelta, timezone

import httpx
from authlib.integrations.starlette_client import OAuth, OAuthError
from fastapi import APIRouter, BackgroundTasks, Depends, HTTPException, Request
from fastapi.responses import JSONResponse, RedirectResponse
from sqlalchemy import delete, func, select
from sqlalchemy.orm import Session

from app import email as email_mod
from app import sysconfig
from app.config import settings
from app.db import get_db
from app.models import AuthToken, DemoWhitelist, Invite, OAuthToken, User
from app.ratelimit import RateLimiter
from app.security import decrypt, encrypt, hash_password, hash_token, verify_password
from app.utils import valid_email

# Email+password auth tuning.
PASSWORD_MIN_LEN = 10
VERIFY_TTL = timedelta(hours=24)
RESET_TTL = timedelta(hours=1)
_register_limiter = RateLimiter(max_events=5, window_seconds=300)
_login_limiter = RateLimiter(max_events=10, window_seconds=300)
_reset_limiter = RateLimiter(max_events=5, window_seconds=300)
# At most one "your account is suspended" email per address per hour, so a suspended user who
# keeps retrying (or a script with their valid credentials) can't be used to mailbomb them.
_suspend_email_limiter = RateLimiter(max_events=1, window_seconds=3600)


def operator_contact() -> str | None:
    """Who a blocked user should contact — the first configured admin email, if any."""
    return next(iter(sorted(settings.admin_email_set)), None)


def _notify_suspended(background: BackgroundTasks, email: str) -> None:
    """Schedule the one-per-hour 'account suspended' email to a user whose (otherwise valid)
    sign-in was just blocked. Gated by verified credentials at the call site, so this can't be
    triggered by someone who doesn't control the account."""
    if _suspend_email_limiter.allow(email):
        background.add_task(email_mod.send_account_suspended, email, operator_contact())

# The single shared demo account's stable identity. The row is created lazily on first
# demo login (see get_or_create_demo_user); these are just its sentinel keys.
DEMO_GOOGLE_SUB = "__demo__"
DEMO_EMAIL = "demo@siftlode.local"

# Throttle the hidden demo-login endpoint per client IP: enough for a real paste-and-wait,
# stingy enough that the field can't be hammered as an enumeration oracle or for DoS.
_demo_limiter = RateLimiter(max_events=5, window_seconds=60)

# How many recently-used accounts to keep switchable in one browser session.
MAX_SESSION_ACCOUNTS = 6

# Base sign-in requests only the non-sensitive OpenID scopes (profile/email). These do
# NOT trigger Google's "unverified app" warning and do NOT expire after 7 days, so a new
# user gets a clean, familiar consent. YouTube access is granted later, one step at a
# time, by the onboarding wizard via /auth/upgrade (incremental authorization):
#   - read  -> youtube.readonly  (read subscriptions, build the feed)
#   - write -> youtube           (full: unsubscribe / playlist export)
# Both YouTube scopes are "sensitive"; the wizard explains the Google warning before
# sending the user there, so the extra permission never feels like a surprise.
WRITE_SCOPE = "https://www.googleapis.com/auth/youtube"
READ_SCOPE = f"{WRITE_SCOPE}.readonly"
BASE_SCOPES = "openid email profile"
READ_SCOPES = f"{BASE_SCOPES} {READ_SCOPE}"
WRITE_SCOPES = f"{BASE_SCOPES} {READ_SCOPE} {WRITE_SCOPE}"

log = logging.getLogger("siftlode.auth")

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

# The Google OAuth client is built lazily from the CURRENT credentials (DB override via sysconfig,
# else env), so the install wizard / admin can set or change them at runtime without a restart. The
# authlib registry is rebuilt only when the creds change (cheap, first use after a change).
_oauth = OAuth()
_oauth_creds: tuple[str, str] | None = None


def google_oauth(db: Session):
    """The configured authlib Google client, or None when no credentials are set (Google sign-in
    is then disabled — email+password still works)."""
    global _oauth, _oauth_creds
    cid = sysconfig.get_str(db, "google_client_id")
    csec = sysconfig.get_str(db, "google_client_secret")
    if not cid or not csec:
        return None
    if _oauth_creds != (cid, csec):
        _oauth = OAuth()
        _oauth.register(
            name="google",
            client_id=cid,
            client_secret=csec,
            server_metadata_url="https://accounts.google.com/.well-known/openid-configuration",
            client_kwargs={"scope": BASE_SCOPES},
        )
        _oauth_creds = (cid, csec)
    return _oauth.google


def google_enabled(db: Session) -> bool:
    """Whether Sign in with Google is available (both client id + secret resolve to non-empty)."""
    return bool(
        sysconfig.get_str(db, "google_client_id")
        and sysconfig.get_str(db, "google_client_secret")
    )


def has_read_scope(user: User) -> bool:
    """Whether the user's stored grant lets us read their YouTube data (build the feed).
    The full write scope is a superset, so it implies read."""
    tok = user.token
    if tok is None or not tok.scopes:
        return False
    granted = tok.scopes.split()
    return READ_SCOPE in granted or WRITE_SCOPE in granted


def has_write_scope(user: User) -> bool:
    """Whether the user's stored grant includes YouTube write (unsubscribe / export)."""
    tok = user.token
    if tok is None or not tok.scopes:
        return False
    return WRITE_SCOPE in tok.scopes.split()


def is_allowed(db: Session, email: str) -> bool:
    """May this email sign in? An approved Invite is the source of truth; the env
    ALLOWED_EMAILS/ADMIN_EMAILS remain a bootstrap fallback (e.g. a fresh DB)."""
    if not email:
        return False
    email = email.lower()
    inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
    if inv is not None:
        return inv.status == "approved"
    return email in settings.allowed_email_set or email in settings.admin_email_set


def is_demo_allowed(db: Session, email: str) -> bool:
    """Whether this email may enter the shared demo account (no Google sign-in)."""
    if not email:
        return False
    return (
        db.execute(
            select(DemoWhitelist).where(DemoWhitelist.email == email.lower())
        ).scalar_one_or_none()
        is not None
    )


def get_or_create_demo_user(db: Session) -> User:
    """The one shared demo user, created on first use. No OAuth token / YouTube scope, so
    has_read_scope/has_write_scope are False and every YouTube-touching path is closed to it."""
    user = db.query(User).filter(User.is_demo.is_(True)).one_or_none()
    if user is None:
        user = User(
            google_sub=DEMO_GOOGLE_SUB,
            email=DEMO_EMAIL,
            display_name="Demo",
            role="user",
            is_demo=True,
            preferences={"language": "en"},
        )
        db.add(user)
        db.commit()
    return user


def _client_ip(request: Request) -> str:
    """Best-effort client IP for rate limiting. Behind our reverse proxy (Caddy/NPM) the
    real client is the first X-Forwarded-For hop; fall back to the socket peer otherwise."""
    xff = request.headers.get("x-forwarded-for")
    if xff:
        return xff.split(",")[0].strip()
    return request.client.host if request.client else "unknown"


def upsert_pending_invite(db: Session, email: str) -> Invite | None:
    """Idempotently record an access request. Returns the Invite if a *new* pending row
    was created (so the caller can notify admins), else None for already-known emails."""
    email = (email or "").lower()
    if not valid_email(email):
        return None
    inv = db.execute(select(Invite).where(Invite.email == email)).scalar_one_or_none()
    if inv is not None:
        # Let a previously-denied person ask again; don't disturb approved/pending rows.
        if inv.status == "denied":
            inv.status = "pending"
            inv.requested_at = datetime.now(timezone.utc)
            inv.decided_at = None
            inv.decided_by = None
            db.commit()
            return inv
        return None
    inv = Invite(email=email, status="pending")
    db.add(inv)
    db.commit()
    return inv


@router.get("/login")
async def login(request: Request, db: Session = Depends(get_db)):
    # access_type=offline ensures a refresh_token on first authorization. We avoid
    # prompt=consent so returning users get a quick sign-in; the stored refresh token
    # is kept when Google doesn't re-issue one (see the callback).
    client = google_oauth(db)
    if client is None:
        # Google sign-in isn't configured on this instance — land back on the login page (the UI
        # hides the button, so this is just a safety net for a stale/direct hit).
        return RedirectResponse(url="/")
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="select_account",
        include_granted_scopes="true",
        scope=BASE_SCOPES,
    )


@router.get("/callback")
async def callback(
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
):
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/")
    try:
        token = await client.authorize_access_token(request)
    except OAuthError as exc:
        raise HTTPException(status_code=400, detail=f"OAuth error: {exc.error}")

    userinfo = token.get("userinfo")
    if not userinfo or not userinfo.get("sub"):
        raise HTTPException(status_code=400, detail="No user info returned by Google")

    sub = userinfo["sub"]
    email = (userinfo.get("email") or "").lower()

    # Linking mode: an already-authenticated account is attaching this Google identity (a
    # password account enabling YouTube or connecting SSO). The marker is set by /auth/link and
    # /auth/upgrade; when present we attach to the current user instead of identifying by sub.
    link_uid = request.session.pop("oauth_link_uid", None)
    if link_uid is not None:
        return _complete_link(request, db, link_uid, sub, userinfo, token)

    if not is_allowed(db, email):
        log.warning("Login denied (not approved): %s", email or "<no email>")
        # A denied Google login doubles as an access request: record it for the admin and
        # land the user on a friendly "request received" screen instead of a raw 403.
        if email:
            inv = upsert_pending_invite(db, email)
            if inv is not None and settings.admin_email_set:
                background.add_task(
                    email_mod.send_admin_new_request,
                    sorted(settings.admin_email_set),
                    email,
                )
            return RedirectResponse(url="/?access=requested")
        return RedirectResponse(url="/?access=denied")

    user = db.query(User).filter(User.google_sub == sub).one_or_none()
    if user is None:
        # No account carries this Google identity yet. If one already exists for this email —
        # e.g. an email+password registration — adopt the Google identity onto it instead of
        # inserting a duplicate (which would collide on the unique email and 500). Google has
        # verified the address, so this safely unifies password + Google sign-in on one account.
        existing = db.query(User).filter(User.email == email).one_or_none()
        if existing is not None:
            if existing.google_sub is not None and existing.google_sub != sub:
                # A different Google identity is already bound to this email — refuse rather than
                # hijack it. (Near-impossible with real Google accounts: email↔sub is stable.)
                log.warning("Google login email collision (different sub): %s", email)
                return RedirectResponse(url="/?access=denied")
            user = existing
            user.google_sub = sub
            # Google verified the email and is_allowed granted access, so finish activating a
            # previously pending password registration (otherwise current_user would bounce it).
            user.email_verified = True
            user.is_active = True
        else:
            user = User(google_sub=sub, email=email)
            db.add(user)
    # A suspended account can't sign in by any method. Block before establishing the session and
    # tell the (verified) owner why — a successful Google auth proves they control this account.
    if user.is_suspended:
        log.info("Suspended account blocked at Google login: %s", email)
        _notify_suspended(background, user.email)
        return RedirectResponse(url="/?login=suspended")
    user.email = email
    user.display_name = userinfo.get("name")
    user.avatar_url = userinfo.get("picture")
    # ADMIN_EMAILS (env) is the bootstrap admin list and always wins, so the configured admin can't
    # be locked out. Everyone else's role is managed in the admin UI and stored in the DB, so a
    # routine login must NOT reset it (that would clobber a UI promotion/demotion). Only seed a
    # default for a brand-new account (its role attribute is still unset before flush).
    if email in settings.admin_email_set:
        user.role = "admin"
    elif not user.role:
        user.role = "user"
    # Google attests the email, so a Google sign-in implicitly verifies it. Setting this also lets
    # a Google user who later adds a password sign in (password-login gates on email_verified).
    if userinfo.get("email_verified", True):
        user.email_verified = True
    # Default UI language from the Google-reported locale on first login (if the user hasn't
    # picked one yet); unsupported locales fall back to English.
    prefs = dict(user.preferences or {})
    if not prefs.get("language"):
        loc = (userinfo.get("locale") or "").split("-")[0].lower()
        prefs["language"] = loc if loc in {"en", "hu", "de"} else "en"
        user.preferences = prefs
    db.flush()

    _store_token(db, user, token)
    db.commit()

    # Multi-session: remember every account that has authenticated in this browser so the
    # user can switch between them without going through Google again. Only accounts that
    # actually completed OAuth here land in this list (it's the proof they may be switched to).
    accounts = [a for a in (request.session.get("account_ids") or []) if a != user.id]
    accounts.append(user.id)
    request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    request.session["user_id"] = user.id
    log.info("Login: %s (id=%s, role=%s)", email, user.id, user.role)
    return RedirectResponse(url="/")


def _store_token(db: Session, user: User, token: dict) -> None:
    """Persist the OAuth grant on the user's single token row, creating it if absent. Google
    only returns a refresh_token on (re)consent, so the previous one is kept otherwise.
    include_granted_scopes=true means `scope` is the union of everything granted, so this
    reflects read/write upgrades correctly."""
    tok = user.token or OAuthToken(user=user)
    if token.get("refresh_token"):
        tok.refresh_token_enc = encrypt(token["refresh_token"])
    tok.access_token = token.get("access_token")
    expires_at = token.get("expires_at")
    tok.expiry = datetime.fromtimestamp(expires_at, tz=timezone.utc) if expires_at else None
    tok.scopes = token.get("scope") or BASE_SCOPES
    db.add(tok)


def revoke_google_token(token: str | None) -> None:
    """Best-effort revocation of a Google OAuth grant at Google's endpoint. Revoking the refresh
    token tears down the whole grant, so a later sign-in starts from a clean consent (no scopes
    silently carried over via include_granted_scopes). Fail-soft: never blocks the caller — used
    from account deletion, where our own data is already gone regardless of Google's response."""
    if not token:
        return
    try:
        resp = httpx.post(
            "https://oauth2.googleapis.com/revoke",
            data={"token": token},
            headers={"Content-Type": "application/x-www-form-urlencoded"},
            timeout=10,
        )
        log.info("Google token revoke: status=%s", resp.status_code)
    except Exception:
        log.warning("Google token revoke failed", exc_info=True)


def purge_user(db: Session, user: User, background: BackgroundTasks) -> None:
    """Hard-delete a user and ALL their personal data (subscriptions, tags, video states,
    playlists, notifications, tokens — all cascade on the users row), erase their access-request
    row, and revoke their Google grant. Shared by GDPR self-deletion and admin deletion. Does NOT
    touch any browser session — each caller handles its own."""
    email = user.email.lower()
    user_id = user.id
    tok = user.token
    google_token = (decrypt(tok.refresh_token_enc) or tok.access_token) if tok else None
    # Raw delete so Postgres applies ON DELETE CASCADE / SET NULL on every dependent table.
    db.execute(delete(User).where(User.id == user_id))
    db.execute(delete(Invite).where(Invite.email == email))
    db.commit()
    if google_token:
        background.add_task(revoke_google_token, google_token)
    # Confirm the erasure to the (now former) user — GDPR good practice, and covers both the
    # self-service and admin deletion paths since both funnel through here.
    background.add_task(email_mod.send_account_deleted, email, operator_contact())


def _complete_link(
    request: Request, db: Session, link_uid: int, sub: str, userinfo: dict, token: dict
):
    """Attach the just-authorized Google identity to the already-signed-in account that started
    the link/upgrade. No is_allowed gate — they're already an active user. Refuses to hijack a
    Google identity owned by another account, or to silently swap a different linked identity."""
    # Consume the companion marker up front so it never lingers across the early returns below.
    explicit = request.session.pop("oauth_link_explicit", False)
    # The marker must match the live session user (guards a stale/forged marker).
    if link_uid != request.session.get("user_id"):
        return RedirectResponse(url="/?link=error")
    user = db.get(User, link_uid)
    if user is None or user.is_demo:
        return RedirectResponse(url="/?link=error")
    other = (
        db.query(User).filter(User.google_sub == sub, User.id != user.id).one_or_none()
    )
    if other is not None:
        return RedirectResponse(url="/?link=conflict")
    if user.google_sub is not None and user.google_sub != sub:
        return RedirectResponse(url="/?link=mismatch")
    user.google_sub = sub
    if not user.display_name:
        user.display_name = userinfo.get("name")
    if not user.avatar_url:
        user.avatar_url = userinfo.get("picture")
    _store_token(db, user, token)
    db.commit()
    log.info("Google linked: uid=%s scopes=%s", user.id, token.get("scope"))
    # An explicit "Connect Google" lands on Settings with a confirmation; a YouTube upgrade
    # (wizard / Settings access rows) returns to the app so the onboarding flow resumes as before.
    return RedirectResponse(url="/?link=ok" if explicit else "/")


@router.post("/logout")
async def logout(request: Request):
    """Sign out the active account. If other accounts have authenticated in this browser,
    switch to the most recent one instead of fully logging out."""
    active = request.session.get("user_id")
    remaining = [a for a in (request.session.get("account_ids") or []) if a != active]
    if remaining:
        request.session["account_ids"] = remaining
        request.session["user_id"] = remaining[-1]
        return JSONResponse({"ok": True, "switched": True})
    request.session.clear()
    return JSONResponse({"ok": True, "switched": False})


@router.post("/request-access")
async def request_access(
    payload: dict,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Public: ask for access. Idempotent — repeat requests for the same email don't
    duplicate or re-spam admins (upsert returns None for an already-pending row)."""
    email = (payload.get("email") or "").strip().lower()
    if not valid_email(email):
        raise HTTPException(status_code=400, detail="Enter a valid email address.")
    if is_allowed(db, email):
        return {"status": "approved"}  # already allowed — just sign in
    inv = upsert_pending_invite(db, email)
    if inv is not None and settings.admin_email_set:
        background.add_task(
            email_mod.send_admin_new_request, sorted(settings.admin_email_set), email
        )
    return {"status": "pending"}


@router.post("/demo")
def demo_login(payload: dict, request: Request, db: Session = Depends(get_db)) -> dict:
    """Hidden demo entry: a whitelisted email logs straight into the shared demo account,
    no Google OAuth. The login page fires this quietly (debounced) as the user types/pastes,
    so there is no visible 'demo login' button. Rate-limited per IP; when blocked it answers
    exactly like a non-match (authenticated=false) so the field can't be hammered as an
    enumeration oracle. On a match it sets the session and the client reloads into the app."""
    if not _demo_limiter.allow(_client_ip(request)):
        return {"authenticated": False}
    email = (payload.get("email") or "").strip().lower()
    if valid_email(email) and is_demo_allowed(db, email):
        demo = get_or_create_demo_user(db)
        request.session["user_id"] = demo.id
        log.info("Demo login (key=%s, demo_id=%s)", email, demo.id)
        return {"authenticated": True}
    return {"authenticated": False}


def _establish_session(request: Request, user: User) -> None:
    """Sign `user` in for this browser session, mirroring the Google callback's multi-account
    handling so password sign-in joins the same account switcher."""
    accounts = [a for a in (request.session.get("account_ids") or []) if a != user.id]
    accounts.append(user.id)
    request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    request.session["user_id"] = user.id


def _app_base() -> str:
    """Public origin of the deployed app (OAUTH_REDIRECT_URL is .../auth/callback)."""
    return settings.app_base


def _issue_token(db: Session, user: User, kind: str, ttl: timedelta) -> str:
    """Create a single-use token of `kind`; only its hash is stored. Returns the raw token
    (goes only into the emailed link)."""
    raw = secrets.token_urlsafe(32)
    db.add(
        AuthToken(
            user_id=user.id,
            kind=kind,
            token_hash=hash_token(raw),
            expires_at=datetime.now(timezone.utc) + ttl,
        )
    )
    db.commit()
    return raw


def _consume_token(db: Session, raw: str | None, kind: str) -> User | None:
    """Validate + burn a token. Returns its user, or None if missing/expired/used/wrong kind."""
    if not raw:
        return None
    row = db.execute(
        select(AuthToken).where(
            AuthToken.token_hash == hash_token(raw), AuthToken.kind == kind
        )
    ).scalar_one_or_none()
    if row is None or row.used_at is not None or row.expires_at < datetime.now(timezone.utc):
        return None
    row.used_at = datetime.now(timezone.utc)
    db.commit()
    return db.get(User, row.user_id)


@router.post("/register")
def register(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Email+password registration. Creates a pending (inactive, unverified) account, sends a
    verification link, and records an access request for admin approval. Two gates before
    sign-in: verified email AND admin approval. Responses are uniform once input is valid, so
    the endpoint can't be used to discover which emails are registered."""
    email = (payload.get("email") or "").strip().lower()
    password = payload.get("password") or ""
    # Input validation is safe to reveal (independent of whether the email exists).
    if not valid_email(email):
        raise HTTPException(status_code=400, detail="Enter a valid email address.")
    if len(password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    if not sysconfig.get_bool(db, "allow_registration"):
        raise HTTPException(status_code=403, detail="Registration is currently closed.")
    if not _register_limiter.allow(_client_ip(request)):
        return {"status": "ok"}  # silently throttle; uniform response

    existing = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
    if existing is None:
        # Without working SMTP we can't deliver a verification link, so email ownership can't
        # gate sign-in. Admin approval stays the real gate (is_active=False); mark the account
        # verified so the flow still completes on a no-SMTP self-host. See email.email_enabled().
        email_ok = email_mod.email_enabled()
        user = User(
            email=email,
            password_hash=hash_password(password),
            is_active=False,
            email_verified=not email_ok,
        )
        db.add(user)
        db.flush()
        upsert_pending_invite(db, email)  # admin-approval gate
        if email_ok:
            raw = _issue_token(db, user, "verify", VERIFY_TTL)
            background.add_task(
                email_mod.send_verify_email, email, f"{_app_base()}/auth/verify?token={raw}"
            )
        if settings.admin_email_set:
            background.add_task(
                email_mod.send_admin_new_request, sorted(settings.admin_email_set), email
            )
    # Existing email → do nothing visible (no enumeration). Uniform success either way.
    return {"status": "ok"}


@router.get("/verify")
def verify_email(token: str, db: Session = Depends(get_db)):
    """Confirm an email-verification link, then bounce to a friendly status on the app."""
    user = _consume_token(db, token, "verify")
    if user is None:
        return RedirectResponse(url="/?verify=invalid")
    user.email_verified = True
    db.commit()
    return RedirectResponse(url="/?verify=ok")


@router.post("/password-login")
def password_login(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Email+password sign-in. Wrong email/password give a single uniform 401 (no enumeration).
    Once the password is proven correct, the owner gets a specific reason if their account is
    still pending or suspended — that's not an enumeration leak (they already hold the password)."""
    if not _login_limiter.allow(_client_ip(request)):
        raise HTTPException(status_code=429, detail="Too many attempts. Try again shortly.")
    email = (payload.get("email") or "").strip().lower()
    password = payload.get("password") or ""
    user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
    if user is None or user.is_demo or not verify_password(password, user.password_hash):
        raise HTTPException(status_code=401, detail="Invalid email or password.")
    if user.is_suspended:
        _notify_suspended(background, user.email)
        op = operator_contact()
        detail = "Your account has been suspended."
        if op:
            detail += f" If you have questions, contact the operator at {op}."
        raise HTTPException(status_code=403, detail=detail)
    if not user.email_verified:
        raise HTTPException(status_code=403, detail="Please verify your email first (check your inbox).")
    if not user.is_active:
        raise HTTPException(status_code=403, detail="Your account is awaiting admin approval.")
    _establish_session(request, user)
    log.info("Password login: %s (id=%s, role=%s)", email, user.id, user.role)
    return {"ok": True}


@router.post("/password-reset/request")
def password_reset_request(
    payload: dict,
    request: Request,
    background: BackgroundTasks,
    db: Session = Depends(get_db),
) -> dict:
    """Request a password-reset link. Uniform response regardless of whether the email has a
    password account, so it can't probe for registered emails."""
    if not _reset_limiter.allow(_client_ip(request)):
        return {"status": "ok"}
    email = (payload.get("email") or "").strip().lower()
    if valid_email(email):
        user = db.execute(select(User).where(User.email == email)).scalar_one_or_none()
        if user is not None and user.password_hash and not user.is_demo:
            raw = _issue_token(db, user, "reset", RESET_TTL)
            background.add_task(email_mod.send_password_reset, email, f"{_app_base()}/?reset={raw}")
    return {"status": "ok"}


@router.post("/password-reset/confirm")
def password_reset_confirm(payload: dict, db: Session = Depends(get_db)) -> dict:
    """Set a new password from a valid reset token, then invalidate any other reset tokens."""
    token = payload.get("token")
    new_password = payload.get("password") or ""
    if len(new_password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    user = _consume_token(db, token, "reset")
    if user is None:
        raise HTTPException(status_code=400, detail="This reset link is invalid or has expired.")
    user.password_hash = hash_password(new_password)
    # Burn any other outstanding reset tokens for this user.
    for row in db.execute(
        select(AuthToken).where(
            AuthToken.user_id == user.id, AuthToken.kind == "reset", AuthToken.used_at.is_(None)
        )
    ).scalars():
        row.used_at = datetime.now(timezone.utc)
    db.commit()
    return {"ok": True}


def current_user(request: Request, db: Session = Depends(get_db)) -> User:
    user_id = request.session.get("user_id")
    if not user_id:
        raise HTTPException(status_code=401, detail="Not authenticated")
    user = db.get(User, user_id)
    if user is None or not user.is_active or user.is_suspended:
        # Suspended/deactivated mid-session → drop the session so the block takes effect at once.
        request.session.clear()
        raise HTTPException(status_code=401, detail="Not authenticated")
    # Always keep the active account in the switchable list — covers sessions created before
    # multi-session existed (their account_ids was never seeded), so the switcher isn't empty.
    accounts = request.session.get("account_ids") or []
    if user_id not in accounts:
        accounts.append(user_id)
        request.session["account_ids"] = accounts[-MAX_SESSION_ACCOUNTS:]
    return user


def require_human(user: User = Depends(current_user)) -> User:
    """Reject the shared demo account from actions that need a real Google/YouTube identity
    or spend the shared quota. Most YouTube paths are already closed to it implicitly (it has
    no token, so has_read_scope/has_write_scope are False); this makes the boundary explicit
    where there's no scope check to lean on."""
    if user.is_demo:
        raise HTTPException(status_code=403, detail="Not available in the demo account.")
    return user


def admin_user(user: User = Depends(current_user)) -> User:
    """Dependency: require the signed-in user to be an admin. Single source of truth for the
    admin gate — every admin route/action depends on this rather than re-checking the role."""
    if user.role != "admin":
        raise HTTPException(status_code=403, detail="Admin only")
    return user


def count_admins(db: Session, *, active_only: bool = False) -> int:
    """How many admins exist (optionally only un-suspended ones). Used by the 'can't remove /
    suspend / delete the last admin' guards so they don't each re-spell the count query."""
    q = select(func.count()).select_from(User).where(User.role == "admin")
    if active_only:
        q = q.where(User.is_suspended.is_(False))
    return db.scalar(q) or 0


@router.get("/link")
async def link_google(
    request: Request,
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
):
    """Start linking a Google account to the signed-in (e.g. password) account. Requests only
    the identity scopes — YouTube access is granted separately via /auth/upgrade. The callback
    sees `oauth_link_uid` and attaches the identity instead of creating a new account."""
    if user.is_demo:
        return RedirectResponse(url="/")
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/?link=error")
    request.session["oauth_link_uid"] = user.id
    request.session["oauth_link_explicit"] = True
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="select_account",
        include_granted_scopes="true",
        scope=BASE_SCOPES,
    )


@router.post("/set-password")
def set_password(
    payload: dict, user: User = Depends(current_user), db: Session = Depends(get_db)
) -> dict:
    """Set or change the signed-in account's password. Setting a first password (Google-only
    account) needs no current password — the session already proves identity. Changing an
    existing one requires the current password, so a hijacked session can't silently rotate it."""
    if user.is_demo:
        raise HTTPException(status_code=403, detail="Not available in the demo account.")
    new_password = payload.get("password") or ""
    if len(new_password) < PASSWORD_MIN_LEN:
        raise HTTPException(
            status_code=400,
            detail=f"Password must be at least {PASSWORD_MIN_LEN} characters.",
        )
    if user.password_hash:
        if not verify_password(payload.get("current_password") or "", user.password_hash):
            raise HTTPException(status_code=403, detail="Current password is incorrect.")
    user.password_hash = hash_password(new_password)
    db.commit()
    log.info("Password set/changed: uid=%s", user.id)
    return {"ok": True}


@router.get("/upgrade")
async def upgrade(
    request: Request,
    access: str = "read",
    user: User = Depends(current_user),
    db: Session = Depends(get_db),
):
    """Incremental consent for the onboarding wizard. `access=read` grants YouTube
    read-only (enough to build the feed); `access=write` additionally grants management
    (unsubscribe). The shared callback stores whatever Google returns, so `can_read` /
    `can_write` flip on once granted. Anything other than an explicit `write` defaults to
    the least-privileged read scope."""
    # This is the browser-facing redirect entry point, so the shared demo account is bounced
    # straight home (no YouTube link is ever attached to it) rather than shown a raw 403.
    if user.is_demo:
        return RedirectResponse(url="/")
    client = google_oauth(db)
    if client is None:
        return RedirectResponse(url="/")
    # Attach the grant to THIS account in the callback. Without it a password account (no
    # google_sub) would get a brand-new, separate Google user created instead of being linked.
    request.session["oauth_link_uid"] = user.id
    scope = WRITE_SCOPES if access == "write" else READ_SCOPES
    return await client.authorize_redirect(
        request,
        settings.oauth_redirect_url,
        access_type="offline",
        prompt="consent",
        include_granted_scopes="true",
        scope=scope,
    )


@router.get("/me")
async def me(user: User = Depends(current_user)) -> dict:
    return {
        "id": user.id,
        "email": user.email,
        "display_name": user.display_name,
        "avatar_url": user.avatar_url,
        "role": user.role,
    }


@router.get("/config")
def auth_config(db: Session = Depends(get_db)) -> dict:
    """Public: which sign-in options this instance offers, so the login page can hide what's off
    (e.g. no Google button when Google OAuth isn't configured)."""
    return {
        "google_enabled": google_enabled(db),
        "allow_registration": sysconfig.get_bool(db, "allow_registration"),
    }