from __future__ import annotations

import asyncio
import base64
import hashlib
import json
import re
import time
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import quote, urlencode

import httpx
from pydantic import BaseModel, ValidationError

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

_logger = get_logger(__name__)
_settings = get_settings()

# Scopes needed to perform full object/bucket management via the JSON API.
_GCS_SCOPES = (
    "https://www.googleapis.com/auth/devstorage.full_control "
    "https://www.googleapis.com/auth/cloud-platform"
)

# Transient HTTP status codes that are safe to retry against Google.
_RETRYABLE_STATUS = frozenset({429, 500, 502, 503, 504})

# Cloud Storage bucket naming rules (3-63 chars, lowercase letters/digits/_/.-).
BUCKET_NAME_PATTERN = re.compile(r"^[a-z0-9][a-z0-9._-]{1,61}[a-z0-9]$")

_SIGNING_ALGORITHM = "GOOG4-RSA-SHA256"
_SIGNING_REGION = "auto"
_SIGNING_SERVICE = "storage"


class GCSCredentialsError(Exception):
    """Raised when service-account credentials are missing or malformed."""

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


class GCSError(Exception):
    """Raised for upstream Cloud Storage failures mapped to client-facing errors."""

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


class GCSCredentials(BaseModel):
    """Validated Google service-account JSON key material."""

    type: str = "service_account"
    project_id: str = ""
    private_key_id: str = ""
    private_key: str
    client_email: str
    client_id: str = ""
    auth_uri: str = "https://accounts.google.com/o/oauth2/auth"
    token_uri: str = "https://oauth2.googleapis.com/token"
    auth_provider_x509_cert_url: str = "https://www.googleapis.com/oauth2/v1/certs"
    client_x509_cert_url: str = ""
    universe_domain: str = "googleapis.com"

    @property
    def scope_key(self) -> str:
        """Unique key used to share cached access tokens across requests."""
        return f"{self.client_email}|{self.private_key_id}"


class _TokenCache:
    """In-memory cache of short-lived Google access tokens, keyed per service account."""

    def __init__(self) -> None:
        self._entries: Dict[str, Dict[str, Any]] = {}
        self._lock = asyncio.Lock()

    def _key(self, creds: GCSCredentials) -> str:
        return f"{creds.client_email}|{creds.private_key_id}"

    def get(self, creds: GCSCredentials) -> Optional[str]:
        entry = self._entries.get(self._key(creds))
        if entry and entry["expires_at"] > time.time():
            return entry["token"]
        return None

    async def set(self, creds: GCSCredentials, token: str, expires_in: int) -> None:
        self._entries[self._key(creds)] = {
            "token": token,
            "expires_at": time.time() + max(expires_in - 60, 60),
        }


class GCSService:
    """Asynchronous, connection-pooled client for the Google Cloud Storage JSON API."""

    def __init__(self) -> None:
        self._http = SharedAsyncClient(timeout=_settings.gcs_timeout)
        self._token_cache = _TokenCache()

    # ------------------------------------------------------------------
    # HTTP client management
    # ------------------------------------------------------------------

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

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

    # ------------------------------------------------------------------
    # Credential resolution (JSON body / JSON string / URL / file / env)
    # ------------------------------------------------------------------

    async def resolve_credentials(
        self,
        *,
        payload: Optional[Any] = None,
        url: Optional[str] = None,
        file_bytes: Optional[bytes] = None,
        file_name: Optional[str] = None,
    ) -> GCSCredentials:
        """Resolve a validated service-account credential from any supported source.

        Resolution precedence: inline ``payload`` (dict or JSON string), then a
        URL to fetch, then uploaded ``file_bytes``, then the env-configured path.
        """
        raw: Optional[str] = None
        source: str = ""

        if file_bytes is not None:
            source = f"uploaded file '{file_name or 'credentials'}'"
            raw = file_bytes.decode("utf-8")
        elif payload is not None:
            if isinstance(payload, dict):
                return self._validate_credentials(payload, "request body")
            if isinstance(payload, str) and payload.strip():
                source = "request body"
                raw = payload
            else:
                raise GCSCredentialsError(
                    "credentials must be a JSON object or a JSON string."
                )
        elif url:
            source = f"url '{url}'"
            raw = await self._fetch_credentials_url(url)
        elif _settings.gcs_service_account_key_path:
            source = f"file '{_settings.gcs_service_account_key_path}'"
            try:
                with open(_settings.gcs_service_account_key_path, "r", encoding="utf-8") as f:
                    raw = f.read()
            except OSError as exc:
                raise GCSCredentialsError(
                    f"Unable to read service account file at "
                    f"'{_settings.gcs_service_account_key_path}': {exc}"
                )
        else:
            raise GCSCredentialsError(
                "No GCS service account credentials provided. Pass 'credentials' "
                "(JSON object) or 'credentials_json' in the body, 'credentials_url', "
                "upload a credentials file, or set GCS_SERVICE_ACCOUNT_KEY_PATH."
            )

        try:
            data = json.loads(raw)
        except (json.JSONDecodeError, TypeError) as exc:
            raise GCSCredentialsError(
                f"Service account data from {source} is not valid JSON: {exc}"
            )
        return self._validate_credentials(data, source)

    @staticmethod
    def _validate_credentials(data: Any, source: str) -> GCSCredentials:
        if not isinstance(data, dict):
            raise GCSCredentialsError(f"Service account data from {source} must be a JSON object.")
        try:
            return GCSCredentials(**data)
        except ValidationError as exc:
            raise GCSCredentialsError(
                f"Service account data from {source} is missing required fields: "
                f"{', '.join(e['loc'][0] for e in exc.errors())}"
            )

    @staticmethod
    def _validate_http_url(url: str) -> None:
        if not isinstance(url, str) or not url.strip():
            raise GCSCredentialsError("A non-empty URL string is required.")
        if len(url) > 2048:
            raise GCSCredentialsError("URL must be at most 2048 characters.")
        parts = url.split("://", 1)
        if len(parts) != 2 or parts[0].lower() not in ("http", "https"):
            raise GCSCredentialsError("URL must use the http or https scheme.")

    async def _fetch_credentials_url(self, url: str) -> str:
        self._validate_http_url(url)
        client = await self._get_client()
        try:
            response = await client.get(url)
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise GCSCredentialsError(f"Failed to fetch service account JSON from '{url}': {exc}")
        return response.text

    async def fetch_url_content(self, url: str) -> Tuple[bytes, str]:
        """Fetch arbitrary content from a URL for use as an object body."""
        self._validate_http_url(url)
        client = await self._get_client()
        try:
            response = await client.get(url)
            response.raise_for_status()
        except httpx.HTTPError as exc:
            raise GCSError(f"Failed to fetch content from '{url}': {exc}", status_code=400)
        return response.content, response.headers.get("content-type", "application/octet-stream")

    # ------------------------------------------------------------------
    # Access token acquisition (service account JWT -> bearer token)
    # ------------------------------------------------------------------

    async def get_access_token(self, creds: GCSCredentials) -> str:
        cached = self._token_cache.get(creds)
        if cached:
            return cached
        async with self._token_cache._lock:
            cached = self._token_cache.get(creds)
            if cached:
                return cached
            token, expires_in = await self._fetch_access_token(creds)
            await self._token_cache.set(creds, token, expires_in)
            return token

    async def _fetch_access_token(self, creds: GCSCredentials) -> Tuple[str, int]:
        assertion = await run_in_executor(self._build_signed_jwt, creds)
        body = urlencode({
            "grant_type": "urn:ietf:params:oauth:grant-type:jwt-bearer",
            "assertion": assertion,
        })
        client = await self._get_client()
        last_error: Optional[str] = None
        for attempt in range(1 + _settings.gcs_max_retries):
            try:
                response = await client.request(
                    "POST",
                    creds.token_uri or _settings.gcs_token_uri,
                    content=body,
                    headers={"Content-Type": "application/x-www-form-urlencoded"},
                )
                if response.status_code == 200:
                    data = response.json()
                    token: str = data.get("access_token", "")
                    if not token:
                        raise GCSCredentialsError(
                            "Google did not return an access token for the service account."
                        )
                    return token, int(data.get("expires_in", 3600))

                last_error = self._format_token_error(response)
                _logger.warning(
                    "GCS token exchange failed (attempt %d/%d): %s",
                    attempt + 1, 1 + _settings.gcs_max_retries, last_error,
                )
                if response.status_code in _RETRYABLE_STATUS and attempt < _settings.gcs_max_retries:
                    await asyncio.sleep(2.0 ** attempt)
                    continue
                raise GCSCredentialsError(last_error)

            except httpx.TimeoutException:
                last_error = "Google token endpoint timed out."
            except httpx.RequestError as exc:
                last_error = f"Google token endpoint request failed: {exc}"
            except GCSCredentialsError:
                raise
            except Exception as exc:
                last_error = f"Unexpected error during token exchange: {exc}"

            if attempt < _settings.gcs_max_retries:
                await asyncio.sleep(2.0 ** attempt)

        raise GCSCredentialsError(last_error or "Unknown token exchange failure.")

    @staticmethod
    def _format_token_error(response: httpx.Response) -> str:
        try:
            body = response.json()
            error_desc = body.get("error_description", "") or body.get("error", "")
            if error_desc:
                return error_desc if isinstance(error_desc, str) else str(error_desc)
        except Exception:
            pass
        return f"Google OAuth token exchange error (HTTP {response.status_code})."

    def _build_signed_jwt(self, creds: GCSCredentials) -> str:
        now = int(time.time())
        header = {"alg": "RS256", "typ": "JWT"}
        claims = {
            "iss": creds.client_email,
            "scope": _GCS_SCOPES,
            "aud": creds.token_uri or _settings.gcs_token_uri,
            "iat": now,
            "exp": now + 3600,
        }
        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import padding
        from cryptography.hazmat.primitives.serialization import load_pem_private_key

        signing_input = (
            base64.urlsafe_b64encode(json.dumps(header, separators=(",", ":")).encode()).rstrip(b"=")
            + b"."
            + base64.urlsafe_b64encode(json.dumps(claims, separators=(",", ":")).encode()).rstrip(b"=")
        )
        private_key = load_pem_private_key(creds.private_key.encode("utf-8"), password=None)
        signature = private_key.sign(signing_input, padding.PKCS1v15(), hashes.SHA256())
        return (
            signing_input
            + b"."
            + base64.urlsafe_b64encode(signature).rstrip(b"=")
        ).decode("ascii")

    # ------------------------------------------------------------------
    # Generic authenticated JSON API request
    # ------------------------------------------------------------------

    async def _request(
        self,
        method: str,
        path: str,
        creds: GCSCredentials,
        *,
        params: Optional[Dict[str, Any]] = None,
        json_body: Optional[Any] = None,
        extra_headers: Optional[Dict[str, str]] = None,
        content: Optional[bytes] = None,
        upload: bool = False,
    ) -> Dict[str, Any]:
        token = await self.get_access_token(creds)
        base_url = _settings.gcs_upload_base_url if upload else _settings.gcs_api_base_url
        url = f"{base_url}{path}"
        headers = {"Authorization": f"Bearer {token}"}
        if extra_headers:
            headers.update(extra_headers)

        client = await self._get_client()
        last_error: Optional[str] = None
        last_status: int = 0
        for attempt in range(1 + _settings.gcs_max_retries):
            try:
                response = await client.request(
                    method,
                    url,
                    params=params,
                    json=json_body,
                    headers=headers,
                    content=content,
                )
                if response.status_code < 400:
                    return self._success_result(response)

                last_status = response.status_code
                last_error = self._format_gcs_error(response)
                _logger.warning(
                    "GCS API error on %s %s: HTTP %s -> %s",
                    method, path, response.status_code, last_error,
                )

                if response.status_code in _RETRYABLE_STATUS:
                    if attempt < _settings.gcs_max_retries:
                        await asyncio.sleep(2.0 ** attempt)
                    continue
                if 400 <= response.status_code < 500:
                    raise GCSError(last_error, status_code=response.status_code)

            except GCSError:
                raise
            except httpx.TimeoutException:
                last_error = "Request timed out"
                _logger.warning("GCS API timeout on %s %s (attempt %d/%d)", method, path, attempt + 1, 1 + _settings.gcs_max_retries)
            except httpx.RequestError as exc:
                last_error = f"Request failed: {exc}"
                _logger.warning("GCS API request error on %s %s: %s (attempt %d/%d)", method, path, last_error, attempt + 1, 1 + _settings.gcs_max_retries)
            except Exception as exc:
                last_error = f"Unexpected error: {exc}"
                _logger.error("GCS API unexpected error on %s %s: %s", method, path, last_error)
                break

            if attempt < _settings.gcs_max_retries:
                await asyncio.sleep(1.0 * (attempt + 1))

        raise GCSError(last_error or f"GCS API error (HTTP {last_status}).", status_code=502)

    @staticmethod
    def _success_result(response: httpx.Response) -> Dict[str, Any]:
        content_type = response.headers.get("content-type", "")
        result: Dict[str, Any] = {
            "success": True,
            "status_code": response.status_code,
            "content_type": content_type,
            "headers": dict(response.headers),
            "error": None,
        }
        if response.content and content_type.startswith("application/json"):
            result["data"] = response.json()
        else:
            result["data"] = response.content
        return result

    @staticmethod
    def _format_gcs_error(response: httpx.Response) -> str:
        status = response.status_code
        message = ""
        reason = ""
        try:
            body = response.json()
            error = body.get("error", {}) if isinstance(body, dict) else {}
            message = error.get("message", "") or ""
            errors = error.get("errors", [])
            if errors and isinstance(errors[0], dict):
                reason = errors[0].get("reason", "") or ""
        except Exception:
            pass

        if status == 401:
            return "Invalid or expired Google service account credentials."
        if status == 403:
            if "permission" in message.lower() or reason in ("forbidden", "required", "storagePermissionDenied"):
                return "Permission denied. The service account lacks the required IAM role for this operation."
            return f"Access forbidden: {message}".rstrip(".")
        if status == 404:
            return "The requested bucket or object was not found."
        if status == 409:
            return "Conflict. A bucket or object with the same name already exists, or a generation condition failed."
        if status == 429:
            return "API rate limit exceeded. Please wait and retry."
        if message:
            return message.rstrip(".") + "."
        return f"Google Cloud Storage error (HTTP {status})."

    # ------------------------------------------------------------------
    # Buckets
    # ------------------------------------------------------------------

    async def create_bucket(
        self,
        creds: GCSCredentials,
        name: str,
        *,
        project: Optional[str] = None,
        bucket_body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        project_id = project or creds.project_id
        body: Dict[str, Any] = dict(bucket_body or {})
        body["name"] = name
        params = {"project": project_id}
        return await self._request("POST", "/b", creds, params=params, json_body=body)

    async def get_bucket(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
        return await self._request("GET", f"/b/{quote(bucket, safe='')}", creds)

    async def list_buckets(
        self,
        creds: GCSCredentials,
        *,
        project: Optional[str] = None,
        prefix: Optional[str] = None,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"project": project or creds.project_id}
        if prefix:
            params["prefix"] = prefix
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        return await self._request("GET", "/b", creds, params=params)

    async def patch_bucket(
        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
        *, if_metageneration_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
        return await self._request("PATCH", f"/b/{quote(bucket, safe='')}", creds, params=params, json_body=body)

    async def update_bucket(
        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
        *, if_metageneration_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
        return await self._request("PUT", f"/b/{quote(bucket, safe='')}", creds, params=params, json_body=body)

    async def delete_bucket(
        self, creds: GCSCredentials, bucket: str,
        *, if_metageneration_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
        return await self._request("DELETE", f"/b/{quote(bucket, safe='')}", creds, params=params)

    # ------------------------------------------------------------------
    # Bucket IAM & permissions
    # ------------------------------------------------------------------

    async def get_bucket_iam(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
        return await self._request("GET", f"/b/{quote(bucket, safe='')}/iam", creds)

    async def set_bucket_iam(
        self, creds: GCSCredentials, bucket: str, policy: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request("PUT", f"/b/{quote(bucket, safe='')}/iam", creds, json_body=policy)

    async def test_bucket_permissions(
        self, creds: GCSCredentials, bucket: str, permissions: List[str],
    ) -> Dict[str, Any]:
        params = {"permissions": permissions}
        return await self._request("GET", f"/b/{quote(bucket, safe='')}/iam/testPermissions", creds, params=params)

    # ------------------------------------------------------------------
    # Default object ACLs
    # ------------------------------------------------------------------

    async def list_default_object_acl(self, creds: GCSCredentials, bucket: str) -> Dict[str, Any]:
        return await self._request("GET", f"/b/{quote(bucket, safe='')}/defaultObjectAcl", creds)

    async def insert_default_object_acl(
        self, creds: GCSCredentials, bucket: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request("POST", f"/b/{quote(bucket, safe='')}/defaultObjectAcl", creds, json_body=body)

    async def get_default_object_acl(
        self, creds: GCSCredentials, bucket: str, entity: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "GET", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds
        )

    async def patch_default_object_acl(
        self, creds: GCSCredentials, bucket: str, entity: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "PATCH", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds, json_body=body
        )

    async def update_default_object_acl(
        self, creds: GCSCredentials, bucket: str, entity: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "PUT", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds, json_body=body
        )

    async def delete_default_object_acl(
        self, creds: GCSCredentials, bucket: str, entity: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "DELETE", f"/b/{quote(bucket, safe='')}/defaultObjectAcl/{quote(entity, safe='')}", creds
        )

    # ------------------------------------------------------------------
    # Objects
    # ------------------------------------------------------------------

    async def list_objects(
        self,
        creds: GCSCredentials,
        bucket: str,
        *,
        prefix: Optional[str] = None,
        delimiter: Optional[str] = None,
        max_results: Optional[int] = None,
        page_token: Optional[str] = None,
        versions: Optional[bool] = None,
        match_glob: Optional[str] = None,
        start_offset: Optional[str] = None,
        end_offset: Optional[str] = None,
        include_trailing_delimiter: Optional[bool] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if prefix is not None:
            params["prefix"] = prefix
        if delimiter is not None:
            params["delimiter"] = delimiter
        if max_results is not None:
            params["maxResults"] = max_results
        if page_token:
            params["pageToken"] = page_token
        if versions is not None:
            params["versions"] = versions
        if match_glob:
            params["matchGlob"] = match_glob
        if start_offset is not None:
            params["startOffset"] = start_offset
        if end_offset is not None:
            params["endOffset"] = end_offset
        if include_trailing_delimiter is not None:
            params["includeTrailingDelimiter"] = include_trailing_delimiter
        return await self._request("GET", f"/b/{quote(bucket, safe='')}/o", creds, params=params)

    async def upload_object(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        content: bytes,
        *,
        content_type: str = "application/octet-stream",
        metadata: Optional[Dict[str, Any]] = None,
        if_generation_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"uploadType": "media", "name": name}
        if if_generation_match is not None:
            params["ifGenerationMatch"] = if_generation_match
        if metadata:
            for key, value in metadata.items():
                params[key] = value
        headers = {"Content-Type": content_type}
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o", creds,
            params=params, content=content, extra_headers=headers, upload=True,
        )

    async def download_object(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        generation: Optional[int] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {"alt": "media"}
        if generation is not None:
            params["generation"] = generation
        return await self._request(
            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params
        )

    async def get_object(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        generation: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"generation": generation} if generation is not None else None
        return await self._request("GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params)

    async def patch_object(
        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
        *, if_generation_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifGenerationMatch": if_generation_match} if if_generation_match is not None else None
        return await self._request(
            "PATCH", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params, json_body=body
        )

    async def update_object(
        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
        *, if_generation_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifGenerationMatch": if_generation_match} if if_generation_match is not None else None
        return await self._request(
            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params, json_body=body
        )

    async def delete_object(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        generation: Optional[int] = None,
        if_generation_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params: Dict[str, Any] = {}
        if generation is not None:
            params["generation"] = generation
        if if_generation_match is not None:
            params["ifGenerationMatch"] = if_generation_match
        params = params or None
        return await self._request("DELETE", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}", creds, params=params)

    async def copy_object(
        self,
        creds: GCSCredentials,
        source_bucket: str,
        source_name: str,
        destination_bucket: str,
        destination_name: str,
        *,
        source_generation: Optional[int] = None,
        body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        path = (
            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
            f"/copyTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
        )
        params = {"sourceGeneration": source_generation} if source_generation is not None else None
        return await self._request("POST", path, creds, params=params, json_body=body)

    async def move_object(
        self,
        creds: GCSCredentials,
        source_bucket: str,
        source_name: str,
        destination_bucket: str,
        destination_name: str,
        *,
        source_generation: Optional[int] = None,
        if_generation_match: Optional[int] = None,
        body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        path = (
            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
            f"/moveTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
        )
        params: Dict[str, Any] = {}
        if source_generation is not None:
            params["sourceGeneration"] = source_generation
        if if_generation_match is not None:
            params["ifGenerationMatch"] = if_generation_match
        return await self._request("POST", path, creds, params=params or None, json_body=body)

    async def rewrite_object(
        self,
        creds: GCSCredentials,
        source_bucket: str,
        source_name: str,
        destination_bucket: str,
        destination_name: str,
        *,
        rewrite_token: Optional[str] = None,
        max_bytes_rewritten_per_call: Optional[int] = None,
        body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        path = (
            f"/b/{quote(source_bucket, safe='')}/o/{quote(source_name, safe='')}"
            f"/rewriteTo/b/{quote(destination_bucket, safe='')}/o/{quote(destination_name, safe='')}"
        )
        params: Dict[str, Any] = {}
        if rewrite_token:
            params["rewriteToken"] = rewrite_token
        if max_bytes_rewritten_per_call is not None:
            params["maxBytesRewrittenPerCall"] = max_bytes_rewritten_per_call
        return await self._request("POST", path, creds, params=params or None, json_body=body)

    async def compose_object(
        self,
        creds: GCSCredentials,
        bucket: str,
        destination_name: str,
        source_objects: List[str],
        *,
        body: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        compose_body: Dict[str, Any] = {
            "sourceObjects": [{"name": name} for name in source_objects]
        }
        if body:
            compose_body["destination"] = body
        path = f"/b/{quote(bucket, safe='')}/o/{quote(destination_name, safe='')}/compose"
        return await self._request("POST", path, creds, json_body=compose_body)

    async def initiate_resumable_upload(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        content_type: str = "application/octet-stream",
        object_metadata: Optional[Dict[str, Any]] = None,
    ) -> Dict[str, Any]:
        """Create a resumable upload session and return its URI.

        The client can then ``PUT`` the object bytes to ``session_uri`` (optionally
        in chunks with ``Content-Range``) to complete the upload.
        """
        params: Dict[str, Any] = {"uploadType": "resumable", "name": name}
        body: Dict[str, Any] = dict(object_metadata or {})
        body.setdefault("name", name)
        body.setdefault("contentType", content_type)
        result = await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o", creds,
            params=params, json_body=body,
            extra_headers={"Content-Type": "application/json; charset=UTF-8"},
            upload=True,
        )
        session_uri = (result.get("headers") or {}).get("location")
        return {
            "success": True,
            "session_uri": session_uri,
            "data": result.get("data"),
            "error": None,
        }

    # ------------------------------------------------------------------
    # Object holds & retention policy
    # ------------------------------------------------------------------

    async def set_object_hold(
        self, creds: GCSCredentials, bucket: str, name: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/hold", creds
        )

    async def release_object_hold(
        self, creds: GCSCredentials, bucket: str, name: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/releaseHold", creds
        )

    async def lock_object_retention(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        if_metageneration_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/lockRetentionPolicy",
            creds, params=params,
        )

    # ------------------------------------------------------------------
    # Notifications (object change watch)
    # ------------------------------------------------------------------

    async def watch_all_objects(
        self, creds: GCSCredentials, bucket: str, channel_body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o/watch", creds, json_body=channel_body
        )

    # ------------------------------------------------------------------
    # Bucket restore (soft-deleted buckets)
    # ------------------------------------------------------------------

    async def restore_bucket(
        self,
        creds: GCSCredentials,
        bucket: str,
        *,
        if_metageneration_match: Optional[int] = None,
    ) -> Dict[str, Any]:
        params = {"ifMetagenerationMatch": if_metageneration_match} if if_metageneration_match is not None else None
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/restore", creds, params=params
        )

    # ------------------------------------------------------------------
    # Object IAM
    # ------------------------------------------------------------------

    async def get_object_iam(
        self, creds: GCSCredentials, bucket: str, name: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/iam", creds
        )

    async def set_object_iam(
        self, creds: GCSCredentials, bucket: str, name: str, policy: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/iam", creds, json_body=policy
        )

    # ------------------------------------------------------------------
    # Object ACLs
    # ------------------------------------------------------------------

    async def list_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl", creds
        )

    async def insert_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "POST", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl", creds, json_body=body
        )

    async def get_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str, entity: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "GET", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}", creds
        )

    async def patch_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str, entity: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "PATCH", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}",
            creds, json_body=body,
        )

    async def update_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str, entity: str, body: Dict[str, Any],
    ) -> Dict[str, Any]:
        return await self._request(
            "PUT", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}",
            creds, json_body=body,
        )

    async def delete_object_acl(
        self, creds: GCSCredentials, bucket: str, name: str, entity: str,
    ) -> Dict[str, Any]:
        return await self._request(
            "DELETE", f"/b/{quote(bucket, safe='')}/o/{quote(name, safe='')}/acl/{quote(entity, safe='')}", creds
        )

    # ------------------------------------------------------------------
    # Public URLs
    # ------------------------------------------------------------------

    @staticmethod
    def public_url(bucket: str, name: str) -> str:
        return f"https://storage.googleapis.com/{bucket}/{name}"

    # ------------------------------------------------------------------
    # Signed URLs (V4, RSA-SHA256)
    # ------------------------------------------------------------------

    def sign_url_v4(
        self,
        creds: GCSCredentials,
        method: str,
        bucket: str,
        name: str,
        *,
        expires_in_seconds: Optional[int] = None,
        content_type: Optional[str] = None,
        query_params: Optional[Dict[str, str]] = None,
        response_content_type: Optional[str] = None,
        response_disposition: Optional[str] = None,
    ) -> Tuple[str, int]:
        """Generate a V4 signed URL for a single GCS operation.

        Returns ``(signed_url, effective_expires_seconds)``.
        """
        expires = (
            expires_in_seconds
            if expires_in_seconds is not None
            else _settings.gcs_default_expires_seconds
        )
        if expires <= 0:
            raise GCSError("expires_in_seconds must be greater than 0.", status_code=400)
        if expires > _settings.gcs_max_expires_seconds:
            raise GCSError(
                f"expires_in_seconds cannot exceed {_settings.gcs_max_expires_seconds} seconds (7 days).",
                status_code=400,
            )

        now = datetime.now(timezone.utc)
        request_timestamp = now.strftime("%Y%m%dT%H%M%SZ")
        datestamp = now.strftime("%Y%m%d")
        credential_scope = f"{datestamp}/{_SIGNING_REGION}/{_SIGNING_SERVICE}/goog4_request"
        credential = f"{creds.client_email}/{credential_scope}"

        host = "storage.googleapis.com"
        canonical_uri = f"/{bucket}/{name}"

        headers: Dict[str, str] = {"host": host}
        if content_type:
            headers["content-type"] = content_type

        ordered_headers = {k.lower(): str(v).lower() for k, v in sorted(headers.items())}
        canonical_headers = "".join(f"{k}:{v}\n" for k, v in ordered_headers.items())
        signed_headers = ";".join(ordered_headers.keys())

        query: Dict[str, Any] = {
            "X-Goog-Algorithm": _SIGNING_ALGORITHM,
            "X-Goog-Credential": credential,
            "X-Goog-Date": request_timestamp,
            "X-Goog-Expires": str(expires),
            "X-Goog-SignedHeaders": signed_headers,
        }
        if response_content_type:
            query["response-content-type"] = response_content_type
        if response_disposition:
            query["response-content-disposition"] = response_disposition
        if query_params:
            query.update(query_params)

        canonical_query_string = "&".join(
            f"{quote(str(k), safe='')}={quote(str(v), safe='')}"
            for k, v in sorted(query.items())
        )

        canonical_request = "\n".join([
            method.upper(),
            canonical_uri,
            canonical_query_string,
            canonical_headers,
            signed_headers,
            "UNSIGNED-PAYLOAD",
        ])

        canonical_request_hash = hashlib.sha256(canonical_request.encode("utf-8")).hexdigest()
        string_to_sign = "\n".join([
            _SIGNING_ALGORITHM,
            request_timestamp,
            credential_scope,
            canonical_request_hash,
        ])

        from cryptography.hazmat.primitives import hashes
        from cryptography.hazmat.primitives.asymmetric import padding
        from cryptography.hazmat.primitives.serialization import load_pem_private_key

        private_key = load_pem_private_key(creds.private_key.encode("utf-8"), password=None)
        signature = private_key.sign(string_to_sign.encode("utf-8"), padding.PKCS1v15(), hashes.SHA256())
        signature_hex = signature.hex()

        signed_url = (
            f"https://{host}{canonical_uri}?{canonical_query_string}"
            f"&X-Goog-Signature={signature_hex}"
        )
        return signed_url, expires

    async def download_signed_url(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        expires_in_seconds: Optional[int] = None,
        response_content_type: Optional[str] = None,
        response_disposition: Optional[str] = None,
    ) -> Tuple[str, int]:
        return await run_in_executor(
            self.sign_url_v4,
            creds,
            "GET",
            bucket,
            name,
            expires_in_seconds=expires_in_seconds,
            response_content_type=response_content_type,
            response_disposition=response_disposition,
        )

    async def upload_signed_url(
        self,
        creds: GCSCredentials,
        bucket: str,
        name: str,
        *,
        expires_in_seconds: Optional[int] = None,
        content_type: Optional[str] = None,
    ) -> Tuple[str, int]:
        return await run_in_executor(
            self.sign_url_v4,
            creds,
            "PUT",
            bucket,
            name,
            expires_in_seconds=expires_in_seconds,
            content_type=content_type,
        )