"""Ops-only bootstrap input contract for data-access BigQuery datasets.

DAX-004 defines the transition input shape used before catalog federation is
available. The parser is deliberately side-effect free: it validates operator
input and returns a normalized projection shape, but it does not write rows,
create a manual catalog UI, submit approval requests, or call BigQuery.
"""
from __future__ import annotations

import re
from collections.abc import Iterable, Mapping
from dataclasses import dataclass
from typing import Any, Final
from uuid import UUID

from app.domains.data_access.constants import (
    FIELD_OWNER_PERSON_ID,
    FIELD_OWNER_SOURCE,
    FIELD_SOURCE_ETAG,
    FIELD_SOURCE_VERSION,
    RESOURCE_TYPE_DATASET,
    assert_canonical,
)

BOOTSTRAP_OWNER_SOURCE: Final = "bootstrap"

OWNER_RESOLUTION_RESOLVED: Final = "resolved"
OWNER_RESOLUTION_UNRESOLVED: Final = "unresolved_owner"
OWNER_RESOLUTION_FALLBACK: Final = "fallback_owner"
OWNER_RESOLUTION_STATUS_FIELD: Final = "owner_resolution_status"

OWNER_RESOLUTION_STATUSES: Final = (
    OWNER_RESOLUTION_RESOLVED,
    OWNER_RESOLUTION_UNRESOLVED,
    OWNER_RESOLUTION_FALLBACK,
)
UNRESOLVED_OWNER_STATUSES: Final = frozenset({
    OWNER_RESOLUTION_UNRESOLVED,
    OWNER_RESOLUTION_FALLBACK,
})

_PROJECT_ID_RE = re.compile(r"^[a-z][a-z0-9-]{4,28}[a-z0-9]$")
_DATASET_ID_RE = re.compile(r"^[A-Za-z0-9_]{1,1024}$")
_ENTRY_KEYS: Final = frozenset({
    "provider",
    "resource_type",
    "locator",
    "catalog_id",
    FIELD_OWNER_PERSON_ID,
    OWNER_RESOLUTION_STATUS_FIELD,
    FIELD_OWNER_SOURCE,
    FIELD_SOURCE_VERSION,
    FIELD_SOURCE_ETAG,
    "display_name",
    "requestable",
})
_LOCATOR_KEYS: Final = frozenset({"project_id", "dataset_id"})
_FORBIDDEN_SHAPE_KEYS: Final = frozenset({
    "table",
    "table_id",
    "table_ref",
    "dataset",
    "dataset_ref",
    "project",
    "project_id",
    "project_number",
    "resource",
    "resource_name",
    "fully_qualified_name",
})
_FORBIDDEN_LOCATOR_KEYS: Final = (
    _FORBIDDEN_SHAPE_KEYS - _LOCATOR_KEYS
)


@dataclass(frozen=True, slots=True)
class BigQueryDatasetLocator:
    """Canonical provider locator for one BigQuery dataset."""

    project_id: str
    dataset_id: str

    @property
    def key(self) -> tuple[str, str]:
        return (self.project_id, self.dataset_id)


@dataclass(frozen=True, slots=True)
class BootstrapDatasetResource:
    """Normalized ops-bootstrap projection for a future resource row."""

    provider: str
    resource_type: str
    locator: BigQueryDatasetLocator
    owner_person_id: str | None
    owner_resolution_status: str
    owner_source: str
    source_version: str
    source_etag: str
    catalog_id: str | None = None
    display_name: str | None = None
    requestable: bool = True

    @property
    def resource_owner_authority_enabled(self) -> bool:
        return (
            self.owner_resolution_status == OWNER_RESOLUTION_RESOLVED
            and self.owner_person_id is not None
        )

    def as_resource_projection(self) -> dict[str, Any]:
        """Return the future resource-row projection, without writing it."""
        return {
            "provider": self.provider,
            "resource_type": self.resource_type,
            "project_id": self.locator.project_id,
            "dataset_id": self.locator.dataset_id,
            "catalog_resource_id": self.catalog_id,
            FIELD_OWNER_PERSON_ID: self.owner_person_id,
            OWNER_RESOLUTION_STATUS_FIELD: self.owner_resolution_status,
            FIELD_OWNER_SOURCE: self.owner_source,
            FIELD_SOURCE_VERSION: self.source_version,
            FIELD_SOURCE_ETAG: self.source_etag,
            "display_name": self.display_name,
            "requestable": self.requestable,
            "resource_owner_authority_enabled": (
                self.resource_owner_authority_enabled
            ),
        }


def parse_bootstrap_resource(raw: Mapping[str, Any]) -> BootstrapDatasetResource:
    """Validate and normalize one ops-bootstrap BigQuery dataset entry."""
    if not isinstance(raw, Mapping):
        raise ValueError("bootstrap resource must be an object")
    locator = _parse_dataset_locator(raw.get("locator"))
    _validate_entry_keys(raw)
    provider = assert_canonical(
        "provider",
        _required_string(raw, "provider"),
    )
    resource_type = assert_canonical(
        "resource_type",
        _required_string(raw, "resource_type"),
    )
    if resource_type != RESOURCE_TYPE_DATASET:
        raise ValueError("bootstrap resources are dataset-scoped only")
    owner_person_id = _parse_owner_person_id(raw.get(FIELD_OWNER_PERSON_ID))
    owner_resolution_status = _parse_owner_resolution_status(
        raw.get(OWNER_RESOLUTION_STATUS_FIELD),
        owner_person_id=owner_person_id,
    )
    owner_source = _required_string(raw, FIELD_OWNER_SOURCE)
    if owner_source != BOOTSTRAP_OWNER_SOURCE:
        raise ValueError(
            f"{FIELD_OWNER_SOURCE} must be {BOOTSTRAP_OWNER_SOURCE!r} "
            "for ops-bootstrap data-access input",
        )
    requestable = raw.get("requestable", True)
    if not isinstance(requestable, bool):
        raise ValueError("requestable must be a boolean when provided")

    return BootstrapDatasetResource(
        provider=provider,
        resource_type=resource_type,
        locator=locator,
        owner_person_id=owner_person_id,
        owner_resolution_status=owner_resolution_status,
        owner_source=owner_source,
        source_version=_required_string(raw, FIELD_SOURCE_VERSION),
        source_etag=_required_string(raw, FIELD_SOURCE_ETAG),
        catalog_id=_optional_string(raw.get("catalog_id")),
        display_name=_optional_string(raw.get("display_name")),
        requestable=requestable,
    )


def parse_bootstrap_resources(
    entries: Iterable[Mapping[str, Any]],
) -> tuple[BootstrapDatasetResource, ...]:
    """Validate a batch and reject duplicate active dataset identities."""
    parsed: list[BootstrapDatasetResource] = []
    seen: set[tuple[str, str, str]] = set()
    seen_catalog_ids: set[tuple[str, str, str]] = set()
    for index, entry in enumerate(entries):
        try:
            resource = parse_bootstrap_resource(entry)
        except ValueError as exc:
            raise ValueError(f"bootstrap resource #{index}: {exc}") from exc
        key = (
            resource.provider,
            resource.locator.project_id,
            resource.locator.dataset_id,
        )
        if key in seen:
            raise ValueError(
                "duplicate bootstrap BigQuery dataset locator: "
                f"{resource.locator.project_id}.{resource.locator.dataset_id}",
            )
        seen.add(key)
        if resource.catalog_id is not None:
            catalog_key = (
                resource.provider,
                resource.resource_type,
                resource.catalog_id,
            )
            if catalog_key in seen_catalog_ids:
                raise ValueError(
                    "duplicate bootstrap BigQuery catalog id: "
                    f"{resource.catalog_id}",
                )
            seen_catalog_ids.add(catalog_key)
        parsed.append(resource)
    return tuple(parsed)


def _validate_entry_keys(raw: Mapping[str, Any]) -> None:
    keys = frozenset(str(k) for k in raw.keys())
    forbidden = keys & _FORBIDDEN_SHAPE_KEYS
    if forbidden:
        raise ValueError(
            "bootstrap data-access entries must be catalog-shaped dataset "
            "resources; top-level table/project/freeform fields are "
            f"forbidden: {sorted(forbidden)!r}",
        )
    extra = keys - _ENTRY_KEYS
    if extra:
        raise ValueError(
            "unexpected bootstrap data-access fields: "
            f"{sorted(extra)!r}",
        )


def _parse_dataset_locator(raw: Any) -> BigQueryDatasetLocator:
    if not isinstance(raw, Mapping):
        raise ValueError(
            "locator must be an object with project_id and dataset_id; "
            "freeform dataset strings are not accepted",
        )
    keys = frozenset(str(k) for k in raw.keys())
    forbidden = keys & _FORBIDDEN_LOCATOR_KEYS
    if forbidden:
        raise ValueError(
            "bootstrap data-access locators are dataset-scoped only; "
            f"forbidden locator fields: {sorted(forbidden)!r}",
        )
    extra = keys - _LOCATOR_KEYS
    if extra:
        raise ValueError(
            "locator must contain only project_id and dataset_id; "
            f"unexpected fields: {sorted(extra)!r}",
        )
    project_id = _required_string(raw, "project_id")
    dataset_id = _required_string(raw, "dataset_id")
    if not _PROJECT_ID_RE.fullmatch(project_id):
        raise ValueError(
            "locator.project_id must be a canonical GCP project id, "
            "not a freeform project entry",
        )
    if "." in dataset_id or ":" in dataset_id or "/" in dataset_id:
        raise ValueError(
            "locator.dataset_id must name a dataset only, not a table path",
        )
    if not _DATASET_ID_RE.fullmatch(dataset_id):
        raise ValueError("locator.dataset_id is not a valid BigQuery dataset id")
    return BigQueryDatasetLocator(project_id=project_id, dataset_id=dataset_id)


def _parse_owner_person_id(raw: Any) -> str | None:
    if raw in (None, ""):
        return None
    if not isinstance(raw, str):
        raise ValueError(f"{FIELD_OWNER_PERSON_ID} must be a UUID string")
    try:
        return str(UUID(raw.strip()))
    except ValueError as exc:
        raise ValueError(
            f"{FIELD_OWNER_PERSON_ID} must be a resolvable UUID",
        ) from exc


def _parse_owner_resolution_status(
    raw: Any,
    *,
    owner_person_id: str | None,
) -> str:
    if owner_person_id is not None:
        if raw in (None, "", OWNER_RESOLUTION_RESOLVED):
            return OWNER_RESOLUTION_RESOLVED
        raise ValueError(
            f"{OWNER_RESOLUTION_STATUS_FIELD} cannot be {raw!r} when "
            f"{FIELD_OWNER_PERSON_ID} is present",
        )
    status = _required_string(
        {OWNER_RESOLUTION_STATUS_FIELD: raw},
        OWNER_RESOLUTION_STATUS_FIELD,
    )
    if status == OWNER_RESOLUTION_RESOLVED:
        raise ValueError(
            f"{FIELD_OWNER_PERSON_ID} is required when "
            f"{OWNER_RESOLUTION_STATUS_FIELD} is {OWNER_RESOLUTION_RESOLVED!r}",
        )
    if status not in UNRESOLVED_OWNER_STATUSES:
        raise ValueError(
            f"{OWNER_RESOLUTION_STATUS_FIELD} must be one of "
            f"{OWNER_RESOLUTION_STATUSES!r}",
        )
    return status


def _required_string(raw: Mapping[str, Any], field: str) -> str:
    value = raw.get(field)
    if not isinstance(value, str) or not value.strip():
        raise ValueError(f"{field} is required")
    return value.strip()


def _optional_string(value: Any) -> str | None:
    if value in (None, ""):
        return None
    if not isinstance(value, str):
        raise ValueError("optional string field must be a string when provided")
    stripped = value.strip()
    return stripped or None