"""Read-only catalog cutover validation for data-access resources."""
from __future__ import annotations

import uuid
from collections import Counter
from collections.abc import Iterable
from dataclasses import dataclass
from typing import Final

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.data_access import constants as c
from app.domains.data_access.models import (
    BINDING_STATUS_ACTIVE,
    BINDING_STATUS_REVOKED,
    BINDING_STATUS_REVOKED_PENDING,
    EFFECT_DESIRED_ABSENT,
    OWNER_RESOLUTION_FALLBACK,
    OWNER_RESOLUTION_RESOLVED,
    OWNER_RESOLUTION_STATUSES,
    OWNER_RESOLUTION_UNRESOLVED,
    OWNER_SOURCE_BOOTSTRAP,
    OWNER_SOURCE_BOOTSTRAP_TRANSFERRED,
    OWNER_SOURCE_CATALOG,
    OWNER_SOURCE_CATALOG_TRANSFERRED,
    OWNER_SOURCES,
    RESOURCE_STATUS_ACTIVE,
    RESOURCE_STATUS_DELETED,
    RESOURCE_STATUS_SUSPENDED,
    RESOURCE_STATUSES,
    DataAccessBinding,
    DataAccessCompiledEffect,
    DataAccessEffectRun,
    DataAccessRequestItem,
    DataAccessResource,
)

CUTOVER_FINDING_BOOTSTRAP_REMAINING: Final = "bootstrap_remaining"
CUTOVER_FINDING_BOOTSTRAP_REQUESTABLE: Final = "bootstrap_requestable"
CUTOVER_FINDING_BOOTSTRAP_HISTORICAL: Final = "bootstrap_historical"
CUTOVER_FINDING_UNMATCHED_BOOTSTRAP: Final = "unmatched_bootstrap"
CUTOVER_FINDING_CATALOG_FED_ACTIVE: Final = "catalog_fed_active"
CUTOVER_FINDING_SUSPENDED: Final = "suspended"
CUTOVER_FINDING_DUPLICATE_LOCATOR: Final = "duplicate_locator"
CUTOVER_FINDING_DUPLICATE_CATALOG_ID: Final = "duplicate_catalog_id"
CUTOVER_FINDING_LOCATOR_DRIFT_WITH_HISTORY: Final = (
    "locator_drift_with_history"
)
CUTOVER_FINDING_OWNER_MISMATCH: Final = "owner_mismatch"
CUTOVER_FINDING_UNRESOLVED_OWNER_FALLBACK: Final = (
    "unresolved_owner_fallback"
)
CUTOVER_FINDING_INVALID: Final = "invalid"
CUTOVER_FINDING_UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE: Final = (
    "unsafe_unmatched_bootstrap_revoke"
)

CUTOVER_FINDINGS: Final = (
    CUTOVER_FINDING_BOOTSTRAP_REMAINING,
    CUTOVER_FINDING_BOOTSTRAP_REQUESTABLE,
    CUTOVER_FINDING_BOOTSTRAP_HISTORICAL,
    CUTOVER_FINDING_UNMATCHED_BOOTSTRAP,
    CUTOVER_FINDING_CATALOG_FED_ACTIVE,
    CUTOVER_FINDING_SUSPENDED,
    CUTOVER_FINDING_DUPLICATE_LOCATOR,
    CUTOVER_FINDING_DUPLICATE_CATALOG_ID,
    CUTOVER_FINDING_LOCATOR_DRIFT_WITH_HISTORY,
    CUTOVER_FINDING_OWNER_MISMATCH,
    CUTOVER_FINDING_UNRESOLVED_OWNER_FALLBACK,
    CUTOVER_FINDING_INVALID,
    CUTOVER_FINDING_UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE,
)

CUTOVER_BLOCKING_FINDINGS: Final = frozenset({
    CUTOVER_FINDING_BOOTSTRAP_REQUESTABLE,
    CUTOVER_FINDING_DUPLICATE_LOCATOR,
    CUTOVER_FINDING_DUPLICATE_CATALOG_ID,
    CUTOVER_FINDING_LOCATOR_DRIFT_WITH_HISTORY,
    CUTOVER_FINDING_OWNER_MISMATCH,
    CUTOVER_FINDING_UNRESOLVED_OWNER_FALLBACK,
    CUTOVER_FINDING_INVALID,
    CUTOVER_FINDING_UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE,
})

UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE_REASONS: Final = frozenset({
    "bootstrap_unmatched",
    "bootstrap_not_in_catalog",
    "catalog_cutover_unmatched",
    "catalog_cutover_cleanup",
})


@dataclass(frozen=True, slots=True)
class DataAccessCutoverResourceHistory:
    request_item_count: int = 0
    binding_count: int = 0
    active_binding_count: int = 0
    revoked_binding_count: int = 0
    absent_effect_count: int = 0
    effect_run_count: int = 0

    @property
    def has_preserved_history(self) -> bool:
        return any((
            self.request_item_count,
            self.binding_count,
            self.absent_effect_count,
            self.effect_run_count,
        ))


@dataclass(frozen=True, slots=True)
class DataAccessCutoverReportItem:
    resource_id: uuid.UUID
    integration_id: str
    provider: str
    resource_type: str
    project_id: str
    dataset_id: str
    locator: str
    display_name: str | None
    catalog_resource_id: str | None
    status: str
    requestable: bool
    owner_source: str
    owner_person_id: str | None
    owner_resolution_status: str
    source_version: str
    source_etag: str
    findings: tuple[str, ...]
    blocking_findings: tuple[str, ...]
    invalid_reasons: tuple[str, ...]
    history: DataAccessCutoverResourceHistory


@dataclass(frozen=True, slots=True)
class DataAccessCutoverReportSummary:
    total_resources: int
    cutover_ready: bool
    blocking_count: int
    category_counts: dict[str, int]
    blocking_category_counts: dict[str, int]
    provider_mutation_attempted: bool = False
    request_handlers_mutate_provider: bool = False
    feishu_callbacks_mutate_provider: bool = False


@dataclass(frozen=True, slots=True)
class DataAccessCutoverReport:
    integration_id: str | None
    integration_count: int
    summary: DataAccessCutoverReportSummary
    items: tuple[DataAccessCutoverReportItem, ...]


async def build_data_access_cutover_report(
    db: AsyncSession,
    *,
    integration_id: str | None = None,
) -> DataAccessCutoverReport:
    """Classify projected resources for the catalog cutover gate.

    The report reads Helix semantic rows only. It does not call provider APIs,
    mutate request/binding/effect state, or read raw catalog/bootstrap input.
    """
    rows = await _load_resource_rows(db, integration_id=integration_id)
    resource_ids = [row.id for row in rows]
    history_by_resource = await _load_history(db, resource_ids)
    unsafe_revoke_resource_ids = await _unsafe_unmatched_bootstrap_revoke_ids(
        db,
        rows,
    )
    duplicate_locator_keys = _duplicate_locator_keys(rows)
    duplicate_catalog_keys = _duplicate_catalog_keys(rows)

    items = tuple(
        _classify_resource(
            row,
            history=history_by_resource.get(
                row.id,
                DataAccessCutoverResourceHistory(),
            ),
            duplicate_locator_keys=duplicate_locator_keys,
            duplicate_catalog_keys=duplicate_catalog_keys,
            unsafe_revoke_resource_ids=unsafe_revoke_resource_ids,
        )
        for row in rows
    )
    items = tuple(sorted(items, key=_item_sort_key))
    category_counts = {
        finding: sum(finding in item.findings for item in items)
        for finding in CUTOVER_FINDINGS
    }
    blocking_category_counts = {
        finding: category_counts[finding]
        for finding in CUTOVER_FINDINGS
        if finding in CUTOVER_BLOCKING_FINDINGS and category_counts[finding]
    }
    blocking_count = sum(bool(item.blocking_findings) for item in items)
    summary = DataAccessCutoverReportSummary(
        total_resources=len(items),
        cutover_ready=blocking_count == 0,
        blocking_count=blocking_count,
        category_counts=category_counts,
        blocking_category_counts=blocking_category_counts,
    )
    return DataAccessCutoverReport(
        integration_id=integration_id,
        integration_count=len({str(row.integration_id) for row in rows}),
        summary=summary,
        items=items,
    )


async def _load_resource_rows(
    db: AsyncSession,
    *,
    integration_id: str | None,
) -> list[DataAccessResource]:
    conditions = []
    if integration_id is not None:
        conditions.append(DataAccessResource.integration_id == integration_id)
    query = select(DataAccessResource).order_by(
        DataAccessResource.integration_id,
        DataAccessResource.project_id,
        DataAccessResource.dataset_id,
        DataAccessResource.id,
    )
    if conditions:
        query = query.where(*conditions)
    return list((await db.scalars(query)).all())


async def _load_history(
    db: AsyncSession,
    resource_ids: list[uuid.UUID],
) -> dict[uuid.UUID, DataAccessCutoverResourceHistory]:
    if not resource_ids:
        return {}
    request_counts = await _resource_count_map(
        db,
        select(DataAccessRequestItem.resource_id).where(
            DataAccessRequestItem.resource_id.in_(resource_ids),
        ),
    )
    run_counts = await _resource_count_map(
        db,
        select(DataAccessEffectRun.resource_id).where(
            DataAccessEffectRun.resource_id.in_(resource_ids),
        ),
    )
    bindings = (
        await db.scalars(
            select(DataAccessBinding).where(
                DataAccessBinding.resource_id.in_(resource_ids),
            ),
        )
    ).all()
    effects = (
        await db.scalars(
            select(DataAccessCompiledEffect).where(
                DataAccessCompiledEffect.resource_id.in_(resource_ids),
            ),
        )
    ).all()

    binding_counts: Counter[uuid.UUID] = Counter()
    active_binding_counts: Counter[uuid.UUID] = Counter()
    revoked_binding_counts: Counter[uuid.UUID] = Counter()
    absent_effect_counts: Counter[uuid.UUID] = Counter()
    for binding in bindings:
        binding_counts[binding.resource_id] += 1
        if binding.status == BINDING_STATUS_ACTIVE:
            active_binding_counts[binding.resource_id] += 1
        if binding.status in {
            BINDING_STATUS_REVOKED,
            BINDING_STATUS_REVOKED_PENDING,
        }:
            revoked_binding_counts[binding.resource_id] += 1
    for effect in effects:
        if effect.desired_state == EFFECT_DESIRED_ABSENT:
            absent_effect_counts[effect.resource_id] += 1

    history: dict[uuid.UUID, DataAccessCutoverResourceHistory] = {}
    for resource_id in resource_ids:
        history[resource_id] = DataAccessCutoverResourceHistory(
            request_item_count=request_counts[resource_id],
            binding_count=binding_counts[resource_id],
            active_binding_count=active_binding_counts[resource_id],
            revoked_binding_count=revoked_binding_counts[resource_id],
            absent_effect_count=absent_effect_counts[resource_id],
            effect_run_count=run_counts[resource_id],
        )
    return history


async def _resource_count_map(
    db: AsyncSession,
    query,
) -> Counter[uuid.UUID]:
    counts: Counter[uuid.UUID] = Counter()
    for resource_id in (await db.scalars(query)).all():
        counts[resource_id] += 1
    return counts


async def _unsafe_unmatched_bootstrap_revoke_ids(
    db: AsyncSession,
    rows: Iterable[DataAccessResource],
) -> set[uuid.UUID]:
    unmatched_bootstrap_ids = [
        row.id
        for row in rows
        if row.owner_source == OWNER_SOURCE_BOOTSTRAP
        and row.catalog_resource_id is None
    ]
    if not unmatched_bootstrap_ids:
        return set()
    bindings = (
        await db.scalars(
            select(DataAccessBinding).where(
                DataAccessBinding.resource_id.in_(unmatched_bootstrap_ids),
                DataAccessBinding.status.in_(
                    (BINDING_STATUS_REVOKED, BINDING_STATUS_REVOKED_PENDING),
                ),
                DataAccessBinding.revoke_reason.in_(
                    tuple(UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE_REASONS),
                ),
            ),
        )
    ).all()
    return {binding.resource_id for binding in bindings}


def _duplicate_locator_keys(
    rows: Iterable[DataAccessResource],
) -> set[tuple[str, str, str, str, str]]:
    counts: Counter[tuple[str, str, str, str, str]] = Counter(
        _locator_key(row)
        for row in rows
        if row.status != RESOURCE_STATUS_DELETED
    )
    return {key for key, count in counts.items() if count > 1}


def _duplicate_catalog_keys(
    rows: Iterable[DataAccessResource],
) -> set[tuple[str, str, str, str]]:
    counts: Counter[tuple[str, str, str, str]] = Counter(
        _catalog_key(row)
        for row in rows
        if row.status != RESOURCE_STATUS_DELETED
        and row.catalog_resource_id is not None
    )
    return {key for key, count in counts.items() if count > 1}


def _classify_resource(
    row: DataAccessResource,
    *,
    history: DataAccessCutoverResourceHistory,
    duplicate_locator_keys: set[tuple[str, str, str, str, str]],
    duplicate_catalog_keys: set[tuple[str, str, str, str]],
    unsafe_revoke_resource_ids: set[uuid.UUID],
) -> DataAccessCutoverReportItem:
    findings: list[str] = []
    invalid_reasons = _invalid_reasons(row)

    if row.owner_source in {
        OWNER_SOURCE_BOOTSTRAP,
        OWNER_SOURCE_BOOTSTRAP_TRANSFERRED,
    }:
        findings.append(CUTOVER_FINDING_BOOTSTRAP_REMAINING)
        if row.status == RESOURCE_STATUS_ACTIVE and row.requestable:
            findings.append(CUTOVER_FINDING_BOOTSTRAP_REQUESTABLE)
        else:
            findings.append(CUTOVER_FINDING_BOOTSTRAP_HISTORICAL)
        if row.catalog_resource_id is None:
            findings.append(CUTOVER_FINDING_UNMATCHED_BOOTSTRAP)

    if (
        row.owner_source in {
            OWNER_SOURCE_CATALOG,
            OWNER_SOURCE_CATALOG_TRANSFERRED,
        }
        and row.status == RESOURCE_STATUS_ACTIVE
    ):
        findings.append(CUTOVER_FINDING_CATALOG_FED_ACTIVE)

    if row.status == RESOURCE_STATUS_SUSPENDED:
        findings.append(CUTOVER_FINDING_SUSPENDED)

    if _locator_key(row) in duplicate_locator_keys:
        findings.append(CUTOVER_FINDING_DUPLICATE_LOCATOR)

    if (
        row.catalog_resource_id is not None
        and _catalog_key(row) in duplicate_catalog_keys
    ):
        findings.append(CUTOVER_FINDING_DUPLICATE_CATALOG_ID)

    if _looks_like_locator_drift_with_history(row, history):
        findings.append(CUTOVER_FINDING_LOCATOR_DRIFT_WITH_HISTORY)

    if _looks_like_owner_mismatch(row):
        findings.append(CUTOVER_FINDING_OWNER_MISMATCH)

    if _has_unresolved_or_fallback_owner(row):
        findings.append(CUTOVER_FINDING_UNRESOLVED_OWNER_FALLBACK)

    if invalid_reasons:
        findings.append(CUTOVER_FINDING_INVALID)

    if row.id in unsafe_revoke_resource_ids:
        findings.append(CUTOVER_FINDING_UNSAFE_UNMATCHED_BOOTSTRAP_REVOKE)

    blocking = tuple(
        finding for finding in findings if finding in CUTOVER_BLOCKING_FINDINGS
    )
    return DataAccessCutoverReportItem(
        resource_id=row.id,
        integration_id=str(row.integration_id),
        provider=row.provider,
        resource_type=row.resource_type,
        project_id=row.project_id,
        dataset_id=row.dataset_id,
        locator=f"{row.project_id}.{row.dataset_id}",
        display_name=row.display_name,
        catalog_resource_id=row.catalog_resource_id,
        status=row.status,
        requestable=row.requestable,
        owner_source=row.owner_source,
        owner_person_id=row.owner_person_id,
        owner_resolution_status=row.owner_resolution_status,
        source_version=row.source_version,
        source_etag=row.source_etag,
        findings=tuple(findings),
        blocking_findings=blocking,
        invalid_reasons=invalid_reasons,
        history=history,
    )


def _invalid_reasons(row: DataAccessResource) -> tuple[str, ...]:
    reasons: list[str] = []
    if row.provider != c.PROVIDER_BIGQUERY:
        reasons.append("non_bigquery_provider")
    if row.resource_type != c.RESOURCE_TYPE_DATASET:
        reasons.append("non_dataset_resource_type")
    if row.status not in RESOURCE_STATUSES:
        reasons.append("invalid_status")
    if row.owner_source not in OWNER_SOURCES:
        reasons.append("invalid_owner_source")
    if row.owner_resolution_status not in OWNER_RESOLUTION_STATUSES:
        reasons.append("invalid_owner_resolution_status")
    if not row.project_id:
        reasons.append("missing_project_id")
    if not row.dataset_id:
        reasons.append("missing_dataset_id")
    if any(sep in row.dataset_id for sep in (".", ":", "/")):
        reasons.append("dataset_id_not_dataset_scoped")
    if row.status == RESOURCE_STATUS_DELETED and row.requestable:
        reasons.append("deleted_resource_requestable")
    if (
        row.status == RESOURCE_STATUS_ACTIVE
        and row.requestable
        and row.owner_source in {
            OWNER_SOURCE_CATALOG,
            OWNER_SOURCE_CATALOG_TRANSFERRED,
        }
        and (
            row.owner_person_id is None
            or row.owner_resolution_status != OWNER_RESOLUTION_RESOLVED
        )
    ):
        reasons.append("requestable_catalog_resource_without_resolved_owner")
    return tuple(reasons)


def _looks_like_locator_drift_with_history(
    row: DataAccessResource,
    history: DataAccessCutoverResourceHistory,
) -> bool:
    return (
        row.catalog_resource_id is not None
        and row.status == RESOURCE_STATUS_SUSPENDED
        and row.requestable is False
        and row.owner_person_id is not None
        and row.owner_resolution_status == OWNER_RESOLUTION_RESOLVED
        and history.has_preserved_history
    )


def _looks_like_owner_mismatch(row: DataAccessResource) -> bool:
    return (
        row.owner_source == OWNER_SOURCE_BOOTSTRAP
        and row.catalog_resource_id is not None
        and row.status == RESOURCE_STATUS_SUSPENDED
        and row.requestable is False
        and row.owner_person_id is None
        and row.owner_resolution_status == OWNER_RESOLUTION_UNRESOLVED
    )


def _has_unresolved_or_fallback_owner(row: DataAccessResource) -> bool:
    return (
        row.status != RESOURCE_STATUS_DELETED
        and (
            row.owner_resolution_status in {
                OWNER_RESOLUTION_UNRESOLVED,
                OWNER_RESOLUTION_FALLBACK,
            }
            or row.owner_person_id is None
        )
    )


def _locator_key(row: DataAccessResource) -> tuple[str, str, str, str, str]:
    return (
        str(row.integration_id),
        row.provider,
        row.resource_type,
        row.project_id,
        row.dataset_id,
    )


def _catalog_key(row: DataAccessResource) -> tuple[str, str, str, str]:
    return (
        str(row.integration_id),
        row.provider,
        row.resource_type,
        str(row.catalog_resource_id),
    )


def _item_sort_key(item: DataAccessCutoverReportItem) -> tuple[int, str, str, str]:
    return (
        0 if item.blocking_findings else 1,
        item.integration_id,
        item.project_id,
        item.dataset_id,
    )