"""HTTP routes for Spotify 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 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 Spotify Authorization Code with PKCE login.""" application_settings = cast(Settings, request.app.state.settings) 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 Spotify login and establish an opaque cookie session.""" if error is not None or code is None or state is None: return _login_error_redirect() application_settings = cast(Settings, request.app.state.settings) 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.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 cookie session or the installed live seed 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)