mnote/wolai-backend/app/deps.py

from __future__ import annotations

from dataclasses import dataclass

from typing import Optional

try:  # Python 3.9+
    from typing import Annotated  # type: ignore[attr-defined]
except ImportError:  # Python 3.8 fallback
    from typing_extensions import Annotated

import httpx
from fastapi import Depends, Header, HTTPException, status

from app.config import settings


@dataclass
class AuthContext:
    user_id: str
    access_token: str


async def get_current_user(
    authorization: Annotated[Optional[str], Header(convert_underscores=False)] = None,
    x_supabase_access_token: Annotated[
        Optional[str], Header(convert_underscores=False, alias="x-supabase-access-token")
    ] = None,
) -> AuthContext:
    """
    验证 Supabase JWT，stage0 直接依赖 service_role 解析 token。
    生产环境应通过 API Gateway 注入 user。
    """
    header_token = authorization or (f"Bearer {x_supabase_access_token}" if x_supabase_access_token else None)
    if not header_token or not header_token.startswith("Bearer "):
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Missing bearer token")

    token = header_token.replace("Bearer ", "", 1).strip()
    if not token:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Empty token")
    print(f"[auth] Validating Supabase token prefix={token[:8]}")

    auth_url = f"{settings.supabase_url.rstrip('/')}/auth/v1/user"
    apikey = settings.supabase_anon_key or settings.supabase_service_role_key
    headers = {
        "Authorization": f"Bearer {token}",
        "apikey": apikey,
    }
    async with httpx.AsyncClient(timeout=10.0) as client:
        try:
            response = await client.get(auth_url, headers=headers)
            response.raise_for_status()
        except httpx.HTTPError as exc:  # pragma: no cover - 网络异常
            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Supabase token") from exc

    data = response.json()
    user_id = data.get("id") or data.get("user", {}).get("id")
    if not user_id:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Supabase token")

    return AuthContext(user_id=str(user_id), access_token=token)


AuthDep = Annotated[AuthContext, Depends(get_current_user)]