from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path
from typing import Any

import yaml


@dataclass(frozen=True)
class PipelineConfig:
    path: Path
    data: dict[str, Any]

    @property
    def root(self) -> Path:
        return self.path.parent

    @property
    def name(self) -> str:
        return str(self.data["source"]["name"])

    @property
    def source_type(self) -> str:
        return str(self.data["source"]["type"])

    @property
    def target_project(self) -> str:
        return str(self.data["target"]["project"])

    @property
    def location(self) -> str:
        return str(self.data["target"]["location"])

    @property
    def raw_dataset(self) -> str:
        return str(self.data["target"]["datasets"]["raw"])

    @property
    def final_dataset(self) -> str | None:
        value = self.data["target"].get("datasets", {}).get("final")
        return str(value) if value else None

    @property
    def control_dataset(self) -> str:
        return str(self.data["target"]["datasets"]["control"])

    @property
    def bucket(self) -> str:
        return str(self.data["gcs"]["bucket"])

    @property
    def snapshot_prefix(self) -> str:
        return str(self.data["gcs"]["snapshot_prefix"]).rstrip("/")

    @property
    def manifest_prefix(self) -> str:
        return str(self.data["gcs"]["manifest_prefix"]).rstrip("/")

    @property
    def runtime_log_prefix(self) -> str:
        return str(self.data["gcs"]["runtime_log_prefix"]).rstrip("/")

    @property
    def source_project(self) -> str:
        return str(self.data["source"].get("project", self.target_project))

    @property
    def source_dataset(self) -> str:
        return str(self.data["source"].get("dataset", self.data["source"].get("database", "")))

    @property
    def source_database(self) -> str:
        return str(self.data["source"].get("database", self.source_dataset or self.name))

    @property
    def source_dsn_env(self) -> str:
        return str(self.data["source"].get("connection_env", "TIDB_SOURCE_DSN"))

    @property
    def tidb_cluster_id(self) -> str:
        return str(self.data["source"]["tidb_cloud"]["cluster_id"])

    @property
    def tidb_cli_profile(self) -> str:
        return str(self.data["source"].get("tidb_cloud_cli_profile", self.name))

    @property
    def tidb_public_key_env(self) -> str:
        return str(self.data["source"].get("tidb_cloud", {}).get("public_key_env", "TIDB_CLOUD_PUBLIC_KEY"))

    @property
    def tidb_private_key_env(self) -> str:
        return str(self.data["source"].get("tidb_cloud", {}).get("private_key_env", "TIDB_CLOUD_PRIVATE_KEY"))

    @property
    def tidb_gcs_service_account_key_env(self) -> str:
        return str(
            self.data["source"]
            .get("tidb_cloud", {})
            .get("gcs_service_account_key_env", "TIDB_EXPORT_GCS_SERVICE_ACCOUNT_KEY")
        )

    @property
    def selected_tables(self) -> list[str]:
        return list(self.data.get("tables", {}).get("include", []))

    @property
    def excluded_tables(self) -> list[str]:
        return list(self.data.get("tables", {}).get("exclude", []))

    @property
    def require_primary_key(self) -> bool:
        return bool(self.data.get("tables", {}).get("require_primary_key", False))

    @property
    def watermark_candidates(self) -> list[str]:
        return list(self.data["source"].get("watermark_candidates", []))

    @property
    def declared_primary_keys(self) -> dict[str, list[str]]:
        return {str(k): list(v) for k, v in self.data["source"].get("primary_keys", {}).items()}

    @property
    def external(self) -> dict[str, Any]:
        return dict(self.data["source"].get("external", {}))

    @property
    def external_uri(self) -> str:
        return str(self.external["uri"]).rstrip("/") + "/"

    @property
    def external_table(self) -> str:
        return str(self.external["table"])

    @property
    def external_row_estimate(self) -> int | None:
        value = self.external.get("row_estimate")
        return int(value) if value is not None else None

    @property
    def external_columns(self) -> list[dict[str, Any]]:
        return list(self.external.get("columns", []))

    @property
    def query(self) -> dict[str, Any]:
        return dict(self.data["source"].get("query", {}))

    @property
    def query_tables(self) -> list[dict[str, Any]]:
        """Return normalized metadata for every managed MySQL query table.

        The single-table ``query`` + ``external`` shape remains supported for
        existing configurations. New multi-table sources keep each SQL file,
        primary key, and column contract together under ``query.tables``.
        """

        configured = self.query.get("tables")
        if configured is not None:
            return [dict(table) for table in configured]
        if not self.external:
            return []
        return [{**self.external, **self.query}]

    def query_table_for(self, table_name: str) -> dict[str, Any] | None:
        for table in self.query_tables:
            if str(table.get("table")) == table_name:
                return table
        return None

    def query_sql_path_for(self, table: dict[str, Any]) -> Path:
        sql_file = table.get("sql_file")
        if not sql_file:
            raise RuntimeError(
                f"query table {table.get('table', '<unknown>')} in source {self.name} has no sql_file configured"
            )
        return (self.root / str(sql_file)).resolve()

    @property
    def query_sql_path(self) -> Path:
        sql_file = self.query.get("sql_file")
        if sql_file:
            return (self.root / str(sql_file)).resolve()
        if not self.query_tables:
            raise RuntimeError(f"query source {self.name} has no sql_file configured")
        return self.query_sql_path_for(self.query_tables[0])

    @property
    def s3_region(self) -> str:
        return str(self.data.get("s3", {}).get("region", "us-east-1"))

    @property
    def s3_profile(self) -> str | None:
        value = self.data.get("s3", {}).get("profile")
        return str(value) if value else None

    @property
    def s3_credentials(self) -> dict[str, str]:
        return dict(self.data.get("s3", {}).get("credentials", {}))

    @property
    def s3_access_key_id_env(self) -> str:
        return self.s3_credentials.get("access_key_id_env", "AWS_ACCESS_KEY_ID")

    @property
    def s3_secret_access_key_env(self) -> str:
        return self.s3_credentials.get("secret_access_key_env", "AWS_SECRET_ACCESS_KEY")

    @property
    def s3_session_token_env(self) -> str | None:
        return self.s3_credentials.get("session_token_env")

    @property
    def databend(self) -> dict[str, Any]:
        return dict(self.data["source"].get("databend", {}))

    @property
    def databend_dsn_env(self) -> str:
        return str(self.databend.get("dsn_env", "DATABEND_DSN"))

    @property
    def databend_database(self) -> str:
        return str(self.databend["database"])

    @property
    def databend_table(self) -> str:
        return str(self.databend["table"])

    @property
    def parquet_compression(self) -> str:
        return str(self.data.get("export", {}).get("parquet_compression", "ZSTD"))

    @property
    def tidb_filter_max_bytes(self) -> int:
        return int(self.data.get("export", {}).get("tidb_filter_max_bytes", 400))

    @property
    def tidb_export_fault_domains(self) -> list[dict[str, Any]]:
        value = self.data.get("export", {}).get("tidb_fault_domains", [])
        if not isinstance(value, list):
            raise ValueError("export.tidb_fault_domains must be a list")
        return list(value)

    @property
    def manifest_freshness_max_age_seconds(self) -> int | None:
        value = self.data.get("load", {}).get("manifest_freshness_max_age_hours")
        if value is None:
            return None
        seconds = int(float(value) * 60 * 60)
        if seconds <= 0:
            raise ValueError("load.manifest_freshness_max_age_hours must be greater than zero")
        return seconds

    @property
    def full_snapshot_delete(self) -> bool:
        return bool(self.data.get("load", {}).get("full_snapshot_delete", True))

    @property
    def materialize_final(self) -> bool:
        return bool(self.data.get("load", {}).get("materialize_final", True)) and self.final_dataset is not None

    def final_table_name_for(self, source_table: str) -> str | None:
        configured = self.data.get("load", {}).get("final_table_names", {}).get(source_table)
        return str(configured) if configured else None

    def primary_key_for(self, table_name: str) -> list[str]:
        configured = self.declared_primary_keys.get(table_name, [])
        if configured:
            return list(configured)
        query_table = self.query_table_for(table_name)
        if query_table is not None:
            return list(query_table.get("primary_key", []))
        if self.external and self.external_table == table_name:
            return list(self.external.get("primary_key", []))
        return []

    def gcs_run_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.bucket}/{self.snapshot_prefix}/"
            f"source={self.name}/run_date={run_date}/run_id={run_id}/"
        )

    def table_gcs_uri(self, table_name: str, run_date: str, run_id: str) -> str:
        return f"{self.gcs_run_uri(run_date, run_id)}table={table_name}/"

    def manifest_gcs_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.bucket}/{self.manifest_prefix}/"
            f"source={self.name}/run_date={run_date}/run_id={run_id}/manifest.json"
        )

    def runtime_log_gcs_uri(self, run_date: str, run_id: str) -> str:
        return (
            f"gs://{self.bucket}/{self.runtime_log_prefix}/"
            f"source={self.name}/run_date={run_date}/run_id={run_id}/events.jsonl"
        )


def load_config(path: str | Path) -> PipelineConfig:
    config_path = Path(path)
    return PipelineConfig(path=config_path, data=yaml.safe_load(config_path.read_text()))