import asyncio
from datetime import UTC, datetime
from decimal import Decimal

import pytest
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker

from app.intelligence.catalog import METRIC_CATALOG, metric_definition, validate_catalog
from app.intelligence.cloud_scope_scoring import (
    CLOUD_SCOPE_METRICS,
    CLOUD_SCOPE_SCORE_VERSION,
    score_cloud_scope,
)
from app.intelligence.privacy import EvidenceObject, UnsafeFactPackage
from app.intelligence.scoring import MetricInput, MetricKind
from app.intelligence.snapshot_service import (
    MetricObservation,
    ProviderResult,
    ProviderUnavailable,
    SnapshotSubject,
    build_snapshot,
    latest_snapshot,
    persist_compiled_snapshot,
    snapshot_history,
)

NOW = datetime(2026, 8, 15, 4, tzinfo=UTC)


class StaticProvider:
    def __init__(self, dimension: str, result: ProviderResult):
        self.dimension = dimension
        self.result = result

    async def collect(self, db, subject):
        return self.result


class FailingProvider:
    dimension = "lifecycle"

    async def collect(self, db, subject):
        raise ProviderUnavailable("upstream unavailable: must not enter snapshot")


class BrokenProvider:
    dimension = "lifecycle"

    async def collect(self, db, subject):
        raise KeyError("programming defect")


def people_result(*, org_coverage=96, evidence_id="person-local-id") -> ProviderResult:
    person = EvidenceObject("person", evidence_id)
    return ProviderResult(
        dimension="people_org",
        metrics=(
            MetricObservation("people_org.hris_freshness", 0, evidence=(person,), data_as_of=NOW),
            MetricObservation(
                "people_org.org_coverage", org_coverage, evidence=(person,), data_as_of=NOW
            ),
            MetricObservation("people_org.manager_coverage", 97, data_as_of=NOW),
            MetricObservation("people_org.prediction_data_coverage", 95, data_as_of=NOW),
        ),
        aggregate_facts={
            "active_count": 500,
            "org_coverage": org_coverage,
            "monthly_cost": Decimal("1234567890123456.1234"),
        },
        data_as_of=NOW,
    )


def lifecycle_result() -> ProviderResult:
    person = EvidenceObject("person", "person-local-id")
    return ProviderResult(
        dimension="lifecycle",
        metrics=(
            MetricObservation("lifecycle.day1_ready", 95, evidence=(person,)),
            MetricObservation("lifecycle.onboarding_sla", 90),
            MetricObservation("lifecycle.leaver_failure_rate", 4),
            MetricObservation("lifecycle.owner_continuity", 96),
        ),
        aggregate_facts={"joiner_count": 20, "blocked_leaver_count": 2},
        data_as_of=NOW,
    )


def test_catalog_is_versioned_complete_for_registered_dimensions():
    validate_catalog()
    assert sum(
        row.weight for row in METRIC_CATALOG.values() if row.dimension == "people_org"
    ) == Decimal("1.00")
    assert sum(
        row.weight for row in METRIC_CATALOG.values() if row.dimension == "lifecycle"
    ) == Decimal("1.00")
    assert metric_definition("people_org.org_coverage", "people_org").label_ja


@pytest.mark.parametrize(
    "kwargs",
    [
        {"subject_type": "unknown", "subject_id": "x", "audience_key": "shared"},
        {"subject_type": "platform", "subject_id": "", "audience_key": "shared"},
        {"subject_type": "platform", "subject_id": "x", "audience_key": ""},
    ],
)
def test_snapshot_subject_rejects_ambiguous_identity(kwargs):
    with pytest.raises(ValueError):
        SnapshotSubject(**kwargs)


@pytest.mark.asyncio
async def test_snapshot_scores_available_dimensions_and_preserves_minimal_evidence(db_session):
    subject = SnapshotSubject("platform", "platform", "shared")
    result = await build_snapshot(
        db_session,
        subject=subject,
        providers=[
            StaticProvider("people_org", people_result()),
            StaticProvider("lifecycle", lifecycle_result()),
        ],
    )
    assert result.created is True
    assert result.provider_failures == ()
    # health-v2 arithmetic. people_org = 100.00 (all four metrics score 100).
    # lifecycle = 100*.30 + 80*.20 + 50*.30 + 100*.20 = 81.00. Only these two
    # dimensions are present, so the overall normalizes across their weights:
    # (100.00*0.10 + 81.00*0.15) / 0.25 = 88.60. Under v1 the same fixture read
    # 86.50 — the retired transfer_recovery metric scored 60 at .15.
    assert result.snapshot.overall_score == Decimal("88.60")
    assert result.snapshot.confidence_score == Decimal("25.00")
    assert result.snapshot.grade == "healthy"
    assert result.snapshot.reference_map == {
        "person_01": {"kind": "person", "object_id": "person-local-id"},
    }
    assert result.snapshot.aggregate_facts["people_org"]["monthly_cost"] == (
        "1234567890123456.1234"
    )
    people = next(row for row in result.snapshot.dimensions if row["key"] == "people_org")
    org_metric = next(row for row in people["metrics"] if row["key"] == "people_org.org_coverage")
    assert org_metric["evidence_refs"] == ["person_01"]
    assert "person-local-id" not in str(result.snapshot.aggregate_facts)


@pytest.mark.asyncio
async def test_same_facts_reuse_immutable_snapshot_and_changed_fact_writes_new(db_session):
    subject = SnapshotSubject("platform", "platform", "shared")
    provider = StaticProvider("people_org", people_result(org_coverage=96))
    first = await build_snapshot(db_session, subject=subject, providers=[provider])
    second = await build_snapshot(db_session, subject=subject, providers=[provider])
    assert first.created is True
    assert second.created is False
    assert second.snapshot.id == first.snapshot.id

    changed = await build_snapshot(
        db_session,
        subject=subject,
        providers=[StaticProvider("people_org", people_result(org_coverage=89))],
    )
    assert changed.created is True
    assert changed.snapshot.id != first.snapshot.id
    assert changed.snapshot.facts_hash != first.snapshot.facts_hash
    latest = await latest_snapshot(db_session, subject=subject)
    assert latest is not None
    assert latest.id == changed.snapshot.id
    history = await snapshot_history(db_session, subject=subject, limit=9999)
    assert len(history) == 2
    assert {row.id for row in history} == {first.snapshot.id, changed.snapshot.id}


@pytest.mark.asyncio
async def test_changed_local_evidence_mapping_writes_a_new_snapshot(db_session):
    subject = SnapshotSubject("platform", "platform", "evidence-change")
    first = await build_snapshot(
        db_session,
        subject=subject,
        providers=[StaticProvider("people_org", people_result(evidence_id="person-a"))],
    )
    changed = await build_snapshot(
        db_session,
        subject=subject,
        providers=[StaticProvider("people_org", people_result(evidence_id="person-b"))],
    )
    assert changed.created is True
    assert changed.snapshot.id != first.snapshot.id
    assert changed.snapshot.facts_hash != first.snapshot.facts_hash
    assert changed.snapshot.reference_map["person_01"]["object_id"] == "person-b"


@pytest.mark.asyncio
async def test_provider_local_aggregate_refs_are_rebased_snapshot_wide(db_session):
    local = ProviderResult(
        dimension="lifecycle",
        metrics=lifecycle_result().metrics,
        aggregate_facts={"forecast": {"candidate_refs": ["person_01"]}},
        data_as_of=NOW,
        reference_map={
            "person_01": {"kind": "person", "object_id": "person-z"},
        },
    )
    result = await build_snapshot(
        db_session,
        subject=SnapshotSubject("platform", "platform", "provider-ref-rebase"),
        providers=[
            StaticProvider("people_org", people_result(evidence_id="person-a")),
            StaticProvider("lifecycle", local),
        ],
    )
    assert result.snapshot.reference_map == {
        "person_01": {"kind": "person", "object_id": "person-a"},
        "person_02": {"kind": "person", "object_id": "person-local-id"},
        "person_03": {"kind": "person", "object_id": "person-z"},
    }
    assert result.snapshot.aggregate_facts["lifecycle"]["forecast"] == {
        "candidate_refs": ["person_03"],
    }


@pytest.mark.asyncio
async def test_provider_outage_reduces_confidence_without_erasing_good_dimension(db_session):
    result = await build_snapshot(
        db_session,
        subject=SnapshotSubject("platform", "platform", "shared"),
        providers=[
            StaticProvider("people_org", people_result()),
            FailingProvider(),
        ],
    )
    assert result.provider_failures == ("lifecycle",)
    assert result.snapshot.overall_score == Decimal("100.00")
    assert result.snapshot.confidence_score == Decimal("10.00")
    assert result.snapshot.aggregate_facts["lifecycle"] == {
        "available": False,
        "reason_code": "provider_unavailable",
    }
    assert result.snapshot.aggregate_facts["people_org"]["available"] is True


@pytest.mark.asyncio
async def test_domain_outage_normalizes_confidence_over_declared_provider_set(db_session):
    result = await build_snapshot(
        db_session,
        subject=SnapshotSubject("domain", "workforce", "hr"),
        providers=[
            StaticProvider("people_org", people_result()),
            FailingProvider(),
        ],
    )
    assert result.provider_failures == ("lifecycle",)
    assert result.snapshot.overall_score == Decimal("100.00")
    assert result.snapshot.confidence_score == Decimal("40.00")
    assert [row["key"] for row in result.snapshot.dimensions] == [
        "people_org", "lifecycle",
    ]


@pytest.mark.asyncio
async def test_zero_sample_provider_does_not_poison_common_data_cutoff(db_session):
    empty_lifecycle = ProviderResult(
        dimension="lifecycle",
        metrics=(
            MetricObservation("lifecycle.day1_ready", 0, confidence=0),
            MetricObservation("lifecycle.onboarding_sla", 0, confidence=0),
            MetricObservation("lifecycle.leaver_failure_rate", 0, confidence=0),
            MetricObservation("lifecycle.owner_continuity", 0, confidence=0),
        ),
        aggregate_facts={"sample_count": 0},
        data_as_of=None,
    )
    result = await build_snapshot(
        db_session,
        subject=SnapshotSubject("platform", "platform", "cutoff-test"),
        providers=[
            StaticProvider("people_org", people_result()),
            StaticProvider("lifecycle", empty_lifecycle),
        ],
    )
    assert result.snapshot.data_as_of.replace(tzinfo=UTC) == NOW


@pytest.mark.asyncio
async def test_all_empty_snapshot_has_no_false_fresh_cutoff(db_session):
    empty = ProviderResult(
        dimension="lifecycle",
        metrics=(
            MetricObservation("lifecycle.day1_ready", 0, confidence=0),
            MetricObservation("lifecycle.onboarding_sla", 0, confidence=0),
            MetricObservation("lifecycle.leaver_failure_rate", 0, confidence=0),
            MetricObservation("lifecycle.owner_continuity", 0, confidence=0),
        ),
        aggregate_facts={"sample_count": 0},
        data_as_of=None,
    )
    result = await build_snapshot(
        db_session,
        subject=SnapshotSubject("platform", "platform", "all-empty"),
        providers=[StaticProvider("lifecycle", empty)],
    )
    assert result.snapshot.data_as_of is None


@pytest.mark.asyncio
async def test_malformed_provider_contract_fails_loudly(db_session):
    duplicate = ProviderResult(
        dimension="people_org",
        metrics=(
            MetricObservation("people_org.org_coverage", 95),
            MetricObservation("people_org.org_coverage", 96),
        ),
        aggregate_facts={},
        data_as_of=NOW,
    )
    with pytest.raises(ValueError, match="duplicate metric"):
        await build_snapshot(
            db_session,
            subject=SnapshotSubject("platform", "platform", "shared"),
            providers=[StaticProvider("people_org", duplicate)],
        )


@pytest.mark.asyncio
async def test_programming_error_is_not_mislabeled_as_source_outage(db_session):
    with pytest.raises(KeyError, match="programming defect"):
        await build_snapshot(
            db_session,
            subject=SnapshotSubject("platform", "platform", "shared"),
            providers=[BrokenProvider()],
        )


@pytest.mark.asyncio
async def test_snapshot_builder_refuses_session_with_unrelated_pending_writes(db_session):
    from app.intelligence.models import IntelligenceGenerationLease

    db_session.add(
        IntelligenceGenerationLease(
            lease_key="unrelated",
            owner_token="unrelated",
            expires_at=NOW,
        )
    )
    with pytest.raises(RuntimeError, match="clean, dedicated"):
        await build_snapshot(
            db_session,
            subject=SnapshotSubject("platform", "platform", "shared"),
            providers=[StaticProvider("people_org", people_result())],
        )


@pytest.mark.asyncio
async def test_sensitive_aggregate_contract_fails_closed(db_session):
    unsafe = ProviderResult(
        dimension="people_org",
        metrics=(MetricObservation("people_org.org_coverage", 95),),
        aggregate_facts={"email": "person@example.com"},
        data_as_of=NOW,
    )
    with pytest.raises(UnsafeFactPackage):
        await build_snapshot(
            db_session,
            subject=SnapshotSubject("platform", "platform", "shared"),
            providers=[StaticProvider("people_org", unsafe)],
        )


@pytest.mark.asyncio
async def test_compiled_snapshot_rejects_unknown_anonymous_reference(db_session):
    metrics = [
        MetricInput(
            definition.key,
            definition.kind,
            0 if definition.kind in {MetricKind.FAILURE_RATE, MetricKind.FRESHNESS} else 100,
            confidence=100,
        )
        for definition in CLOUD_SCOPE_METRICS.values()
    ]
    with pytest.raises(UnsafeFactPackage, match="unknown evidence refs"):
        await persist_compiled_snapshot(
            db_session,
            subject=SnapshotSubject("cloud_scope", "scope", "business_shared"),
            score_version=CLOUD_SCOPE_SCORE_VERSION,
            health=score_cloud_scope(metrics),
            aggregate_facts={"risk_refs": ["server_01"]},
            reference_map={},
            data_as_of=NOW,
        )


@pytest.mark.asyncio
async def test_concurrent_builders_single_write_same_fact_snapshot(db_engine):
    factory = async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False)
    subject = SnapshotSubject("platform", "platform", "shared")
    provider = StaticProvider("people_org", people_result())

    async def run_one():
        async with factory() as session:
            return await build_snapshot(session, subject=subject, providers=[provider])

    first, second = await asyncio.gather(run_one(), run_one())
    assert sorted([first.created, second.created]) == [False, True]
    assert first.snapshot.id == second.snapshot.id