app/services/gmail_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.models.schemas import (
    GmailComposeMessageRequest,
    GmailFilterCreateRequest,
    GmailLabelRequest,
    GmailParsedMessage,
)
from app.utils.gmail_utils import (
    GmailMessageValidationError,
    build_mime_message,
    parse_message,
)
from app.utils.http_utils import SharedAsyncClient

_logger = get_logger(__name__)
_settings = get_settings()

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


class GmailAPIError(Exception):
    """Raised for upstream Gmail 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 GmailCredentials:
    """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.gmail_refresh_buffer_seconds


class GmailService:
    """Stateless, fully-async abstraction over the Gmail 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 ``GmailCredentials.refreshed_access_token``.
    """

    def __init__(self) -> None:
        self._http = SharedAsyncClient(
            timeout=httpx.Timeout(_settings.gmail_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: GmailCredentials,
        method: str,
        path: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Any] = None,
    ) -> Any:
        """Send a Gmail 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
            )
            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(
                    "Gmail 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.gmail_max_retries:
                await self._backoff(response, attempt)
                attempt += 1
                continue

            raise self._map_error(response, creds)

    async def _send(
        self,
        creds: GmailCredentials,
        method: str,
        path: str,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Any] = None,
    ) -> httpx.Response:
        client = await self._get_client()
        url = f"{_settings.gmail_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 GmailAPIError(
                f"Gmail API request timed out for {method} {path}.", status_code=504
            ) from exc
        except httpx.RequestError as exc:
            raise GmailAPIError(
                f"Failed to reach Gmail 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.gmail_base_backoff_seconds * (2 ** attempt)
            delay = min(base, _settings.gmail_max_backoff_seconds) * (0.5 + random.random())
        delay = min(delay, _settings.gmail_max_backoff_seconds + 5.0)
        _logger.warning(
            "Gmail transient HTTP %s; retrying in %.2fs (attempt %d/%d)",
            response.status_code,
            delay,
            attempt + 1,
            _settings.gmail_max_retries,
        )
        await asyncio.sleep(delay)

    def _map_error(self, response: httpx.Response, creds: GmailCredentials) -> GmailAPIError:
        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(
            "Gmail API error: HTTP %s reason=%s detail=%s", status, reason, detail
        )

        if status == 401:
            if creds.can_refresh:
                return GmailAPIError(
                    "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 GmailAPIError(
                "Access token is invalid or expired. Refresh it via "
                "POST /api/v1/google/oauth/refresh (or /api/v1/gmail/token/refresh) "
                "and retry, or supply a fresh access token.",
                status_code=401,
                reason=reason,
            )
        if status == 403 and reason == "PERMISSION_DENIED":
            return GmailAPIError(
                "Permission denied. The access token may be missing the required "
                "Gmail scope (see /api/v1/google/gmail/scopes).",
                status_code=403,
                reason=reason,
            )
        if status == 429:
            return GmailAPIError(
                "Gmail API rate limit exceeded. Please retry after a short delay.",
                status_code=429,
                reason=reason,
            )
        if status in _RETRYABLE_STATUS:
            return GmailAPIError(
                detail or f"Gmail API service error (HTTP {status}).",
                status_code=status,
                reason=reason,
            )
        return GmailAPIError(
            detail or f"Gmail 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 the mailbox (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[GmailCredentials],
        *,
        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 GmailAPIError(
                "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 GmailAPIError(
                "Google token refresh timed out.", status_code=504
            ) from exc
        except httpx.RequestError as exc:
            raise GmailAPIError(
                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 GmailAPIError(
                    "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("Gmail 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 GmailAPIError(
                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 GmailAPIError(
                description or "Invalid client_id or client_secret.",
                status_code=401,
                reason=error,
            )
        raise GmailAPIError(
            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,
        )

    # ------------------------------------------------------------------
    # Users profile & scopes
    # ------------------------------------------------------------------

    async def get_profile(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/profile")

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

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

    # ------------------------------------------------------------------
    # Messages
    # ------------------------------------------------------------------

    async def list_messages(
        self,
        creds: GmailCredentials,
        user_id: str = "me",
        *,
        q: Optional[str] = None,
        label_ids: Optional[List[str]] = None,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
        include_spam_trash: bool = False,
        batch_size: Optional[int] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if q is not None:
            params["q"] = q
        if label_ids:
            params["labelIds"] = label_ids
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        if include_spam_trash:
            params["includeSpamTrash"] = "true"
        if batch_size is not None:
            params["batchSize"] = batch_size
        return await self._request(
            creds, "GET", f"/users/{user_id}/messages", params=params or None
        )

    async def get_message(
        self,
        creds: GmailCredentials,
        message_id: str,
        user_id: str = "me",
        *,
        format: str = "full",
        metadata_headers: Optional[List[str]] = None,
        headers: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"format": format}
        if metadata_headers:
            params["metadataHeaders"] = metadata_headers
        if headers:
            params["headers"] = headers
        return await self._request(
            creds, "GET", f"/users/{user_id}/messages/{message_id}", params=params
        )

    async def get_parsed_message(
        self,
        creds: GmailCredentials,
        message_id: str,
        user_id: str = "me",
        *,
        format: str = "full",
    ) -> GmailParsedMessage:
        data = await self.get_message(creds, message_id, user_id=user_id, format=format)
        return parse_message(data)

    async def send_raw(
        self,
        creds: GmailCredentials,
        raw: str,
        user_id: str = "me",
        *,
        thread_id: Optional[str] = None,
        labels: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"raw": raw}
        if thread_id:
            body["threadId"] = thread_id
        if labels:
            body["labelIds"] = labels
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages/send", json_body=body
        )

    async def send_composed(
        self,
        creds: GmailCredentials,
        request: GmailComposeMessageRequest,
        user_id: str = "me",
    ) -> Dict[str, Any]:
        try:
            raw = build_mime_message(request)
        except GmailMessageValidationError as exc:
            raise GmailAPIError(str(exc), status_code=400) from exc
        return await self.send_raw(
            creds,
            raw,
            user_id=user_id,
            thread_id=request.thread_id,
            labels=request.labels,
        )

    async def insert_message(
        self,
        creds: GmailCredentials,
        raw: str,
        user_id: str = "me",
        *,
        label_ids: Optional[List[str]] = None,
        internal_date_source: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"raw": raw}
        if label_ids:
            body["labelIds"] = label_ids
        if internal_date_source:
            body["internalDateSource"] = internal_date_source
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages", json_body=body
        )

    async def import_message(
        self,
        creds: GmailCredentials,
        raw: str,
        user_id: str = "me",
        *,
        label_ids: Optional[List[str]] = None,
        internal_date_source: Optional[str] = None,
        never_mark_spam: bool = False,
        process_for_calendar: bool = False,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"raw": raw}
        if label_ids:
            body["labelIds"] = label_ids
        if internal_date_source:
            body["internalDateSource"] = internal_date_source
        if never_mark_spam:
            body["neverMarkSpam"] = True
        if process_for_calendar:
            body["processForCalendar"] = True
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages/import", json_body=body
        )

    async def modify_message(
        self,
        creds: GmailCredentials,
        message_id: str,
        user_id: str = "me",
        *,
        add_label_ids: Optional[List[str]] = None,
        remove_label_ids: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if add_label_ids:
            body["addLabelIds"] = add_label_ids
        if remove_label_ids:
            body["removeLabelIds"] = remove_label_ids
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages/{message_id}/modify", json_body=body
        )

    async def batch_modify(
        self,
        creds: GmailCredentials,
        ids: List[str],
        user_id: str = "me",
        *,
        add_label_ids: Optional[List[str]] = None,
        remove_label_ids: Optional[List[str]] = None,
    ) -> None:
        body: Dict[str, Any] = {"ids": ids}
        if add_label_ids:
            body["addLabelIds"] = add_label_ids
        if remove_label_ids:
            body["removeLabelIds"] = remove_label_ids
        await self._request(creds, "POST", f"/users/{user_id}/messages/batchModify", json_body=body)

    async def batch_delete(
        self, creds: GmailCredentials, ids: List[str], user_id: str = "me"
    ) -> None:
        await self._request(
            creds, "POST", f"/users/{user_id}/messages/batchDelete", json_body={"ids": ids}
        )

    async def trash_message(
        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages/{message_id}/trash"
        )

    async def untrash_message(
        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/messages/{message_id}/untrash"
        )

    async def delete_message(
        self, creds: GmailCredentials, message_id: str, user_id: str = "me"
    ) -> None:
        await self._request(creds, "DELETE", f"/users/{user_id}/messages/{message_id}")

    async def get_attachment(
        self,
        creds: GmailCredentials,
        message_id: str,
        attachment_id: str,
        user_id: str = "me",
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "GET",
            f"/users/{user_id}/messages/{message_id}/attachments/{attachment_id}",
        )

    # ------------------------------------------------------------------
    # Push notifications (watch / stop)
    # ------------------------------------------------------------------

    async def watch(
        self,
        creds: GmailCredentials,
        user_id: str = "me",
        *,
        topic_name: str,
        label_ids: Optional[List[str]] = None,
        label_filter_action: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"topicName": topic_name}
        if label_ids:
            body["labelIds"] = label_ids
        if label_filter_action:
            body["labelFilterAction"] = label_filter_action
        return await self._request(creds, "POST", f"/users/{user_id}/watch", json_body=body)

    async def stop_watch(self, creds: GmailCredentials, user_id: str = "me") -> None:
        await self._request(creds, "POST", f"/users/{user_id}/stop")

    # ------------------------------------------------------------------
    # History
    # ------------------------------------------------------------------

    async def list_history(
        self,
        creds: GmailCredentials,
        user_id: str = "me",
        *,
        start_history_id: int,
        label_id: Optional[str] = None,
        history_types: Optional[List[str]] = None,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"startHistoryId": start_history_id}
        if label_id:
            params["labelId"] = label_id
        if history_types:
            params["historyTypes"] = history_types
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        return await self._request(creds, "GET", f"/users/{user_id}/history", params=params)

    # ------------------------------------------------------------------
    # Drafts
    # ------------------------------------------------------------------

    async def list_drafts(
        self,
        creds: GmailCredentials,
        user_id: str = "me",
        *,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
        q: Optional[str] = None,
        include_spam_trash: bool = False,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        if q:
            params["q"] = q
        if include_spam_trash:
            params["includeSpamTrash"] = "true"
        return await self._request(creds, "GET", f"/users/{user_id}/drafts", params=params or None)

    async def get_draft(
        self,
        creds: GmailCredentials,
        draft_id: str,
        user_id: str = "me",
        *,
        format: str = "full",
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "GET", f"/users/{user_id}/drafts/{draft_id}", params={"format": format}
        )

    async def create_draft(
        self,
        creds: GmailCredentials,
        raw: str,
        user_id: str = "me",
        *,
        thread_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"message": {"raw": raw}}
        if thread_id:
            body["message"]["threadId"] = thread_id
        return await self._request(creds, "POST", f"/users/{user_id}/drafts", json_body=body)

    async def update_draft(
        self,
        creds: GmailCredentials,
        draft_id: str,
        raw: str,
        user_id: str = "me",
        *,
        thread_id: Optional[str] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"message": {"raw": raw}}
        if thread_id:
            body["message"]["threadId"] = thread_id
        return await self._request(
            creds, "PUT", f"/users/{user_id}/drafts/{draft_id}", json_body=body
        )

    async def send_draft(
        self, creds: GmailCredentials, draft_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/drafts/send", json_body={"id": draft_id}
        )

    async def delete_draft(
        self, creds: GmailCredentials, draft_id: str, user_id: str = "me"
    ) -> None:
        await self._request(creds, "DELETE", f"/users/{user_id}/drafts/{draft_id}")

    # ------------------------------------------------------------------
    # Threads
    # ------------------------------------------------------------------

    async def list_threads(
        self,
        creds: GmailCredentials,
        user_id: str = "me",
        *,
        q: Optional[str] = None,
        label_ids: Optional[List[str]] = None,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
        include_spam_trash: bool = False,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if q:
            params["q"] = q
        if label_ids:
            params["labelIds"] = label_ids
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        if include_spam_trash:
            params["includeSpamTrash"] = "true"
        return await self._request(creds, "GET", f"/users/{user_id}/threads", params=params or None)

    async def get_thread(
        self,
        creds: GmailCredentials,
        thread_id: str,
        user_id: str = "me",
        *,
        format: str = "full",
        metadata_headers: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"format": format}
        if metadata_headers:
            params["metadataHeaders"] = metadata_headers
        return await self._request(
            creds, "GET", f"/users/{user_id}/threads/{thread_id}", params=params
        )

    async def modify_thread(
        self,
        creds: GmailCredentials,
        thread_id: str,
        user_id: str = "me",
        *,
        add_label_ids: Optional[List[str]] = None,
        remove_label_ids: Optional[List[str]] = None,
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {}
        if add_label_ids:
            body["addLabelIds"] = add_label_ids
        if remove_label_ids:
            body["removeLabelIds"] = remove_label_ids
        return await self._request(
            creds, "POST", f"/users/{user_id}/threads/{thread_id}/modify", json_body=body
        )

    async def trash_thread(
        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/threads/{thread_id}/trash"
        )

    async def untrash_thread(
        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/threads/{thread_id}/untrash"
        )

    async def delete_thread(
        self, creds: GmailCredentials, thread_id: str, user_id: str = "me"
    ) -> None:
        await self._request(creds, "DELETE", f"/users/{user_id}/threads/{thread_id}")

    # ------------------------------------------------------------------
    # Labels
    # ------------------------------------------------------------------

    async def list_labels(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/labels")

    async def get_label(
        self, creds: GmailCredentials, label_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/labels/{label_id}")

    async def create_label(
        self, creds: GmailCredentials, label: GmailLabelRequest, user_id: str = "me"
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"name": label.name}
        if label.label_list_visibility:
            body["labelListVisibility"] = label.label_list_visibility
        if label.message_list_visibility:
            body["messageListVisibility"] = label.message_list_visibility
        if label.color:
            body["color"] = label.color
        return await self._request(creds, "POST", f"/users/{user_id}/labels", json_body=body)

    async def update_label(
        self,
        creds: GmailCredentials,
        label_id: str,
        label: GmailLabelRequest,
        user_id: str = "me",
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"name": label.name}
        if label.label_list_visibility:
            body["labelListVisibility"] = label.label_list_visibility
        if label.message_list_visibility:
            body["messageListVisibility"] = label.message_list_visibility
        if label.color:
            body["color"] = label.color
        return await self._request(
            creds, "PUT", f"/users/{user_id}/labels/{label_id}", json_body=body
        )

    async def patch_label(
        self,
        creds: GmailCredentials,
        label_id: str,
        payload: Dict[str, Any],
        user_id: str = "me",
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PATCH", f"/users/{user_id}/labels/{label_id}", json_body=payload
        )

    async def delete_label(
        self, creds: GmailCredentials, label_id: str, user_id: str = "me"
    ) -> None:
        await self._request(creds, "DELETE", f"/users/{user_id}/labels/{label_id}")

    # ------------------------------------------------------------------
    # Settings: auto-forwarding / vacation / filters / forwarding / send-as / delegates
    # ------------------------------------------------------------------

    async def get_auto_forwarding(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/autoForwarding")

    async def update_auto_forwarding(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PUT", f"/users/{user_id}/settings/autoForwarding", json_body=payload
        )

    async def get_vacation(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/vacation")

    async def update_vacation(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PUT", f"/users/{user_id}/settings/vacation", json_body=payload
        )

    async def get_imap(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/imap")

    async def update_imap(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PUT", f"/users/{user_id}/settings/imap", json_body=payload
        )

    async def get_pop(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/pop")

    async def update_pop(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PUT", f"/users/{user_id}/settings/pop", json_body=payload
        )

    async def get_language(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/language")

    async def update_language(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PUT", f"/users/{user_id}/settings/language", json_body=payload
        )

    async def list_filters(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/filters")

    async def get_filter(
        self, creds: GmailCredentials, filter_id: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/filters/{filter_id}")

    async def create_filter(
        self, creds: GmailCredentials, req: GmailFilterCreateRequest, user_id: str = "me"
    ) -> Dict[str, Any]:
        body: Dict[str, Any] = {"criteria": req.criteria or {}, "action": req.action or {}}
        return await self._request(creds, "POST", f"/users/{user_id}/settings/filters", json_body=body)

    async def delete_filter(
        self, creds: GmailCredentials, filter_id: str, user_id: str = "me"
    ) -> None:
        await self._request(creds, "DELETE", f"/users/{user_id}/settings/filters/{filter_id}")

    async def list_forwarding_addresses(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/forwardingAddresses")

    async def get_forwarding_address(
        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "GET", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}"
        )

    async def create_forwarding_address(
        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/users/{user_id}/settings/forwardingAddresses",
            json_body={"forwardingEmail": forwarding_email},
        )

    async def delete_forwarding_address(
        self, creds: GmailCredentials, forwarding_email: str, user_id: str = "me"
    ) -> None:
        await self._request(
            creds, "DELETE", f"/users/{user_id}/settings/forwardingAddresses/{forwarding_email}"
        )

    async def list_send_as(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/sendAs")

    async def get_send_as(
        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}"
        )

    async def create_send_as(
        self, creds: GmailCredentials, payload: Dict[str, Any], user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "POST", f"/users/{user_id}/settings/sendAs", json_body=payload
        )

    async def update_send_as(
        self,
        creds: GmailCredentials,
        send_as_email: str,
        payload: Dict[str, Any],
        user_id: str = "me",
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "PATCH", f"/users/{user_id}/settings/sendAs/{send_as_email}", json_body=payload
        )

    async def verify_send_as(
        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
    ) -> None:
        await self._request(
            creds, "POST", f"/users/{user_id}/settings/sendAs/{send_as_email}/verify"
        )

    async def delete_send_as(
        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
    ) -> None:
        await self._request(
            creds, "DELETE", f"/users/{user_id}/settings/sendAs/{send_as_email}"
        )

    async def list_delegates(self, creds: GmailCredentials, user_id: str = "me") -> Dict[str, Any]:
        return await self._request(creds, "GET", f"/users/{user_id}/settings/delegates")

    async def get_delegate(
        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "GET", f"/users/{user_id}/settings/delegates/{delegate_email}"
        )

    async def create_delegate(
        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/users/{user_id}/settings/delegates",
            json_body={"delegateEmail": delegate_email},
        )

    async def delete_delegate(
        self, creds: GmailCredentials, delegate_email: str, user_id: str = "me"
    ) -> None:
        await self._request(
            creds, "DELETE", f"/users/{user_id}/settings/delegates/{delegate_email}"
        )

    async def list_smime_info(
        self, creds: GmailCredentials, send_as_email: str, user_id: str = "me"
    ) -> Dict[str, Any]:
        return await self._request(
            creds, "GET", f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo"
        )

    async def get_smime_info(
        self,
        creds: GmailCredentials,
        send_as_email: str,
        smime_info_id: str,
        user_id: str = "me",
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "GET",
            f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}",
        )

    async def insert_smime_info(
        self,
        creds: GmailCredentials,
        send_as_email: str,
        payload: Dict[str, Any],
        user_id: str = "me",
    ) -> Dict[str, Any]:
        return await self._request(
            creds,
            "POST",
            f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo",
            json_body=payload,
        )

    async def delete_smime_info(
        self,
        creds: GmailCredentials,
        send_as_email: str,
        smime_info_id: str,
        user_id: str = "me",
    ) -> None:
        await self._request(
            creds,
            "DELETE",
            f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}",
        )

    async def set_default_smime_info(
        self,
        creds: GmailCredentials,
        send_as_email: str,
        smime_info_id: str,
        user_id: str = "me",
    ) -> None:
        await self._request(
            creds,
            "POST",
            f"/users/{user_id}/settings/sendAs/{send_as_email}/smimeInfo/{smime_info_id}/setDefault",
        )