"""Build the versioned dbt-to-BigQuery lineage catalog consumed by DAX.

This is deliberately a build-time projection.  DAX request pages must not
start BigQuery jobs or infer dependencies from SQL at request time.  The dbt
manifest is the one dependency source, while the generated catalog is the
small, deployable read model used by the request UI and MCP read tool.
"""
from __future__ import annotations

from collections import defaultdict
from datetime import UTC, datetime
from typing import Any

CATALOG_SCHEMA_VERSION = 1
_SUPPORTED_RESOURCE_TYPES = frozenset({"model", "seed", "snapshot", "source"})
_RESOURCE_TYPE_PRIORITY = {
    "model": 0,
    "snapshot": 1,
    "seed": 2,
    "source": 3,
}


def build_dbt_lineage_catalog(
    manifest: dict[str, Any],
    *,
    catalog_id: str,
    source_version: str,
    generated_at: datetime | None = None,
) -> dict[str, Any]:
    """Project dbt manifest nodes and direct dependency edges to BigQuery.

    The projection contains metadata only: relation names, dbt node kinds,
    materializations, and direct ``ref``/``source`` edges.  It never contains
    query text, table rows, credentials, or a BigQuery execution result.
    """
    if not catalog_id.strip():
        raise ValueError("catalog_id is required")
    if not source_version.strip():
        raise ValueError("source_version is required")

    documents: dict[str, dict[str, Any]] = {}
    for collection_name in ("nodes", "sources"):
        collection = manifest.get(collection_name) or {}
        if not isinstance(collection, dict):
            raise ValueError(f"dbt manifest {collection_name} must be an object")
        for node_id, document in collection.items():
            if not isinstance(node_id, str) or not isinstance(document, dict):
                continue
            if document.get("resource_type") not in _SUPPORTED_RESOURCE_TYPES:
                continue
            database = _text(document.get("database"))
            schema = _text(document.get("schema"))
            table = _relation_name(document)
            if not database or not schema or not table:
                continue
            documents[node_id] = document

    canonical_id_by_node_id, canonical_documents = _canonical_relations(documents)
    nodes = [
        _catalog_node(node_id, document)
        for node_id, document in sorted(canonical_documents.items())
    ]
    known_node_ids = set(documents)
    parent_map = manifest.get("parent_map") or {}
    if not isinstance(parent_map, dict):
        raise ValueError("dbt manifest parent_map must be an object")
    node_type_by_id = {
        node_id: str(document["resource_type"])
        for node_id, document in canonical_documents.items()
    }
    canonical_edges = {
        (
            canonical_id_by_node_id[upstream_id],
            canonical_id_by_node_id[downstream_id],
        )
        for downstream_id, upstream_ids in sorted(parent_map.items())
        if downstream_id in known_node_ids and isinstance(upstream_ids, list)
        for upstream_id in sorted(upstream_ids)
        if upstream_id in known_node_ids
        and canonical_id_by_node_id[upstream_id]
        != canonical_id_by_node_id[downstream_id]
    }
    edges = [
        {
            "upstream_node_id": upstream_id,
            "downstream_node_id": downstream_id,
            "relationship": (
                "source"
                if node_type_by_id[upstream_id] == "source"
                else "ref"
            ),
        }
        for upstream_id, downstream_id in sorted(canonical_edges)
    ]
    timestamp = (generated_at or datetime.now(UTC)).astimezone(UTC)
    manifest_metadata = manifest.get("metadata") or {}
    manifest_generated_at = _text(
        manifest_metadata.get("generated_at")
        if isinstance(manifest_metadata, dict)
        else None,
    )
    return {
        "schema_version": CATALOG_SCHEMA_VERSION,
        "catalog_id": catalog_id.strip(),
        "provider": "bigquery",
        "source_version": source_version.strip(),
        "source_generated_at": manifest_generated_at,
        "generated_at": timestamp.isoformat().replace("+00:00", "Z"),
        "nodes": nodes,
        "edges": edges,
    }


def _canonical_relations(
    documents: dict[str, dict[str, Any]],
) -> tuple[dict[str, str], dict[str, dict[str, Any]]]:
    """Collapse dbt aliases that point at one physical BigQuery relation.

    A dbt project may declare the same table through several source names.
    DAX is relation-addressed, so exposing those internal aliases as separate
    nodes makes one table locator ambiguous. Prefer a dbt-produced node when a
    relation is also declared as a source, then choose the stable node id.
    All dependencies are remapped to that canonical physical relation.
    """
    node_ids_by_locator: dict[tuple[str, str, str], list[str]] = defaultdict(list)
    for node_id, document in documents.items():
        node_ids_by_locator[_relation_locator(document)].append(node_id)

    canonical_id_by_node_id: dict[str, str] = {}
    canonical_documents: dict[str, dict[str, Any]] = {}
    for node_ids in node_ids_by_locator.values():
        canonical_id = min(
            node_ids,
            key=lambda node_id: (
                _RESOURCE_TYPE_PRIORITY[str(documents[node_id]["resource_type"])],
                node_id,
            ),
        )
        canonical_documents[canonical_id] = documents[canonical_id]
        for node_id in node_ids:
            canonical_id_by_node_id[node_id] = canonical_id
    return canonical_id_by_node_id, canonical_documents


def _relation_locator(document: dict[str, Any]) -> tuple[str, str, str]:
    return (
        _text(document.get("database")),
        _text(document.get("schema")),
        _relation_name(document),
    )


def _catalog_node(node_id: str, document: dict[str, Any]) -> dict[str, Any]:
    config = document.get("config") or {}
    materialized = (
        _text(config.get("materialized")) if isinstance(config, dict) else None
    )
    return {
        "node_id": node_id,
        "project_id": _text(document.get("database")),
        "dataset_id": _text(document.get("schema")),
        "table_id": _relation_name(document),
        "node_type": _text(document.get("resource_type")),
        "materialization": materialized,
        "description": _text(document.get("description")),
    }


def _relation_name(document: dict[str, Any]) -> str:
    return (
        _text(document.get("alias"))
        or _text(document.get("identifier"))
        or _text(document.get("name"))
    )


def _text(value: object) -> str:
    return str(value or "").strip()