"""Admin Health Console — single aggregate endpoint feeding /admin/health.

Four panels, each a sub-object on the response:

  * ``cron`` — scheduler liveness + window run aggregates (success rate,
    unacked failures, daily trend, top failures). Built from
    ``scheduler_health_summary`` + ``analytics_summary``.
  * ``offboarding`` — leaver-blocked KPIs + T+x bucket distribution
    (so admins see the in-flight queue shape at a glance) + by
    blocker resource_type.
  * ``notifications`` — outbox by_status, 24h sent/failed, daily
    trend.
  * ``data_health`` — under-licensed software plans, seat-drift plan
    count, leaver-residual holders, open compliance incidents,
    pending dual-sign overrides.

Sub-objects are computed independently; if one slice raises the
endpoint still returns the others (the failing slice surfaces
``status='error'`` so the UI degrades gracefully instead of 500-ing
the whole page).
"""
from __future__ import annotations

import logging
from collections import defaultdict
from datetime import UTC, date, datetime, timedelta
from typing import Any

from sqlalchemy import func, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.compliance.models import INCIDENT_OPEN, ComplianceIncident
from app.core.exceptions import format_exception_for_log
from app.core.timeseries import day_axis
from app.domains.asset.models import (
    EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS,
    ExternalAccountRegistry,
    SoftwareExternalHolder,
    SoftwareLicensePlan,
    SoftwareSeatSyncStat,
)
from app.domains.leaver_blocked.dashboard_service import (
    LeaverBlockedDashboardService,
)
from app.domains.leaver_blocked.models import (
    DUAL_SIGN_PENDING_SECONDARY,
    LeaverDualSignOverride,
)
from app.domains.offboarding.models import (
    PROC_CANCELLED,
    PROC_COMPLETED,
    RESO_PENDING,
    OffboardingBlocker,
    OffboardingProcess,
)
from app.notifications.models import (
    STATUS_DEAD,
    STATUS_DELIVERED,
    STATUS_FAILED,
    STATUS_PARTIAL,
    STATUS_PENDING,
    NotificationOutbox,
)
from app.scheduled_jobs.service import (
    analytics_summary as _scheduled_analytics,
)
from app.scheduled_jobs.service import (
    scheduler_health_summary as _scheduler_health,
)

logger = logging.getLogger(__name__)


# ── T+x bucket boundaries (days from today, where today = leaver_date) ──
#
# A process whose ``leaver_date`` is in the future is "pre-leaver"
# (HRIS gave us advance notice — happy path). On-or-after is "in
# flight". Buckets are intentionally coarse — the cockpit page
# (PR-B) renders the fine-grained per-process timeline; here we just
# want the queue shape.
_T_MINUS_GT_7 = "t_minus_gt_7"          # > 7 days before leaver_date
_T_MINUS_3_TO_7 = "t_minus_3_to_7"      # 3–7 days before
_T_MINUS_0_TO_3 = "t_minus_0_to_3"      # 0–3 days before (incl. today)
_T_PLUS_0_TO_1 = "t_plus_0_to_1"        # 0–1 day past leaver_date
_T_PLUS_1_TO_3 = "t_plus_1_to_3"        # 1–3 days past
_T_PLUS_GT_3 = "t_plus_gt_3"            # > 3 days past (SLA territory)

_BUCKET_ORDER = [
    _T_MINUS_GT_7,
    _T_MINUS_3_TO_7,
    _T_MINUS_0_TO_3,
    _T_PLUS_0_TO_1,
    _T_PLUS_1_TO_3,
    _T_PLUS_GT_3,
]


def _bucket_for(leaver_date: date, today: date) -> str:
    delta = (leaver_date - today).days
    if delta > 7:
        return _T_MINUS_GT_7
    if delta > 3:
        return _T_MINUS_3_TO_7
    if delta >= 0:
        return _T_MINUS_0_TO_3
    # leaver_date already passed
    days_past = -delta
    if days_past <= 1:
        return _T_PLUS_0_TO_1
    if days_past <= 3:
        return _T_PLUS_1_TO_3
    return _T_PLUS_GT_3


# ── Panel: cron ──────────────────────────────────────────────────


async def _cron_panel(db: AsyncSession, window_days: int) -> dict[str, Any]:
    """Liveness + window run KPIs. Trend is 7-day daily ok/failed."""
    scheduler = await _scheduler_health(db)
    runs = await _scheduled_analytics(db, window_days=window_days)
    totals = runs.get("totals", {})

    # Pick the KPIs we want surfaced one screen up. The detail page
    # (ScheduledJobsAnalyticsTab) gets the full payload.
    return {
        "scheduler": scheduler,
        "window_days": window_days,
        "total_jobs": totals.get("total_jobs", 0),
        "enabled_jobs": totals.get("enabled_jobs", 0),
        "runs_in_window": totals.get("runs_in_window", 0),
        "ok_in_window": totals.get("ok_in_window", 0),
        "failed_in_window": totals.get("failed_in_window", 0),
        "success_rate": totals.get("success_rate", 100.0),
        "unacked_failures": totals.get("unacked_failures", 0),
        "daily_series": runs.get("daily_series", []),
        "top_failures": runs.get("top_failing_jobs", [])[:5],
        "slowest": runs.get("slowest_jobs", [])[:5],
    }


# ── Panel: offboarding ───────────────────────────────────────────


async def _offboarding_panel(db: AsyncSession) -> dict[str, Any]:
    summary = await LeaverBlockedDashboardService(db).get_summary()

    # Bucket in-flight processes by their T+x position. In-flight =
    # NOT completed and NOT cancelled — same definition as the
    # leaver-blocked dashboard so the totals line up.
    in_flight = (
        await db.execute(
            select(OffboardingProcess).where(
                OffboardingProcess.status.notin_([PROC_COMPLETED, PROC_CANCELLED]),
            )
        )
    ).scalars().all()
    today = datetime.now(UTC).date()
    bucket_counts: dict[str, int] = {b: 0 for b in _BUCKET_ORDER}
    for p in in_flight:
        bucket_counts[_bucket_for(p.leaver_date, today)] += 1

    # Group pending blockers by resource_type so admin sees which class
    # of resource is the long pole. Filtered to processes still
    # in-flight to avoid counting historical resolved blockers.
    proc_ids = [p.id for p in in_flight]
    by_resource_type: dict[str, int] = {}
    if proc_ids:
        rows = (
            await db.execute(
                select(
                    OffboardingBlocker.resource_type,
                    func.count(OffboardingBlocker.id),
                )
                .where(
                    OffboardingBlocker.process_id.in_(proc_ids),
                    OffboardingBlocker.resolution == RESO_PENDING,
                )
                .group_by(OffboardingBlocker.resource_type)
            )
        ).all()
        for rtype, n in rows:
            by_resource_type[rtype] = int(n)

    return {
        "total_in_flight": len(in_flight),
        "total_blocked": summary.total_blocked,
        "sla_warn": summary.sla_warn,
        "sla_urgent": summary.sla_urgent,
        "sla_breach": summary.sla_breach,
        "pending_overrides": summary.pending_overrides,
        "approved_overrides_ready_to_execute": (
            summary.approved_overrides_ready_to_execute
        ),
        "today_completed": summary.today_completed,
        "by_bucket": [
            {"bucket": b, "count": bucket_counts[b]} for b in _BUCKET_ORDER
        ],
        "by_resource_type": by_resource_type,
    }


# ── Panel: notifications ─────────────────────────────────────────


_UNHEALTHY_STATUSES = (
    STATUS_PENDING,
    STATUS_FAILED,
    STATUS_PARTIAL,
    STATUS_DEAD,
)


async def _notifications_panel(db: AsyncSession) -> dict[str, Any]:
    # Lifetime status distribution (the badge / pie source). Same query
    # the existing /notifications/outbox/summary uses — repeated here so
    # the health page is a single round-trip; intentional small dupe.
    status_rows = (
        await db.execute(
            select(NotificationOutbox.status, func.count(NotificationOutbox.id))
            .group_by(NotificationOutbox.status)
        )
    ).all()
    by_status = {status: int(count) for status, count in status_rows}
    unhealthy = sum(by_status.get(s, 0) for s in _UNHEALTHY_STATUSES)

    # 24h window + 7-day daily trend keyed on ``created_at``.
    now = datetime.now(UTC)
    h24_start = now - timedelta(hours=24)
    h24_rows = (
        await db.execute(
            select(NotificationOutbox.status, func.count(NotificationOutbox.id))
            .where(NotificationOutbox.created_at >= h24_start)
            .group_by(NotificationOutbox.status)
        )
    ).all()
    h24_by_status = {status: int(count) for status, count in h24_rows}
    h24_total = sum(h24_by_status.values())
    # Round-62 P2: classify PARTIAL alongside FAILED/DEAD for the 24h
    # rollup so it matches the lifetime ``unhealthy`` badge taxonomy.
    h24_failed = sum(
        h24_by_status.get(s, 0)
        for s in (STATUS_FAILED, STATUS_DEAD, STATUS_PARTIAL)
    )
    h24_delivered = h24_by_status.get(STATUS_DELIVERED, 0)

    # 7-day daily ok/failed trend, INCLUDING today. Delegates day-axis
    # construction to ``app.core.timeseries.day_axis`` — see that module
    # for the lock-step contract between SQL filter and bucket loop.
    # (The previous inline implementation here re-introduced the
    # off-by-one bug class twice; the shared helper prevents recurrence
    # by keeping window_start anchored to days[0].)
    #
    # Round-62 P2 fix (cross-panel consistency): STATUS_PARTIAL is in
    # ``_UNHEALTHY_STATUSES`` for the lifetime badge but was missing
    # from the 24h-failed and daily-series classifiers — a "5 unhealthy
    # / 0 failures on chart" mismatch. Add STATUS_PARTIAL to the failed
    # set so the three rollups use the same status taxonomy.
    axis = day_axis(window_days=7, now=now)
    daily_rows = (
        await db.execute(
            select(
                func.date(NotificationOutbox.created_at).label("day"),
                NotificationOutbox.status,
                func.count(),
            )
            .where(NotificationOutbox.created_at >= axis.window_start)
            .group_by("day", NotificationOutbox.status)
            .order_by("day")
        )
    ).all()
    daily_map: dict[str, dict[str, int]] = defaultdict(lambda: {"ok": 0, "failed": 0})
    failed_for_chart = (STATUS_FAILED, STATUS_DEAD, STATUS_PARTIAL)
    for day, status, n in daily_rows:
        key = day.isoformat() if hasattr(day, "isoformat") else str(day)
        if status == STATUS_DELIVERED:
            daily_map[key]["ok"] += int(n)
        elif status in failed_for_chart:
            daily_map[key]["failed"] += int(n)
    daily_series = []
    for d in axis.days:
        key = d.isoformat()
        slot = daily_map.get(key, {"ok": 0, "failed": 0})
        daily_series.append({"date": key, "ok": slot["ok"], "failed": slot["failed"]})

    return {
        "by_status": by_status,
        "unhealthy": unhealthy,
        "last_24h_total": h24_total,
        "last_24h_delivered": h24_delivered,
        "last_24h_failed": h24_failed,
        "daily_series": daily_series,
    }


# ── Panel: data_health ───────────────────────────────────────────


async def _data_health_panel(db: AsyncSession) -> dict[str, Any]:
    """Cross-domain data-integrity speed-read.

    The deep forensic view (per-row drill / one-click recon trigger)
    lives in PR-E's /admin/reconciliation. Here we just count.
    """
    # Under-licensed plans: vendor_license_holders > plan.max_seats (the
    # exact signal PR-#610 added; surfaces Confluence-style over-use).
    under_licensed_q = (
        select(func.count(SoftwareSeatSyncStat.id))
        .join(
            SoftwareLicensePlan,
            SoftwareLicensePlan.id == SoftwareSeatSyncStat.plan_id,
        )
        .where(
            SoftwareLicensePlan.max_seats.isnot(None),
            SoftwareSeatSyncStat.vendor_license_holders > SoftwareLicensePlan.max_seats,
        )
    )
    under_licensed_plans = int(
        (await db.execute(under_licensed_q)).scalar() or 0
    )

    # Seat-drift plans: vendor != matched+external+no_email+leaver_residual
    # +vendor_inactive+off_paid_site (the 6-bucket identity). Drift here
    # is a sync identity violation — the cron already raises on it, but
    # we count plans with the drift just in case (typically 0).
    all_stats = (
        await db.execute(
            select(
                SoftwareSeatSyncStat.id,
                SoftwareSeatSyncStat.vendor_license_holders,
                SoftwareSeatSyncStat.holders_matched,
                SoftwareSeatSyncStat.holders_external,
                SoftwareSeatSyncStat.holders_no_email,
                SoftwareSeatSyncStat.holders_leaver_residual,
                SoftwareSeatSyncStat.holders_vendor_inactive,
                SoftwareSeatSyncStat.holders_off_paid_site,
            )
        )
    ).all()
    seat_drift_plans = 0
    leaver_residual_holders = 0
    for row in all_stats:
        bucket_sum = (
            (row.holders_matched or 0)
            + (row.holders_external or 0)
            + (row.holders_no_email or 0)
            + (row.holders_leaver_residual or 0)
            + (row.holders_vendor_inactive or 0)
            + (row.holders_off_paid_site or 0)
        )
        if (row.vendor_license_holders or 0) != bucket_sum:
            seat_drift_plans += 1
        leaver_residual_holders += row.holders_leaver_residual or 0

    # Net out documented keeps, same rule the seat-reconciliation panel
    # applies (2026-07-29): a leaver-residual holder whose email carries
    # a KNOWN registry classification is an approved retention (that row
    # is what offboarding's own retention path writes, and what #1983's
    # leaver-assurance reads to stop paging IT). Counting them here would
    # light the health tile red for keeps the platform already signed off
    # on — and would disagree with the drawer showing the same holders as
    # 「另已知保留」.
    leaver_residual_known = int((
        await db.execute(
            select(func.count())
            .select_from(SoftwareExternalHolder)
            .join(
                ExternalAccountRegistry,
                func.lower(ExternalAccountRegistry.email)
                == func.lower(SoftwareExternalHolder.email),
            )
            .where(
                SoftwareExternalHolder.category == "leaver_residual",
                SoftwareExternalHolder.resolved.is_(False),
                ExternalAccountRegistry.classification.in_(
                    EXTERNAL_ACCOUNT_KNOWN_CLASSIFICATIONS,
                ),
            )
        )
    ).scalar() or 0)
    leaver_residual_holders = max(
        0, leaver_residual_holders - leaver_residual_known,
    )

    # Open compliance incidents — leaver_orphan, manager_missing,
    # notification_dead etc. Surfaced as one number; the detail page
    # (/admin/compliance) does the breakdown.
    open_incidents_q = (
        select(func.count(ComplianceIncident.id))
        .where(ComplianceIncident.status == INCIDENT_OPEN)
    )
    open_incidents = int((await db.execute(open_incidents_q)).scalar() or 0)

    # Pending dual-sign overrides (blocks leaver-revoke; visibility =
    # someone must act).
    #
    # Round-62 P2: previously this counted BLOCKING_DUAL_SIGN_STATUSES
    # = {PENDING_SECONDARY, APPROVED}, double-counting against the
    # offboarding panel's separate ``pending_overrides`` (PENDING_SECONDARY
    # only) and ``approved_overrides_ready_to_execute`` (APPROVED only)
    # KPIs. Narrow to PENDING_SECONDARY so the label '待批的双签覆写'
    # matches the count and the two panels reconcile.
    pending_dual_sign_q = (
        select(func.count(LeaverDualSignOverride.id))
        .where(LeaverDualSignOverride.status == DUAL_SIGN_PENDING_SECONDARY)
    )
    pending_dual_sign = int(
        (await db.execute(pending_dual_sign_q)).scalar() or 0
    )

    return {
        "under_licensed_plans": under_licensed_plans,
        "seat_drift_plans": seat_drift_plans,
        "leaver_residual_holders": leaver_residual_holders,
        "open_compliance_incidents": open_incidents,
        "pending_dual_sign_overrides": pending_dual_sign,
    }


# ── Top-level aggregator ─────────────────────────────────────────


async def get_health_overview(
    db: AsyncSession,
    *,
    window_days: int = 7,
) -> dict[str, Any]:
    """Build the /admin/health payload.

    Each panel is computed independently; a per-panel exception is
    recorded as ``{"status": "error", "error": "<message>"}`` so one
    bad slice doesn't 500 the whole page.
    """
    # Clamp window_days; matches the existing analytics endpoints.
    window_days = max(1, min(int(window_days or 7), 180))

    panels: dict[str, Any] = {}
    for name, builder in (
        ("cron", _cron_panel(db, window_days)),
        ("offboarding", _offboarding_panel(db)),
        ("notifications", _notifications_panel(db)),
        ("data_health", _data_health_panel(db)),
    ):
        try:
            panels[name] = await builder
        except Exception as exc:
            logger.exception("health_overview.panel_failed", extra={"panel": name})
            panels[name] = {
                "status": "error",
                "error": format_exception_for_log(exc),
            }

    return {
        "generated_at": datetime.now(UTC).isoformat(),
        "window_days": window_days,
        "panels": panels,
    }