"""Cloud subscription, resource-efficiency, access, and network facts."""

from __future__ import annotations

from collections.abc import Callable, Sequence
from datetime import UTC, datetime, timedelta
from decimal import Decimal

from sqlalchemy import or_, select
from sqlalchemy.exc import SQLAlchemyError
from sqlalchemy.ext.asyncio import AsyncSession

from app.domains.cmdb.cert_service import effective_cert_status
from app.domains.cmdb.efficiency import NON_RUNNING_SERVER_STATUSES
from app.domains.cmdb.efficiency_service import _effective_triage
from app.domains.cmdb.models import (
    CERT_STATUS_VALID,
    EFFICIENCY_RECO_NONE,
    EFFICIENCY_TRIAGE_OPEN,
    EFFICIENCY_VERDICT_BUSY,
    EFFICIENCY_VERDICT_HEALTHY,
    EFFICIENCY_VERDICT_IDLE,
    EFFICIENCY_VERDICT_OVERSIZED,
    GRANT_STATUS_ACTIVE,
    PROD_ENVIRONMENTS,
    SCOPE_STATUS_ACTIVE,
    CmdbCertificate,
    CmdbCloudAccessGrant,
    CmdbCloudScope,
    CmdbDomain,
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerEfficiencyTriage,
)
from app.identity.person_models import IdentityPerson
from app.integrations.models import Integration
from app.intelligence.privacy import EvidenceObject
from app.intelligence.snapshot_service import (
    MetricObservation,
    ProviderResult,
    ProviderUnavailable,
    SnapshotSubject,
)

_EVIDENCE_LIMIT = 50
_ACTIVE_SERVER_EXCLUSIONS = NON_RUNNING_SERVER_STATUSES | {"decommissioned", "deleted"}
_CLASSIFIED_VERDICTS = frozenset({
    EFFICIENCY_VERDICT_IDLE,
    EFFICIENCY_VERDICT_OVERSIZED,
    EFFICIENCY_VERDICT_HEALTHY,
    EFFICIENCY_VERDICT_BUSY,
})


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


def _pct(value: int, total: int) -> Decimal:
    return (Decimal(value) * 100 / Decimal(total)).quantize(Decimal("0.01"))


def _latest(values: Sequence[datetime]) -> datetime | None:
    return max((_utc(value) for value in values), default=None)


class CloudNetworkFactProvider:
    """Collect global cloud/network health without re-deriving cloud savings."""

    dimension = "cloud_network"

    def __init__(self, *, now: Callable[[], datetime] | None = None) -> None:
        self._now = now or (lambda: datetime.now(UTC))

    @staticmethod
    def _validate_subject(subject: SnapshotSubject) -> None:
        if subject.subject_type == "platform" and subject.subject_id == "platform":
            return
        if subject.subject_type == "domain" and subject.subject_id == "cloud_network":
            return
        raise ValueError("Cloud/network facts only support platform or cloud_network snapshots")

    async def collect(self, db: AsyncSession, subject: SnapshotSubject) -> ProviderResult:
        self._validate_subject(subject)
        try:
            return await self._collect(db)
        except SQLAlchemyError as exc:
            raise ProviderUnavailable("cloud/network datastore unavailable") from exc

    async def _collect(self, db: AsyncSession) -> ProviderResult:
        now = _utc(self._now())
        enabled_integration_ids = {
            str(value)
            for value in (
                await db.execute(
                    select(Integration.id).where(
                        Integration.kind == "cloud", Integration.enabled.is_(True),
                    )
                )
            ).scalars()
        }
        scopes = list(
            (
                await db.execute(
                    select(CmdbCloudScope).where(
                        CmdbCloudScope.integration_id.in_(enabled_integration_ids),
                        CmdbCloudScope.deleted_at.is_(None),
                        CmdbCloudScope.status == SCOPE_STATUS_ACTIVE,
                    )
                )
            ).scalars()
        ) if enabled_integration_ids else []
        scope_ids = {str(row.id) for row in scopes}
        servers = list(
            (
                await db.execute(
                    select(CmdbServer).where(
                        CmdbServer.integration_id.in_(enabled_integration_ids),
                        CmdbServer.deleted_at.is_(None),
                    )
                )
            ).scalars()
        ) if enabled_integration_ids else []
        servers = [
            row for row in servers if row.status.lower() not in _ACTIVE_SERVER_EXCLUSIONS
        ]
        server_ids = {str(row.id) for row in servers}
        domains = list(
            (
                await db.execute(
                    select(CmdbDomain).where(
                        CmdbDomain.integration_id.in_(enabled_integration_ids),
                        CmdbDomain.deleted_at.is_(None),
                        CmdbDomain.status == "active",
                    )
                )
            ).scalars()
        ) if enabled_integration_ids else []
        certificate_integration_clause = (
            or_(
                CmdbCertificate.integration_id.is_(None),
                CmdbCertificate.integration_id.in_(enabled_integration_ids),
            )
            if enabled_integration_ids
            else CmdbCertificate.integration_id.is_(None)
        )
        certificates = list(
            (
                await db.execute(
                    select(CmdbCertificate).where(
                        CmdbCertificate.deleted_at.is_(None),
                        certificate_integration_clause,
                    )
                )
            ).scalars()
        )

        # One daily expected cycle: within one missed cycle remains healthy;
        # two and three+ use the immutable FRESHNESS thresholds in
        # ``scoring.score_metric`` (unchanged from health-v1 through v2).
        #
        # ``last_synced_at`` does not mean the same thing on every scope, and
        # reading it uniformly is a category error:
        #   * A discovery-synced scope (``last_seen_sync_job_id`` set) carries
        #     the last time a connector actually confirmed it cloud-side. A gap
        #     here is a broken sync job or credential — the real signal.
        #   * A CHILD scope (``parent_scope_id`` set, e.g. the six
        #     ``azure_resource_group`` rows Helix provisions inside a tracked
        #     subscription for an Azure OpenAI key request) is never enumerated
        #     by the discovery sync, because that sync walks SUBSCRIPTIONS. Its
        #     ``last_synced_at`` never moves off its creation stamp no matter
        #     how healthy everything is, so it inherits its parent's freshness
        #     — the parent is the row a connector actually refreshes.
        #     Production 2026-08-18: one such row (an Azure OpenAI resource
        #     group whose key is valid to 2026-12-31, inside a subscription
        #     synced hours earlier) reported "20 missed cycles" and pinned the
        #     whole platform at the worst band while all 233 discovery-synced
        #     scopes were under a day old, and sent the reader to check a
        #     connector that was never broken.
        #   * A ROOT scope that discovery has never confirmed keeps its own
        #     stamp and stays strict. Nothing legitimate refreshes it, so it
        #     really is stale inventory and must keep scoring as such.
        def _effective_sync_time(
            scope: CmdbCloudScope, seen: frozenset[str] = frozenset(),
        ) -> datetime:
            if scope.last_seen_sync_job_id is not None:
                return _utc(scope.last_synced_at)
            if scope.parent_scope_id is None:
                return _utc(scope.last_synced_at)
            parent_id = str(scope.parent_scope_id)
            parent = scope_by_id_all.get(parent_id)
            # ``seen`` breaks a cycle if parent pointers are ever corrupted;
            # falling back to the row's own stamp keeps the strict reading.
            if parent is None or parent_id in seen:
                return _utc(scope.last_synced_at)
            return _effective_sync_time(parent, seen | {parent_id})

        scope_by_id_all = {str(row.id): row for row in scopes}
        sync_times = [_effective_sync_time(row) for row in scopes]
        sync_as_of = min(sync_times, default=None)
        missed_cycles = (
            max(0, int((now - sync_as_of).total_seconds() // 86_400))
            if sync_as_of is not None else 0
        )
        stale_scopes = [
            row for row in scopes
            if now - _effective_sync_time(row) >= timedelta(days=2)
        ]
        inherited_scopes = [
            row for row in scopes
            if row.last_seen_sync_job_id is None and row.parent_scope_id is not None
        ]
        never_discovered_root_scopes = [
            row for row in scopes
            if row.last_seen_sync_job_id is None and row.parent_scope_id is None
        ]

        efficiency = list(
            (
                await db.execute(
                    select(CmdbServerEfficiency).where(
                        CmdbServerEfficiency.server_id.in_(server_ids)
                    )
                )
            ).scalars()
        ) if server_ids else []
        classified = [row for row in efficiency if row.verdict in _CLASSIFIED_VERDICTS]
        # "Actionable" must mean the same thing here and on the efficiency
        # workspace's 待处理 count, or an owner can work the queue to zero
        # while this metric keeps reporting the original backlog — the loop
        # the triage table was built to close would never reach the score.
        # ``_effective_triage`` is the shared freshness-aware reading: a
        # stored decision only suppresses the row while the verdict it was
        # made against still holds, and an elapsed snooze reopens it.
        triage_by_server = {
            str(row.server_id): row
            for row in (
                (
                    await db.execute(
                        select(CmdbServerEfficiencyTriage).where(
                            CmdbServerEfficiencyTriage.server_id.in_(server_ids)
                        )
                    )
                ).scalars()
                if server_ids else []
            )
        }
        flagged = [
            row for row in classified
            if row.verdict in (EFFICIENCY_VERDICT_IDLE, EFFICIENCY_VERDICT_OVERSIZED)
            and row.recommendation != EFFICIENCY_RECO_NONE
        ]

        def _is_open(row: CmdbServerEfficiency) -> bool:
            triage = _effective_triage(
                row, triage_by_server.get(str(row.server_id)), now,
            )
            return bool(triage["triage_status"] == EFFICIENCY_TRIAGE_OPEN)

        actionable = [row for row in flagged if _is_open(row)]
        # Split the open backlog by whether it carries a cost basis. A row
        # with no attributed spend cannot be quantified or prioritised (the
        # owner digest already drops those), so the two counts travel as
        # separate facts instead of being silently averaged into one rate.
        priced_savings: list[Decimal] = [
            saving for row in actionable
            if (saving := row.estimated_monthly_savings_usd) is not None
            and saving > 0
        ]
        priced_actionable = [
            row for row in actionable
            if row.estimated_monthly_savings_usd is not None
            and row.estimated_monthly_savings_usd > 0
        ]
        actionable_savings = sum(priced_savings, Decimal("0"))
        efficiency_confidence = _pct(len(classified), len(servers)) if servers else 0
        efficiency_as_of = _latest([row.computed_at for row in efficiency])

        grants = list(
            (
                await db.execute(
                    select(CmdbCloudAccessGrant).where(
                        CmdbCloudAccessGrant.scope_id.in_(scope_ids),
                        CmdbCloudAccessGrant.status == GRANT_STATUS_ACTIVE,
                    )
                )
            ).scalars()
        ) if scope_ids else []
        person_ids = {
            str(value)
            for value in (
                *(row.owner_person_id for row in scopes),
                *(row.owner_person_id for row in servers),
                *(row.owner_person_id for row in domains),
                *(row.person_id for row in grants),
            )
            if value is not None
        }
        active_people = {
            str(value)
            for value in (
                await db.execute(
                    select(IdentityPerson.id).where(
                        IdentityPerson.id.in_(person_ids),
                        IdentityPerson.deleted_at.is_(None),
                        IdentityPerson.employment_status == "active",
                    )
                )
            ).scalars()
        } if person_ids else set()
        # Cloud access governance is scored against Helix's own grant policy,
        # not against "does a future expiry timestamp exist". Three shipped
        # policy positions make a bare expiry test wrong, and together they
        # put 100% out of reach no matter how well the platform is operated:
        #
        #   * Production scopes are not managed by Helix at all — the R&D
        #     security team owns prod cloud access end-to-end, and both expiry
        #     sweeps skip ``PROD_ENVIRONMENTS`` (no reminder, no revoke). Prod
        #     grants therefore drift past ``expires_at`` by design. They leave
        #     the scored population and travel as their own facts so the
        #     read-only prod-overdue count stays visible.
        #   * Perpetual holders (``cloud_admin`` / ``global_admin``) hold
        #     ``expires_at IS NULL`` deliberately — see
        #     ``load_perpetual_person_ids``, the single source of truth every
        #     writer and sweeper already shares. Scoring them as ungoverned
        #     re-introduces exactly the confusion the 2026-07-29 incident
        #     produced (UI says 永久授权, automation says overdue).
        #   * A scope owner's grant on their own scope is exempt from both
        #     sweeps (``_is_owner_grant``): the owner always has access as
        #     part of holding the subscription.
        #
        # What remains in the gap set is the set a human can actually close:
        # a grant whose holder is no longer active, or a managed grant with no
        # valid expiry and no policy reason to be perpetual.
        from app.domains.cmdb.lifecycle_service import load_perpetual_person_ids

        perpetual_person_ids = await load_perpetual_person_ids(db)
        scope_by_id = {str(row.id): row for row in scopes}
        prod_environments = {value.lower() for value in PROD_ENVIRONMENTS}

        def _is_prod_grant(row: CmdbCloudAccessGrant) -> bool:
            scope = scope_by_id.get(str(row.scope_id))
            environment = (scope.environment if scope else None) or "dev"
            return environment.lower() in prod_environments

        def _is_owner_grant(row: CmdbCloudAccessGrant) -> bool:
            scope = scope_by_id.get(str(row.scope_id))
            return (
                scope is not None
                and scope.owner_person_id is not None
                and row.person_id is not None
                and str(scope.owner_person_id) == str(row.person_id)
            )

        prod_grants = [row for row in grants if _is_prod_grant(row)]
        prod_overdue_grants = [
            row for row in prod_grants
            if row.expires_at is not None and _utc(row.expires_at) <= now
        ]
        managed_grants = [row for row in grants if not _is_prod_grant(row)]
        governed_grants = [
            row for row in managed_grants
            if row.person_id is not None and str(row.person_id) in active_people
            and (
                (row.expires_at is not None and _utc(row.expires_at) > now)
                or str(row.person_id) in perpetual_person_ids
                or _is_owner_grant(row)
            )
        ]
        governed_ids = {str(row.id) for row in governed_grants}
        grant_gaps = [
            row for row in managed_grants if str(row.id) not in governed_ids
        ]
        perpetual_grants = [
            row for row in managed_grants
            if row.person_id is not None
            and str(row.person_id) in perpetual_person_ids
        ]
        owner_grants = [row for row in managed_grants if _is_owner_grant(row)]
        access_as_of = _latest([row.updated_at for row in grants])

        # A server's owner is its own ``owner_person_id`` when one is stored,
        # otherwise the owner it INHERITS from its cloud subscription. That
        # inheritance is a shipped, live-resolved product fact — the server
        # list, the efficiency page, the owner digest and savings attribution
        # all render it (``resolve_inherited_server_owners``) — so a metric
        # that reads only the stored column reports a gap the product does not
        # have. Production 2026-08-18: 13 of 6503 active servers carried a
        # stored owner (0.2%), while 6486 resolved to an active subscription
        # owner; almost all of the rest are Kubernetes / autoscaling-group
        # nodes that churn with their nodepool and are never individually
        # assigned. Nothing is written back: the stored column stays the
        # source of truth for the owner-confirmation cron.
        from app.domains.cmdb.service import resolve_inherited_server_owners

        inherited_owners = (
            await resolve_inherited_server_owners(db, servers) if servers else {}
        )
        # NOTE: not a walrus in a comprehension — a comprehension's walrus
        # binds in the ENCLOSING function scope, and ``value`` is already used
        # further down for a datetime, which would collide.
        inherited_person_ids: set[str] = set()
        for inherited_row in inherited_owners.values():
            inherited_person = inherited_row.get("person_id")
            if inherited_person is not None:
                inherited_person_ids.add(str(inherited_person))
        inherited_active_people = {
            str(value)
            for value in (
                await db.execute(
                    select(IdentityPerson.id).where(
                        IdentityPerson.id.in_(inherited_person_ids),
                        IdentityPerson.deleted_at.is_(None),
                        IdentityPerson.employment_status == "active",
                    )
                )
            ).scalars()
        } if inherited_person_ids else set()

        def _server_has_owner(row: CmdbServer) -> bool:
            if (
                row.owner_person_id is not None
                and str(row.owner_person_id) in active_people
            ):
                return True
            person_id = (inherited_owners.get(str(row.id)) or {}).get("person_id")
            return person_id is not None and str(person_id) in inherited_active_people

        scope_owner_gaps = [
            row for row in scopes
            if row.owner_person_id is None or str(row.owner_person_id) not in active_people
        ]
        server_owner_gaps = [row for row in servers if not _server_has_owner(row)]
        directly_owned_servers = [
            row for row in servers
            if row.owner_person_id is not None
            and str(row.owner_person_id) in active_people
        ]
        domain_owner_gaps = [
            row for row in domains
            if row.owner_person_id is None or str(row.owner_person_id) not in active_people
        ]
        owner_total = len(scopes) + len(servers) + len(domains)
        owner_gap_count = (
            len(scope_owner_gaps) + len(server_owner_gaps) + len(domain_owner_gaps)
        )
        owner_covered_count = owner_total - owner_gap_count
        owner_as_of = _latest([
            *[row.updated_at for row in scopes],
            *[row.last_synced_at for row in servers],
            *[row.last_synced_at for row in domains],
        ])

        cutoff = now + timedelta(days=30)
        healthy_domains = [
            row for row in domains
            if row.expiry_date is not None and row.expiry_date > cutoff.date()
        ]
        # Read the status the row would DISPLAY, not the stored column: it is
        # written at probe time and never re-derived, so a host whose probe
        # stopped keeps a stale answer (production 2026-08-18: three rows read
        # ``expiring`` while their cert had expired 52 days earlier). The
        # ``not_after > cutoff`` test already made those three count as
        # unhealthy, so this does not move the score — it stops the metric and
        # the certificate page from disagreeing about the same row.
        healthy_certificates = [
            row for row in certificates
            if effective_cert_status(row, now) == CERT_STATUS_VALID
            and row.not_after is not None
            and _utc(row.not_after) > cutoff
        ]
        network_total = len(domains) + len(certificates)
        network_healthy = len(healthy_domains) + len(healthy_certificates)
        network_gaps = [
            *[row for row in domains if row not in healthy_domains],
            *[row for row in certificates if row not in healthy_certificates],
        ]
        network_as_of = _latest([
            *[row.last_synced_at for row in domains],
            *[
                value for row in certificates
                if (value := row.last_probed_at or row.last_synced_at) is not None
            ],
        ])

        def refs(kind: str, rows: Sequence[object]) -> tuple[EvidenceObject, ...]:
            return tuple(
                EvidenceObject(kind, str(getattr(row, "id")))
                for row in rows[:_EVIDENCE_LIMIT]
            )

        efficiency_refs = tuple(
            EvidenceObject("server", str(row.server_id))
            for row in actionable[:_EVIDENCE_LIMIT]
        )

        owner_evidence = (
            refs("scope", scope_owner_gaps)
            + refs("server", server_owner_gaps)
            + refs("domain", domain_owner_gaps)
        )[:_EVIDENCE_LIMIT]
        network_evidence = (
            refs("domain", [row for row in network_gaps if isinstance(row, CmdbDomain)])
            + refs(
                "certificate",
                [row for row in network_gaps if isinstance(row, CmdbCertificate)],
            )
        )[:_EVIDENCE_LIMIT]

        metric_times = [
            value for value in (
                sync_as_of, efficiency_as_of, access_as_of, owner_as_of, network_as_of,
            ) if value is not None
        ]
        return ProviderResult(
            dimension=self.dimension,
            metrics=(
                MetricObservation(
                    "cloud_network.sync_freshness", missed_cycles,
                    confidence=100 if scopes else 0,
                    # Never-discovered root scopes lead the evidence: they are
                    # the condition a reader has to act on first, and the
                    # truncation limit must not drop them behind rows whose
                    # connector is merely lagging.
                    evidence=(
                        refs("scope", [
                            row for row in stale_scopes
                            if row.last_seen_sync_job_id is None
                            and row.parent_scope_id is None
                        ])
                        + refs("scope", [
                            row for row in stale_scopes
                            if row.last_seen_sync_job_id is not None
                            or row.parent_scope_id is not None
                        ])
                    )[:_EVIDENCE_LIMIT],
                    data_as_of=sync_as_of,
                ),
                MetricObservation(
                    "cloud_network.resource_waste_rate",
                    _pct(len(actionable), len(classified)) if classified else 0,
                    confidence=efficiency_confidence,
                    evidence=efficiency_refs, data_as_of=efficiency_as_of,
                ),
                MetricObservation(
                    "cloud_network.access_governance",
                    (
                        _pct(len(governed_grants), len(managed_grants))
                        if managed_grants else 0
                    ),
                    confidence=100 if managed_grants else 0,
                    evidence=refs("grant", grant_gaps), data_as_of=access_as_of,
                ),
                MetricObservation(
                    "cloud_network.owner_coverage",
                    _pct(owner_covered_count, owner_total) if owner_total else 0,
                    confidence=100 if owner_total else 0,
                    evidence=owner_evidence, data_as_of=owner_as_of,
                ),
                MetricObservation(
                    "cloud_network.domain_certificate_health",
                    _pct(network_healthy, network_total) if network_total else 0,
                    confidence=100 if network_total else 0,
                    evidence=network_evidence, data_as_of=network_as_of,
                ),
            ),
            aggregate_facts={
                "enabled_cloud_integration_count": len(enabled_integration_ids),
                "active_scope_count": len(scopes),
                "stale_scope_count": len(stale_scopes),
                "child_scope_inheriting_freshness_count": len(inherited_scopes),
                "never_discovered_root_scope_count": len(never_discovered_root_scopes),
                "sync_missed_cycles": missed_cycles,
                "active_server_count": len(servers),
                "classified_server_count": len(classified),
                "actionable_efficiency_count": len(actionable),
                "flagged_efficiency_count": len(flagged),
                "triaged_efficiency_count": len(flagged) - len(actionable),
                "priced_actionable_efficiency_count": len(priced_actionable),
                "actionable_monthly_savings_usd": actionable_savings,
                "active_grant_count": len(grants),
                "managed_grant_count": len(managed_grants),
                "governed_grant_count": len(governed_grants),
                "perpetual_grant_count": len(perpetual_grants),
                "owner_grant_count": len(owner_grants),
                "prod_grant_count": len(prod_grants),
                "prod_overdue_grant_count": len(prod_overdue_grants),
                "owner_object_count": owner_total,
                "owner_covered_count": owner_covered_count,
                # The stored owner FK is not the effective-owner contract:
                # servers may inherit the current cloud-account owner without
                # being backfilled. Keep this explicit for the LLM consumer
                # so a low direct-owner count cannot become a false ownerless
                # resource claim.
                "owner_coverage_basis": "effective_owner_direct_or_cloud_scope_inherited",
                "owner_gap_count": owner_gap_count,
                "directly_owned_server_count": len(directly_owned_servers),
                "inherited_owner_server_count": (
                    len(servers) - len(server_owner_gaps) - len(directly_owned_servers)
                ),
                "active_domain_count": len(domains),
                "certificate_count": len(certificates),
                "healthy_network_object_count": network_healthy,
            },
            data_as_of=min(metric_times) if metric_times else None,
        )