app/services/google_scope_service.py

from __future__ import annotations

from typing import Any, Dict, List, Optional

from app.services.google_scopes_data import GOOGLE_APIS

# OIDC short names that are conventionally sent without a full URI. They are
# already present as scope names for the oauth_signin plan, but accepting them
# explicitly keeps validation predictable for all APIs.
_OIDC_SHORT_NAMES = frozenset({"openid", "email", "profile"})


class GoogleScopeError(Exception):
    """Raised for invalid scope configuration requests."""

    def __init__(self, message: str, status_code: int = 400) -> None:
        super().__init__(message)
        self.message = message
        self.status_code = status_code


def _default_scopes(plan: Dict[str, Any]) -> List[str]:
    """Derive the recommended default scopes for a plan."""
    pn = plan.get("production_notes") or {}
    dflt = pn.get("default_scope_request")
    if dflt:
        return [s.strip() for s in dflt.split() if s.strip()]
    recs = plan.get("recommended_combinations") or {}
    if recs:
        first = next(iter(recs.values()))
        return list(first)
    return []


def _allowed_scopes(plan: Dict[str, Any]) -> frozenset:
    """Set of scope identifiers (URI or short name) accepted for a plan."""
    allowed = set(_OIDC_SHORT_NAMES)
    for scope in plan.get("scopes") or []:
        if scope.get("uri"):
            allowed.add(scope["uri"])
        if scope.get("name"):
            allowed.add(scope["name"])
    return frozenset(allowed)


class GoogleScopeService:
    """Serves the Google per-API scope catalog entirely from memory.

    The catalog is a Python object structure (``google_scopes_data.GOOGLE_APIS``)
    and the currently selected scopes are held as an in-memory dict of objects.
    No JSON files are read or written.
    """

    def __init__(self) -> None:
        self._plans: Dict[str, Dict[str, Any]] = dict(GOOGLE_APIS)
        self._current_scopes: Optional[Dict[str, List[str]]] = None

    @property
    def catalog(self) -> Dict[str, Dict[str, Any]]:
        return self._plans

    @property
    def current_scopes(self) -> Dict[str, List[str]]:
        return self._state()

    def _state(self) -> Dict[str, List[str]]:
        if self._current_scopes is None:
            self._current_scopes = {
                pid: _default_scopes(plan) for pid, plan in self._plans.items()
            }
        return self._current_scopes

    async def load_all_plans(self) -> Dict[str, Dict[str, Any]]:
        return self._plans

    async def load_state(self) -> Dict[str, List[str]]:
        return self._state()

    async def set_current_scopes(
        self, updates: Dict[str, List[str]], reset: bool = False
    ) -> Dict[str, List[str]]:
        """Validate and store the requested current scopes in memory.

        Raises GoogleScopeError with a 400 status for unknown API ids or scopes
        that are not defined in the catalog.
        """
        if reset:
            self._current_scopes = {
                pid: _default_scopes(plan) for pid, plan in self._plans.items()
            }
            return self._current_scopes

        current = self._state()
        errors: List[str] = []
        for pid, scopes in (updates or {}).items():
            if pid not in self._plans:
                errors.append(f"Unknown API id '{pid}'. Valid ids: {', '.join(sorted(self._plans))}")
                continue
            allowed = _allowed_scopes(self._plans[pid])
            cleaned = [str(s).strip() for s in scopes if str(s).strip()]
            bad = [s for s in cleaned if s not in allowed]
            if bad:
                errors.append(f"Invalid scopes for '{pid}': {', '.join(bad)}")
                continue
            current[pid] = cleaned

        if errors:
            raise GoogleScopeError("; ".join(errors), status_code=400)

        self._current_scopes = current
        return current

    @staticmethod
    def _scope_text(scope: Dict[str, Any]) -> str:
        return " ".join(
            str(scope.get(k) or "") for k in ("name", "uri", "description")
        ).lower()

    @staticmethod
    def _scope_matches(
        scope: Dict[str, Any],
        *,
        scope_name: Optional[str] = None,
        permission_level: Optional[List[str]] = None,
        required: Optional[bool] = None,
        recommended: Optional[bool] = None,
    ) -> bool:
        if scope_name:
            needle = scope_name.lower()
            if needle not in (str(scope.get("name") or "").lower()) and needle not in (
                str(scope.get("uri") or "").lower()
            ):
                return False
        if permission_level:
            level = str(scope.get("permission_level") or "").lower()
            if level not in {p.lower() for p in permission_level}:
                return False
        if required is not None and bool(scope.get("required")) is not required:
            return False
        if recommended is not None and bool(scope.get("recommended")) is not recommended:
            return False
        return True

    @classmethod
    def filter_data(
        cls,
        data: List[Dict[str, Any]],
        *,
        api_name: Optional[str] = None,
        oauth_only: bool = False,
        no_scope: bool = False,
        search: Optional[str] = None,
        scope_name: Optional[str] = None,
        permission_level: Optional[List[str]] = None,
        required: Optional[bool] = None,
        recommended: Optional[bool] = None,
        scope_name_only: bool = False,
    ) -> List[Dict[str, Any]]:
        """Apply discoverability filters to the built catalog entries.

        - ``api_name`` keeps a single API (case-insensitive id match).
        - ``oauth_only`` / ``no_scope`` filter by whether scopes are required.
        - ``search`` free-text matches the API id/name and scope fields.
        - ``scope_name``, ``permission_level``, ``required``, ``recommended``
          filter the ``scopes`` array of each entry; entries left with zero
          matching scopes are dropped unless ``api_name`` was requested.
        - ``scope_name_only`` replaces each ``scopes`` array with the list of
          scope names (falling back to the URI).
        """
        api_name_l = api_name.strip().lower() if api_name else None
        scope_filters_active = any(
            [scope_name, permission_level, required is not None, recommended is not None]
        )
        needle = search.strip().lower() if search else None
        levels = [p.strip().lower() for p in (permission_level or []) if p.strip()]

        filtered: List[Dict[str, Any]] = []
        for entry in data:
            if api_name_l and str(entry.get("id") or "").lower() != api_name_l:
                continue
            if oauth_only and not entry.get("scope_required"):
                continue
            if no_scope and entry.get("scope_required"):
                continue

            haystack = (
                f"{entry.get('id') or ''} {entry.get('api') or ''} "
                + " ".join(cls._scope_text(s) for s in entry.get("scopes") or [])
            ).lower()
            if needle and needle not in haystack:
                continue

            scopes = entry.get("scopes") or []
            if scope_filters_active:
                scopes = [
                    s for s in scopes
                    if cls._scope_matches(
                        s,
                        scope_name=scope_name,
                        permission_level=levels or None,
                        required=required,
                        recommended=recommended,
                    )
                ]
                if not scopes and not api_name_l:
                    continue

            item = dict(entry)
            if scope_name_only:
                item["scopes"] = [str(s.get("name") or s.get("uri")) for s in scopes]
            else:
                item["scopes"] = scopes
            filtered.append(item)
        return filtered

    @staticmethod
    def build_data(
        plans: Dict[str, Dict[str, Any]],
        state: Dict[str, List[str]],
    ) -> List[Dict[str, Any]]:
        """Merge catalog plans with current scopes into the API response payload."""
        data: List[Dict[str, Any]] = []
        for pid in sorted(plans):
            plan = plans[pid]
            data.append(
                {
                    "id": pid,
                    "api": plan.get("api"),
                    "version": plan.get("version"),
                    "base_url": plan.get("base_url"),
                    "batch_endpoint": plan.get("batch_endpoint"),
                    "auth_model": plan.get("auth_model"),
                    "scope_required": plan.get("scope_required", False),
                    "docs_url": plan.get("docs_url"),
                    "info": plan.get("info"),
                    "current_scopes": state.get(pid, []),
                    "default_scopes": _default_scopes(plan),
                    "scopes": plan.get("scopes", []),
                    "recommended_combinations": plan.get("recommended_combinations", {}),
                }
            )
        return data