"""Carry a confirmed decommission past "the scope still has stuff in it".

## Why this exists

``close_scope_on_provider`` is cloud-first, and several clouds refuse the
close while the scope still contains resources:

| cloud  | refusal                                  | what has to go first |
|--------|------------------------------------------|----------------------|
| Azure  | ``400 ResourcesExistingOnSubscription``  | every resource group |
| GCP    | ``liens_exist``                          | every project lien   |

Until 2026-09-07 both came back to the admin as a red modal with a CLI
one-liner in it — Helix answering "下线这个订阅" with "here is how to
decommission a subscription". The sibling module
``decommission_reconcile`` already states the rule this violates:

> One click is the whole interaction … Reporting work back to the person
> who asked you to do it is not diligence.

That module owns everything AFTER the provider accepts. This one owns
everything BETWEEN the admin's confirmation and the provider accepting:
delete the blockers, retry the close, and tell the person once — when it
is done, or when it genuinely needs them.

## The consent invariant

Cleanup deletes **only** keys the admin was shown in the confirmation
dialog and confirmed (``decommission_cleanup_keys``). Every tick
re-reads the live blockers and intersects them with that list again:

* live ∩ confirmed → delete (idempotent; already-gone counts as done)
* live ∖ confirmed → **stop**. Somebody created something after the
  dialog was rendered. ``needs_reconfirm``, notify once, and wait for a
  human to re-confirm the new list. An automatic cleanup that deletes
  what nobody looked at is a different feature from the one that was
  approved.
* confirmed ∖ live → nothing to do; it is already gone.

## What it deliberately does NOT do

**It never widens the scope of a deletion.** See above. This is the
whole safety story of the feature.

**It never touches ``status``.** The scope stays ``active`` for the
whole cleanup window and flips to ``closed`` only through the normal
``decommission_scope`` path, i.e. only after the provider acknowledges.
A scope mid-cleanup is not "closed-ish"; it is a live account with a
takedown in flight, and every predicate that reads ``status`` should
keep seeing the truth.

**It never gives up quietly.** Both terminal failure states
(``blocked``, ``stalled``) notify the person who clicked, exactly once,
and say what is left. Silence would be the worst outcome: the admin
believes the account is going away, and it is not.

**It never treats "could not read" as "nothing left".** A failed
preflight keeps the scope in ``cleaning`` and burns one attempt; it
never advances to the close as if the scope were empty.
"""
from __future__ import annotations

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

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

from app.core import database as _database
from app.core.exceptions import AppException
from app.core.safe_logging import get_safe_logger
from app.domains.cmdb.lifecycle_service import (
    CLEANUP_STATE_BLOCKED,
    CLEANUP_STATE_CLEANING,
    CLEANUP_STATE_NEEDS_RECONFIRM,
    CLEANUP_STATE_STALLED,
    MAX_CLEANUP_ATTEMPTS,
    CloudLifecycleService,
    _load_driver_for_scope,
    describe_blocker,
    plan_blocker_cleanup,
)
from app.domains.cmdb.models import CmdbCloudScope
from app.domains.cmdb.prod_write_lock import (
    ProdCloudLockedError,
    ensure_prod_scope_writable,
)
from app.identity.person_models import IdentityPerson
from app.integrations.driver_base import ClosePreflight, CloudScopeRef
from app.notifications.service import NotificationEnvelope, NotificationService

logger = get_safe_logger(__name__)

SCOPE_STATUS_ACTIVE = "active"

# One tick handles at most this many scopes. Pending decommissions are
# rare (single digits at a time in prod), and the cap is here so a
# pathological backlog degrades to "the queue drains over several ticks"
# rather than to a 5-minute cron overrunning its own schedule.
MAX_SCOPES_PER_RUN = 20


async def resume_pending_decommissions(
    limit: int = MAX_SCOPES_PER_RUN,
) -> dict[str, Any]:
    """Advance every decommission waiting on blocker cleanup.

    **Session-per-scope, by design.** The advance path calls
    ``CloudLifecycleService.decommission_scope``, which raises
    ``AppException`` when the provider still refuses — a normal,
    expected outcome here. Sharing one session across scopes would mean
    one scope's rollback discarding the attempt counters and state
    transitions already written for the others in the same tick. Each
    scope therefore gets its own session and its own commit, and a
    failure isolates to the scope that caused it.
    """
    summary: dict[str, Any] = {
        "checked": 0,
        "closed": 0,
        "still_cleaning": 0,
        "needs_reconfirm": 0,
        "blocked": 0,
        "stalled": 0,
        "unreadable": 0,
        "notified": 0,
        "errors": [],
    }

    # ``_database.async_session_factory()`` — resolved through the module
    # on every call, never bound at import time. The test harness swaps
    # that attribute to a factory pointing at the test database; a
    # ``from … import async_session_factory`` here would keep the
    # original object and every session this job opens would land on the
    # DEV database instead (which is how it was caught: an "unknown
    # column" for a migration the dev DB had never seen).
    async with _database.async_session_factory() as session:
        scope_ids = list((await session.execute(
            select(CmdbCloudScope.id)
            .where(
                CmdbCloudScope.deleted_at.is_(None),
                CmdbCloudScope.status == SCOPE_STATUS_ACTIVE,
                CmdbCloudScope.decommission_pending_since.isnot(None),
                CmdbCloudScope.decommission_cleanup_state
                == CLEANUP_STATE_CLEANING,
            )
            .order_by(CmdbCloudScope.decommission_pending_since.asc())
            .limit(limit)
        )).scalars().all())

    for scope_id in scope_ids:
        summary["checked"] += 1
        try:
            async with _database.async_session_factory() as session:
                await _advance_one(session, str(scope_id), summary)
                await session.commit()
        except Exception as exc:  # noqa: BLE001
            # One scope's driver blowing up must not stop the queue.
            summary["errors"].append(f"{scope_id}: {exc}")
            logger.warning(
                "decommission cleanup resume failed for %s: %s",
                scope_id, exc,
            )
    return summary


async def _advance_one(
    session: AsyncSession, scope_id: str, summary: dict[str, Any],
) -> None:
    scope = await session.get(CmdbCloudScope, uuid.UUID(scope_id))
    if scope is None or scope.decommission_pending_since is None:
        return
    now = datetime.now(UTC)
    attempts = int(scope.decommission_cleanup_attempts or 0) + 1
    scope.decommission_cleanup_attempts = attempts
    scope.updated_at = now

    if attempts > MAX_CLEANUP_ATTEMPTS:
        scope.decommission_cleanup_state = CLEANUP_STATE_STALLED
        summary["stalled"] += 1
        if await _notify_actor(
            session, scope,
            title_suffix="的下线还没走完",
            lines=[
                "Helix 已经按你确认的清单清理了这个 scope 里的资源,但云端"
                f"在 {MAX_CLEANUP_ATTEMPTS} 次重试(约 6 小时)之后仍然不"
                "接受关闭。",
                "",
                f"最后一次的云端说法:{scope.decommission_cleanup_error or '—'}",
                "",
                "这一步需要你看一眼:到云控制台确认还剩什么,或者在 Helix 上"
                "重新点一次下线(会重新读取最新清单)。",
            ],
        ):
            summary["notified"] += 1
        return

    # Honour the prod-cloud kill-switch on every tick, not only at the
    # click. The lock means "stop touching production cloud accounts
    # NOW" — an in-flight cleanup is exactly the kind of write it is
    # there to stop, and the earlier confirmation cannot have known the
    # lock would be engaged. Deliberately does NOT burn an attempt: the
    # budget measures the provider being slow, not an operator holding
    # a switch down, and cancelling a decommission over a maintenance
    # window would be a decision nobody made.
    try:
        await ensure_prod_scope_writable(session, scope)
    except ProdCloudLockedError:
        scope.decommission_cleanup_attempts = attempts - 1
        scope.decommission_cleanup_error = (
            "paused: prod cloud writes are locked"
        )[:500]
        summary["still_cleaning"] += 1
        return

    # A driver that cannot even be built (integration deleted, secrets
    # rotated out) must still burn an attempt — otherwise the scope sits
    # in the queue forever, the cron goes red every 5 minutes, and
    # nobody is ever told, because the escalation is gated on the
    # attempt budget this branch would have skipped.
    try:
        _integration, driver = await _load_driver_for_scope(session, scope)
    except Exception as exc:  # noqa: BLE001
        scope.decommission_cleanup_error = (
            f"driver unavailable: {exc}"
        )[:500]
        summary["unreadable"] += 1
        summary["still_cleaning"] += 1
        return

    scope_ref = CloudScopeRef(
        scope_type=(scope.scope_type or "").strip(),
        external_id=(scope.external_id or "").strip(),
        display_name=scope.display_name,
    )

    try:
        preflight: ClosePreflight = await driver.preflight_close_scope(
            scope_ref,
        )
    except Exception as exc:  # noqa: BLE001
        preflight = ClosePreflight(
            supported=True, checked=False,
            error_code="preflight_failed", error_message=str(exc)[:300],
        )

    if not preflight.supported:
        # The driver lost its preflight between the click and now (a
        # deploy changing scope_type handling, a driver swap). Not
        # something a retry fixes.
        scope.decommission_cleanup_state = CLEANUP_STATE_STALLED
        scope.decommission_cleanup_error = (
            "driver no longer supports close preflight"
        )
        summary["stalled"] += 1
        if await _notify_actor(
            session, scope,
            title_suffix="的下线需要人工继续",
            lines=[
                "Helix 无法再自动确认这个 scope 里还剩什么(驱动不支持),"
                "所以这次下线停在这里了。请到云控制台处理,或在 Helix 上"
                "重新点一次下线。",
            ],
        ):
            summary["notified"] += 1
        return

    if not preflight.checked and not preflight.blockers:
        # Could not look. Keep the scope in the queue and say why —
        # never advance to the close on the strength of an empty list we
        # do not believe.
        scope.decommission_cleanup_error = (
            preflight.error_message or "preflight unreadable"
        )[:500]
        summary["unreadable"] += 1
        summary["still_cleaning"] += 1
        return

    plan = plan_blocker_cleanup(
        preflight.blockers, list(scope.decommission_cleanup_keys or []),
    )

    if plan.unseen:
        scope.decommission_cleanup_state = CLEANUP_STATE_NEEDS_RECONFIRM
        scope.decommission_cleanup_error = (
            f"{len(plan.unseen)} unconfirmed blocker(s) appeared"
        )
        summary["needs_reconfirm"] += 1
        if await _notify_actor(
            session, scope,
            title_suffix="的下线需要你重新确认",
            lines=[
                "这个 scope 里出现了你确认下线时还不存在的内容,Helix 不会"
                "自动删除没被确认过的东西,所以停下来了:",
                "",
                *(f"· {describe_blocker(b)}" for b in plan.unseen[:10]),
                "",
                "如果这些也该一起删掉,在 Helix 上重新点一次下线,弹窗里会"
                "列出最新的清单。",
            ],
        ):
            summary["notified"] += 1
        return

    if plan.unremovable and not plan.deletable:
        scope.decommission_cleanup_state = CLEANUP_STATE_BLOCKED
        scope.decommission_cleanup_error = (
            "; ".join(
                f"{b.name}: {b.blocked_reason or 'not removable'}"
                for b in plan.unremovable[:3]
            )
        )[:500]
        summary["blocked"] += 1
        if await _notify_actor(
            session, scope,
            title_suffix="的下线被云端保护挡住了",
            lines=[
                "下线被以下内容挡住,而 Helix 没有权限(或不应自行)删除它们:",
                "",
                *(
                    f"· {describe_blocker(b)}"
                    + (f" —— {b.blocked_reason}" if b.blocked_reason else "")
                    for b in plan.unremovable[:10]
                ),
                "",
                "请在云控制台处理掉这些,然后在 Helix 上重新点一次下线。",
            ],
        ):
            summary["notified"] += 1
        return

    if plan.deletable:
        try:
            clear = await driver.clear_close_blockers(
                scope_ref, keys=plan.deletable,
            )
        except Exception as exc:  # noqa: BLE001
            scope.decommission_cleanup_error = (
                f"cleanup call failed: {exc}"
            )[:500]
            summary["still_cleaning"] += 1
            return
        scope.decommission_cleanup_error = (
            (clear.error_message or "")[:500] or None
        )
        if clear.in_progress or clear.failed:
            # Deletion in flight (or partly refused) → nothing more to
            # do this tick. The close is attempted only once the
            # blockers are actually gone: Azure keeps answering
            # ``ResourcesExistingOnSubscription`` while a group sits in
            # ``Deleting``, and each such call is a wasted write against
            # a rate-limited API.
            summary["still_cleaning"] += 1
            return
        # Everything the provider deletes synchronously is already gone
        # (GCP liens; an Azure group that had vanished before we asked).
        # Falling through retries the close in the SAME tick rather than
        # parking a finished cleanup for another 5 minutes.

    # No live blockers left → the close should be acceptable now.
    service = CloudLifecycleService(session)
    trace_id = (
        scope.decommission_cleanup_trace_id
        or f"cmdb.decommission_cleanup_resume:{scope.id}"
    )
    actor_raw = (scope.decommission_pending_by or "").strip()
    try:
        actor_id = uuid.UUID(actor_raw)
    except ValueError:
        # Only reachable if the column was hand-edited: the click path
        # always writes ``str(actor_user_id)``. Finishing the closure is
        # still right (it is what was asked for), and the nil uuid is an
        # honest "the actor was lost" marker in the audit row — better
        # than attributing an irreversible action to a real person we
        # guessed at. ``audit_logs.actor_user_id`` is nullable with no
        # FK, so this cannot fail the write.
        actor_id = uuid.UUID(int=0)
    reason = scope.decommission_pending_reason or "(cleanup resume)"
    try:
        _scope, disabled, close_result = await service.decommission_scope(
            str(scope.id),
            reason=reason,
            actor_user_id=actor_id,
            trace_id=trace_id,
            # Cleanup is finished by definition on this path; passing
            # True would re-enter ``_begin_blocker_cleanup`` and could
            # re-stamp a pending state on top of a successful close.
            cleanup_blockers=False,
        )
    except AppException as exc:
        # Provider still refusing. Keep the scope in ``cleaning``; the
        # attempt budget above is what eventually escalates it.
        #
        # ``decommission_scope`` wrote its failure audit through a
        # separate session, so the row survives this session's state;
        # but the raise itself aborted whatever it had started, so
        # re-apply the attempt bookkeeping on a fresh read.
        await session.rollback()
        scope = await session.get(CmdbCloudScope, uuid.UUID(scope_id))
        if scope is not None:
            scope.decommission_cleanup_attempts = attempts
            scope.decommission_cleanup_error = str(exc.message)[:500]
            scope.updated_at = datetime.now(UTC)
        summary["still_cleaning"] += 1
        return

    if close_result.ok:
        summary["closed"] += 1
        if await _notify_actor(
            session, _scope,
            title_suffix="的资源已清理完毕,云端已接受关闭",
            lines=[
                "你在 Helix 上下线的这个云账号,里面的资源已经按你确认的"
                "清单清理完毕,云厂商也接受了关闭请求。",
                "",
                f"云端状态:{close_result.provider_state or '—'}",
                f"随之停用的权限:{disabled} 条",
                "",
                "云厂商的彻底删除还要走它自己的宽限期(AWS/Azure 约 90 天,"
                "GCP 约 30 天);走完之后 Helix 会再通知你一次。无需再操作。",
            ],
        ):
            summary["notified"] += 1
    else:
        summary["still_cleaning"] += 1


async def _notify_actor(
    session: AsyncSession,
    scope: CmdbCloudScope,
    *,
    title_suffix: str,
    lines: list[str],
) -> bool:
    """Message the person who clicked 下线 — and only them.

    Same rule as ``decommission_reconcile._notify_closure_done``: this
    is the answer to somebody's action, not a governance broadcast. The
    admin group already has the scope list and the audit trail; a card
    in four inboxes about an account they did not touch is how a useful
    notification becomes an ignored one.
    """
    actor_id = (
        scope.decommission_pending_by or 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"云账号「{name}」({scope.external_id}){title_suffix}。",
        "",
        f"下线理由:{scope.decommission_pending_reason or scope.decommission_reason or '—'}",
        "",
        *lines,
    ])
    try:
        await NotificationService(session).dispatch(
            NotificationEnvelope(
                recipient=person,
                title=f"[云治理] 「{name}」{title_suffix}",
                body_text=body,
                url=f"/admin/cloud/scopes/{scope.id}",
                channels=["feishu", "email"],
                related_resource_type="cmdb_cloud_scope",
                related_resource_id=None,
            ),
            trace_id=(
                scope.decommission_cleanup_trace_id
                or "cmdb.decommission_cleanup_resume"
            ),
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning(
            "decommission cleanup: notice failed for %s: %s",
            person.work_email, exc,
        )
        return False
    return True