"""Scheduled jobs for cloud-scope owner governance.

Two crons:

* ``cmdb.owner_semiannual_confirmation`` (daily 09:00 Asia/Shanghai)
  — for every non-prod active scope whose owner is past their 180-day
    confirmation window, send a confirmation card.
  — ALSO scans pending prompts of either kind that have been
    outstanding >14 days and escalates them to IT admins.

* ``cmdb.zero_spend_alert`` (monthly 09:00 on the 1st)
  — for every non-prod active scope whose previous calendar month had
    $0 spend, send a zero-spend alert card.

Both handlers:
  * Iterate eligible scopes with their own per-scope DB session — a
    single Feishu push failure does not roll back the whole sweep.
  * Skip scopes whose owner is a leaver (employment_status=left /
    leaver_pending) — that scope already has the leaver-blocked
    governance attention.
  * Are idempotent — a "outstanding pending row of same kind" guard in
    :class:`OwnerConfirmationService.list_scopes_due_*` short-circuits
    re-runs within the same day / month.
"""
from __future__ import annotations

import uuid
from datetime import datetime
from typing import Any
from zoneinfo import ZoneInfo

from sqlalchemy.ext.asyncio import AsyncSession

from app.core.safe_logging import get_safe_logger
from app.scheduled_jobs.registry import register_scheduled_job

logger = get_safe_logger(__name__)


@register_scheduled_job(
    key="cmdb.owner_semiannual_confirmation",
    display_name="云订阅 Owner 半年确认告警",
    description=(
        "每日 09:00（Asia/Shanghai）扫描所有非生产环境（environment≠prod）"
        "的云订阅，找出 owner 距离上次确认已超过 180 天的发送飞书卡片，"
        "owner 必须确认或转移。卡片含最近 30/180 天花费、人数、活跃成员"
        "等丰富数据。同一次任务还会把所有**尚未响应**的待确认（半年 / "
        "零花费）重新推给该 owner —— 每天一次，直到他确认或转让为止；"
        "不再在 14 天后升级给 IT Admin（2026-08-24 起）。"
        "订阅已下线、或期间换过 owner 的待确认会被作废，"
        "owner 已离职的跳过不推。仅 environment ∈ (dev, staging) 进入。"
        "首次初始化扫描：通过管理员"
        "`POST /scheduled-jobs/{id}/trigger` 携带 "
        "`config_override={\"min_age_days\": 0}` 绕过 30 天新生保护，"
        "一次性给所有合格 scope 推送卡片；默认 30 天保护不变。"
        "积压首轮想限速：`config_override={\"reminder_max_per_run\": 20}`。"
    ),
)
async def run_owner_semiannual_confirmation(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Send semiannual cards + re-send every unanswered prompt to its owner.

    Returns a summary dict for ``last_result_summary``:
      ``scopes_prompted`` — how many owners got a new semiannual card
      ``reminded`` — how many unanswered prompts were re-sent to their
        owner this run (at most one per prompt per day)
      ``voided_owner_changed`` / ``voided_scope_closed`` — prompts closed
        instead of reminded, because the addressee is no longer the owner
        or the subscription is gone
      ``skipped_owner_inactive`` — prompts left open but NOT pushed,
        because the addressee has left the company
      ``deferred_over_cap`` — prompts a ``reminder_max_per_run`` override
        left for the next tick
      ``errors`` — per-scope error count (best-effort sweep)
    """
    from app.core.database import async_session_factory
    from app.domains.cmdb.models import CmdbCloudScope
    from app.domains.cmdb.owner_confirmation_service import (
        DEFAULT_NEWBORN_GUARD_DAYS,
        OwnerConfirmationService,
    )
    from app.identity.person_models import IdentityPerson

    trace_id = uuid.uuid4().hex
    scopes_prompted = 0
    scopes_deduped = 0
    errors = 0

    # Newborn-guard override — admins can pass ``min_age_days`` via the
    # ``POST /scheduled-jobs/{id}/trigger`` ``config_override`` payload
    # for a one-shot init sweep on a tenant whose scopes were all
    # bulk-imported within the last 30 days. Default = the 30-day
    # production behaviour; the default cron config dict does not set
    # this key, so scheduled ticks stay unchanged.
    raw = config.get("min_age_days", DEFAULT_NEWBORN_GUARD_DAYS)
    try:
        min_age_days = int(raw)
    except (TypeError, ValueError) as exc:
        raise ValueError(
            f"min_age_days must be a non-negative int, got {raw!r}",
        ) from exc
    if min_age_days < 0:
        raise ValueError(
            f"min_age_days must be >= 0, got {min_age_days}",
        )

    # Reminder fan-out cap. Unset by default, which is the intended
    # steady state: one card per open prompt per day, and open prompts
    # are bounded by the subscription count. It exists for the FIRST
    # pass over a backlog — the escalation sweep this replaced was inert
    # from 2026-06-03 to 2026-08-24, so ~96 prompts are sitting
    # unanswered, and an operator may want to ramp rather than reach 90
    # owners on day one. ``config_override={"reminder_max_per_run": 20}``.
    raw_cap = config.get("reminder_max_per_run")
    reminder_max_per_run: int | None = None
    if raw_cap is not None:
        try:
            reminder_max_per_run = int(raw_cap)
        except (TypeError, ValueError) as exc:
            raise ValueError(
                "reminder_max_per_run must be a positive int, got "
                f"{raw_cap!r}",
            ) from exc
        if reminder_max_per_run < 1:
            raise ValueError(
                "reminder_max_per_run must be >= 1, got "
                f"{reminder_max_per_run}",
            )

    # First pass — identify eligible scopes using the shared session.
    async with async_session_factory() as scan_session:
        scan_service = OwnerConfirmationService(scan_session)
        candidates = await scan_service.list_scopes_due_semiannual(
            min_age_days=min_age_days,
        )
        scope_ids = [str(s.id) for s in candidates]

    # Second pass — one transaction per scope so a single failure
    # doesn't roll back the whole sweep.
    for scope_id in scope_ids:
        try:
            async with async_session_factory() as scope_session:
                service = OwnerConfirmationService(scope_session)
                scope = await scope_session.get(
                    CmdbCloudScope, uuid.UUID(scope_id),
                )
                if scope is None or scope.deleted_at is not None:
                    continue
                if scope.owner_person_id is None:
                    continue
                owner = await scope_session.get(
                    IdentityPerson, uuid.UUID(scope.owner_person_id),
                )
                if owner is None or owner.deleted_at is not None:
                    continue
                if owner.employment_status in ("left", "leaver_pending"):
                    # Leaver-blocked sweep handles this scope separately.
                    continue
                _row, snapshot = await service.create_semiannual_pending(
                    scope, owner, trace_id=trace_id,
                )
                await scope_session.commit()
                # Distinguish "fresh card dispatched" from "race-safe
                # short-circuit on existing pending row". Both are
                # success-shapes but the operator-facing telemetry
                # should not double-count the same scope across
                # parallel cron ticks.
                if snapshot.get("deduplicated"):
                    scopes_deduped += 1
                else:
                    scopes_prompted += 1
        except Exception:
            logger.exception(
                "semiannual confirmation failed for scope=%s", scope_id,
            )
            errors += 1

    # Third pass — re-send every unanswered prompt (both kinds) to its
    # owner. Daily cadence; the service enforces at-most-once-per-day
    # per prompt.
    reminded = 0
    voided_owner_changed = 0
    voided_scope_closed = 0
    skipped_owner_inactive = 0
    deferred_over_cap = 0
    try:
        async with async_session_factory() as rem_session:
            rem_service = OwnerConfirmationService(rem_session)
            rem_summary = await rem_service.remind_unanswered_owners(
                trace_id=trace_id, max_per_run=reminder_max_per_run,
            )
            await rem_session.commit()
            reminded = rem_summary.get("reminded", 0)
            voided_owner_changed = rem_summary.get("voided_owner_changed", 0)
            voided_scope_closed = rem_summary.get("voided_scope_closed", 0)
            skipped_owner_inactive = rem_summary.get(
                "skipped_owner_inactive", 0,
            )
            deferred_over_cap = rem_summary.get("deferred_over_cap", 0)
            errors += rem_summary.get("errors", 0)
    except Exception:
        logger.exception("owner reminder sweep failed")
        errors += 1

    if deferred_over_cap:
        logger.info(
            "owner reminder sweep capped at %s; %s unanswered prompts "
            "deferred to the next tick",
            reminder_max_per_run, deferred_over_cap,
        )
    if skipped_owner_inactive:
        logger.info(
            "owner reminder sweep skipped %s prompt(s) whose addressee has "
            "left — awaiting offboarding force-transfer",
            skipped_owner_inactive,
        )

    summary = {
        "scopes_prompted": scopes_prompted,
        "scopes_deduped": scopes_deduped,
        "reminded": reminded,
        # Prompts closed instead of reminded — the addressee is no longer
        # the owner, or the subscription is gone.
        "voided_owner_changed": voided_owner_changed,
        "voided_scope_closed": voided_scope_closed,
        # Left open but NOT pushed: never send a card to a leaver.
        "skipped_owner_inactive": skipped_owner_inactive,
        # Never let a capped run read as a drained queue.
        "deferred_over_cap": deferred_over_cap,
        "errors": errors,
        "trace_id": trace_id,
    }
    # Pre-launch hardening 2026-05-16 (LS3 P0): raise when EVERY scope
    # failed or escalation failed AND nothing succeeded. The previous
    # behaviour returned ``errors > 0`` silently → cron status='ok' →
    # failure_notifier stayed quiet for systemic failures (Feishu
    # webhook down, open_id resolver hang). The threshold is "did
    # anything succeed?" — partial errors that still produced real
    # work continue to return ok with a non-zero ``errors`` counter
    # so a single transient row doesn't page the SRE at 3am.
    # Round-1 review follow-up (Review-D P2-1): pre-fix carried a
    # dead ``(1 if escalated >= 0 else 0)`` clause — ``escalated`` is
    # a counter and always >= 0, so the term was always 1. Simplified
    # to the substantive condition.
    # Pre-launch hardening 2026-05-17 (Round-19 audit, finding H):
    # also raise when a majority of scopes fail. The previous gate
    # "anything succeeded" was too lenient — a sweep that pushed 5
    # cards but errored on 95 scopes lands ``status='ok'`` and the
    # operator never notices the partial outage until the affected
    # owners report they didn't receive their card. A ≥50% per-tick
    # failure rate is a real systemic signal worth paging on.
    total_attempted = scopes_prompted + scopes_deduped + errors
    majority_failed = (
        total_attempted >= 4 and errors * 2 >= total_attempted
    )
    if (
        errors > 0
        and scopes_prompted == 0
        and reminded == 0
        and voided_owner_changed == 0
        and voided_scope_closed == 0
    ) or majority_failed:
        raise RuntimeError(
            f"cmdb.semiannual_confirmation: errors={errors} "
            f"scopes_prompted={scopes_prompted} "
            f"scopes_deduped={scopes_deduped} reminded={reminded} — "
            "systemic failure (check Feishu webhook + open_id resolver)."
        )
    return summary


@register_scheduled_job(
    key="cmdb.zero_spend_alert",
    display_name="云订阅零花费告警",
    description=(
        "每月 1 号 09:00（Asia/Shanghai）扫描所有非生产环境（environment≠prod）"
        "的云订阅，找出上自然月花费 = $0 的发送飞书卡片，owner 选择"
        "保留（写理由）/ 我来清理 / 联系 IT 处理。保留过的下个月仍 $0 "
        "会再次告警（不做静默期）。仅 environment ∈ (dev, staging) 进入。"
    ),
)
async def run_zero_spend_alert(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    """Send zero-spend cards for last calendar month's $0-spend scopes.

    The escalation sweep is NOT done here — that's owned by the daily
    semiannual handler so zero-spend prompts don't have to wait up to
    30 days for their first escalation chance.
    """
    from app.core.database import async_session_factory
    from app.domains.cmdb.models import CmdbCloudScope
    from app.domains.cmdb.owner_confirmation_service import (
        OwnerConfirmationService,
    )
    from app.identity.person_models import IdentityPerson

    trace_id = uuid.uuid4().hex

    # Resolve "last calendar month" relative to today. Cron fires on
    # the 1st of each month — "last month" is today.month-1 (rolling
    # back through January → December of previous year).
    # Anchor to Asia/Shanghai (matches the cron's configured timezone +
    # billing_sync helper) — a manual trigger at Jan 1 00:30 Asia/Shanghai
    # (= Dec 31 16:30 UTC) used to compute today.month==12 and scan
    # November's zero-spend, off-by-one. Same TZ fix as PR #443.
    today = datetime.now(ZoneInfo("Asia/Shanghai")).date()
    if today.month == 1:
        billing_year = today.year - 1
        billing_month = 12
    else:
        billing_year = today.year
        billing_month = today.month - 1

    async with async_session_factory() as scan_session:
        scan_service = OwnerConfirmationService(scan_session)
        candidates = await scan_service.list_scopes_with_zero_spend_last_month(
            billing_year=billing_year, billing_month=billing_month,
        )
        scope_ids = [str(s.id) for s in candidates]

    scopes_prompted = 0
    scopes_deduped = 0
    errors = 0
    for scope_id in scope_ids:
        try:
            async with async_session_factory() as scope_session:
                service = OwnerConfirmationService(scope_session)
                scope = await scope_session.get(
                    CmdbCloudScope, uuid.UUID(scope_id),
                )
                if scope is None or scope.deleted_at is not None:
                    continue
                if scope.owner_person_id is None:
                    continue
                owner = await scope_session.get(
                    IdentityPerson, uuid.UUID(scope.owner_person_id),
                )
                if owner is None or owner.deleted_at is not None:
                    continue
                if owner.employment_status in ("left", "leaver_pending"):
                    continue
                _row, snapshot = await service.create_zero_spend_pending(
                    scope, owner,
                    billing_year=billing_year,
                    billing_month=billing_month,
                    trace_id=trace_id,
                )
                await scope_session.commit()
                # Same dedup distinction as semiannual.
                if snapshot.get("deduplicated"):
                    scopes_deduped += 1
                else:
                    scopes_prompted += 1
        except Exception:
            logger.exception(
                "zero-spend alert failed for scope=%s month=%d-%d",
                scope_id, billing_year, billing_month,
            )
            errors += 1

    summary = {
        "scopes_prompted": scopes_prompted,
        "scopes_deduped": scopes_deduped,
        "billing_period": f"{billing_year}-{billing_month:02d}",
        "errors": errors,
        "trace_id": trace_id,
    }
    # Same defence-in-depth as the semiannual handler — raise if
    # nothing succeeded but errors are non-zero (systemic Feishu /
    # resolver outage, not a per-scope hiccup).
    # Pre-launch hardening 2026-05-17 (Round-19 audit, finding H):
    # additionally raise when a majority of scopes fail in a single
    # tick. Pre-fix: 50 errors + 5 prompts landed ``status='ok'`` and
    # nobody noticed the 90% outage until owners complained their
    # card never arrived.
    total_attempted = scopes_prompted + scopes_deduped + errors
    majority_failed = (
        total_attempted >= 4 and errors * 2 >= total_attempted
    )
    if (
        (errors > 0 and scopes_prompted == 0 and len(scope_ids) > 0)
        or majority_failed
    ):
        raise RuntimeError(
            f"cmdb.zero_spend_alert: errors={errors} "
            f"scopes_prompted={scopes_prompted} "
            f"scopes_deduped={scopes_deduped} "
            f"candidates={len(scope_ids)} — systemic failure."
        )
    return summary