app/services/sheets_service.py

from __future__ import annotations

import asyncio
import random
import time
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional

import httpx

from app.config import get_settings
from app.core.logger import get_logger
from app.utils.http_utils import SharedAsyncClient

_logger = get_logger(__name__)
_settings = get_settings()

_RETRYABLE_STATUS = frozenset({
    int(code)
    for code in _settings.sheets_retryable_statuses.split(",")
    if code.strip().isdigit()
})


class SheetsAPIError(Exception):
    """Raised for upstream Google Sheets API failures that map to a client-facing error."""

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


@dataclass
class SheetsCredentials:
    """Client-supplied OAuth credentials (never persisted by this backend).

    ``refreshed_access_token`` is populated whenever the service transparently
    refreshes the access token, so the stateless client can persist it itself.
    """

    access_token: str
    refresh_token: Optional[str] = None
    client_id: Optional[str] = None
    client_secret: Optional[str] = None
    expires_at: Optional[float] = None
    refreshed_access_token: Optional[str] = field(default=None, repr=False)

    @property
    def can_refresh(self) -> bool:
        return bool(self.refresh_token and self.client_id and self.client_secret)

    @property
    def needs_proactive_refresh(self) -> bool:
        if self.expires_at is None:
            return False
        return time.time() >= self.expires_at - _settings.sheets_refresh_buffer_seconds


class SheetsService:
    """Stateless, fully-async abstraction over the Google Sheets REST API.

    Credentials are supplied by the caller on every invocation. Access tokens
    are transparently refreshed against Google's token endpoint when a
    ``refresh_token`` + ``client_id`` + ``client_secret`` are provided, and the
    resulting token is surfaced via ``SheetsCredentials.refreshed_access_token``.
    """

    def __init__(self) -> None:
        self._http = SharedAsyncClient(
            timeout=httpx.Timeout(_settings.sheets_timeout),
            follow_redirects=False,
        )

    async def _get_client(self) -> httpx.AsyncClient:
        return await self._http.get()

    async def close(self) -> None:
        await self._http.close()

    # ------------------------------------------------------------------
    # Core request pipeline: auth, retry, backoff, error mapping
    # ------------------------------------------------------------------

    async def _request(
        self,
        creds: SheetsCredentials,
        method: str,
        path: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Any] = None,
        base_url: Optional[str] = None,
    ) -> Any:
        """Send a Sheets/Drive API request with proactive + on-401 refresh and retries."""
        if creds.needs_proactive_refresh:
            await self._refresh(creds, context=f"{method} {path}")

        refresh_done = False
        attempt = 0
        while True:
            response = await self._send(
                creds, method, path, params=params, json_body=json_body, base_url=base_url
            )
            if 200 <= response.status_code < 300:
                return self._decode(response)

            if response.status_code == 401 and creds.can_refresh and not refresh_done:
                _logger.warning(
                    "Sheets 401 for %s %s; refreshing access token once",
                    method,
                    path,
                )
                await self._refresh(creds, context=f"{method} {path}")
                refresh_done = True
                continue

            if response.status_code in _RETRYABLE_STATUS and attempt < _settings.sheets_max_retries:
                await self._backoff(response, attempt)
                attempt += 1
                continue

            raise self._map_error(response, creds)

    async def _send(
        self,
        creds: SheetsCredentials,
        method: str,
        path: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Any] = None,
        base_url: Optional[str] = None,
    ) -> httpx.Response:
        client = await self._get_client()
        url = f"{base_url or _settings.sheets_api_base_url}{path}"
        headers = {
            "Authorization": f"Bearer {creds.access_token}",
            "Accept": "application/json",
        }
        try:
            return await client.request(
                method,
                url,
                params=params,
                json=json_body,
                headers=headers,
            )
        except httpx.TimeoutException as exc:
            raise SheetsAPIError(
                f"Sheets API request timed out for {method} {path}.", status_code=504
            ) from exc
        except httpx.RequestError as exc:
            raise SheetsAPIError(
                f"Failed to reach Sheets API: {exc}", status_code=502
            ) from exc

    @staticmethod
    def _decode(response: httpx.Response) -> Any:
        if not response.content:
            return None
        try:
            return response.json()
        except ValueError:
            return response.text

    async def _backoff(self, response: httpx.Response, attempt: int) -> None:
        retry_after = response.headers.get("Retry-After")
        delay: Optional[float] = None
        if retry_after:
            try:
                delay = float(retry_after)
            except (TypeError, ValueError):
                delay = None
        if delay is None:
            base = _settings.sheets_base_backoff_seconds * (2 ** attempt)
            delay = min(base, _settings.sheets_max_backoff_seconds) * (0.5 + random.random())
        delay = min(delay, _settings.sheets_max_backoff_seconds + 5.0)
        _logger.warning(
            "Sheets transient HTTP %s; retrying in %.2fs (attempt %d/%d)",
            response.status_code,
            delay,
            attempt + 1,
            _settings.sheets_max_retries,
        )
        await asyncio.sleep(delay)

    def _map_error(self, response: httpx.Response, creds: SheetsCredentials) -> SheetsAPIError:
        status = response.status_code
        reason = ""
        detail = ""
        try:
            body = response.json()
            error = body.get("error") or {}
            if isinstance(error, dict):
                detail = error.get("message", "")
                errors = error.get("errors") or []
                if errors and isinstance(errors[0], dict):
                    reason = errors[0].get("reason", "")
        except Exception:
            body = {}
            detail = response.text[:500]

        _logger.error(
            "Sheets API error: HTTP %s reason=%s detail=%s", status, reason, detail
        )

        if status == 401:
            if creds.can_refresh:
                return SheetsAPIError(
                    "Access token is invalid or expired and could not be refreshed. "
                    "Re-authorize via POST /api/v1/google/oauth/auth-url.",
                    status_code=401,
                    reason=reason,
                )
            return SheetsAPIError(
                "Access token is invalid or expired. Refresh it via "
                "POST /api/v1/google/oauth/refresh (or /api/v1/google/sheets/token/refresh) "
                "and retry, or supply a fresh access token.",
                status_code=401,
                reason=reason,
            )
        if status == 403 and reason == "PERMISSION_DENIED":
            return SheetsAPIError(
                "Permission denied. The access token may be missing the required "
                "Sheets scope (see /api/v1/google/sheets/scopes), or the spreadsheet "
                "is not shared with the authenticated account.",
                status_code=403,
                reason=reason,
            )
        if status == 404:
            return SheetsAPIError(
                detail or "Spreadsheet or range not found.",
                status_code=404,
                reason=reason,
            )
        if status == 429:
            return SheetsAPIError(
                "Sheets API rate limit exceeded. Please retry after a short delay.",
                status_code=429,
                reason=reason,
            )
        if status in _RETRYABLE_STATUS:
            return SheetsAPIError(
                detail or f"Sheets API service error (HTTP {status}).",
                status_code=status,
                reason=reason,
            )
        return SheetsAPIError(
            detail or f"Sheets API error (HTTP {status}).",
            status_code=status,
            reason=reason,
        )

    # ------------------------------------------------------------------
    # Token lifecycle
    # ------------------------------------------------------------------

    async def refresh_access_token(
        self,
        *,
        client_id: str,
        client_secret: str,
        refresh_token: str,
    ) -> Dict[str, Any]:
        """Refresh an access token without touching any spreadsheet (standalone)."""
        tokens = await self._refresh(
            creds=None,
            client_id=client_id,
            client_secret=client_secret,
            refresh_token=refresh_token,
            context="standalone refresh",
        )
        return tokens

    async def _refresh(
        self,
        creds: Optional[SheetsCredentials],
        *,
        context: str,
        client_id: Optional[str] = None,
        client_secret: Optional[str] = None,
        refresh_token: Optional[str] = None,
    ) -> Dict[str, Any]:
        if creds is not None:
            client_id = creds.client_id
            client_secret = creds.client_secret
            refresh_token = creds.refresh_token

        if not (client_id and client_secret and refresh_token):
            raise SheetsAPIError(
                "Token refresh requires refresh_token, client_id and client_secret.",
                status_code=400,
            )

        client = await self._get_client()
        data = {
            "client_id": client_id,
            "client_secret": client_secret,
            "refresh_token": refresh_token,
            "grant_type": "refresh_token",
        }
        try:
            response = await client.post(_settings.google_oauth_token_url, data=data)
        except httpx.TimeoutException as exc:
            raise SheetsAPIError(
                "Google token refresh timed out.", status_code=504
            ) from exc
        except httpx.RequestError as exc:
            raise SheetsAPIError(
                f"Failed to reach Google token endpoint: {exc}", status_code=502
            ) from exc

        if response.status_code == 200:
            tokens = response.json()
            new_token = tokens.get("access_token")
            if not new_token:
                raise SheetsAPIError(
                    "Google did not return an access token.", status_code=502
                )
            expires_in = int(tokens.get("expires_in", 3600))
            if creds is not None:
                creds.access_token = new_token
                creds.expires_at = time.time() + expires_in
                creds.refreshed_access_token = new_token
            _logger.info("Sheets access token refreshed (%s)", context)
            return tokens

        try:
            body = response.json()
            error = body.get("error", "")
            description = body.get("error_description", "")
        except Exception:
            error = ""
            description = ""
            body = {}

        if error == "invalid_grant":
            raise SheetsAPIError(
                description or "Refresh token is invalid, expired, or revoked. "
                "Re-authorize via POST /api/v1/google/oauth/auth-url.",
                status_code=401,
                reason=error,
            )
        if error == "invalid_client":
            raise SheetsAPIError(
                description or "Invalid client_id or client_secret.",
                status_code=401,
                reason=error,
            )
        raise SheetsAPIError(
            description or f"Google token refresh failed (HTTP {response.status_code}).",
            status_code=response.status_code if response.status_code >= 400 else 502,
            reason=error,
        )

    # ------------------------------------------------------------------
    # Scopes
    # ------------------------------------------------------------------

    async def list_available_scopes(self) -> List[Dict[str, Any]]:
        from app.models.schemas import SheetsScope

        return [
            {
                "name": scope.name,
                "uri": scope.value,
                "permission_level": scope.permission_level,
            }
            for scope in SheetsScope
        ]

    # ------------------------------------------------------------------
    # Spreadsheets
    # ------------------------------------------------------------------

    async def create_spreadsheet(
        self,
        creds: SheetsCredentials,
        *,
        title: str,
        sheets: Optional[List[Dict[str, Any]]] = None,
        properties: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"properties": {"title": title}}
        if properties:
            body["properties"].update(properties)
        if sheets:
            body["sheets"] = sheets
        return await self._request(creds, "POST", "/spreadsheets", json_body=body)

    _SPREADSHEET_MIME_TYPE = "application/vnd.google-apps.spreadsheet"

    async def get_drive_file(
        self,
        creds: SheetsCredentials,
        file_id: str,
    ) -> Dict[str, Any]:
        """Read a Drive file's metadata (drive.files.get) — used to validate the
        target before deletion so arbitrary non-sheet files cannot be removed."""
        return await self._request(
            creds,
            "GET",
            f"/files/{file_id}",
            params={"fields": "id,name,mimeType,trashed"},
            base_url=_settings.drive_api_base_url,
        )

    async def list_spreadsheets(
        self,
        creds: SheetsCredentials,
        *,
        page_size: int = 100,
        page_token: Optional[str] = None,
        include_trashed: bool = False,
    ) -> Dict[str, Any]:
        """List the user's Google Sheets via drive.files.list.

        Filters to ``application/vnd.google-apps.spreadsheet`` so only
        spreadsheets are returned (no arbitrary Drive files). The Sheets API has
        no list endpoint; Drive's ``files.list`` with a ``q`` filter is the
        documented way to enumerate spreadsheets.
        """
        query = f"mimeType='{self._SPREADSHEET_MIME_TYPE}'"
        if not include_trashed:
            query += " and trashed=false"

        params: Dict[str, Any] = {
            "q": query,
            "pageSize": page_size,
            "fields": "nextPageToken,files(id,name,mimeType,trashed,modifiedTime)",
        }
        if page_token:
            params["pageToken"] = page_token
        return await self._request(
            creds,
            "GET",
            "/files",
            params=params,
            base_url=_settings.drive_api_base_url,
        )


    async def delete_spreadsheet(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
    ) -> Dict[str, Any]:
        """Delete a spreadsheet via drive.files.delete.

        Pre-validates with drive.files.get that the target file is a Google
        Sheet (``mimeType == application/vnd.google-apps.spreadsheet``) so this
        endpoint can never remove an arbitrary Drive file. Requires the
        ``drive.file`` (or broader ``drive``) OAuth scope — the Sheets API has no
        delete method.
        """
        file_meta = await self.get_drive_file(creds, spreadsheet_id)
        mime = (file_meta or {}).get("mimeType")
        if mime != self._SPREADSHEET_MIME_TYPE:
            raise SheetsAPIError(
                f"Refusing to delete '{spreadsheet_id}': it is not a Google Sheet "
                f"(mimeType={mime or 'unknown'}). Only spreadsheets can be removed "
                "through the spreadsheet delete API.",
                status_code=400,
                reason="INVALID_MIME_TYPE",
            )

        await self._request(
            creds,
            "DELETE",
            f"/files/{spreadsheet_id}",
            base_url=_settings.drive_api_base_url,
        )
        return {
            "spreadsheetId": spreadsheet_id,
            "mimeType": mime,
            "name": file_meta.get("name"),
            "deleted": True,
        }


    async def get_spreadsheet(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        ranges: Optional[List[str]] = None,
        include_grid_data: bool = False,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if ranges:
            params["ranges"] = ranges
        if include_grid_data:
            params["includeGridData"] = "true"
        return await self._request(
            creds, "GET", f"/spreadsheets/{spreadsheet_id}", params=params or None
        )

    async def get_sheet(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        sheet_id: int,
        *,
        ranges: Optional[List[str]] = None,
        include_grid_data: bool = False,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if ranges:
            params["ranges"] = ranges
        if include_grid_data:
            params["includeGridData"] = "true"
        return await self._request(
            creds,
            "GET",
            f"/spreadsheets/{spreadsheet_id}/sheets/{sheet_id}",
            params=params or None,
        )

    async def batch_update(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        requests: List[Dict[str, Any]],
        response_include_grid_data: Optional[bool] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"requests": requests}
        if response_include_grid_data is not None:
            body["responseIncludeGridData"] = response_include_grid_data
        return await self._request(
            creds, "POST", f"/spreadsheets/{spreadsheet_id}:batchUpdate", json_body=body
        )

    # ------------------------------------------------------------------
    # Values
    # ------------------------------------------------------------------

    async def get_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        range_: str,
        *,
        major_dimension: Optional[str] = None,
        value_render_option: Optional[str] = None,
        date_time_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if major_dimension:
            params["majorDimension"] = major_dimension
        if value_render_option:
            params["valueRenderOption"] = value_render_option
        if date_time_render_option:
            params["dateTimeRenderOption"] = date_time_render_option
        return await self._request(
            creds,
            "GET",
            f"/spreadsheets/{spreadsheet_id}/values/{range_}",
            params=params or None,
        )

    async def batch_get_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        ranges: List[str],
        major_dimension: Optional[str] = None,
        value_render_option: Optional[str] = None,
        date_time_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"ranges": ranges}
        if major_dimension:
            params["majorDimension"] = major_dimension
        if value_render_option:
            params["valueRenderOption"] = value_render_option
        if date_time_render_option:
            params["dateTimeRenderOption"] = date_time_render_option
        return await self._request(
            creds,
            "GET",
            f"/spreadsheets/{spreadsheet_id}/values:batchGet",
            params=params,
        )

    async def update_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        range_: str,
        *,
        values: List[List[Any]],
        major_dimension: Optional[str] = None,
        value_input_option: str = "RAW",
        include_values_in_response: Optional[bool] = None,
        value_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"values": values}
        if major_dimension:
            body["majorDimension"] = major_dimension
        params: Dict[str, Any] = {"valueInputOption": value_input_option}
        if include_values_in_response is not None:
            params["includeValuesInResponse"] = "true" if include_values_in_response else "false"
        if value_render_option:
            params["valueRenderOption"] = value_render_option
        return await self._request(
            creds,
            "PUT",
            f"/spreadsheets/{spreadsheet_id}/values/{range_}",
            params=params,
            json_body=body,
        )

    async def append_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        range_: str,
        *,
        values: List[List[Any]],
        major_dimension: Optional[str] = None,
        value_input_option: str = "USER_ENTERED",
        insert_data_option: Optional[str] = None,
        include_values_in_response: Optional[bool] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"values": values}
        if major_dimension:
            body["majorDimension"] = major_dimension
        params: Dict[str, Any] = {"valueInputOption": value_input_option}
        if insert_data_option:
            params["insertDataOption"] = insert_data_option
        if include_values_in_response is not None:
            params["includeValuesInResponse"] = "true" if include_values_in_response else "false"
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values/{range_}:append",
            params=params,
            json_body=body,
        )

    async def clear_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        range_: str,
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values/{range_}:clear",
            json_body={},
        )

    async def batch_update_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data: List[Dict[str, Any]],
        value_input_option: str = "RAW",
        include_values_in_response: Optional[bool] = None,
        value_render_option: Optional[str] = None,
        response_value_render_option: Optional[str] = None,
        response_date_time_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"data": data}
        if include_values_in_response is not None:
            body["includeValuesInResponse"] = include_values_in_response
        if value_render_option:
            body["valueRenderOption"] = value_render_option
        if response_value_render_option:
            body["responseValueRenderOption"] = response_value_render_option
        if response_date_time_render_option:
            body["responseDateTimeRenderOption"] = response_date_time_render_option
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values:batchUpdate",
            params={"valueInputOption": value_input_option},
            json_body=body,
        )

    async def batch_clear_values(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        ranges: List[str],
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values:batchClear",
            json_body={"ranges": ranges},
        )

    async def get_spreadsheet_by_data_filter(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data_filters: List[Dict[str, Any]],
        include_grid_data: bool = False,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"dataFilters": data_filters}
        if include_grid_data:
            body["includeGridData"] = True
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}:getByDataFilter",
            json_body=body,
        )

    async def copy_sheet(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        sheet_id: int,
        *,
        destination_spreadsheet_id: str,
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/sheets/{sheet_id}:copyTo",
            json_body={"destinationSpreadsheetId": destination_spreadsheet_id},
        )

    async def batch_get_values_by_data_filter(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data_filters: List[Dict[str, Any]],
        major_dimension: Optional[str] = None,
        value_render_option: Optional[str] = None,
        date_time_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"dataFilters": data_filters}
        if major_dimension:
            body["majorDimension"] = major_dimension
        if value_render_option:
            body["valueRenderOption"] = value_render_option
        if date_time_render_option:
            body["dateTimeRenderOption"] = date_time_render_option
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values:batchGetByDataFilter",
            json_body=body,
        )

    async def batch_update_values_by_data_filter(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data: List[Dict[str, Any]],
        value_input_option: str = "RAW",
        include_values_in_response: Optional[bool] = None,
        value_render_option: Optional[str] = None,
        response_value_render_option: Optional[str] = None,
        response_date_time_render_option: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"data": data}
        if include_values_in_response is not None:
            body["includeValuesInResponse"] = include_values_in_response
        if value_render_option:
            body["valueRenderOption"] = value_render_option
        if response_value_render_option:
            body["responseValueRenderOption"] = response_value_render_option
        if response_date_time_render_option:
            body["responseDateTimeRenderOption"] = response_date_time_render_option
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values:batchUpdateByDataFilter",
            params={"valueInputOption": value_input_option},
            json_body=body,
        )

    async def batch_clear_values_by_data_filter(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data_filters: List[Dict[str, Any]],
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/values:batchClearByDataFilter",
            json_body={"dataFilters": data_filters},
        )

    async def get_developer_metadata(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        metadata_id: int,
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "GET",
            f"/spreadsheets/{spreadsheet_id}/developerMetadata/{metadata_id}",
        )

    async def search_developer_metadata(
        self,
        creds: SheetsCredentials,
        spreadsheet_id: str,
        *,
        data_filters: Optional[List[Dict[str, Any]]] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if data_filters:
            body["dataFilters"] = data_filters
        return await self._request(
            creds,
            "POST",
            f"/spreadsheets/{spreadsheet_id}/developerMetadata:search",
            json_body=body,
        )