"""Software seat and API-key capacity/governance facts."""

from __future__ import annotations

from collections.abc import Callable, Sequence
from datetime import UTC, datetime, timedelta
from decimal import Decimal
from typing import Any
from zoneinfo import ZoneInfo

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

from app.domains.apikey.expiry import effective_assignment_status
from app.domains.apikey.models import ApiKeyAssignment
from app.domains.asset.models import (
    EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS,
    AssetSoftwareCatalog,
    ExternalAccountRegistry,
    SoftwareExternalHolder,
    SoftwareLicensePlan,
    SoftwareSeatAllocation,
    SoftwareSeatSyncStat,
)
from app.identity.person_models import IdentityPerson
from app.intelligence.privacy import EvidenceObject
from app.intelligence.snapshot_service import (
    MetricObservation,
    ProviderResult,
    ProviderUnavailable,
    SnapshotSubject,
)
from app.scheduled_jobs.models import ScheduledJob

_APP_TZ = ZoneInfo("Asia/Shanghai")
_EVIDENCE_LIMIT = 50
_LIVE_KEY_STATUSES = frozenset({"active", "approved"})
# ``detail.source`` values written by the console-key adoption reconcilers
# (``bailian_adopt`` + its manual twin). Helix never issued these
# credentials and never held their plaintext, so it structurally cannot
# stamp a validity window on them — see ``key_expiry_coverage`` below.
_ADOPTED_KEY_SOURCES = frozenset({"adopted", "adopted_manual"})
# Holder categories whose KNOWN-classified emails the shipped seat
# reconciliation nets out of the anomaly count. Mirrors
# ``asset.router._netted_categories`` — the two must not drift, or the
# health metric contradicts the panel that owns the same rows.
_NETTED_HOLDER_CATEGORIES = ("external", "leaver_residual")


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


def _is_adopted_key(row: ApiKeyAssignment) -> bool:
    detail = row.detail if isinstance(row.detail, dict) else {}
    return str(detail.get("source") or "") in _ADOPTED_KEY_SOURCES


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


class SoftwareApiFactProvider:
    dimension = "software_api"

    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 == "software_api":
            return
        raise ValueError("Software/API facts only support platform or software_api 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("software/API datastore unavailable") from exc

    async def _collect(self, db: AsyncSession) -> ProviderResult:
        now = _utc(self._now())
        today = now.astimezone(_APP_TZ).date()
        software = list(
            (
                await db.execute(
                    select(AssetSoftwareCatalog).where(AssetSoftwareCatalog.status == "active")
                )
            ).scalars()
        )
        software_ids = {row.id for row in software}
        plans = list(
            (
                await db.execute(
                    select(SoftwareLicensePlan).where(
                        SoftwareLicensePlan.software_id.in_(software_ids),
                        SoftwareLicensePlan.status == "active",
                    )
                )
            ).scalars()
        ) if software_ids else []
        plan_ids = {row.id for row in plans}
        seats = list(
            (
                await db.execute(
                    select(SoftwareSeatAllocation).where(
                        SoftwareSeatAllocation.plan_id.in_(plan_ids),
                        SoftwareSeatAllocation.status == "active",
                    )
                )
            ).scalars()
        ) if plan_ids else []
        seats_by_plan: dict[Any, int] = {}
        for seat in seats:
            seats_by_plan[seat.plan_id] = seats_by_plan.get(seat.plan_id, 0) + 1

        idle_job = (
            await db.execute(
                select(ScheduledJob).where(ScheduledJob.name == "software_seat.idle_reclaim")
            )
        ).scalar_one_or_none()
        summary = idle_job.last_result_summary if idle_job and idle_job.last_run_at else None
        scanned = self._summary_count(summary, "seats_scanned")
        unobservable = sum(
            self._summary_count(summary, key)
            for key in ("undetermined", "admin_skipped", "leaver_skipped")
        )
        idle_seats = [
            row for row in seats
            if row.idle_warned_at is not None
            or row.idle_manual_review_at is not None
        ]
        observable = max(max(0, scanned - unobservable), len(idle_seats))
        if scanned:
            observable = min(observable, scanned)
        known_idle_count = min(len(idle_seats), observable)
        utilized = max(0, observable - known_idle_count)
        utilization_confidence = _pct(observable, scanned) if scanned else 0

        constrained = [row for row in plans if row.max_seats is not None]
        headroom = [row for row in constrained if seats_by_plan.get(row.id, 0) < int(row.max_seats or 0)]
        full_plans = [row for row in constrained if row not in headroom]
        # A recorded quota the ledger has already blown past is a stale
        # number, not a capacity wall: the seats got there through vendor
        # seat_sync, which imports what the vendor actually entitles us to
        # regardless of ``max_seats``. Counting the plan as "no headroom"
        # stays correct (it is at/over quota, and ``over_quota_policy
        # ='reject'`` really will refuse the next grant), but the operator's
        # next step is to correct the quota, not to buy seats — so name the
        # subset instead of leaving the generic capacity wording to mislead.
        stale_quota_plans = [
            row for row in constrained
            if seats_by_plan.get(row.id, 0) > int(row.max_seats or 0)
        ]

        assignments = list((await db.execute(select(ApiKeyAssignment))).scalars())
        live_keys = [
            row for row in assignments
            if effective_assignment_status(row.status, row.valid_until, today=today)
            in _LIVE_KEY_STATUSES
        ]
        person_ids = {
            value
            for row in software
            for value in (row.owner_person_id, row.it_contact_person_id)
            if value is not None
        } | {row.assigned_to for row in live_keys}
        active_people = set(
            (
                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()
        software_owned = [
            row for row in software
            if row.owner_person_id in active_people and row.it_contact_person_id in active_people
        ]
        key_owned = [row for row in live_keys if row.assigned_to in active_people]
        owner_total = len(software) + len(live_keys)
        owner_covered = len(software_owned) + len(key_owned)
        software_owner_gaps = [row for row in software if row not in software_owned]
        key_owner_gaps = [row for row in live_keys if row not in key_owned]

        stats = list(
            (
                await db.execute(
                    select(SoftwareSeatSyncStat).where(
                        SoftwareSeatSyncStat.software_id.in_(software_ids)
                    )
                )
            ).scalars()
        ) if software_ids else []
        family_stats: dict[tuple[object, str], SoftwareSeatSyncStat] = {}
        for stat in stats:
            family = stat.product_key or f"plan:{stat.plan_id}"
            key = (stat.software_id, family)
            current = family_stats.get(key)
            # A vendor product-family snapshot is repeated on every plan
            # bound to that family.  Mirror the billing/dashboard authority:
            # choose one complete representative row (largest vendor total),
            # never combine matched buckets from duplicate family snapshots.
            if current is None or stat.vendor_license_holders > current.vendor_license_holders:
                family_stats[key] = stat
        vendor_holders = sum(row.vendor_license_holders for row in family_stats.values())
        matched_holders = sum(row.holders_matched for row in family_stats.values())
        # ``holders_matched`` alone is not "reconciled": the shipped
        # reconciliation panel (``asset.router`` seat-reconciliation
        # summary) nets KNOWN-classified external accounts —
        # ``shared_service`` / ``external_guest`` / ``contractor`` in
        # ``external_account_registry`` — out of the anomaly count,
        # because an admin has already filed them as 备案保留. Scoring
        # them as unreconciled put the 95% band out of arithmetic reach
        # (105 of production's 326 unmatched vendor seats are registered
        # reservations) and made the health metric contradict the panel
        # that owns the rows. Same shape as the cloud
        # ``access_governance`` fix — measure against shipped policy, not
        # against a bare column read.
        known_reserved_holders = int(
            (
                await db.execute(
                    select(func.count())
                    .select_from(SoftwareExternalHolder)
                    .join(
                        ExternalAccountRegistry,
                        func.lower(ExternalAccountRegistry.email)
                        == func.lower(SoftwareExternalHolder.email),
                    )
                    .where(
                        SoftwareExternalHolder.plan_id.in_(
                            [row.plan_id for row in family_stats.values()]
                        ),
                        SoftwareExternalHolder.resolved.is_(False),
                        SoftwareExternalHolder.category.in_(
                            _NETTED_HOLDER_CATEGORIES
                        ),
                        ExternalAccountRegistry.classification.in_(
                            EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS
                        ),
                    )
                )
            ).scalar_one()
        ) if family_stats else 0
        # Bounded by the vendor total: the netting query counts holder
        # rows on the representative plans while the buckets come from
        # the sync stat, so a timing slip between the two must never
        # produce a >100% coverage value.
        attributed_holders = min(
            matched_holders + known_reserved_holders, vendor_holders,
        )
        known_reserved_emails = select(
            func.lower(ExternalAccountRegistry.email)
        ).where(
            ExternalAccountRegistry.classification.in_(
                EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS
            )
        )
        anomalies = list(
            (
                await db.execute(
                    select(SoftwareExternalHolder).where(
                        SoftwareExternalHolder.software_id.in_(software_ids),
                        SoftwareExternalHolder.resolved.is_(False),
                        # Evidence must match the scored population, or the
                        # "next step" card points the operator at rows the
                        # metric already counted as handled.
                        or_(
                            SoftwareExternalHolder.category.notin_(
                                _NETTED_HOLDER_CATEGORIES
                            ),
                            SoftwareExternalHolder.email.is_(None),
                            func.lower(SoftwareExternalHolder.email).notin_(
                                known_reserved_emails
                            ),
                        ),
                    )
                )
            ).scalars()
        ) if software_ids else []

        # ``key_expiry_coverage`` measures credentials Helix can actually
        # govern the validity of. Two populations were wrongly in the
        # denominator and together made the 95% band unreachable (44 of
        # 61 "live" rows could never carry a Helix expiry):
        #
        # * ``status='approved'`` — approved but NOT yet issued. The
        #   admin fills ``key_value`` + ``valid_from`` / ``valid_until``
        #   at issuance, so no credential exists yet and there is nothing
        #   to expire. (They are surfaced as their own fact instead —
        #   production carries 15 sitting unissued since 2026-04-22.)
        # * adopted console keys (``detail.source`` in
        #   ``_ADOPTED_KEY_SOURCES``) — minted directly in the vendor
        #   console and pulled in by the adoption reconciler, which
        #   deliberately leaves ``key_value`` NULL because Helix never
        #   held the plaintext. Helix cannot stamp or enforce a validity
        #   window on a secret it does not hold.
        #
        # The numerator is "carries a bounded validity", not "expires
        # more than 30 days out". Applications are capped at one year, so
        # under the old rule every correctly-governed key fell out of
        # coverage for its final month with no action that could keep it
        # in — the metric penalised the alert cron doing its job.
        issued_keys = [
            row for row in live_keys
            if row.status == "active" and not _is_adopted_key(row)
        ]
        adopted_keys = [row for row in live_keys if _is_adopted_key(row)]
        pending_issuance_keys = [
            row for row in live_keys
            if row.status != "active" and not _is_adopted_key(row)
        ]
        expiry_ok = [row for row in issued_keys if row.valid_until is not None]
        expiry_gaps = [row for row in issued_keys if row.valid_until is None]
        expiring_soon = [
            row for row in expiry_ok
            if row.valid_until is not None
            and row.valid_until <= today + timedelta(days=30)
        ]
        overdue_rotation = [
            row for row in live_keys
            if row.rotation_retire_at is not None and _utc(row.rotation_retire_at) <= now
        ]
        software_as_of_values = [
            _utc(updated)
            for row in (*software, *plans, *seats)
            if (updated := getattr(row, "updated_at", None)) is not None
        ]
        key_as_of_values = [_utc(row.updated_at) for row in live_keys]
        stat_as_of_values = [_utc(row.synced_at) for row in stats]
        software_as_of = max(software_as_of_values, default=None)
        key_as_of = max(key_as_of_values, default=None)
        stat_as_of = max(stat_as_of_values, default=None)
        owner_as_of = max(
            [value for value in (software_as_of, key_as_of) if value is not None],
            default=None,
        )
        as_of_values = [
            value for value in (software_as_of, key_as_of, stat_as_of) if value is not None
        ]
        if idle_job and idle_job.last_run_at:
            as_of_values.append(_utc(idle_job.last_run_at))

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

        owner_evidence = (
            refs("software", software_owner_gaps)
            + refs("key", key_owner_gaps)
        )[:_EVIDENCE_LIMIT]

        return ProviderResult(
            dimension=self.dimension,
            metrics=(
                MetricObservation(
                    "software_api.seat_utilization",
                    _pct(utilized, observable) if observable else 0,
                    confidence=utilization_confidence,
                    evidence=refs("seat", idle_seats),
                    data_as_of=_utc(idle_job.last_run_at) if idle_job and idle_job.last_run_at else None,
                ),
                MetricObservation(
                    "software_api.capacity_headroom",
                    _pct(len(headroom), len(constrained)) if constrained else 0,
                    confidence=100 if constrained else 0,
                    evidence=refs("plan", full_plans), data_as_of=software_as_of,
                ),
                MetricObservation(
                    "software_api.owner_coverage",
                    _pct(owner_covered, owner_total) if owner_total else 0,
                    confidence=100 if owner_total else 0,
                    evidence=owner_evidence, data_as_of=owner_as_of,
                ),
                MetricObservation(
                    "software_api.reconciliation_coverage",
                    _pct(attributed_holders, vendor_holders) if vendor_holders else 0,
                    confidence=100 if vendor_holders else 0,
                    evidence=refs("holder", anomalies),
                    data_as_of=stat_as_of,
                ),
                MetricObservation(
                    "software_api.key_expiry_coverage",
                    _pct(len(expiry_ok), len(issued_keys)) if issued_keys else 0,
                    confidence=100 if issued_keys else 0,
                    evidence=refs("key", expiry_gaps), data_as_of=key_as_of,
                ),
                MetricObservation(
                    "software_api.rotation_failure_rate",
                    _pct(len(overdue_rotation), len(live_keys)) if live_keys else 0,
                    confidence=100 if live_keys else 0,
                    evidence=refs("key", overdue_rotation), data_as_of=key_as_of,
                ),
            ),
            aggregate_facts={
                "software_count": len(software),
                "active_plan_count": len(plans),
                "active_seat_count": len(seats),
                "observable_seat_count": observable,
                "known_idle_seat_count": known_idle_count,
                "constrained_plan_count": len(constrained),
                "plan_with_headroom_count": len(headroom),
                "over_recorded_quota_plan_count": len(stale_quota_plans),
                "owner_object_count": owner_total,
                "owner_covered_count": owner_covered,
                "vendor_holder_count": vendor_holders,
                "matched_holder_count": matched_holders,
                "known_reserved_holder_count": known_reserved_holders,
                "attributed_holder_count": attributed_holders,
                "open_reconciliation_anomaly_count": len(anomalies),
                "live_key_count": len(live_keys),
                "issued_key_count": len(issued_keys),
                "adopted_key_count": len(adopted_keys),
                "pending_issuance_key_count": len(pending_issuance_keys),
                "expiry_governed_key_count": len(expiry_ok),
                "expiring_within_30d_key_count": len(expiring_soon),
                "overdue_rotation_count": len(overdue_rotation),
            },
            data_as_of=min(as_of_values) if as_of_values else None,
        )

    @staticmethod
    def _summary_count(summary: dict[str, Any] | None, key: str) -> int:
        value = summary.get(key) if isinstance(summary, dict) else 0
        return max(0, int(value)) if isinstance(value, int | float) else 0