"""Console-side realized-savings detector (sync/recompute observer).

The audit-hook tracker (``savings_tracker``) only credits actions executed
INSIDE Helix (``cmdb.server.destroy`` / ``power_off``). Owners routinely
act in the cloud console directly — terminate, stop, or resize a machine —
and none of that ever reached 「已优化成本」. This module closes that gap
(2026-07-22 decision, see DECISIONS.md): once a day, right BEFORE the
efficiency recompute overwrites yesterday's verdicts, compare each flagged
(idle / oversized — live verdict or a triage made against one) server's
freshly-synced state against its efficiency snapshot and record:

* server soft-deleted since the flag           → ``destroy`` (console)
* server now in a known-stopped status         → ``stop``    (console)
* server resized to a smaller spec (vcpu ↓, or
  equal vcpu with memory ↓ — conservative)     → ``downsize`` (console),
  with from/to evidence breadcrumbs

Rules shared with the tracker:
* **Only flagged machines count** — same conservative attribution.
* **One event per server** (``uq_savings_event_server``); a server the
  Helix-side hook already credited is skipped, so nothing double-counts.
* **Baseline = the flagged-time monthly cost** (the efficiency row's
  ``monthly_cost_usd`` snapshot — pre-action by construction since the
  detector runs before recompute), falling back to the trailing-30d cost
  facts window. Realized stays read-time-derived in ``savings_service``
  (baseline − actual post-action cost), so a restarted / re-upsized box
  self-corrects toward 0 with no cron.

Ordering contract: call ``detect_console_savings_for_integration`` at the
START of ``cmdb.efficiency_compute`` (handlers.py), before
``compute_efficiency_for_integration`` refreshes verdicts and spec
snapshots — afterwards the pre-action state is gone. Detection is
best-effort per server; one bad row never fails the compute job.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, datetime
from typing import Any

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

from app.domains.cmdb.models import (
    SAVINGS_METHOD_DESTROY,
    SAVINGS_METHOD_DOWNSIZE,
    SAVINGS_METHOD_STOP,
    SAVINGS_SOURCE_CONSOLE,
    CmdbEfficiencySavingsEvent,
    CmdbServer,
    CmdbServerEfficiency,
    CmdbServerEfficiencyTriage,
)
from app.domains.cmdb.savings_tracker import _baseline_monthly, snapshot_owner

logger = logging.getLogger(__name__)

_ACTIONABLE = ("idle", "oversized")

# Mirrors efficiency.py's known-stopped allowlist (kept as a copy of the
# vocabulary, not an import, to avoid coupling the detector to the rule
# engine module; the closed set changes ~never).
_STOPPED_STATUSES = frozenset({
    "stopped", "stopping", "terminated", "terminating",
    "deallocated", "deallocating", "shutting-down", "shutdown",
})


def _classify(
    eff: CmdbServerEfficiency, server: CmdbServer,
) -> tuple[str, dict[str, Any] | None, datetime | None] | None:
    """Decide whether this flagged server was acted on console-side.
    Returns ``(method, evidence, acted_at_hint)`` or None."""
    if server.deleted_at is not None:
        return (
            SAVINGS_METHOD_DESTROY,
            {"observed": "server_disappeared_from_sync"},
            server.deleted_at,
        )
    if (server.status or "").lower() in _STOPPED_STATUSES:
        return (
            SAVINGS_METHOD_STOP,
            {"observed": "status_stopped", "status": server.status},
            None,
        )
    # Downsize: strictly-smaller spec vs the flagged-time snapshot.
    # Conservative — vcpu decreased, or vcpu equal with memory decreased;
    # a lateral type change with equal specs is NOT a saving.
    if (
        eff.vcpu is not None
        and server.vcpu is not None
        and (
            server.vcpu < eff.vcpu
            or (
                server.vcpu == eff.vcpu
                and eff.memory_mb is not None
                and server.memory_mb is not None
                and server.memory_mb < eff.memory_mb
            )
        )
    ):
        return (
            SAVINGS_METHOD_DOWNSIZE,
            {
                "observed": "spec_shrunk",
                "from_instance_type": eff.instance_type,
                "to_instance_type": server.instance_type,
                "from_vcpu": eff.vcpu,
                "to_vcpu": server.vcpu,
                "from_memory_mb": eff.memory_mb,
                "to_memory_mb": server.memory_mb,
            },
            None,
        )
    return None


async def detect_console_savings_for_integration(
    db: AsyncSession, integration_id: str, *, now: datetime | None = None,
) -> dict[str, Any]:
    """Scan one integration's flagged servers for console-side outcomes and
    record savings events (source='console'). Commits its own rows so a
    later recompute failure can't roll the observations back. Returns a
    summary dict for the scheduled-job result."""
    now = now or datetime.now(UTC)

    t = CmdbServerEfficiencyTriage
    rows = (
        await db.execute(
            select(CmdbServerEfficiency, CmdbServer, t.verdict_at_decision)
            .join(CmdbServer, CmdbServer.id == CmdbServerEfficiency.server_id)
            .join(t, t.server_id == CmdbServerEfficiency.server_id, isouter=True)
            .where(
                CmdbServerEfficiency.integration_id == integration_id,
                or_(
                    CmdbServerEfficiency.verdict.in_(_ACTIONABLE),
                    t.verdict_at_decision.in_(_ACTIONABLE),
                ),
            )
        )
    ).all()
    if not rows:
        return {"scanned": 0, "detected": 0}

    server_ids = [str(eff.server_id) for eff, _, _ in rows]
    already = {
        str(sid)
        for sid in (
            await db.execute(
                select(CmdbEfficiencySavingsEvent.server_id).where(
                    CmdbEfficiencySavingsEvent.server_id.in_(server_ids),
                )
            )
        ).scalars()
    }

    detected = 0
    by_method: dict[str, int] = {}
    for eff, server, triage_verdict in rows:
        sid = str(eff.server_id)
        if sid in already:
            continue
        try:
            hit = _classify(eff, server)
            if hit is None:
                continue
            method, evidence, acted_hint = hit
            verdict = (
                eff.verdict if eff.verdict in _ACTIONABLE else triage_verdict
            )
            # Flagged-time cost snapshot is the preferred baseline (strictly
            # pre-action); trailing cost facts are the fallback for rows
            # computed before the snapshot columns existed.
            fallback, currency, scope_id = await _baseline_monthly(
                db, str(server.integration_id), server.instance_id, now,
            )
            baseline = (
                eff.monthly_cost_usd
                if eff.monthly_cost_usd is not None
                else fallback
            )
            owner_person_id, owner_dept = await snapshot_owner(db, server)
            db.add(CmdbEfficiencySavingsEvent(
                id=uuid.uuid4(),
                server_id=sid,
                instance_id=server.instance_id,
                integration_id=str(server.integration_id),
                scope_id=scope_id,
                method=method,
                source=SAVINGS_SOURCE_CONSOLE,
                evidence=evidence,
                verdict_at_action=verdict,
                monthly_cost_baseline_usd=baseline,
                cost_currency=(eff.cost_currency or currency or "USD"),
                owner_person_id=owner_person_id,
                owner_department=owner_dept,
                acted_by_person_id=None,
                acted_at=(acted_hint or now),
            ))
            await db.commit()
            already.add(sid)
            detected += 1
            by_method[method] = by_method.get(method, 0) + 1
        except Exception as exc:  # noqa: BLE001 — per-server best-effort
            logger.warning(
                "console-savings detection for server %s failed: %s",
                sid, str(exc)[:200],
            )
            await db.rollback()

    return {"scanned": len(rows), "detected": detected, "by_method": by_method}