"""Give ``cmdb.server_metrics_sync`` rows an explicit 1h timeout_secs.

Revision ID: 460
Revises: 459
Create Date: 2026-08-11

``execute_job`` arms ``asyncio.wait_for`` at ``config['timeout_secs']``,
defaulting to 1800 s. The AliCloud 海外 utilization pull has been running
1 514 s → 1 633 s over the last three weeks (583 instances × 7
CloudMonitor metrics × master + every Resource Directory member account,
14-day hourly window) — 91% of that budget and climbing with the fleet.
Crossing it would kill a nearly-complete multi-minute pull with a
timeout, which is exactly the failure shape the 2026-08-10 connection
drop had.

Reconcile (``sync_jobs_for_integration``) only refreshes display_name /
description on existing rows — it never rewrites ``config`` — so the new
code default in ``_build_metrics_rows`` would otherwise apply to future
integrations only.

Patches ONLY rows that carry no explicit ``timeout_secs`` (an admin value
is left untouched) and is idempotent: on a second run every row already
has the key.
"""
from __future__ import annotations

import json

import sqlalchemy as sa

from alembic import op

revision = "460"
down_revision = "459"
branch_labels = None
depends_on = None

_HANDLER_KEY = "cmdb.server_metrics_sync"
# Keep in sync with ``_METRICS_TIMEOUT_SECS`` in
# app/domains/cmdb/scheduled_jobs_provisioning.py — migrations must not
# import app code.
_TIMEOUT_SECS = 3600


def _rows(bind):
    return bind.execute(
        sa.text(
            "SELECT id, config FROM scheduled_jobs WHERE handler_key = :k"
        ),
        {"k": _HANDLER_KEY},
    ).fetchall()


def _as_dict(config) -> dict | None:
    """``config`` comes back as a JSON string on pymysql, dict on some
    drivers. Anything else (NULL, a JSON scalar) is left alone."""
    if isinstance(config, dict):
        return dict(config)
    if isinstance(config, str | bytes | bytearray):
        try:
            parsed = json.loads(config)
        except ValueError:
            return None
        return dict(parsed) if isinstance(parsed, dict) else None
    return None


def _write(bind, row_id, payload: dict) -> None:
    bind.execute(
        sa.text(
            "UPDATE scheduled_jobs SET config = :c, updated_at = NOW() "
            "WHERE id = :i"
        ),
        {"c": json.dumps(payload), "i": row_id},
    )


def upgrade() -> None:
    bind = op.get_bind()
    for row_id, config in _rows(bind):
        payload = _as_dict(config)
        if payload is None or "timeout_secs" in payload:
            continue
        payload["timeout_secs"] = _TIMEOUT_SECS
        _write(bind, row_id, payload)


def downgrade() -> None:
    bind = op.get_bind()
    for row_id, config in _rows(bind):
        payload = _as_dict(config)
        # Only drop the value this migration could have written; an
        # admin-tuned timeout survives the downgrade.
        if payload is None or payload.get("timeout_secs") != _TIMEOUT_SECS:
            continue
        payload.pop("timeout_secs", None)
        _write(bind, row_id, payload)