from __future__ import annotations

import json
import os
import subprocess
import tempfile
import tomllib
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Protocol
from urllib import error, parse, request

from config import CONFIG_SNAPSHOT_PREFIX, CONFIG_TRUNCATE_BEFORE_SNAPSHOT, ConnectorConfig
from errors import ConfigurationError
from metadata import TableName
from state import (
    PHASE_SNAPSHOT,
    RESYNC_CURRENT_FILE,
    RESYNC_NEXT_ROW,
    RESYNC_PHASE,
    RESYNC_REQUEST_ID,
    RESYNC_SNAPSHOT_PREFIX,
    RESYNC_SNAPSHOT_TSO,
    RESYNC_TABLES,
    RESYNC_TRUNCATE_DONE,
    STATE_INCREMENTAL,
    STATE_RESYNC,
    STATE_SNAPSHOT,
    STATE_SYNC_TASK_ID,
    STATE_VERSION,
)

FIVETRAN_CONFIG = "config"
FIVETRAN_PAUSED = "paused"
FIVETRAN_STATE = "state"


class DeployerError(Exception):
    def __init__(self, code: str, message: str):
        super().__init__(message)
        self.code = code
        self.message = message


@dataclass(frozen=True)
class JobConfig:
    api_key_b64: str
    destination_name: str
    connection_name: str
    connector_config: dict[str, Any]
    request_id: str | None
    connector_name: str
    connector_version: str
    project_path: str
    python_version: str
    termination_log_path: str
    base_url: str

    @classmethod
    def from_env(cls) -> JobConfig:
        api_key_b64 = _require_env("FIVETRAN_API_KEY_BASE64")
        destination_name = _require_env("DESTINATION_NAME")
        connection_name = _require_env("CONNECTION_NAME")
        connector_config = _load_connector_config()
        _validate_connector_config(connector_config)
        project_path = os.environ.get("PROJECT_PATH", os.getcwd())
        connector_version = os.environ.get("CONNECTOR_VERSION") or _read_project_version(project_path)
        expected_version = _read_project_version(project_path)
        if connector_version != expected_version:
            raise DeployerError(
                "connector_version_mismatch",
                f"CONNECTOR_VERSION={connector_version} does not match project version {expected_version}",
            )
        return cls(
            api_key_b64=api_key_b64,
            destination_name=destination_name,
            connection_name=connection_name,
            connector_config=connector_config,
            request_id=os.environ.get("REQUEST_ID"),
            connector_name=os.environ.get("CONNECTOR_NAME", "tidb-fivetran-connector"),
            connector_version=connector_version,
            project_path=project_path,
            python_version=os.environ.get("PYTHON_VERSION", "3.13"),
            termination_log_path=os.environ.get("TERMINATION_LOG_PATH", "/dev/termination-log"),
            base_url=os.environ.get("FIVETRAN_BASE_URL", "https://api.fivetran.com/v1"),
        )


@dataclass(frozen=True)
class DeploymentResult:
    status: str
    request_id: str | None
    connector_name: str
    connector_version: str
    destination_name: str
    destination_id: str
    connection_name: str
    connection_id: str
    dashboard_url: str

    def to_dict(self) -> dict[str, Any]:
        return {
            "status": self.status,
            "request_id": self.request_id,
            "connector_name": self.connector_name,
            "connector_version": self.connector_version,
            "destination_name": self.destination_name,
            "destination_id": self.destination_id,
            "connection_name": self.connection_name,
            "connection_id": self.connection_id,
            "dashboard_url": self.dashboard_url,
        }


class FivetranApiClient:
    def __init__(self, api_key_b64: str, base_url: str):
        self._api_key_b64 = api_key_b64
        self._base_url = base_url.rstrip("/")

    def get_group_id_by_name(self, group_name: str) -> str:
        payload = self._get_json("/groups")
        items = payload["data"]["items"]
        matches = [item for item in items if item.get("name") == group_name]
        if not matches:
            raise DeployerError("destination_not_found", f"destination {group_name} not found")
        if len(matches) > 1:
            raise DeployerError("destination_ambiguous", f"destination {group_name} resolved to multiple groups")
        return str(matches[0]["id"])

    def get_connection_id_by_name(self, group_id: str, connection_name: str) -> str:
        query = parse.urlencode({"group_id": group_id})
        payload = self._get_json(f"/connections?{query}")
        items = payload["data"]["items"]
        matches = [item for item in items if item.get("schema") == connection_name]
        if not matches:
            raise DeployerError(
                "connection_not_found",
                f"connection {connection_name} not found in destination group {group_id}",
            )
        if len(matches) > 1:
            raise DeployerError(
                "connection_ambiguous",
                f"connection {connection_name} resolved to multiple IDs in destination group {group_id}",
            )
        return str(matches[0]["id"])

    def get_connection(self, connection_id: str) -> dict[str, Any]:
        payload = self._get_json(f"/connections/{parse.quote(connection_id)}")
        data = payload.get("data")
        if not isinstance(data, dict):
            raise DeployerError("fivetran_api_error", "connection response missing data object")
        return data

    def patch_connection(self, connection_id: str, body: dict[str, Any]) -> dict[str, Any]:
        payload = self._request_json("PATCH", f"/connections/{parse.quote(connection_id)}", body)
        data = payload.get("data")
        if not isinstance(data, dict):
            raise DeployerError("fivetran_api_error", "connection patch response missing data object")
        return data

    def set_connection_paused(self, connection_id: str, paused: bool) -> dict[str, Any]:
        return self.patch_connection(connection_id, {FIVETRAN_PAUSED: paused})

    def get_connection_state(self, connection_id: str) -> dict[str, Any]:
        payload = self._get_json(f"/connections/{parse.quote(connection_id)}/state")
        data = payload.get("data")
        if not isinstance(data, dict):
            raise DeployerError("fivetran_api_error", "connection state response missing data object")
        return data

    def patch_connection_state(self, connection_id: str, state: dict[str, Any]) -> dict[str, Any]:
        payload = self._request_json(
            "PATCH",
            f"/connections/{parse.quote(connection_id)}/state",
            {FIVETRAN_STATE: state},
        )
        data = payload.get("data")
        if not isinstance(data, dict):
            raise DeployerError("fivetran_api_error", "connection state patch response missing data object")
        return data

    def trigger_resync(self, connection_id: str) -> dict[str, Any]:
        payload = self._request_json("POST", f"/connections/{parse.quote(connection_id)}/resync", {})
        data = payload.get("data", {})
        if not isinstance(data, dict):
            raise DeployerError("fivetran_api_error", "connection resync response data must be an object")
        return data

    def _get_json(self, path: str) -> dict[str, Any]:
        return self._request_json("GET", path)

    def _request_json(
        self,
        method: str,
        path: str,
        body: dict[str, Any] | None = None,
    ) -> dict[str, Any]:
        encoded_body = None
        headers = {
            "Authorization": f"Basic {self._api_key_b64}",
            "Accept": "application/json",
        }
        if body is not None:
            encoded_body = json.dumps(body).encode("utf-8")
            headers["Content-Type"] = "application/json"
        req = request.Request(
            f"{self._base_url}{path}",
            data=encoded_body,
            headers=headers,
            method=method,
        )
        try:
            with request.urlopen(req) as resp:
                payload = json.load(resp)
        except error.HTTPError as exc:
            error_body = exc.read().decode("utf-8", errors="replace")
            raise DeployerError("fivetran_http_error", f"Fivetran API HTTP {exc.code}: {error_body}") from exc
        except error.URLError as exc:
            raise DeployerError("fivetran_network_error", f"failed to call Fivetran API: {exc.reason}") from exc

        if payload.get("code") != "Success":
            raise DeployerError(
                "fivetran_api_error",
                f"Fivetran API returned {payload.get('code')}: {payload.get('message')}",
            )
        return payload


class ControlPlaneApi(Protocol):
    def get_connection(self, connection_id: str) -> dict[str, Any]: ...

    def patch_connection(self, connection_id: str, body: dict[str, Any]) -> dict[str, Any]: ...

    def set_connection_paused(self, connection_id: str, paused: bool) -> dict[str, Any]: ...

    def get_connection_state(self, connection_id: str) -> dict[str, Any]: ...

    def patch_connection_state(self, connection_id: str, state: dict[str, Any]) -> dict[str, Any]: ...

    def trigger_resync(self, connection_id: str) -> dict[str, Any]: ...


def build_full_resync_config(
    connector_config: dict[str, Any],
    *,
    snapshot_prefix: str,
) -> dict[str, Any]:
    updated = dict(connector_config)
    updated[CONFIG_SNAPSHOT_PREFIX] = snapshot_prefix.strip("/")
    updated[CONFIG_TRUNCATE_BEFORE_SNAPSHOT] = True
    _validate_connector_config(updated)
    return updated


def _validate_connector_config(connector_config: dict[str, Any]) -> None:
    try:
        ConnectorConfig.from_dict(connector_config)
    except ConfigurationError as exc:
        raise DeployerError("invalid_connector_config", str(exc)) from exc


def run_full_resync(
    *,
    api: ControlPlaneApi,
    connection_id: str,
    connector_config: dict[str, Any],
    snapshot_prefix: str,
) -> None:
    updated_config = build_full_resync_config(connector_config, snapshot_prefix=snapshot_prefix)
    api.set_connection_paused(connection_id, True)
    patched_config = False
    completed = False
    try:
        api.get_connection(connection_id)
        api.get_connection_state(connection_id)
        api.patch_connection(connection_id, {FIVETRAN_CONFIG: updated_config})
        patched_config = True
        read_back = api.get_connection(connection_id)
        read_back_config = read_back.get(FIVETRAN_CONFIG)
        if not isinstance(read_back_config, dict):
            raise DeployerError(
                "config_readback_mismatch",
                "patched config was not visible in Fivetran read-back",
            )
        for key, value in updated_config.items():
            if read_back_config.get(key) != value:
                raise DeployerError(
                    "config_readback_mismatch",
                    f"patched config key {key!r} was not visible in Fivetran read-back",
                )
        api.trigger_resync(connection_id)
        completed = True
    finally:
        if completed or not patched_config:
            api.set_connection_paused(connection_id, False)


def build_table_resync_state(
    current_state: dict[str, Any],
    *,
    request_id: str,
    tables: list[str],
    snapshot_tso: int,
    snapshot_prefix: str,
) -> dict[str, Any]:
    normalized_tables, normalized_snapshot_tso, normalized_snapshot_prefix = _normalize_table_resync_request(
        request_id=request_id,
        tables=tables,
        snapshot_tso=snapshot_tso,
        snapshot_prefix=snapshot_prefix,
    )
    _validate_initialized_connector_state(current_state)
    patched = deepcopy_json(current_state)
    patched[STATE_RESYNC] = {
        RESYNC_REQUEST_ID: request_id,
        RESYNC_PHASE: PHASE_SNAPSHOT,
        RESYNC_TABLES: normalized_tables,
        RESYNC_SNAPSHOT_TSO: str(normalized_snapshot_tso),
        RESYNC_SNAPSHOT_PREFIX: normalized_snapshot_prefix,
        RESYNC_TRUNCATE_DONE: False,
        RESYNC_CURRENT_FILE: None,
        RESYNC_NEXT_ROW: 0,
    }
    return patched


def _validate_initialized_connector_state(current_state: dict[str, Any]) -> None:
    if not current_state:
        raise DeployerError(
            "invalid_connector_state",
            "table resync requires an initialized connector state; run a normal sync before table resync",
        )
    missing = [
        field
        for field in (STATE_VERSION, STATE_SYNC_TASK_ID, STATE_SNAPSHOT, STATE_INCREMENTAL)
        if field not in current_state
    ]
    if missing:
        raise DeployerError(
            "invalid_connector_state",
            "table resync requires an initialized connector state; missing " + ", ".join(missing),
        )
    if not isinstance(current_state[STATE_SNAPSHOT], dict):
        raise DeployerError("invalid_connector_state", "state.snapshot must be an object before table resync")
    if not isinstance(current_state[STATE_INCREMENTAL], dict):
        raise DeployerError("invalid_connector_state", "state.incremental must be an object before table resync")


def _normalize_table_resync_request(
    *,
    request_id: str,
    tables: list[str],
    snapshot_tso: int,
    snapshot_prefix: str,
) -> tuple[list[str], int, str]:
    if not isinstance(request_id, str) or not request_id.strip():
        raise DeployerError("invalid_resync_request", "table resync request_id must be non-empty")
    if not tables:
        raise DeployerError("invalid_resync_request", "table resync tables must be non-empty")
    normalized_tables: list[str] = []
    seen_tables: set[TableName] = set()
    for table in tables:
        if not isinstance(table, str) or not table.strip():
            raise DeployerError("invalid_resync_request", "table resync tables must be non-empty strings")
        normalized_table = table.strip()
        try:
            table_name = TableName.from_qualified(normalized_table)
        except ValueError as exc:
            raise DeployerError("invalid_resync_request", f"table resync tables {exc}") from exc
        if table_name in seen_tables:
            raise DeployerError("invalid_resync_request", f"duplicate table resync table: {normalized_table}")
        seen_tables.add(table_name)
        normalized_tables.append(table_name.quoted)
    try:
        normalized_snapshot_tso = int(snapshot_tso)
    except (TypeError, ValueError) as exc:
        raise DeployerError("invalid_resync_request", "table resync snapshot_tso must be a positive integer") from exc
    if normalized_snapshot_tso <= 0:
        raise DeployerError("invalid_resync_request", "table resync snapshot_tso must be a positive integer")
    if not isinstance(snapshot_prefix, str):
        raise DeployerError("invalid_resync_request", "table resync snapshot_prefix must be non-empty")
    normalized_snapshot_prefix = snapshot_prefix.strip().strip("/")
    if not normalized_snapshot_prefix:
        raise DeployerError("invalid_resync_request", "table resync snapshot_prefix must be non-empty")
    if any(part in {"", ".", ".."} for part in normalized_snapshot_prefix.split("/")):
        raise DeployerError(
            "invalid_resync_request",
            "table resync snapshot_prefix must be a normalized relative object path",
        )
    return normalized_tables, normalized_snapshot_tso, normalized_snapshot_prefix


def run_table_resync(
    *,
    api: ControlPlaneApi,
    connection_id: str,
    request_id: str,
    tables: list[str],
    snapshot_tso: int,
    snapshot_prefix: str,
) -> None:
    _normalize_table_resync_request(
        request_id=request_id,
        tables=tables,
        snapshot_tso=snapshot_tso,
        snapshot_prefix=snapshot_prefix,
    )
    api.set_connection_paused(connection_id, True)
    state_patch_applied = False
    completed = False
    try:
        latest_state_response = api.get_connection_state(connection_id)
        latest_state = latest_state_response.get(FIVETRAN_STATE, latest_state_response)
        if not isinstance(latest_state, dict):
            raise DeployerError("invalid_state_readback", "Fivetran state read-back must be an object")
        patched_state = build_table_resync_state(
            latest_state,
            request_id=request_id,
            tables=tables,
            snapshot_tso=snapshot_tso,
            snapshot_prefix=snapshot_prefix,
        )
        api.patch_connection_state(connection_id, patched_state)
        state_patch_applied = True
        read_back = api.get_connection_state(connection_id)
        if read_back.get(FIVETRAN_STATE, read_back) != patched_state:
            raise DeployerError("state_readback_mismatch", "patched state was not visible in Fivetran read-back")
        completed = True
    finally:
        if completed or not state_patch_applied:
            api.set_connection_paused(connection_id, False)


def deepcopy_json(value: dict[str, Any]) -> dict[str, Any]:
    return json.loads(json.dumps(value))


def main() -> int:
    try:
        config = JobConfig.from_env()
        result = run_job(config)
    except (DeployerError, json.JSONDecodeError, OSError, tomllib.TOMLDecodeError, ValueError) as exc:
        if isinstance(exc, DeployerError):
            payload = {
                "status": "failed",
                "request_id": os.environ.get("REQUEST_ID"),
                "error_code": exc.code,
                "message": exc.message,
            }
        else:
            payload = {
                "status": "failed",
                "request_id": os.environ.get("REQUEST_ID"),
                "error_code": "runtime_error",
                "message": str(exc),
            }
        _emit_result(payload, os.environ.get("TERMINATION_LOG_PATH", "/dev/termination-log"))
        return 1

    _emit_result(result.to_dict(), config.termination_log_path)
    return 0


def run_job(config: JobConfig) -> DeploymentResult:
    print(
        json.dumps(
            {
                "event": "deployer_start",
                "request_id": config.request_id,
                "destination_name": config.destination_name,
                "connection_name": config.connection_name,
                "connector_name": config.connector_name,
                "connector_version": config.connector_version,
                "config_keys": sorted(config.connector_config.keys()),
            }
        )
    )
    with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as handle:
        json.dump(_render_fivetran_configuration(config.connector_config), handle)
        handle.flush()
        config_path = handle.name

    try:
        _run_fivetran_deploy(config, config_path)
        api = FivetranApiClient(config.api_key_b64, config.base_url)
        destination_id = api.get_group_id_by_name(config.destination_name)
        connection_id = api.get_connection_id_by_name(destination_id, config.connection_name)
        return DeploymentResult(
            status="succeeded",
            request_id=config.request_id,
            connector_name=config.connector_name,
            connector_version=config.connector_version,
            destination_name=config.destination_name,
            destination_id=destination_id,
            connection_name=config.connection_name,
            connection_id=connection_id,
            dashboard_url=f"https://fivetran.com/dashboard/connectors/{connection_id}/status",
        )
    finally:
        Path(config_path).unlink(missing_ok=True)


def _run_fivetran_deploy(config: JobConfig, config_path: str) -> None:
    cmd = [
        "fivetran",
        "deploy",
        config.project_path,
        "--force",
        "--api-key",
        config.api_key_b64,
        "--destination",
        config.destination_name,
        "--connection",
        config.connection_name,
        "--configuration",
        config_path,
        "--python-version",
        config.python_version,
    ]
    proc = subprocess.run(cmd, text=True, capture_output=True)
    if proc.stdout:
        print(proc.stdout, end="" if proc.stdout.endswith("\n") else "\n")
    if proc.stderr:
        print(proc.stderr, end="" if proc.stderr.endswith("\n") else "\n")
    if proc.returncode != 0:
        raise DeployerError("deploy_failed", f"fivetran deploy exited with code {proc.returncode}")


def _render_fivetran_configuration(connector_config: dict[str, Any]) -> dict[str, str]:
    rendered: dict[str, str] = {}
    for key, value in connector_config.items():
        if isinstance(value, str):
            rendered[key] = value
        elif isinstance(value, bool):
            rendered[key] = "true" if value else "false"
        elif value is None:
            rendered[key] = ""
        elif isinstance(value, (dict, list)):
            rendered[key] = json.dumps(value, separators=(",", ":"))
        else:
            rendered[key] = str(value)
    return rendered


def _load_connector_config() -> dict[str, Any]:
    raw = _require_env("CONNECTOR_CONFIG_JSON")
    parsed = json.loads(raw)
    if not isinstance(parsed, dict):
        raise DeployerError("invalid_config", "CONNECTOR_CONFIG_JSON must decode to a JSON object")
    return parsed


def _read_project_version(project_path: str) -> str:
    pyproject_path = Path(project_path) / "pyproject.toml"
    if not pyproject_path.exists():
        raise DeployerError("missing_pyproject", f"pyproject.toml not found under {project_path}")
    with pyproject_path.open("rb") as handle:
        pyproject = tomllib.load(handle)
    project = pyproject.get("project")
    if not isinstance(project, dict):
        raise DeployerError("invalid_pyproject", "pyproject.toml missing [project] section")
    version = project.get("version")
    if not isinstance(version, str) or not version:
        raise DeployerError("invalid_pyproject", "pyproject.toml missing project.version")
    return version


def _require_env(name: str) -> str:
    value = os.environ.get(name)
    if value is None or value == "":
        raise DeployerError("missing_env", f"{name} is required")
    return value


def _emit_result(payload: dict[str, Any], termination_log_path: str) -> None:
    rendered = json.dumps(payload, sort_keys=True)
    print(rendered)
    Path(termination_log_path).write_text(rendered)


if __name__ == "__main__":
    raise SystemExit(main())