"""Signing in to Google APIs — shared by the Google Workspace nodes
(app/crud/google_apps.py) and Firebase push (app/crud/firebase_push.py).

Two kinds of credentials, both held on a Connector:
* a service account's JSON key, optionally acting as a Workspace user
  ("delegated_user", needs domain-wide delegation);
* an OAuth client plus a refresh token ("oauth"), acting as the Google user
  who granted it.

Tokens are always requested from Google's own fixed endpoint — a key file's
"token_uri" is ignored, since a crafted key could otherwise have the signed
assertion sent anywhere. Access tokens are cached per exact credential + scope
set, so a run looping over rows doesn't sign in on every call.
"""

import asyncio
import hashlib
import json
import time
from typing import Any

import httpx
from jose import jwt

TOKEN_URL = "https://oauth2.googleapis.com/token"
TIMEOUT = 30.0
MAX_RETRIES = 2
_MAX_RETRY_WAIT = 5.0
_MAX_CACHED_TOKENS = 512


class GoogleError(Exception):
    """A problem worth showing the task's author, already in plain words."""


def _client(timeout: float = TIMEOUT) -> httpx.AsyncClient:
    """One place every call gets its HTTP client — tests swap in a MockTransport."""
    return httpx.AsyncClient(timeout=timeout)


async def _sleep(seconds: float) -> None:
    await asyncio.sleep(seconds)


_tokens: dict[str, tuple[str, float]] = {}


def service_account(config: dict) -> dict:
    """The parsed key file. Its contents never appear in an error."""
    raw = config.get("service_account_json")
    if isinstance(raw, dict):
        info = raw
    else:
        try:
            info = json.loads(str(raw or "").strip())
        except ValueError as exc:
            raise GoogleError("the service account key isn't valid JSON — paste the whole key file downloaded from Google Cloud") from exc
    if not isinstance(info, dict) or not info.get("client_email") or not info.get("private_key"):
        raise GoogleError("the service account key is missing client_email or private_key — paste the whole key file")
    return info


def _uses_oauth(config: dict) -> bool:
    return config.get("provider") == "oauth"


def _cache_key(config: dict, scopes: tuple[str, ...]) -> str:
    material = {k: config.get(k) for k in ("provider", "service_account_json", "delegated_user", "client_id", "client_secret", "refresh_token")}
    return hashlib.sha256(json.dumps([material, scopes], sort_keys=True, default=str).encode()).hexdigest()


def _token_error(resp: httpx.Response) -> str:
    try:
        body = resp.json()
    except ValueError:
        return f"HTTP {resp.status_code}"
    code, description = body.get("error", ""), body.get("error_description", "")
    if code == "invalid_grant":
        return f"Google refused the credentials ({description or code}) — the refresh token may be revoked or expired, or the key deleted"
    if code == "unauthorized_client":
        return "Google refused this account for these permissions — for a delegated user, allow these scopes under domain-wide delegation in the Workspace admin console"
    return f"Google refused the credentials: {description or code or resp.status_code}"


async def access_token(config: dict, scopes: tuple[str, ...]) -> str:
    key = _cache_key(config, scopes)
    cached = _tokens.get(key)
    if cached and cached[1] > time.time() + 60:
        return cached[0]

    if _uses_oauth(config):
        for field, label in (("client_id", "client ID"), ("client_secret", "client secret"), ("refresh_token", "refresh token")):
            if not str(config.get(field) or "").strip():
                raise GoogleError(f"the connector is missing the OAuth {label}")
        # The refresh token's permissions were fixed when it was granted.
        form = {
            "grant_type": "refresh_token",
            "client_id": str(config["client_id"]).strip(),
            "client_secret": str(config["client_secret"]).strip(),
            "refresh_token": str(config["refresh_token"]).strip(),
        }
    else:
        info = service_account(config)
        now = int(time.time())
        claims: dict[str, Any] = {"iss": info["client_email"], "scope": " ".join(scopes), "aud": TOKEN_URL, "iat": now, "exp": now + 3600}
        delegated = str(config.get("delegated_user") or "").strip()
        if delegated:
            claims["sub"] = delegated
        try:
            assertion = jwt.encode(claims, info["private_key"], algorithm="RS256")
        except Exception as exc:  # noqa: BLE001 - jose wraps bad keys in several error types
            raise GoogleError("the service account's private key couldn't be read — paste the whole key file again") from exc
        form = {"grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer", "assertion": assertion}

    async with _client() as client:
        resp = await client.post(TOKEN_URL, data=form)
    if resp.status_code >= 400:
        raise GoogleError(_token_error(resp))
    body = resp.json()
    token = body.get("access_token")
    if not token:
        raise GoogleError("Google returned no access token")
    if len(_tokens) >= _MAX_CACHED_TOKENS:
        _tokens.clear()
    _tokens[key] = (token, time.time() + float(body.get("expires_in") or 3600))
    return token


def forget_token(config: dict, scopes: tuple[str, ...]) -> None:
    _tokens.pop(_cache_key(config, scopes), None)


def account_label(config: dict) -> str:
    """Who this connector acts as, for "share the file with …" hints."""
    if _uses_oauth(config):
        return "the Google account that granted this connector"
    delegated = str(config.get("delegated_user") or "").strip()
    if delegated:
        return delegated
    try:
        return service_account(config)["client_email"]
    except GoogleError:
        return "this connector's account"


def error_reason(resp: httpx.Response) -> tuple[str, str]:
    """(status/errorCode, message) out of Google's error body."""
    try:
        error = resp.json().get("error")
    except (ValueError, AttributeError):
        return str(resp.status_code), resp.text[:300]
    if isinstance(error, str):
        return error, error
    if not isinstance(error, dict):
        return str(resp.status_code), resp.text[:300]
    code = error.get("status") or str(resp.status_code)
    for item in error.get("details") or []:
        if isinstance(item, dict) and item.get("errorCode"):
            code = item["errorCode"]
    return code, str(error.get("message") or code)[:500]


async def request(config: dict, scopes: tuple[str, ...], method: str, url: str, *, timeout: float = TIMEOUT, **kwargs: Any) -> httpx.Response:
    """One authorised call. A rate limit is waited out briefly and retried —
    Google hasn't taken the request, so that can't double a write — and an
    expired token is replaced once. Any other status is the caller's to read."""
    retries = 0
    refreshed = False
    while True:
        token = await access_token(config, scopes)
        headers = {**kwargs.pop("headers", {}), "Authorization": f"Bearer {token}"}
        async with _client(timeout) as client:
            resp = await client.request(method, url, headers=headers, **kwargs)
        kwargs["headers"] = {k: v for k, v in headers.items() if k != "Authorization"}
        if resp.status_code == 401 and not refreshed:
            refreshed = True
            forget_token(config, scopes)
            continue
        if resp.status_code == 429 and retries < MAX_RETRIES:
            retries += 1
            try:
                wait = float(resp.headers.get("retry-after") or retries)
            except ValueError:
                wait = float(retries)
            await _sleep(min(wait, _MAX_RETRY_WAIT))
            continue
        return resp