"""Daily connection test for every enabled integration.

The gap this fills, stated plainly: ``integrations.last_test_status``
was written **only** by an admin clicking "test connection". Nothing
periodic ever touched it. So the platform's answer to "are our
credentials still good" was whatever somebody last checked by hand,
possibly months ago — and an expired token announced itself as a rising
count of unrelated-looking failures across provisioning, identity sync
and notifications, each domain faithfully reporting a symptom.

Design notes that matter more than the code:

**Only enabled integrations are probed.** A disabled integration is
one somebody turned off; testing it would spend a vendor round trip to
learn something nobody is waiting for. The exception collector still
reports a disabled integration whose *last* test failed, at low
severity, so "turned off and forgotten" stays visible.

**Every probe is savepoint-isolated.** One driver raising must not cost
the other forty their verdict, and must not leave the session unusable.

**Failures are not notified from here.** This cron's job is to
establish the fact; the exception ledger's job is to route it. A cron
that both measures and alerts is how one expired credential became
four notification cards in this codebase before.

**Probes are spread out, not fired at once.** Forty simultaneous vendor
calls is a self-inflicted rate limit. A small delay between probes
keeps the run polite; the run is nightly and has no deadline.
"""
from __future__ import annotations

import asyncio
import logging
import uuid
from typing import Any

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

from app.core.txn import best_effort
from app.integrations.models import Integration
from app.integrations.service import IntegrationService
from app.scheduled_jobs.registry import register_scheduled_job
from app.scheduled_jobs.slo import SLOTarget

logger = logging.getLogger(__name__)

#: Seconds between probes. Small, but enough that a shared vendor
#: (several cloud accounts behind one provider) is not hit in a burst.
_PROBE_SPACING_SECONDS = 1.0

#: Upper bound on probes per run. A guard, not a policy: the estate is
#: dozens of integrations, so this only ever trips if something has gone
#: very wrong with the estate itself, and it trips loudly in the summary
#: rather than silently truncating.
_MAX_PROBES = 200


@register_scheduled_job(
    key="integrations.health_check",
    display_name="集成 · 每日连接体检",
    description=(
        "每日 01:20 UTC 对每个**启用**的 integrations 行调用其驱动的 "
        "test_connection(),把结果写回 last_tested_at / last_test_status / "
        "last_test_message。此前这三列只在管理员手点「测试连接」时才被写过 —— "
        "所以凭证过期在集成这一层永远不会被发现,只会表现为 provisioning / "
        "identity_sync / 通知出箱三个下游域各自报出一堆看起来无关的失败。"
        "本任务只负责**确立事实**,不发任何通知:告警与派工由自动化异常台"
        "(automation_exception.collect)统一承担,避免一次凭证过期又变成四张卡。"
        "每个探测独立 SAVEPOINT,一个驱动抛异常不影响其余;探测之间间隔 1 秒,"
        "避免对同一厂商形成突发调用。停用的集成不探测(没人在等它),但它上一次"
        "失败的结论仍会被异常台以低严重度收录。"
    ),
    slo=SLOTarget(tier="P2"),
    # Dozens of sequential vendor round trips — this holds its DB
    # connection for the length of the whole estate's probes.
    weight="heavy",
)
async def run_integration_health_check(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    spacing = float(config.get("probe_spacing_seconds", _PROBE_SPACING_SECONDS))
    max_probes = int(config.get("max_probes", _MAX_PROBES))

    rows = (
        await db.execute(
            select(Integration)
            .where(Integration.enabled.is_(True))
            .order_by(Integration.kind, Integration.slug),
        )
    ).scalars().all()

    service = IntegrationService(db)
    trace_id = str(uuid.uuid4())

    probed = 0
    ok_count = 0
    failed_count = 0
    errored: list[dict[str, str]] = []
    truncated = False

    for index, row in enumerate(rows):
        if probed >= max_probes:
            truncated = True
            break
        if index and spacing > 0:
            await asyncio.sleep(spacing)

        slug = row.slug
        async with best_effort(db, label=f"probe integration {slug}", log=logger) as step:
            result = await service.test_connection(
                slug,
                trace_id=trace_id,
                actor_user_id=None,
                actor_type="system",
            )
            # Read the primitive out INSIDE the block. After a savepoint
            # rollback the ORM objects touched in here are expired, and
            # reading one outside would raise MissingGreenlet on a lazy
            # refresh — so nothing below this line may touch ``row`` or
            # ``result``.
            probe_ok = bool(result.ok)
        probed += 1
        if not step.ok:
            errored.append({"slug": slug, "error": step.reason() or "unknown"})
            continue
        if probe_ok:
            ok_count += 1
        else:
            failed_count += 1

    return {
        "enabled_integrations": len(rows),
        "probed": probed,
        "ok": ok_count,
        "failed": failed_count,
        # A driver that blew up is neither ok nor failed: we learned
        # nothing about the credential. Kept separate so a rise here
        # reads as "our probe is broken", not "their vendor is down".
        "probe_errors": len(errored),
        "probe_error_detail": errored[:20],
        "truncated": truncated,
    }