import enum
import logging
import time
from typing import Any, TypeVar

import httpx
import pydantic

logger = logging.getLogger(__name__)

FIVETRAN_API_BASE = "https://api.fivetran.com/v1"
DAILY_SYNC_TIME = "16:00"  # PST (UTC-08)

# Retry policy. 429 (rate limit) and 5xx (transient Fivetran server errors —
# the "contact support with log ID" 500s) are retried with backoff rather than
# raising, so a transient blip can never be recorded as a permanent failure.
# 429s honour the Retry-After header and back off long; 5xx back off briefly.
RATE_LIMIT_MAX_RETRIES = 5
RATE_LIMIT_BACKOFF = 60  # initial backoff for 429 (seconds)
RATE_LIMIT_BACKOFF_MAX = 900  # cap for the 429 doubling backoff
SERVER_ERROR_BACKOFF = 10  # initial backoff for 5xx (seconds)
SERVER_ERROR_BACKOFF_MAX = 120  # cap for the 5xx doubling backoff


class ConnectionSetupState(enum.StrEnum):
    INCOMPLETE = "incomplete"
    CONNECTED = "connected"
    BROKEN = "broken"


class ConnectionSyncState(enum.StrEnum):
    SCHEDULED = "scheduled"
    SYNCING = "syncing"
    PAUSED = "paused"
    RESCHEDULED = "rescheduled"


class ConnectionStatus(pydantic.BaseModel):
    setup_state: ConnectionSetupState
    sync_state: ConnectionSyncState
    update_state: str | None = None
    is_historical_sync: bool | None = None


class ShopifyConnectionConfig(pydantic.BaseModel):
    shop: str | None = None


class ShopifyConnection(pydantic.BaseModel):
    id: str
    service: str
    status: ConnectionStatus
    name: str = pydantic.Field(alias="schema")
    config: ShopifyConnectionConfig
    created_at: str | None = None
    succeeded_at: str | None = None
    sync_frequency: int | None = None

    model_config = pydantic.ConfigDict(populate_by_name=True)


class ConnectCard(pydantic.BaseModel):
    token: str
    uri: str


class ConnectCardResponse(pydantic.BaseModel):
    connect_card: ConnectCard


T = TypeVar("T")


class Response[T](pydantic.BaseModel):
    code: str
    message: str | None = None
    data: T


class TableEnabledPatchSettings(pydantic.BaseModel):
    allowed: bool


class TableConfig(pydantic.BaseModel):
    enabled: bool
    name_in_destination: str
    enabled_patch_settings: TableEnabledPatchSettings


class SchemaConfig(pydantic.BaseModel):
    enabled: bool
    name_in_destination: str
    tables: dict[str, TableConfig]


class StandardConfig(pydantic.BaseModel):
    enable_new_by_default: bool
    schemas: dict[str, SchemaConfig]
    schema_change_handling: str


class TableUpdate(pydantic.BaseModel):
    enabled: bool


class SchemaUpdate(pydantic.BaseModel):
    tables: dict[str, TableUpdate]


class StandardConfigUpdate(pydantic.BaseModel):
    schemas: dict[str, SchemaUpdate]


class ConnectorSummary(pydantic.BaseModel):
    id: str
    service: str
    schema_name: str = pydantic.Field(alias="schema")

    model_config = pydantic.ConfigDict(populate_by_name=True)


def _error_detail(response: httpx.Response) -> str:
    """Safe error summary for logging — Fivetran's code/message only.

    Never returns the raw body: connect-card responses embed a live OAuth
    token, so logging response.text could leak a bearer credential.
    """
    try:
        body = response.json()
    except Exception:
        return f"<{len(response.content)} bytes, non-JSON body suppressed>"
    if isinstance(body, dict):
        code, message = body.get("code"), body.get("message")
        if code or message:
            return ": ".join(str(p) for p in (code, message) if p)
    return "<no code/message in body>"


class FivetranClient:
    def __init__(self, api_key: str, api_secret: str) -> None:
        self.auth = (api_key, api_secret)
        self.client = httpx.Client(timeout=300.0)

    def _request(self, method: str, path: str, **kwargs: Any) -> httpx.Response:
        url = FIVETRAN_API_BASE + path
        rl_backoff = RATE_LIMIT_BACKOFF  # 429 backoff
        se_backoff = SERVER_ERROR_BACKOFF  # 5xx backoff
        response: httpx.Response | None = None
        for attempt in range(RATE_LIMIT_MAX_RETRIES + 1):
            response = self.client.request(method, url, auth=self.auth, **kwargs)
            status = response.status_code

            # Retry 429 (rate limit) for any request. Only retry 5xx for
            # idempotent methods to avoid duplicating side effects on POST.
            method_upper = method.upper()
            is_idempotent = method_upper in {"GET", "PUT", "PATCH", "DELETE", "HEAD", "OPTIONS"}
            should_retry = status == 429 or (status >= 500 and is_idempotent)
            if should_retry and attempt < RATE_LIMIT_MAX_RETRIES:
                if status == 429:
                    retry_after = response.headers.get("Retry-After")
                    wait = int(retry_after) if retry_after and retry_after.isdigit() else rl_backoff
                    rl_backoff = min(rl_backoff * 2, RATE_LIMIT_BACKOFF_MAX)
                else:
                    wait = se_backoff
                    se_backoff = min(se_backoff * 2, SERVER_ERROR_BACKOFF_MAX)
                logger.warning(
                    f"Fivetran {status} on {method} {path} — retry {attempt + 1}/{RATE_LIMIT_MAX_RETRIES} in {wait}s"
                )
                time.sleep(wait)
                continue

            if not response.is_success:
                # Log only the Fivetran error envelope (code/message), never the
                # raw body: connect-card responses embed a live OAuth token, and
                # a non-2xx from that endpoint would otherwise leak it to logs.
                logger.error(f"Fivetran error {status} on {method} {path}: {_error_detail(response)}")
            response.raise_for_status()
            return response

        # Retries exhausted on a persistent 429/5xx — surface it to the caller.
        assert response is not None  # loop always runs >= 1 times
        response.raise_for_status()
        return response

    def create_shopify_connector(self, group_id: str, schema: str, shop_domain: str) -> ShopifyConnection:
        payload = {
            "group_id": group_id,
            "service": "shopify",
            "run_setup_tests": False,
            "paused": False,
            "pause_after_trial": False,
            "daily_sync_time": DAILY_SYNC_TIME,
            "sync_frequency": "1440",
            "config": {
                "schema": schema,
                "shop": shop_domain,
            },
        }
        raw = self._request("POST", "/connections", json=payload).json()
        return pydantic.TypeAdapter(Response[ShopifyConnection]).validate_python(raw).data

    def get_connect_card(self, connector_id: str, redirect_uri: str) -> ConnectCard:
        payload = {
            "connect_card_config": {
                "redirect_uri": redirect_uri,
                "hide_setup_guide": True,
            }
        }
        raw = self._request("POST", f"/connections/{connector_id}/connect-card", json=payload).json()
        return pydantic.TypeAdapter(Response[ConnectCardResponse]).validate_python(raw).data.connect_card

    def get_group_connectors(self, group_id: str) -> list[ConnectorSummary]:
        """Return all connectors in the group, handling pagination."""
        results = []
        cursor = None
        while True:
            params = {"cursor": cursor} if cursor else {}
            raw = self._request("GET", f"/groups/{group_id}/connectors", params=params).json()
            items = raw.get("data", {}).get("items", [])
            results.extend(pydantic.TypeAdapter(ConnectorSummary).validate_python(item) for item in items)
            cursor = raw.get("data", {}).get("next_cursor")
            if not cursor:
                break
        return results

    def get_connection_status(self, connector_id: str) -> ShopifyConnection:
        raw = self._request("GET", f"/connections/{connector_id}").json()
        return pydantic.TypeAdapter(Response[ShopifyConnection]).validate_python(raw).data

    def trigger_sync(self, connector_id: str) -> None:
        self._request("POST", f"/connections/{connector_id}/sync")

    def delete_connection(self, connector_id: str) -> None:
        self._request("DELETE", f"/connections/{connector_id}")

    def update_schema_name(self, connector_id: str, schema: str) -> None:
        self._request("PATCH", f"/connections/{connector_id}", json={"config": {"schema": schema}})

    def pause_connector(self, connector_id: str) -> None:
        self._request("PATCH", f"/connections/{connector_id}", json={"paused": True})

    def resume_connector(self, connector_id: str) -> None:
        self._request("PATCH", f"/connections/{connector_id}", json={"paused": False})

    def reload_schema(self, connector_id: str) -> None:
        self._request("POST", f"/connections/{connector_id}/schemas/reload")

    def get_connection_schema_config(self, connector_id: str) -> StandardConfig:
        raw = self._request("GET", f"/connections/{connector_id}/schemas").json()
        return pydantic.TypeAdapter(Response[StandardConfig]).validate_python(raw).data

    def patch_connection_schema_config(self, connector_id: str, config: StandardConfigUpdate) -> StandardConfig:
        raw = self._request(
            "PATCH",
            f"/connections/{connector_id}/schemas",
            json=config.model_dump(exclude_none=True),
        ).json()
        return pydantic.TypeAdapter(Response[StandardConfig]).validate_python(raw).data

    def close(self) -> None:
        self.client.close()

    def __enter__(self) -> "FivetranClient":
        return self

    def __exit__(self, *_: Any) -> None:
        self.close()