"""Scheduled handlers that keep Helix's Bailian mirror current.

Three jobs registered here:

  * ``bailian.workspace_sync`` — pulls workspaces (~04:00 UTC).
  * ``bailian.api_key_sync``   — pulls API keys per workspace (~04:45 UTC).
  * ``bailian.usage_sync``     — pulls last-7-days usage rows (~05:00 UTC).

Each iterates over every enabled AliCloud International cloud
integration and runs the corresponding ``BailianService`` method.
Errors on one integration don't abort the others — the per-integration
error message is accumulated into the summary so the admin can see
partial progress on the scheduled-jobs run-history page.
"""
from __future__ import annotations

import logging
from typing import Any

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

from app.core.exceptions import AppException, format_exception_for_log
from app.scheduled_jobs._seat_sync_helpers import raise_if_total_failure
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)


async def _enabled_bailian_integrations(db: AsyncSession) -> list[str]:
    """Return the integration IDs we should sync.

    Bailian / Model Studio only runs on the 海外 (``alicloud_intl``)
    tenant — the 国内 ``aliyun`` provider stays for CMDB but never has
    Bailian workspaces. :class:`BailianClient` pins to the SG control
    plane (``bailian.ap-southeast-1.aliyuncs.com``).

    We deliberately don't filter on ``slug like '%bailian%'`` — admins
    may run a single AliCloud Intl integration that backs both CMDB
    and Bailian, and the Bailian OpenAPI returns "no workspaces" for
    accounts that haven't enabled Bailian, which is fine.
    """
    from app.domains.bailian.client import _SUPPORTED_PROVIDERS
    from app.integrations.models import Integration

    rows = (await db.execute(
        select(Integration).where(
            Integration.kind == "cloud",
            Integration.provider.in_(_SUPPORTED_PROVIDERS),
            Integration.enabled.is_(True),
        )
    )).scalars().all()
    return [str(r.id) for r in rows]


@register_scheduled_job(
    key="bailian.workspace_sync",
    display_name="阿里云百炼工作空间同步",
    description=(
        "每天拉取所有启用的阿里云集成下的百炼业务空间到 Helix。"
        "基于 ListWorkspaces OpenAPI,可以手动触发。"
    ),
)
async def run_bailian_workspace_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.bailian.service import BailianService

    integration_ids = await _enabled_bailian_integrations(db)
    if not integration_ids:
        return {
            "integrations_processed": 0,
            "errors": [],
            "message": "No enabled AliCloud International cloud integrations.",
        }
    service = BailianService(db)
    per_integration = []
    errors: list[str] = []
    total_upserted = 0
    total_soft_deleted = 0
    for integ_id in integration_ids:
        try:
            # Per-integration SAVEPOINT — sync_workspaces flushes on the
            # shared work_session; an internal flush failure would leave
            # that session in rolled-back-pending state and poison every
            # later integration + the terminal db.flush() this tick. The
            # savepoint rolls back only this integration's writes and keeps
            # the outer transaction usable (per-process SAVEPOINT isolation).
            async with db.begin_nested():
                result = await service.sync_workspaces(integration_id=integ_id)
        except OperationalError:
            # Deadlock (1213) rolls back the whole server-side transaction —
            # no savepoint saves it. Propagate so the fresh-session
            # lock-retry wrapper restarts the idempotent run cleanly.
            raise
        except AppException as exc:
            errors.append(f"{integ_id}: {format_exception_for_log(exc)}")
            continue
        except Exception as exc:  # noqa: BLE001
            # Non-AppException (httpx errors, SDK errors, etc.) used to
            # abort the whole sweep, dropping every healthy integration's
            # sync that tick. Capture and isolate per-integration so a
            # single bad provider tenant doesn't take down the rest;
            # raise_if_total_failure below still fires if EVERY one fails.
            logger.exception(
                "bailian sync unexpected error for integration %s", integ_id,
            )
            errors.append(
                f"{integ_id}: unexpected {format_exception_for_log(exc)}"
            )
            continue
        per_integration.append(result)
        total_upserted += result["upserted"]
        total_soft_deleted += result["soft_deleted"]
    await db.flush()
    summary = {
        "integrations_processed": len(integration_ids),
        "total_upserted": total_upserted,
        "total_soft_deleted": total_soft_deleted,
        "per_integration": per_integration,
        "errors": errors,
    }
    raise_if_total_failure(
        job_key="bailian.workspace_sync",
        summary=summary,
        progress_keys=("total_upserted", "total_soft_deleted"),
    )
    return summary


@register_scheduled_job(
    key="bailian.api_key_sync",
    display_name="阿里云百炼 API Key 同步",
    description=(
        "对所有 BailianWorkspace 调用 ListApiKeys,把 sk-xxx Bailian "
        "API Key 同步到 bailian_api_keys 表。autogrant 创建的 key 也"
        "会在这里被对账（同一 api_key_id 的 row 合并）。需要 workspace_sync "
        "先跑完,所以调度顺序排在它后面。"
    ),
)
async def run_bailian_api_key_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.bailian.service import BailianService

    integration_ids = await _enabled_bailian_integrations(db)
    if not integration_ids:
        return {
            "integrations_processed": 0,
            "errors": [],
            "message": "No enabled AliCloud International cloud integrations.",
        }
    from app.domains.apikey.bailian_adopt import adopt_orphan_bailian_keys

    service = BailianService(db)
    per_integration = []
    errors: list[str] = []
    total_upserted = 0
    total_soft_deleted = 0
    total_adopted = 0
    for integ_id in integration_ids:
        try:
            # Per-integration SAVEPOINT — sync_api_keys flushes on the
            # shared work_session; an internal flush failure would poison
            # the session and take every later integration + the terminal
            # db.flush() with it. Isolate so the outer transaction stays
            # usable (per-process SAVEPOINT isolation).
            async with db.begin_nested():
                result = await service.sync_api_keys(integration_id=integ_id)
        except OperationalError:
            # Deadlock (1213) — propagate to the fresh-session lock-retry.
            raise
        except AppException as exc:
            errors.append(f"{integ_id}: {format_exception_for_log(exc)}")
            continue
        except Exception as exc:  # noqa: BLE001
            # Non-AppException (httpx errors, SDK errors, etc.) used to
            # abort the whole sweep, dropping every healthy integration's
            # sync that tick. Capture and isolate per-integration so a
            # single bad provider tenant doesn't take down the rest;
            # raise_if_total_failure below still fires if EVERY one fails.
            logger.exception(
                "bailian sync unexpected error for integration %s", integ_id,
            )
            errors.append(
                f"{integ_id}: unexpected {format_exception_for_log(exc)}"
            )
            continue
        # Adopt console-created keys that already have an owner pinned
        # into the canonical api_key_assignments registry so every real,
        # owned key is visible to its owner + the admin registry. Pure
        # DB reconciliation (no external calls), idempotent, so a failure
        # here must NOT lose the sync result — isolate + record it.
        #
        # The handler runs on ONE shared work_session that the executor
        # commits once after we return (scheduled_jobs/service.py). A bare
        # except that swallows an adopt flush error would leave that
        # session in a failed-transaction state, poisoning the terminal
        # ``db.flush()`` and every *already-synced* integration's writes
        # this tick. Wrap the adopt sub-work in a SAVEPOINT so a failure
        # rolls back only the adoption and the outer transaction (with the
        # prior sync writes) stays usable. Per-row isolation lives inside
        # ``adopt_orphan_bailian_keys``.
        try:
            async with db.begin_nested():
                adopt_result = await adopt_orphan_bailian_keys(
                    db, integration_id=integ_id,
                )
            result["adopted"] = adopt_result["adopted"]
            result["adopt_skipped_no_owner"] = (
                adopt_result["skipped_no_owner"]
            )
            total_adopted += adopt_result["adopted"]
        except Exception as exc:  # noqa: BLE001
            logger.exception(
                "bailian adopt failed for integration %s", integ_id,
            )
            errors.append(
                f"{integ_id}: adopt {format_exception_for_log(exc)}"
            )
        per_integration.append(result)
        total_upserted += result["upserted"]
        total_soft_deleted += result["soft_deleted"]
    await db.flush()
    summary = {
        "integrations_processed": len(integration_ids),
        "total_upserted": total_upserted,
        "total_soft_deleted": total_soft_deleted,
        "total_adopted": total_adopted,
        "per_integration": per_integration,
        "errors": errors,
    }
    raise_if_total_failure(
        job_key="bailian.api_key_sync",
        summary=summary,
        progress_keys=("total_upserted", "total_soft_deleted"),
    )
    return summary


@register_scheduled_job(
    key="bailian.usage_sync",
    display_name="阿里云百炼调用量同步",
    description=(
        "每天拉取最近 7 天的百炼模型调用量 + 成本明细,upsert 到 "
        "bailian_usage_daily 表。可作为成本看板的数据源。"
    ),
)
async def run_bailian_usage_sync(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.bailian.service import BailianService

    integration_ids = await _enabled_bailian_integrations(db)
    if not integration_ids:
        return {
            "integrations_processed": 0,
            "errors": [],
            "message": "No enabled AliCloud International cloud integrations.",
        }
    # Cap days to prevent admin-triggered unbounded Aliyun BSS API calls
    # that would exhaust rate limits and hold the connection pool.
    days = min(int(config.get("days") or 7), 90)
    service = BailianService(db)
    per_integration = []
    errors: list[str] = []
    total_upserted = 0
    for integ_id in integration_ids:
        try:
            # Per-integration SAVEPOINT — sync_usage_daily flushes on the
            # shared work_session; an internal flush failure would poison
            # the session and take every later integration + the terminal
            # db.flush() with it. Isolate so the outer transaction stays
            # usable (per-process SAVEPOINT isolation).
            async with db.begin_nested():
                result = await service.sync_usage_daily(
                    integration_id=integ_id, days=days,
                )
        except OperationalError:
            # Deadlock (1213) — propagate to the fresh-session lock-retry.
            raise
        except AppException as exc:
            errors.append(f"{integ_id}: {format_exception_for_log(exc)}")
            continue
        except Exception as exc:  # noqa: BLE001
            # Non-AppException (httpx errors, SDK errors, etc.) used to
            # abort the whole sweep, dropping every healthy integration's
            # sync that tick. Capture and isolate per-integration so a
            # single bad provider tenant doesn't take down the rest;
            # raise_if_total_failure below still fires if EVERY one fails.
            logger.exception(
                "bailian sync unexpected error for integration %s", integ_id,
            )
            errors.append(
                f"{integ_id}: unexpected {format_exception_for_log(exc)}"
            )
            continue
        per_integration.append(result)
        total_upserted += result["upserted"]
    await db.flush()
    summary = {
        "integrations_processed": len(integration_ids),
        "total_upserted": total_upserted,
        "per_integration": per_integration,
        "errors": errors,
    }
    raise_if_total_failure(
        job_key="bailian.usage_sync",
        summary=summary,
        progress_keys=("total_upserted",),
    )
    return summary