backend/auth.py

"""
Google OAuth token verification and lightweight user store.

Uses google-auth to verify JWT credentials issued by Google Identity Services.
User profiles are stored in a JSON file (/tmp) — ephemeral on HF Spaces,
which is acceptable for a demo. Client-side localStorage persists the session.
"""

import json
import os
from pathlib import Path
from datetime import datetime, timezone

from google.oauth2 import id_token
from google.auth.transport import requests as google_requests

GOOGLE_CLIENT_ID = os.getenv("GOOGLE_CLIENT_ID", "")
USER_STORE_PATH = Path(os.getenv("USER_STORE_PATH", "/tmp/urbanflow_users.json"))


def _load_users() -> dict:
    if USER_STORE_PATH.exists():
        return json.loads(USER_STORE_PATH.read_text())
    return {}


def _save_users(users: dict):
    USER_STORE_PATH.write_text(json.dumps(users, indent=2))


def verify_google_token(credential: str) -> dict:
    """
    Verify a Google ID token (JWT) and return the decoded payload.
    Raises ValueError on invalid/expired tokens.
    """
    idinfo = id_token.verify_oauth2_token(
        credential,
        google_requests.Request(),
        GOOGLE_CLIENT_ID,
    )
    return {
        "email": idinfo["email"],
        "name": idinfo.get("name", ""),
        "picture": idinfo.get("picture", ""),
    }


def get_or_create_user(email: str, name: str, picture: str) -> dict:
    """
    Look up a user by email. If they don't exist, create a stub record.
    Returns the user record with a `new_user` flag.
    """
    users = _load_users()
    is_new = email not in users

    if is_new:
        users[email] = {
            "username": "",
            "name": name,
            "picture": picture,
            "created_at": datetime.now(timezone.utc).isoformat(),
        }
        _save_users(users)

    user = users[email]
    return {
        "email": email,
        "username": user.get("username", ""),
        "name": user.get("name", name),
        "picture": user.get("picture", picture),
        "new_user": is_new or not user.get("username"),
    }


def set_username(email: str, username: str) -> bool:
    """
    Save a display username for a first-time user.
    Returns True on success, False if the user record doesn't exist.
    """
    users = _load_users()
    if email not in users:
        return False
    users[email]["username"] = username.strip()
    _save_users(users)
    return True