"""Read-only fact-provider orchestration and immutable snapshot persistence."""

from __future__ import annotations

from collections.abc import Sequence
from dataclasses import dataclass, field
from datetime import UTC, datetime
from decimal import Decimal
from typing import Any, Protocol

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

from app.intelligence.catalog import metric_definition
from app.intelligence.models import IntelligenceSnapshot
from app.intelligence.privacy import (
    EvidenceObject,
    assert_llm_safe,
    build_reference_map,
    validate_anonymous_refs,
)
from app.intelligence.scoring import (
    DIMENSION_WEIGHTS,
    SCORE_VERSION,
    MetricInput,
    facts_hash,
    json_value,
    score_dimension,
    score_health,
)

SUBJECT_TYPES = frozenset({
    "platform", "domain", "cloud_scope",
    # Monthly compliance posture (auditor domain). Declared here because
    # ``SnapshotSubject.__post_init__`` rejects anything unlisted — that
    # check exists so a typo'd subject fails at construction rather than
    # quietly writing a snapshot nothing will ever read back.
    "compliance_posture",
})


class ProviderUnavailable(RuntimeError):
    """A recoverable source outage that may publish a degraded snapshot.

    Providers must only translate expected infrastructure/data-source
    availability failures to this type. Programming and contract errors stay
    visible and abort the build instead of being mislabeled as missing data.
    """


@dataclass(frozen=True)
class SnapshotSubject:
    subject_type: str
    subject_id: str
    audience_key: str

    def __post_init__(self) -> None:
        if self.subject_type not in SUBJECT_TYPES:
            raise ValueError(f"Unsupported intelligence subject: {self.subject_type}")
        if not self.subject_id or len(self.subject_id) > 128:
            raise ValueError("subject_id must contain 1-128 characters")
        if not self.audience_key or len(self.audience_key) > 128:
            raise ValueError("audience_key must contain 1-128 characters")


@dataclass(frozen=True)
class MetricObservation:
    key: str
    value: Decimal | int | float
    confidence: Decimal | int | float = Decimal("100")
    evidence: tuple[EvidenceObject, ...] = ()
    data_as_of: datetime | None = None


@dataclass(frozen=True)
class ProviderResult:
    dimension: str
    metrics: tuple[MetricObservation, ...]
    aggregate_facts: dict[str, Any]
    data_as_of: datetime | None
    # Optional provider-local aliases used inside aggregate_facts. The
    # compiler rebases them into the snapshot-wide reference namespace.
    reference_map: dict[str, Any] = field(default_factory=dict)


class FactProvider(Protocol):
    dimension: str

    async def collect(
        self,
        db: AsyncSession,
        subject: SnapshotSubject,
    ) -> ProviderResult: ...


@dataclass(frozen=True)
class SnapshotBuildResult:
    snapshot: IntelligenceSnapshot
    created: bool
    provider_failures: tuple[str, ...] = field(default_factory=tuple)


def _utc(value: datetime) -> datetime:
    return value.replace(tzinfo=UTC) if value.tzinfo is None else value.astimezone(UTC)


def _validate_provider_result(provider: FactProvider, result: ProviderResult) -> None:
    if result.dimension != provider.dimension:
        raise ValueError(f"Provider {provider.dimension!r} returned dimension {result.dimension!r}")
    seen: set[str] = set()
    for observation in result.metrics:
        if observation.key in seen:
            raise ValueError(
                f"Provider {provider.dimension!r} returned duplicate metric {observation.key!r}"
            )
        seen.add(observation.key)
        metric_definition(observation.key, result.dimension)
    assert_llm_safe(result.aggregate_facts)
    for ref, value in result.reference_map.items():
        if not isinstance(ref, str) or not isinstance(value, dict):
            raise ValueError(f"Provider {provider.dimension!r} returned an invalid reference map")
        if set(value) != {"kind", "object_id"} or not all(
            isinstance(value[key], str) and value[key] for key in ("kind", "object_id")
        ):
            raise ValueError(f"Provider {provider.dimension!r} returned an invalid reference map")
    validate_anonymous_refs(result.aggregate_facts, set(result.reference_map))


def _rebase_provider_refs(value: Any, replacements: dict[str, str]) -> Any:
    if isinstance(value, dict):
        return {key: _rebase_provider_refs(child, replacements) for key, child in value.items()}
    if isinstance(value, list):
        return [_rebase_provider_refs(child, replacements) for child in value]
    if isinstance(value, tuple):
        return tuple(_rebase_provider_refs(child, replacements) for child in value)
    if isinstance(value, str):
        return replacements.get(value, value)
    return value


def _compile_results(
    results: Sequence[ProviderResult],
    failed_dimensions: Sequence[str],
    expected_dimensions: Sequence[str],
) -> tuple[
    dict[str, Any],
    list[dict[str, Any]],
    dict[str, Any],
    datetime | None,
    dict[str, Any],
]:
    evidence_objects = [
        evidence
        for result in results
        for observation in result.metrics
        for evidence in observation.evidence
    ]
    evidence_objects.extend(
        EvidenceObject(value["kind"], value["object_id"])
        for result in results
        for value in result.reference_map.values()
    )
    _, reference_map = build_reference_map(evidence_objects)
    reverse_refs = {(row["kind"], row["object_id"]): ref for ref, row in reference_map.items()}

    dimensions: list[dict[str, Any]] = []
    aggregate_facts: dict[str, Any] = {}
    source_times: list[datetime] = []
    for result in sorted(results, key=lambda item: item.dimension):
        metric_inputs: list[MetricInput] = []
        for observation in result.metrics:
            definition = metric_definition(observation.key, result.dimension)
            metric_inputs.append(
                MetricInput(
                    key=observation.key,
                    kind=definition.kind,
                    value=observation.value,
                    weight=definition.weight,
                    confidence=observation.confidence,
                    evidence_refs=tuple(
                        reverse_refs[(item.kind, item.object_id)] for item in observation.evidence
                    ),
                    data_as_of=observation.data_as_of or result.data_as_of,
                )
            )
        dimensions.append(score_dimension(result.dimension, metric_inputs))
        replacements = {
            local_ref: reverse_refs[(value["kind"], value["object_id"])]
            for local_ref, value in result.reference_map.items()
        }
        aggregate_facts[result.dimension] = {
            "available": True,
            **_rebase_provider_refs(result.aggregate_facts, replacements),
        }
        if result.data_as_of is not None:
            source_times.append(_utc(result.data_as_of))

    for dimension in sorted(failed_dimensions):
        aggregate_facts[dimension] = {
            "available": False,
            "reason_code": "provider_unavailable",
        }
    health = score_health(dimensions, expected_dimensions=expected_dimensions)
    # The common cutoff cannot be newer than the stalest participating source.
    # Per-metric timestamps still expose fresher domains independently.
    data_as_of = min(source_times) if source_times else None
    dimension_json = json_value(health["dimensions"])
    facts_json = json_value(aggregate_facts, decimal_as_string=True)
    assert_llm_safe({"dimensions": dimension_json, "facts": facts_json})
    validate_anonymous_refs(
        {"dimensions": dimension_json, "facts": facts_json}, set(reference_map),
    )
    return health, dimension_json, facts_json, data_as_of, reference_map


async def _persist_snapshot(
    db: AsyncSession,
    *,
    subject: SnapshotSubject,
    health: dict[str, Any],
    dimensions: list[dict[str, Any]],
    aggregate_facts: dict[str, Any],
    reference_map: dict[str, Any],
    data_as_of: datetime | None,
    score_version: str = SCORE_VERSION,
) -> tuple[IntelligenceSnapshot, bool]:
    digest = facts_hash(
        {
            "dimensions": dimensions,
            "aggregate_facts": aggregate_facts,
            # Anonymous refs are ordinal within one package. If the backing
            # object changes while the aggregate numbers stay equal, reusing
            # the old row would retain a stale local evidence mapping.
            "reference_map": reference_map,
        },
        score_version=score_version,
    )
    identity = (
        IntelligenceSnapshot.subject_type == subject.subject_type,
        IntelligenceSnapshot.subject_id == subject.subject_id,
        IntelligenceSnapshot.audience_key == subject.audience_key,
        IntelligenceSnapshot.score_version == score_version,
        IntelligenceSnapshot.facts_hash == digest,
    )
    existing = (
        await db.execute(select(IntelligenceSnapshot).where(*identity).limit(1))
    ).scalar_one_or_none()
    if existing is not None:
        return existing, False

    snapshot = IntelligenceSnapshot(
        subject_type=subject.subject_type,
        subject_id=subject.subject_id,
        audience_key=subject.audience_key,
        score_version=score_version,
        facts_hash=digest,
        overall_score=health["overall_score"],
        confidence_score=health["confidence_score"],
        grade=health["grade"],
        dimensions=dimensions,
        aggregate_facts=aggregate_facts,
        reference_map=reference_map,
        data_as_of=data_as_of,
    )
    try:
        async with db.begin_nested():
            db.add(snapshot)
            await db.flush()
    except IntegrityError:
        # A concurrent daily/manual builder won the immutable identity. The
        # savepoint keeps the session structurally usable; roll back the outer
        # read snapshot as well so MySQL REPEATABLE READ can see the winner
        # that committed after our initial existence query.
        await db.rollback()
        existing = (
            await db.execute(select(IntelligenceSnapshot).where(*identity).limit(1))
        ).scalar_one()
        return existing, False
    # This service owns a read-only collection + one immutable insert unit.
    # Commit here so a concurrent builder blocked on the unique identity can
    # observe the winner instead of waiting for a request-scope commit.
    await db.commit()
    return snapshot, True


async def persist_compiled_snapshot(
    db: AsyncSession,
    *,
    subject: SnapshotSubject,
    score_version: str,
    health: dict[str, Any],
    aggregate_facts: dict[str, Any],
    reference_map: dict[str, Any],
    data_as_of: datetime | None,
) -> SnapshotBuildResult:
    """Persist a pre-scored, privacy-safe fact package immutably.

    Specialized subjects such as one cloud subscription have their own
    versioned metric catalogue and therefore cannot use the platform provider
    compiler. They still share the exact same dedupe, race and privacy rules.
    """
    if db.new or db.dirty or db.deleted:
        raise RuntimeError("persist_compiled_snapshot requires a clean, dedicated DB session")
    if not score_version or len(score_version) > 32:
        raise ValueError("score_version must contain 1-32 characters")
    if health.get("score_version") != score_version:
        raise ValueError("Compiled health score_version does not match persistence version")
    overall = health.get("overall_score")
    grade = health.get("grade")
    dimensions = health.get("dimensions")
    if overall is None or not grade or not isinstance(dimensions, list):
        raise ValueError("Compiled health must contain a scored result and dimensions")
    compiled_dimensions = json_value(dimensions)
    compiled_facts = json_value(aggregate_facts, decimal_as_string=True)
    assert_llm_safe({"dimensions": compiled_dimensions, "facts": compiled_facts})
    validate_anonymous_refs(
        {"dimensions": compiled_dimensions, "facts": compiled_facts},
        set(reference_map),
    )
    snapshot, created = await _persist_snapshot(
        db,
        subject=subject,
        health=health,
        dimensions=compiled_dimensions,
        aggregate_facts=compiled_facts,
        reference_map=reference_map,
        data_as_of=data_as_of,
        score_version=score_version,
    )
    return SnapshotBuildResult(snapshot=snapshot, created=created)


async def build_snapshot(
    db: AsyncSession,
    *,
    subject: SnapshotSubject,
    providers: Sequence[FactProvider],
) -> SnapshotBuildResult:
    """Build one immutable snapshot using a dedicated, clean DB session.

    The function commits the immutable insert to make database single-flight
    races observable under MySQL REPEATABLE READ. Refuse sessions carrying
    unrelated writes so a future request handler cannot commit them by
    accident.
    """
    if db.new or db.dirty or db.deleted:
        raise RuntimeError("build_snapshot requires a clean, dedicated DB session")
    dimensions = [provider.dimension for provider in providers]
    unknown = set(dimensions) - set(DIMENSION_WEIGHTS)
    if unknown:
        raise ValueError(f"Unknown fact-provider dimensions: {sorted(unknown)}")
    if len(set(dimensions)) != len(dimensions):
        raise ValueError("Only one fact provider may own a health dimension")

    results: list[ProviderResult] = []
    failures: list[str] = []
    for provider in providers:
        try:  # savepoint-exempt: providers are contractually read-only; rollback recovers DB errors
            result = await provider.collect(db, subject)
        except ProviderUnavailable:
            await db.rollback()
            failures.append(provider.dimension)
            continue
        # A malformed provider contract is a deploy-time programming error,
        # not missing business data. Fail loudly instead of publishing a
        # misleading "source unavailable" snapshot.
        _validate_provider_result(provider, result)
        results.append(result)

    health, compiled_dimensions, facts, data_as_of, reference_map = _compile_results(
        results,
        failures,
        tuple(DIMENSION_WEIGHTS) if subject.subject_type == "platform" else dimensions,
    )
    snapshot, created = await _persist_snapshot(
        db,
        subject=subject,
        health=health,
        dimensions=compiled_dimensions,
        aggregate_facts=facts,
        reference_map=reference_map,
        data_as_of=data_as_of,
    )
    return SnapshotBuildResult(
        snapshot=snapshot,
        created=created,
        provider_failures=tuple(sorted(failures)),
    )


async def latest_snapshot(
    db: AsyncSession,
    *,
    subject: SnapshotSubject,
) -> IntelligenceSnapshot | None:
    """Newest immutable snapshot for one exact RBAC audience."""
    return (
        await db.execute(
            select(IntelligenceSnapshot)
            .where(
                IntelligenceSnapshot.subject_type == subject.subject_type,
                IntelligenceSnapshot.subject_id == subject.subject_id,
                IntelligenceSnapshot.audience_key == subject.audience_key,
            )
            .order_by(
                IntelligenceSnapshot.created_at.desc(),
                IntelligenceSnapshot.id.desc(),
            )
            .limit(1)
        )
    ).scalar_one_or_none()


async def snapshot_history(
    db: AsyncSession,
    *,
    subject: SnapshotSubject,
    limit: int = 90,
) -> list[IntelligenceSnapshot]:
    """Bounded newest-first history; retention itself remains unlimited."""
    bounded_limit = max(1, min(365, limit))
    return list(
        (
            await db.execute(
                select(IntelligenceSnapshot)
                .where(
                    IntelligenceSnapshot.subject_type == subject.subject_type,
                    IntelligenceSnapshot.subject_id == subject.subject_id,
                    IntelligenceSnapshot.audience_key == subject.audience_key,
                )
                .order_by(
                    IntelligenceSnapshot.created_at.desc(),
                    IntelligenceSnapshot.id.desc(),
                )
                .limit(bounded_limit)
            )
        )
        .scalars()
        .all()
    )