"""HTTP routes for login and session management.""" from dataclasses import dataclass from typing import cast import httpx2 from fastapi import APIRouter, Request from fastapi.responses import JSONResponse, RedirectResponse, Response from app.adapters.spotify.login import begin_login, complete_login from app.adapters.spotify.session import PendingLogins, SessionStore, SpotifySession from app.config import AppMode, Settings SESSION_COOKIE_NAME = "discovery_session" router = APIRouter() @dataclass(frozen=True) class ResolvedSession: """A session and stable cache key selected for one request.""" session_id: str session: SpotifySession @router.get("/api/auth/login") def login(request: Request) -> RedirectResponse: """Start the configured live or demo login flow.""" application_settings = cast(Settings, request.app.state.settings) if application_settings.app_mode is AppMode.DEMO: return RedirectResponse("/?login=demo", status_code=307) pending_logins = cast(PendingLogins, request.app.state.pending_logins) authorize_url = begin_login(application_settings, pending_logins) return RedirectResponse(authorize_url, status_code=307) @router.get("/callback") async def callback( request: Request, code: str | None = None, state: str | None = None, error: str | None = None, ) -> RedirectResponse: """Complete live login and establish an opaque cookie session.""" application_settings = cast(Settings, request.app.state.settings) if application_settings.app_mode is AppMode.DEMO: return RedirectResponse("/?login=demo", status_code=307) if error is not None or code is None or state is None: return _login_error_redirect() http = cast(httpx2.AsyncClient, request.app.state.http) pending_logins = cast(PendingLogins, request.app.state.pending_logins) session_store = cast(SessionStore, request.app.state.session_store) session_id = await complete_login( http, application_settings, pending_logins, session_store, code, state, ) if session_id is None: return _login_error_redirect() response = RedirectResponse("/", status_code=307) response.set_cookie( SESSION_COOKIE_NAME, session_id, httponly=True, samesite="lax", path="/", secure=application_settings.session_cookie_secure, ) return response @router.get("/api/auth/me") def current_session(request: Request) -> JSONResponse: """Return the display name for a valid application session.""" resolved = resolve_session(request) if resolved is None: return JSONResponse({"detail": "Not authenticated"}, status_code=401) return JSONResponse({"display_name": resolved.session.display_name}) @router.get("/api/suggestions") def suggestions(request: Request) -> list[dict[str, str]]: """Return shared non-refinement suggestion chips and queries.""" return cast(list[dict[str, str]], request.app.state.suggestions) @router.post("/api/auth/logout", status_code=204) def logout(request: Request) -> Response: """Remove the current application session and clear its cookie.""" session_id = request.cookies.get(SESSION_COOKIE_NAME) session_store = cast(SessionStore, request.app.state.session_store) if session_id is not None: session_store.remove(session_id) application_settings = cast(Settings, request.app.state.settings) response = Response(status_code=204) response.delete_cookie( SESSION_COOKIE_NAME, path="/", secure=application_settings.session_cookie_secure, httponly=True, samesite="lax", ) return response def _login_error_redirect() -> RedirectResponse: return RedirectResponse("/?login=error", status_code=307) def resolve_session(request: Request) -> ResolvedSession | None: """Resolve the stable demo identity, cookie, or live seed session.""" application_settings = cast(Settings, request.app.state.settings) if application_settings.app_mode is AppMode.DEMO: demo_session = cast(SpotifySession, request.app.state.demo_session) return ResolvedSession("demo", demo_session) session_store = cast(SessionStore, request.app.state.session_store) cookie_session_id = request.cookies.get(SESSION_COOKIE_NAME) if cookie_session_id is not None: cookie_session = session_store.get(cookie_session_id) if cookie_session is not None: return ResolvedSession(cookie_session_id, cookie_session) seed_session_id = cast(str | None, getattr(request.app.state, "seed_session_id", None)) seed_session = session_store.get(seed_session_id) if seed_session_id is not None else None if seed_session is None or seed_session_id is None: return None return ResolvedSession(seed_session_id, seed_session)