"""Joiner, mover, leaver, and ownership-continuity deterministic facts."""

from __future__ import annotations

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

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

from app.domains.offboarding.models import (
    PROC_CANCELLED as OFFBOARDING_CANCELLED,
)
from app.domains.offboarding.models import (
    RES_CDN_DOMAIN,
    RES_CLOUD_IAM_USER,
    RES_CLOUD_SCOPE_OWNER,
    RES_CLOUD_SERVER,
    RES_DNS_RECORD,
    RES_DOMAIN,
    RES_DOMAIN_REGISTRATION,
    RES_EMAIL_GROUP,
    RES_GOOGLE_ORPHAN_OWNER,
    RES_IDENTITY_CONSUMER_OWNER,
    RES_LOAD_BALANCER,
    RES_SOFTWARE_IT_CONTACT,
    RES_SOFTWARE_OWNER,
    RES_SSO_GRANT,
    RES_SYSTEM_OWNER,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.domains.offboarding.models import (
    RESO_PENDING as OFFBOARDING_PENDING,
)
from app.domains.onboarding.analytics_service import OnboardingAnalyticsService
from app.domains.onboarding.models import PROC_CANCELLED as ONBOARDING_CANCELLED
from app.domains.onboarding.models import OnboardingProcess
from app.domains.provisioning.models import STATUS_FAILED, ProvisioningTask
from app.intelligence.demand_forecast_facts import collect_demand_forecast_facts
from app.intelligence.leaver_risk_facts import collect_leaver_risk_facts
from app.intelligence.privacy import EvidenceObject, merge_anonymous_fact_packages
from app.intelligence.snapshot_service import (
    MetricObservation,
    ProviderResult,
    ProviderUnavailable,
    SnapshotSubject,
)
from app.intelligence.transfer_forecast_facts import collect_transfer_forecast_facts

_EPOCH = datetime(1970, 1, 1, tzinfo=UTC)
_EVIDENCE_LIMIT = 50
_APP_TZ = ZoneInfo("Asia/Shanghai")
# Post-departure IT grace window before the force-resolve sweep clears a
# leaver's remaining blockers. Kept local rather than imported: the 24h lives
# as an inline comparison in hr_lifecycle.advance_service, not as an exported
# constant, and a fact provider must not reach into another domain's internals.
_LEAVER_FORCE_RESOLVE_GRACE = timedelta(hours=24)
# The onboarding SLA escalates to ``breach`` at 72h past ``hire_date``
# with pending BLOCKER gaps (``OnboardingService.evaluate_sla``), counted
# in whole local days — so the attainment deadline is hire_date + 3 days.
_ONBOARDING_SLA_GRACE = timedelta(days=3)
_OWNER_TYPES = (
    RES_DOMAIN, RES_DNS_RECORD, RES_CLOUD_SERVER, RES_LOAD_BALANCER, RES_CDN_DOMAIN,
    RES_SOFTWARE_OWNER, RES_SOFTWARE_IT_CONTACT, RES_EMAIL_GROUP, RES_CLOUD_SCOPE_OWNER,
    RES_SSO_GRANT, RES_DOMAIN_REGISTRATION, RES_GOOGLE_ORPHAN_OWNER,
    RES_CLOUD_IAM_USER, RES_IDENTITY_CONSUMER_OWNER, RES_SYSTEM_OWNER,
)


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


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


def _six_month_window(today: date) -> date:
    year, month = today.year, today.month - 5
    while month < 1:
        year -= 1
        month += 12
    return date(year, month, 1)


def _day1_ready(process: OnboardingProcess) -> bool:
    if process.hard_ready_at is None:
        return False
    ready_at = _utc(process.hard_ready_at).astimezone(_APP_TZ).date()
    return bool(ready_at <= process.hire_date)


def _sla_deadline(process: OnboardingProcess) -> date:
    """Last day the shipped onboarding SLA allows for hard readiness.

    ``OnboardingService.evaluate_sla`` escalates to ``breach`` once a
    process is 72h past ``hire_date`` with pending BLOCKER-severity gaps,
    counting whole local days from ``hire_date``. The attainment metric
    has to score against that same clock, not a second one.
    """
    return process.hire_date + _ONBOARDING_SLA_GRACE


def _sla_window_closed(process: OnboardingProcess, today: date) -> bool:
    """Has this hire's SLA verdict been decided yet?

    A hire still inside the grace window with work in flight has neither
    met nor missed the SLA — scoring them as a miss is the "常态在途状态
    当失败态" trap that health-v2 removed from the leaver metric. They
    re-enter the population the moment the window closes.
    """
    return process.hard_ready_at is not None or today > _sla_deadline(process)


def _sla_attained(process: OnboardingProcess) -> bool:
    if process.hard_ready_at is None:
        return False
    ready_at = _utc(process.hard_ready_at).astimezone(_APP_TZ).date()
    return bool(ready_at <= _sla_deadline(process))


class LifecycleFactProvider:
    dimension = "lifecycle"

    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 == "lifecycle":
            return
        raise ValueError("Lifecycle facts only support platform or lifecycle domain 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("lifecycle datastore unavailable") from exc

    @staticmethod
    async def _collect_failed_leavers(
        db: AsyncSession,
        offboarding: Sequence[OffboardingProcess],
        *,
        now: datetime,
    ) -> set[object]:
        """Leaver processes whose offboarding is genuinely NOT clean.

        health-v2 口径 (2026-08-18). The v1 predicate was "any blocker with
        ``error_count > 0``, or process status ``blocked``", and both halves
        were wrong:

        * ``error_count`` is documented on the column itself as "total
          failures **since blocker creation**" and is never reset when the
          blocker resolves. A blocker that failed 58 times and then landed
          ``revoked_auto`` is a retried success, not a failure — but v1 kept
          counting it forever. The platform's own single source of truth for
          "is this blocker still a problem", ``compute_triage_bucket``, gates
          on ``resolution != pending`` first for exactly this reason; v1 read
          ``error_count`` without that gate.
        * ``blocked`` is not a failure state on a leaver. ``_recompute_status``
          flips a process to ``blocked`` on **any** pending row, so it is the
          ordinary in-flight status of every offboarding between kickoff and
          the T+0+24h force-resolve — v1 scored every running leaver as failed.

        Net effect in production: 12 of 78 window leavers were reported failed
        while zero had an unresolved failure, and the metric could only
        degrade as retry history accumulated in the rolling window.

        v2 counts a process as failed when it still carries a blocker that is
        unresolved AND either has actually failed execution (mirrors the two
        evidence-of-failure arms of ``compute_triage_bucket``'s ``needs_it``
        rule — deliberately excluding its "needs a human but has not failed"
        arms), or has sat past the T+0+24h force-resolve deadline and is
        therefore stuck regardless of severity.
        """
        process_ids = [row.id for row in offboarding]
        if not process_ids:
            return set()
        # Deadline mirrors the 24h post-departure IT grace window in
        # hr_lifecycle.advance_service (``hours_since_discovery < 24``); a
        # blocker still pending past it should already have been
        # force-resolved, so it is stuck rather than in flight.
        leaver_dt = {
            row.id: datetime.combine(
                row.leaver_date, datetime.min.time(), tzinfo=_APP_TZ,
            )
            for row in offboarding
        }
        rows = (
            await db.execute(
                select(
                    OffboardingBlocker.process_id,
                    OffboardingBlocker.error_count,
                    ProvisioningTask.status,
                )
                .outerjoin(
                    ProvisioningTask,
                    ProvisioningTask.id == OffboardingBlocker.provisioning_task_id,
                )
                .where(
                    OffboardingBlocker.process_id.in_(process_ids),
                    OffboardingBlocker.resolution == OFFBOARDING_PENDING,
                )
            )
        ).all()
        failed: set[object] = set()
        for process_id, error_count, task_status in rows:
            execution_failed = (error_count or 0) >= 1 or task_status == STATUS_FAILED
            deadline = leaver_dt.get(process_id)
            past_deadline = (
                deadline is not None
                and now >= _utc(deadline) + _LEAVER_FORCE_RESOLVE_GRACE
            )
            if execution_failed or past_deadline:
                failed.add(process_id)
        return failed

    async def _collect(self, db: AsyncSession) -> ProviderResult:
        now = _utc(self._now())
        today = now.astimezone(_APP_TZ).date()
        window_date = _six_month_window(today)
        readiness = await OnboardingAnalyticsService(db).day1_readiness(
            months=6, as_of_date=today,
        )
        demand_package = await collect_demand_forecast_facts(db, now=now)
        transfer_package = await collect_transfer_forecast_facts(db, now=now)
        leaver_package = await collect_leaver_risk_facts(db, now=now)
        forecast_facts, forecast_reference_map = merge_anonymous_fact_packages({
            "demand_forecast": (
                demand_package.aggregate_facts(), demand_package.reference_map,
            ),
            "transfer_forecast": (
                transfer_package.aggregate_facts(), transfer_package.reference_map,
            ),
            "leaver_risk": (
                leaver_package.aggregate_facts(), leaver_package.reference_map,
            ),
        })
        onboarding = list(
            (
                await db.execute(
                    select(OnboardingProcess).where(
                        OnboardingProcess.hire_date >= window_date,
                        OnboardingProcess.hire_date <= today,
                        OnboardingProcess.status != ONBOARDING_CANCELLED,
                        OnboardingProcess.backfilled.is_(False),
                    )
                )
            ).scalars()
        )
        offboarding = list(
            (
                await db.execute(
                    select(OffboardingProcess).where(
                        OffboardingProcess.leaver_date >= window_date,
                        OffboardingProcess.leaver_date <= today,
                        OffboardingProcess.status != OFFBOARDING_CANCELLED,
                    )
                )
            ).scalars()
        )
        failed_process_ids = await self._collect_failed_leavers(
            db, offboarding, now=now,
        )
        owner_blockers = list(
            (
                await db.execute(
                    select(OffboardingBlocker)
                    .join(OffboardingProcess, OffboardingProcess.id == OffboardingBlocker.process_id)
                    .where(
                        OffboardingProcess.leaver_date >= window_date,
                        OffboardingProcess.leaver_date <= today,
                        OffboardingProcess.status != OFFBOARDING_CANCELLED,
                        OffboardingBlocker.resource_type.in_(_OWNER_TYPES),
                    )
                )
            ).scalars()
        )

        day1_total = int(readiness["window_total"])
        day1_ready = int(readiness["window_day1_ready"])
        # health-v3 口径 (2026-08-18). v2 read the CURRENT
        # ``sla_breach_level``, which is not an attainment record:
        # ``evaluate_sla`` returns ``none`` for every terminal process and
        # resets to ``none`` the moment the last blocker clears. So a hire
        # whose hard gaps closed 22 days late scored identically to one
        # that was ready on day 1, and the metric was really "share of
        # processes not red right now" — 25 of production's 25 completed
        # hires passed while only 17 actually made the 72h SLA.
        #
        # v3 scores the shipped SLA against history: did hard readiness
        # land within ``hire_date + 72h``, over the processes whose window
        # has already closed.
        sla_scored = [row for row in onboarding if _sla_window_closed(row, today)]
        onboarding_sla_ok = sum(_sla_attained(row) for row in sla_scored)
        owner_resolved = sum(row.resolution != OFFBOARDING_PENDING for row in owner_blockers)
        onboarding_as_of = max((_utc(row.updated_at) for row in onboarding), default=_EPOCH)
        offboarding_as_of = max((_utc(row.updated_at) for row in offboarding), default=_EPOCH)
        owner_as_of = max((_utc(row.updated_at) for row in owner_blockers), default=offboarding_as_of)
        demand_as_of = demand_package.data_as_of
        transfer_as_of = transfer_package.data_as_of
        leaver_as_of = leaver_package.data_as_of

        def refs(kind: str, ids: Sequence[object]) -> tuple[EvidenceObject, ...]:
            return tuple(EvidenceObject(kind, str(value)) for value in ids[:_EVIDENCE_LIMIT])

        day1_gap_ids = [row.id for row in onboarding if not _day1_ready(row)]
        onboarding_sla_gap = [row.id for row in sla_scored if not _sla_attained(row)]
        owner_gap = [row.id for row in owner_blockers if row.resolution == OFFBOARDING_PENDING]
        return ProviderResult(
            dimension=self.dimension,
            metrics=(
                MetricObservation(
                    "lifecycle.day1_ready", _pct(day1_ready, day1_total) if day1_total else 0,
                    confidence=100 if day1_total else 0,
                    evidence=refs("onboarding", day1_gap_ids), data_as_of=onboarding_as_of,
                ),
                MetricObservation(
                    "lifecycle.onboarding_sla",
                    _pct(onboarding_sla_ok, len(sla_scored)) if sla_scored else 0,
                    confidence=100 if sla_scored else 0,
                    evidence=refs("onboarding", onboarding_sla_gap), data_as_of=onboarding_as_of,
                ),
                MetricObservation(
                    "lifecycle.leaver_failure_rate",
                    _pct(len(failed_process_ids), len(offboarding)) if offboarding else 0,
                    confidence=100 if offboarding else 0,
                    evidence=refs("offboarding", sorted(failed_process_ids, key=str)),
                    data_as_of=offboarding_as_of,
                ),
                MetricObservation(
                    "lifecycle.owner_continuity",
                    _pct(owner_resolved, len(owner_blockers)) if owner_blockers else 0,
                    confidence=100 if owner_blockers else 0,
                    evidence=refs("blocker", owner_gap), data_as_of=owner_as_of,
                ),
            ),
            aggregate_facts={
                "day1_total_count": day1_total,
                "day1_ready_count": day1_ready,
                "open_past_due_hard_gap_count": int(readiness["open_past_due_with_hard_gaps"]),
                "onboarding_sla_total_count": len(sla_scored),
                "onboarding_sla_ok_count": onboarding_sla_ok,
                "onboarding_sla_in_window_count": len(onboarding) - len(sla_scored),
                "leaver_total_count": len(offboarding),
                "leaver_failed_count": len(failed_process_ids),
                "ownership_total_count": len(owner_blockers),
                "ownership_resolved_count": owner_resolved,
                **forecast_facts,
            },
            data_as_of=(
                min(
                    as_of
                    for present, as_of in (
                        (bool(onboarding), onboarding_as_of),
                        (bool(offboarding), offboarding_as_of),
                        (bool(owner_blockers), owner_as_of),
                        (demand_as_of is not None, demand_as_of or _EPOCH),
                        (transfer_as_of is not None, transfer_as_of or _EPOCH),
                        (leaver_as_of is not None, leaver_as_of or _EPOCH),
                    )
                    if present
                )
                if (
                    onboarding or offboarding or owner_blockers
                    or demand_as_of or transfer_as_of
                    or leaver_as_of
                )
                else None
            ),
            reference_map=forecast_reference_map,
        )