from __future__ import annotations

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

import yaml

from ingestion.common.contract_overrides import deep_merge
from ingestion.common.env import load_secrets, require_env

# ingestion/common/load_config.py -> repo root is three parents up.
REPO_ROOT = Path(__file__).resolve().parents[2]


@dataclass(frozen=True)
class TableTargets:
    """Raw load target for one source table, inherited from its raw-snapshot contract.

    The load stage materializes only the raw layer; staging and above are transform-layer
    concerns (see ADR 0016), so no current/delta targets are resolved here.
    """

    table: str
    raw_table: str | None
    classifications: list[str]
    product_ids: list[str]


@dataclass(frozen=True)
class LoadConfigBase:
    """Storage-agnostic load-source definition shared by every bulk loader.

    A load source declares where to write (Cloud Lake DSN), the target databases it owns,
    idempotent warehouse setup files, and the source contracts whose target tables and
    governance it inherits. Subclasses add the object-store credentials specific to one
    loader (S3 keys, GCS credential). The physical column schema for COPY comes from the
    run manifest, not from here.
    """

    path: Path
    data: dict[str, Any]

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

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

    @property
    def cloud_lake_dsn(self) -> str:
        return require_env(self.data["cloud_lake"]["dsn_env"])

    @property
    def cloud_lake_databases(self) -> dict[str, str]:
        return dict(self.data["cloud_lake"]["databases"])

    @property
    def setup_files(self) -> list[str]:
        return list(self.data.get("setup", {}).get("databend", {}).get("files", []))

    @property
    def source_contract_refs(self) -> list[str]:
        """Source contracts this load consumes as input (its declared upstreams).

        Each ``ref`` is a repo-relative path that may contain glob wildcards, so a single
        entry can consume every per-table source contract a source produces.
        """
        refs: list[str] = []
        for item in self.data.get("source_contracts", []):
            pattern = item["ref"]
            matches = sorted(str(p.relative_to(REPO_ROOT)) for p in REPO_ROOT.glob(pattern))
            refs.extend(matches or [pattern])
        return refs

    @property
    def generated_contracts_dir(self) -> Path:
        return self.root / "contracts" / "generated"

    # --- External-incremental mode (storage-agnostic; ADR 0018) -------------------------
    # When a top-level `external` block is present, the loader reads a pre-existing,
    # externally-produced bucket (no extract stage / manifest): it lists objects, infers the
    # schema from a sample, and loads incrementally per a pluggable strategy. The storage
    # credential stays under the loader's own `gcs`/`s3` block; everything here is shared.

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

    @property
    def is_external(self) -> bool:
        return self.external is not None

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

    @property
    def external_bucket(self) -> str:
        return self.external_uri.split("://", 1)[1].split("/", 1)[0]

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

    @property
    def delta(self) -> bool:
        return bool(self.external.get("delta", True))

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

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

    @property
    def external_owner(self) -> str:
        return self.external.get("owner", "data-platform")

    @property
    def external_system(self) -> str:
        return self.external.get("system", "external")

    @property
    def external_primary_key(self) -> list[str]:
        return list(self.external.get("primary_key", []))

    @property
    def external_source_contract_overrides(self) -> dict[str, Any]:
        external = self.external or {}
        return dict(external.get("source_contract", {}))

    def raw_contract_override_for(self, table: str) -> dict[str, Any]:
        """Optional source-local overrides for generated raw product contracts.

        Most raw contracts should be pure derivations from their upstream source contract. Rare
        cases, such as reviewed one-off freezes, can still declare executable tests or lifecycle
        metadata in source.yaml so regeneration preserves them.
        """
        overrides: dict[str, Any] = {}
        configured = self.data.get("raw_contracts", {})
        table_keys = {table, table.lower(), table.upper()}
        for key, value in configured.items():
            if key in table_keys or str(key).lower() == table.lower():
                overrides = deep_merge(overrides, value)
        if self.is_external and self.external_table.lower() == table.lower():
            overrides = deep_merge(overrides, self.external.get("raw_contract", {}))
        return overrides

    @property
    def stage_name(self) -> str:
        return self.external.get("stage_name", f"{self.name}_ext_stage")

    @property
    def classification_rules(self) -> dict[str, Any]:
        governance = self.data.get("governance", {})
        return {
            "baseline": list(governance.get("classifications", [])),
            "patterns": dict(governance.get("classification_patterns", {})),
            "name_rules": list(governance.get("classification_name_rules", [])),
        }

    def resolve_contracts(self) -> dict[str, TableTargets]:
        """Build per-table raw load targets from this load source's raw-snapshot contracts.

        The load stage owns and generates only the raw product contracts (under
        ``contracts/generated/``); their target table name, keys, and classifications are
        read here rather than recomputed. Tables are keyed by ``source.identifiers.table``.
        """
        merged: dict[str, dict[str, Any]] = {}
        for contract_path in sorted(self.generated_contracts_dir.glob("*.contract.yaml")):
            contract = yaml.safe_load(contract_path.read_text())
            if contract.get("kind") == "source" or contract.get("layer") != "raw":
                continue
            table = contract.get("source", {}).get("identifiers", {}).get("table")
            if not table:
                raise RuntimeError(f"contract has no source.identifiers.table: {contract_path}")
            entry = merged.setdefault(
                table,
                {"raw_table": None, "classifications": set(), "product_ids": []},
            )
            entry["product_ids"].append(contract["id"])
            entry["classifications"].update(contract.get("governance", {}).get("classifications", []))
            assets = contract.get("assets", [])
            if assets:
                entry["raw_table"] = assets[0]["ref"]
        return {
            table: TableTargets(
                table=table,
                raw_table=entry["raw_table"],
                classifications=sorted(entry["classifications"]),
                product_ids=entry["product_ids"],
            )
            for table, entry in merged.items()
        }


T = TypeVar("T", bound=LoadConfigBase)


def load_config_file(path: str | Path, cls: type[T]) -> T:
    """Parse a load ``source.yaml`` into ``cls`` and load its source-local secrets.

    Shared by every loader's ``load_load_config`` and by the cross-cutting setup runner
    (which only needs the storage-agnostic base), so the same yaml + secrets handling is
    not reimplemented per loader.
    """
    config_path = Path(path)
    data = yaml.safe_load(config_path.read_text())
    secrets_file = data.get("load", {}).get("secrets_file", "secrets.env")
    load_secrets(config_path.parent / secrets_file)
    return cls(path=config_path, data=data)