from __future__ import annotations

from copy import deepcopy
from typing import Any

Path = tuple[str, ...]


def deep_merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    """Merge source-local config fragments without mutating either input.

    This is for combining override config before it reaches a contract generator. Contract
    generators must use the allowlisted helpers below so source config cannot rewrite invariants
    such as schema, identifiers, layer, classifications, or asset refs.
    """
    result = deepcopy(base)
    for key, value in overrides.items():
        if isinstance(value, dict) and isinstance(result.get(key), dict):
            result[key] = deep_merge(result[key], value)
        else:
            result[key] = deepcopy(value)
    return result


SOURCE_CONTRACT_OVERRIDE_PATHS: frozenset[Path] = frozenset(
    {
        ("purpose",),
        ("source", "description"),
        ("assets", "*", "produced_by"),
        ("freshness",),
        ("tests",),
        ("governance", "retention"),
        ("extensions", "lifecycle"),
        ("extensions", "managed_extract"),
        ("extensions", "row_estimate"),
        ("extensions", "known_blind_spots"),
    }
)

RAW_CONTRACT_OVERRIDE_PATHS: frozenset[Path] = frozenset(
    {
        ("freshness",),
        ("tests",),
        ("governance", "retention"),
        ("extensions", "lifecycle"),
        ("extensions", "managed_extract"),
        ("extensions", "row_estimate"),
        ("extensions", "known_blind_spots"),
    }
)


def apply_source_contract_overrides(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    return apply_allowed_overrides(base, overrides, SOURCE_CONTRACT_OVERRIDE_PATHS, "source contract")


def apply_raw_contract_overrides(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    return apply_allowed_overrides(base, overrides, RAW_CONTRACT_OVERRIDE_PATHS, "raw contract")


def apply_allowed_overrides(
    base: dict[str, Any],
    overrides: dict[str, Any],
    allowed_paths: frozenset[Path],
    label: str,
) -> dict[str, Any]:
    """Apply contract overrides only at explicitly allowed paths.

    Lists normally replace wholesale only when the list path is allowlisted (for example
    `tests`). The special `*` path component permits merging a dictionary into each existing
    list item by index, used for `assets[*].produced_by` without letting config replace asset
    ids, refs, or materialization.
    """
    result = deepcopy(base)
    _merge_allowed(result, overrides or {}, (), allowed_paths, label)
    return result


def _merge_allowed(
    target: dict[str, Any],
    overrides: dict[str, Any],
    path: Path,
    allowed_paths: frozenset[Path],
    label: str,
) -> None:
    for key, value in overrides.items():
        next_path = path + (str(key),)
        if _path_allowed(next_path, allowed_paths):
            target[key] = deepcopy(value)
            continue
        if isinstance(value, dict) and isinstance(target.get(key), dict):
            if not _has_allowed_descendant(next_path, allowed_paths):
                raise ValueError(f"{label} override path is not allowed: {'.'.join(next_path)}")
            _merge_allowed(target[key], value, next_path, allowed_paths, label)
            continue
        if isinstance(value, list) and isinstance(target.get(key), list):
            if not _has_allowed_descendant(next_path, allowed_paths):
                raise ValueError(f"{label} override path is not allowed: {'.'.join(next_path)}")
            _merge_allowed_list(target[key], value, next_path, allowed_paths, label)
            continue
        raise ValueError(f"{label} override path is not allowed: {'.'.join(next_path)}")


def _merge_allowed_list(
    target: list[Any],
    overrides: list[Any],
    path: Path,
    allowed_paths: frozenset[Path],
    label: str,
) -> None:
    if len(overrides) > len(target):
        raise ValueError(f"{label} override path has too many entries: {'.'.join(path)}")
    for index, value in enumerate(overrides):
        if not isinstance(value, dict) or not isinstance(target[index], dict):
            raise ValueError(f"{label} override path is not allowed: {'.'.join(path + ('*',))}")
        _merge_allowed(target[index], value, path + ("*",), allowed_paths, label)


def _path_allowed(path: Path, allowed_paths: frozenset[Path]) -> bool:
    return path in allowed_paths


def _has_allowed_descendant(path: Path, allowed_paths: frozenset[Path]) -> bool:
    return any(len(allowed) > len(path) and allowed[: len(path)] == path for allowed in allowed_paths)