Spaces:
Running
Running
| from fastapi import Request, HTTPException | |
| from config.database import get_supabase_admin | |
| async def get_current_user(request: Request): | |
| """ | |
| Check for token in two places: | |
| 1. httpOnly cookie 'access_token' (email/password login) | |
| 2. Authorization: Bearer header (GitHub/Google OAuth) | |
| """ | |
| token = request.cookies.get("access_token") | |
| if not token: | |
| auth_header = request.headers.get("Authorization", "") | |
| if auth_header.startswith("Bearer "): | |
| token = auth_header[7:] | |
| if not token: | |
| raise HTTPException(status_code=401, detail="Not authenticated") | |
| try: | |
| admin_client = get_supabase_admin() | |
| user_response = admin_client.auth.get_user(token) | |
| if not user_response or not user_response.user: | |
| raise HTTPException(status_code=401, detail="Invalid or expired session") | |
| user = user_response.user | |
| # New: Automatically ensure user exists in public table | |
| from services.user_sync import ensure_user_exists | |
| await ensure_user_exists(user) | |
| # New: Fetch role from public.users table | |
| from config.settings import get_settings | |
| settings = get_settings() | |
| profile = admin_client.table("users").select("role").eq("id", str(user.id)).single().execute() | |
| # Force admin role for configured emails (Case-Insensitive) | |
| admin_list = [e.lower() for e in settings.admin_emails] | |
| user_role = "admin" if user.email.lower() in admin_list else (profile.data.get("role", "").lower() if profile.data else "free") | |
| # Attach role to the user object (patch it) | |
| user.role = user_role | |
| return user | |
| except HTTPException: | |
| raise | |
| except Exception: | |
| raise HTTPException(status_code=401, detail="Invalid or expired session") | |
| async def get_current_user_optional(request: Request): | |
| """Returns user if logged in, None if not. Never raises.""" | |
| try: | |
| return await get_current_user(request) | |
| except HTTPException: | |
| return None |