"""Re-read the provider for scopes Helix already marked ``closed``.

## Why this exists

``decommission_scope`` is cloud-first: it calls the provider's close API
and only then flips Helix's row. That makes the *submission* honest. It
does not make the *outcome* honest, because on every cloud worth the
name the closure then runs asynchronously for weeks:

| cloud  | submitted state                      | finishes                |
|--------|--------------------------------------|-------------------------|
| AWS    | ``PENDING_CLOSURE``                  | ~90 days → ``SUSPENDED``|
| GCP    | ``DELETE_REQUESTED``                 | ~30 days → gone         |
| Azure  | ``Warned`` → ``Disabled``            | ~90 days → ``Deleted``  |
| Aliyun | ``Checking`` → silence period        | → ``Success``           |

Every one of those intermediate states is ``pending``. A driver that
classifies one of them as ``alive`` does not merely send a wrong
alert — ``active`` is in ``TERMINAL_PROVIDER_STATES``, so the scope
leaves this queue permanently and its real completion is never
confirmed. Azure's ``Warned`` was mapped that way until 2026-09-02;
when adding a state to a driver's mapping, the question to answer is
"can this state appear on a closure that is working correctly?", and
if yes it is ``pending``, whatever else it might also mean.

Helix wrote ``status='closed'`` at hour zero and then never looked
again. Three things can happen in that window and Helix would learn
about none of them:

1. **It completes.** Benign, but Helix's ``decommission_provider_state``
   stays frozen at the submission-time value forever, so the archive tab
   cannot tell a finished closure from one still in flight.
2. **It fails.** Aliyun ``CheckFailed`` / ``DeleteFailed`` stop dead with
   no retry. The account is alive, billable, and Helix says closed.
3. **A human undoes it.** ``projects.undelete`` (GCP) and subscription
   re-activation (Azure) are self-service for the *entire* grace window.
   Helix has already disabled every grant on that scope, so the account
   comes back **outside Helix's governance** — no owner, no grants, no
   idle sweep (every sweep excludes closed scopes), and no cost
   attribution.

(2) and (3) are the same failure the 2026-09-01 Aliyun fix was about —
Helix asserting a closure the cloud does not agree with — just arriving
by a different route and a few weeks later. Fixing the submission
without adding the read-back would have left the window open.

## One click is the whole interaction

The admin's job ends at the confirm dialog. Everything after it belongs
to this module: when a provider lets a closure die (Aliyun
``CheckFailed`` / ``DeleteFailed`` stop dead with no retry), the
reconciler **re-submits the close itself**, up to
``MAX_AUTO_CLOSE_RETRIES``, and only hands the account to a human once
that budget is spent (``closure_stalled``). On the happy path the person
who clicked gets exactly one message — the one confirming the cloud
actually finished, days or weeks later.

That is a deliberate reversal of this module's first shape (2026-09-01),
which detected a dead closure and *told someone about it*. Correct, and
useless: the admin had already made the only decision there was to make,
and forwarding them a 16-item list of Aliyun check ids asked them to
re-do it. Reporting work back to the person who asked you to do it is
not diligence.

## What it deliberately does NOT do

**It never reopens a scope.** An ``alive`` finding writes
``decommission_provider_state='active'``, an audit row, and an admin
alert — it does not flip ``status`` back, re-enable grants, or re-issue
the close. Reversing an irreversible admin action automatically, on the
strength of one provider read, is a bigger decision than the one being
corrected; a human decides whether the right answer is "close it again"
or "this was intentional, reopen it properly".

**It never re-closes a scope someone brought back.** A ``failed``
closure is unfinished work and Helix finishes it. An ``alive`` one is a
person having deliberately reversed the closure — re-closing would be
Helix overruling them, which is a different act entirely from completing
its own job. That one alerts and stops.

**It never treats a read failure as a closure.** ``unknown`` changes
nothing and leaves the scope in the queue. "We could not check" and "the
provider confirmed it is gone" are precisely the two facts this module
exists to keep apart, so they must never collapse into one.

**It only reconciles closures Helix itself submitted** — rows with a
non-NULL ``decommission_provider_state``. A NULL means a legacy or
manual-console close from before the provider-state column was written;
those have no submission to track, and sweeping them would fan out
thousands of provider reads (prod carries ~1000 historical GCP rows) to
learn nothing anyone asked for.
"""
from __future__ import annotations

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

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

from app.audit.models import AuditLog
from app.audit.service import AuditService
from app.auth.models import AuthRole, AuthUserRole
from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.models import CmdbCloudScope
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import CloudScopeRef, ScopeClosureState
from app.integrations.models import Integration
from app.notifications.service import NotificationEnvelope, NotificationService

logger = get_safe_logger(__name__)

SCOPE_STATUS_CLOSED = "closed"

# Provider states that mean the cloud is DONE. A scope in one of these
# leaves the reconcile queue for good.
#
# ``users_evacuated`` is here because the IAM-evacuation clouds (Tencent,
# UCloud, AWS ``standalone_iam``) have no provider-side after-state at
# all — their "closure" is a set of DeleteUser calls that either happened
# or didn't. Their drivers raise ``NotImplementedError`` from
# ``get_scope_closure_state``; keeping them out of the query means we
# never generate an error row for a question that has no answer.
#
# ``closure_stalled`` and ``active`` are terminal in the opposite
# direction: Helix has exhausted what it can do on its own and a human
# owns it now. Re-alerting every day would train people to ignore the
# alert, which is worse than not sending it.
#
# ``deletion_failed`` is pointedly NOT here. A closure that died is
# unfinished work, and finishing it is Helix's job, not the admin's —
# they clicked 下线 once and that is the whole interaction they signed up
# for. The reconciler re-submits the close (bounded by
# ``MAX_AUTO_CLOSE_RETRIES``) and only escalates to ``closure_stalled``
# when the retries are spent.
TERMINAL_PROVIDER_STATES = frozenset({
    "deleted",
    "deletesuccess",
    "suspended",
    "not_found",
    "not_in_org",
    "not_in_directory",
    "removed",
    "removesuccess",
    "removed_from_org",
    "users_evacuated",
    "manual_console_close",
    "closure_stalled",
    "active",
})

# How many times Helix re-submits a closure that the provider let die
# before handing it to a human. Counted from
# ``cloud.scope.decommission_auto_retry`` audit rows, so the bound
# survives restarts and is visible in the forensic trail rather than
# living in memory. Three is enough to ride out a transient provider
# failure and few enough that a genuinely stuck account reaches someone
# within days instead of being retried forever.
MAX_AUTO_CLOSE_RETRIES = 3
# Absolute ceiling on re-submissions regardless of outcome. Guards the
# one loop the budget above cannot: a closure that reports "in flight"
# on every single run would otherwise never spend its budget and would
# be retried nightly forever.
MAX_AUTO_RETRY_ATTEMPTS = 12
AUTO_RETRY_ACTION = "cloud.scope.decommission_auto_retry"
STALLED_PROVIDER_STATE = "closure_stalled"

# Driver error codes that mean the RE-submission is under way rather
# than refused — the same resumable states the interactive close
# surfaces as a 409 to the admin. Treated as "the retry took" so a
# mid-flight pre-check does not burn a retry budget.
_RESUBMISSION_IN_FLIGHT_CODES = frozenset({
    "deletion_precheck_running",
    "cloud_account_switch_pending",
})

# Coverage cap for one run. The live set is naturally tiny — a scope only
# sits in it between submission and the provider finishing — but a cap
# keeps one bad day from turning into a thousand provider calls. Anything
# dropped is COUNTED AND LOGGED (``skipped_over_limit``): a silent
# truncation would read as "everything reconciled" when it wasn't.
MAX_SCOPES_PER_RUN = 200


async def _load_driver(session: AsyncSession, integration: Integration):
    """Thin wrapper over ``lifecycle_service._load_cloud_driver``.

    Two reasons it is a function here rather than a direct call at the
    use site: it gives the tests one monkeypatch point, and the import
    is deferred so this module does not pull ``lifecycle_service`` (and
    its transitive graph) in at import time.

    It does NOT change the loader's behaviour — a bad credential still
    raises ``AppException``. That is fine and deliberate: the caller
    wraps this in ``except Exception``, so an unreadable integration
    degrades to one skipped scope rather than a failed run.
    """
    from app.domains.cmdb.lifecycle_service import _load_cloud_driver

    return await _load_cloud_driver(session, integration)


async def reconcile_decommissioned_scopes(
    session: AsyncSession,
    *,
    limit: int = MAX_SCOPES_PER_RUN,
) -> dict[str, Any]:
    """Re-read every in-flight closure and write back what the cloud says.

    Returns the run summary persisted as ``scheduled_job_runs.result_summary``.
    """
    summary: dict[str, Any] = {
        "checked": 0,
        "confirmed_closed": 0,
        "still_pending": 0,
        "state_advanced": 0,
        "failed_closures": 0,
        "auto_retried": 0,
        "auto_retry_succeeded": 0,
        "stalled": 0,
        "reopened_alive": 0,
        "unreadable": 0,
        "unsupported": 0,
        "alerted_recipients": 0,
        "completion_notices": 0,
        "skipped_over_limit": 0,
        "errors": [],
    }

    candidates = (await session.execute(
        select(CmdbCloudScope)
        .where(
            CmdbCloudScope.status == SCOPE_STATUS_CLOSED,
            CmdbCloudScope.deleted_at.is_(None),
            CmdbCloudScope.decommissioned_at.is_not(None),
            CmdbCloudScope.decommission_provider_state.is_not(None),
            CmdbCloudScope.decommission_provider_state.not_in(
                tuple(TERMINAL_PROVIDER_STATES),
            ),
        )
        # Oldest submission first: the ones most likely to have reached a
        # terminal state, and the ones whose silent failure has been
        # sitting unnoticed longest.
        .order_by(CmdbCloudScope.decommissioned_at.asc())
    )).scalars().all()

    if len(candidates) > limit:
        summary["skipped_over_limit"] = len(candidates) - limit
        logger.warning(
            "decommission reconcile: %d in-flight closures exceed the "
            "per-run cap of %d — %d not checked this run",
            len(candidates), limit, len(candidates) - limit,
        )
        candidates = list(candidates[:limit])

    if not candidates:
        return summary

    now = datetime.now(UTC)
    audit = AuditService(session)
    # One driver per integration, not per scope: building one decrypts
    # the integration's credentials.
    driver_cache: dict[str, Any] = {}
    alerts: list[dict[str, Any]] = []

    for scope in candidates:
        summary["checked"] += 1
        integration_id = str(scope.integration_id)
        try:
            if integration_id not in driver_cache:
                integration = await session.get(
                    Integration, uuid.UUID(integration_id),
                )
                if integration is None:
                    raise RuntimeError(
                        f"integration {integration_id} is gone",
                    )
                driver_cache[integration_id] = (
                    integration.provider,
                    await _load_driver(session, integration),
                )
            provider, driver = driver_cache[integration_id]
        except Exception as exc:  # noqa: BLE001
            summary["unreadable"] += 1
            summary["errors"].append(
                f"{scope.external_id}: driver unavailable: {exc}"[:200]
            )
            continue

        ref = CloudScopeRef(
            scope_type=(scope.scope_type or "").strip(),
            external_id=(scope.external_id or "").strip(),
            display_name=scope.display_name,
        )
        try:
            state: ScopeClosureState = await driver.get_scope_closure_state(ref)
        except NotImplementedError:
            # Nothing asynchronous to reconcile for this driver/scope_type.
            summary["unsupported"] += 1
            continue
        except Exception as exc:  # noqa: BLE001
            # A driver that raises instead of returning ``unknown`` is a
            # driver bug, but it must not take the whole sweep down with
            # it — one unreadable scope, not zero reconciled scopes.
            summary["unreadable"] += 1
            summary["errors"].append(
                f"{scope.external_id}: {type(exc).__name__}: {exc}"[:200]
            )
            continue

        previous = scope.decommission_provider_state
        disposition = state.disposition

        if disposition == "unknown":
            summary["unreadable"] += 1
            if state.error_message:
                summary["errors"].append(
                    f"{scope.external_id}: {state.error_message}"[:200]
                )
            continue

        # Write the provider's word back, whatever it is. Even a
        # ``pending`` whose state string moved on (AWS PENDING_CLOSURE →
        # SUSPENDED, Aliyun Checking → Deleting) is information the
        # archive tab should show.
        if state.provider_state and state.provider_state != previous:
            scope.decommission_provider_state = state.provider_state
            scope.updated_at = now
            summary["state_advanced"] += 1

        if disposition == "pending":
            summary["still_pending"] += 1
            continue

        if disposition == "closed":
            summary["confirmed_closed"] += 1
            # They clicked once and asked to be told the result. This is
            # the result — the only notification the happy path sends.
            if await _notify_closure_done(session, scope, provider):
                summary["completion_notices"] += 1
            await audit.log(
                trace_id="cmdb.decommission_reconcile",
                actor_type="system",
                action="cloud.scope.decommission_confirmed",
                status="success",
                resource_type="cmdb_cloud_scope",
                resource_id=str(scope.id),
                detail={
                    "provider": provider,
                    "external_id": scope.external_id,
                    "previous_provider_state": previous,
                    "provider_state": state.provider_state,
                    "closure_detail": state.detail,
                },
            )
            continue

        if disposition == "failed":
            # The closure died. That is unfinished work, not news to
            # forward: the admin clicked 下线 once and Helix owns the rest.
            # Re-submit — ``close_scope_on_provider`` is idempotent and,
            # for Aliyun, starts a FRESH deletion check because the dead
            # one's result is stale.
            summary["failed_closures"] += 1
            attempts, total_attempts = await _auto_retry_count(
                session, str(scope.id),
            )
            if (attempts < MAX_AUTO_CLOSE_RETRIES
                    and total_attempts < MAX_AUTO_RETRY_ATTEMPTS):
                retried = await _auto_retry_close(
                    session, audit, scope,
                    driver=driver, provider=provider, ref=ref,
                    attempt=attempts + 1, state=state, now=now,
                )
                summary["auto_retried"] += 1
                if retried:
                    summary["auto_retry_succeeded"] += 1
                # Either way the scope keeps a non-terminal
                # ``provider_state`` and stays in the queue, so the next
                # run reads whatever the re-submission actually did.
                continue
            # Retries spent. NOW it is a human's problem, and saying so
            # once is worth more than saying it every night.
            scope.decommission_provider_state = STALLED_PROVIDER_STATE
            scope.updated_at = now
            summary["stalled"] += 1
        else:
            # ``alive`` is not a stalled job — someone deliberately
            # reversed the closure (GCP undelete / Azure re-activation).
            # Re-closing would be Helix overruling a person, which is a
            # different thing entirely from finishing its own work.
            summary["reopened_alive"] += 1
        await audit.log(
            trace_id="cmdb.decommission_reconcile",
            actor_type="system",
            action="cloud.scope.decommission_contradicted",
            status="failed",
            resource_type="cmdb_cloud_scope",
            resource_id=str(scope.id),
            detail={
                "provider": provider,
                "external_id": scope.external_id,
                "disposition": disposition,
                "previous_provider_state": previous,
                "provider_state": state.provider_state,
                "error_message": state.error_message,
                "closure_detail": state.detail,
            },
        )
        alerts.append({
            "scope_id": str(scope.id),
            "provider": provider,
            "external_id": scope.external_id,
            "display_name": scope.display_name or scope.external_id,
            "disposition": disposition,
            "message": state.error_message or "",
            "decommissioned_at": (
                scope.decommissioned_at.isoformat()
                if scope.decommissioned_at else None
            ),
        })

    if alerts:
        summary["alerted_recipients"] = await _alert_admins(session, alerts)

    return summary


async def _auto_retry_count(
    session: AsyncSession, scope_id: str,
) -> tuple[int, int]:
    """``(unproductive, total)`` re-submissions of this closure so far.

    Counted from the audit trail rather than a column: the bound then
    survives restarts, needs no migration, and — the part that matters —
    is answerable by anyone reading the log later, which a counter in a
    column is not.

    The retry BUDGET spends only on **unproductive** attempts (audit
    ``status='failed'``). An attempt that lands mid-pre-check is progress
    — Aliyun's runs for 3–5 minutes, so on a daily cadence the first
    re-submission almost always returns "running" — and charging the
    budget for it would burn most of the allowance on the mechanism
    working correctly. ``total`` is still returned and still capped, so a
    closure that somehow stays perpetually "in flight" cannot be retried
    forever.
    """
    rows = (await session.execute(
        select(AuditLog.status, func.count())
        .select_from(AuditLog)
        .where(
            AuditLog.action == AUTO_RETRY_ACTION,
            AuditLog.resource_id == scope_id,
        )
        .group_by(AuditLog.status)
    )).all()
    total = sum(int(n) for _, n in rows)
    unproductive = sum(int(n) for st, n in rows if st != "success")
    return unproductive, total


async def _auto_retry_close(
    session: AsyncSession,
    audit: AuditService,
    scope: CmdbCloudScope,
    *,
    driver: Any,
    provider: str,
    ref: CloudScopeRef,
    attempt: int,
    state: ScopeClosureState,
    now: datetime,
) -> bool:
    """Re-submit a closure the provider let die. Returns "it took".

    This is the whole point of the feature: one click from the admin,
    and Helix keeps pushing until the cloud is actually done or until it
    genuinely cannot proceed. Every attempt is audited with the failure
    that prompted it, so the retry chain reads as a story rather than as
    a mystery series of provider calls.
    """
    outcome: dict[str, Any] = {
        "provider": provider,
        "external_id": scope.external_id,
        "attempt": attempt,
        "max_attempts": MAX_AUTO_CLOSE_RETRIES,
        "prompted_by": state.error_message,
        "prompted_by_detail": state.detail,
    }
    took = False
    try:
        result = await driver.close_scope_on_provider(ref)
    except Exception as exc:  # noqa: BLE001
        outcome["error"] = f"{type(exc).__name__}: {exc}"[:300]
    else:
        outcome["ok"] = result.ok
        outcome["error_code"] = result.error_code
        outcome["error_message"] = result.error_message
        outcome["provider_state"] = result.provider_state
        outcome["close_detail"] = result.detail
        if result.ok and result.provider_state:
            scope.decommission_provider_state = result.provider_state
            scope.updated_at = now
            took = True
        elif result.error_code in _RESUBMISSION_IN_FLIGHT_CODES:
            # The re-submission is mid-flight (Aliyun's 3–5 minute
            # pre-check). Not a failure and not done — leave the row
            # where it is so tomorrow's run reads the real outcome.
            took = True
    await audit.log(
        trace_id="cmdb.decommission_reconcile",
        actor_type="system",
        action=AUTO_RETRY_ACTION,
        status="success" if took else "failed",
        resource_type="cmdb_cloud_scope",
        resource_id=str(scope.id),
        detail=outcome,
    )
    return took


async def _notify_closure_done(
    session: AsyncSession, scope: CmdbCloudScope, provider: str,
) -> bool:
    """Tell whoever clicked 下线 that the cloud actually finished.

    Only the person who asked for it — this is the answer to their
    action, not a governance broadcast. Admins already have the archive
    tab and the contradiction alerts; adding a nightly "an account you
    did not close is now closed" card to four inboxes is how a useful
    notification becomes an ignored one.
    """
    actor_id = (scope.decommissioned_by or "").strip()
    if not actor_id:
        return False
    person = await session.get(IdentityPerson, actor_id)
    if not (person and person.work_email and person.deleted_at is None
            and person.employment_status == "active"):
        return False
    name = scope.display_name or scope.external_id
    body = "\n".join([
        f"你在 Helix 上下线的云账号 [{provider}] {name}"
        f"({scope.external_id})已经在云端**彻底删除完成**。",
        "",
        "云厂商的关闭是异步的,提交之后通常还要跑上几周;这条消息是"
        "Helix 每天回读云端状态、确认它真的走完之后才发的。",
        "",
        f"云端终态:{scope.decommission_provider_state}",
        (
            f"下线时间:{scope.decommissioned_at.isoformat()}"
            if scope.decommissioned_at else ""
        ),
        f"下线理由:{scope.decommission_reason or '—'}",
        "",
        "账号内的资源与数据已随账号一并永久删除,不可恢复。无需再做任何操作。",
    ])
    try:
        await NotificationService(session).dispatch(
            NotificationEnvelope(
                recipient=person,
                title=f"[云治理] 「{name}」已在云端彻底删除",
                body_text=body,
                url="/admin/cloud/scopes",
                channels=["feishu", "email"],
                related_resource_type="cmdb_cloud_scope",
                related_resource_id=None,
            ),
            trace_id="cmdb.decommission_reconcile",
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "decommission reconcile: completion notice failed for %s: %s",
            person.work_email, exc,
        )
        return False
    return True


async def _alert_admins(
    session: AsyncSession, alerts: list[dict[str, Any]],
) -> int:
    """One digest to the admins per run, not one card per scope.

    These findings arrive in bursts (a credential that lost a permission
    contradicts every scope under that integration at once), and a burst
    of individual cards is the fastest way to teach people to swipe the
    alert away.
    """
    admin_ids = list((await session.execute(
        select(AuthUserRole.user_id)
        .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
        .where(AuthRole.name.in_(("global_admin", "it_admin")))
    )).scalars().all())
    if not admin_ids:
        return 0

    lines: list[str] = [
        f"{len(alerts)} 个已在 Helix 标记为「已下线」的云账号,"
        "云端说法与之矛盾。",
        "",
        "Helix 不会自动改回去 —— 下面每一条都需要人来判断"
        "「重新关一次」还是「本来就该留着,按流程恢复」。",
        "",
    ]
    for a in alerts[:20]:
        verdict = (
            "云端报告关闭失败,不会自动重试"
            if a["disposition"] == "failed"
            else "云端说这个账号还活着(很可能被人在控制台恢复了)"
        )
        lines.append(f"  · [{a['provider']}] {a['display_name']}")
        lines.append(f"      {verdict}")
        if a["message"]:
            lines.append(f"      {a['message'][:300]}")
        if a["decommissioned_at"]:
            lines.append(f"      Helix 记录的下线时间:{a['decommissioned_at']}")
        lines.append("")
    if len(alerts) > 20:
        lines.append(f"  ... 另有 {len(alerts) - 20} 个未列出")
        lines.append("")
    lines.append(
        "为什么要管:Helix 已经把这些 scope 上的 grant 全部停用了,"
        "闲置巡检也会跳过 closed 的 scope。所以一个「活过来」的账号"
        "此刻没有 owner、没有 grant、也不会再被任何巡检提醒 —— "
        "它在治理之外运行并继续计费。"
    )
    body = "\n".join(lines)
    title = f"[云治理] {len(alerts)} 个「已下线」云账号与云端状态矛盾"

    notify = NotificationService(session)
    sent = 0
    for admin_id in admin_ids:
        person = await session.get(IdentityPerson, str(admin_id))
        if not (person and person.work_email and person.deleted_at is None
                and person.employment_status == "active"):
            continue
        try:
            await notify.dispatch(
                NotificationEnvelope(
                    recipient=person,
                    title=title,
                    body_text=body,
                    url="/admin/cloud/scopes",
                    channels=["feishu", "email"],
                    related_resource_type="cmdb_cloud_scope",
                    related_resource_id=None,
                ),
                trace_id="cmdb.decommission_reconcile",
            )
            sent += 1
        except Exception as exc:  # noqa: BLE001
            logger.warning(
                "decommission reconcile: alert dispatch failed for %s: %s",
                person.work_email, exc,
            )
    return sent