"""Auto-provision ``scheduled_jobs`` rows for cloud integrations.

Each (cloud integration × capability) combination gets its own
``scheduled_jobs`` row pointing at the generic ``cmdb.resource_sync``
handler. When the admin creates a cloud integration or flips which
capabilities it should sync, this module diffs the desired set against
what's in the DB and adds / removes rows accordingly. When an
integration is deleted, its rows are removed too.

Only cloud integrations (``kind='cloud'``) are touched. Other kinds —
HRIS, chat_approval, email, software_provisioning — continue to own
their own scheduled_jobs rows manually.

The row naming convention is ``cmdb.sync.<slug>.<capability>`` so the
admin UI at ``/admin/scheduled-jobs`` sorts all resource-syncs for one
tenant together. The per-capability cron is spread across the hour
(``minute = hash(slug+cap) % 60``) so a fleet-wide 100 accounts × 6
capabilities doesn't stampede the top-of-hour.
"""
from __future__ import annotations

import hashlib
import logging
import uuid
from typing import Any

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

from app.domains.cmdb.providers.base import (
    ProviderCapability,
    get_provider_class,
)
from app.integrations.models import Integration
from app.scheduled_jobs.models import ScheduledJob

logger = logging.getLogger(__name__)


HANDLER_KEY = "cmdb.resource_sync"
BILLING_HANDLER_KEY = "cmdb.billing_sync"

# Cloud-lifecycle scheduled handlers that run per cloud integration
# (PR 2026-04-23). The access_expiry + scope_inactivity sweeps are
# org-wide and seeded by migration 052; they're NOT touched here.
SCOPES_HANDLER_KEY = "cmdb.scopes_sync"
ROLES_HANDLER_KEY = "cmdb.roles_sync"
GRANTS_HANDLER_KEY = "cmdb.grants_sync"
BILLING_DAILY_HANDLER_KEY = "cmdb.billing_daily_sync"
COST_FACTS_HANDLER_KEY = "cmdb.cost_facts_sync"

_LIFECYCLE_HANDLERS = (
    SCOPES_HANDLER_KEY,
    ROLES_HANDLER_KEY,
    GRANTS_HANDLER_KEY,
    BILLING_DAILY_HANDLER_KEY,
    COST_FACTS_HANDLER_KEY,
)

# Cloud-efficiency handlers (2026-07-20). Provisioned per cloud
# integration whose provider class declares the ``metrics`` capability
# (AWS / Aliyun / Azure / GCP). ``server_metrics_sync`` pulls utilization
# from the cloud's monitoring API into cmdb_server_metrics_daily;
# ``efficiency_compute`` turns that + cost facts into per-server verdicts.
# Like ``_LIFECYCLE_HANDLERS``, these keys + names must stay aligned
# across ``_build_metrics_rows``, the ``handler_key.in_`` filter, and the
# name-scan list in ``sync_jobs_for_integration`` or reconcile DELETEs the
# rows every boot.
SERVER_METRICS_HANDLER_KEY = "cmdb.server_metrics_sync"
EFFICIENCY_HANDLER_KEY = "cmdb.efficiency_compute"

_METRICS_HANDLERS = (
    SERVER_METRICS_HANDLER_KEY,
    EFFICIENCY_HANDLER_KEY,
)

# Hard per-run timeout for the utilization pull, overriding
# ``execute_job``'s 1800 s default. The cloud-monitoring leg is the long
# pole and scales with the fleet: AliCloud 海外 sat at 1 633 s of 1 800 s
# on 2026-08-10 (583 instances × 7 CloudMonitor metrics × master + every
# Resource Directory member), up from 1 514 s three weeks earlier. Timing
# out a nearly-complete pull is a pure loss — the run has already spent
# the vendor API quota. Migration 460 backfills this onto existing rows
# (reconcile never rewrites ``config``); keep the two values in sync.
_METRICS_TIMEOUT_SECS = 3600


def _capability_label(cap: str) -> str:
    return {
        "servers": "机器",
        "domains": "域名",
        "dns_records": "DNS 记录",
        "load_balancers": "负载均衡",
        "cdn": "CDN",
        "object_storage": "对象存储",
        "iam": "IAM 用户",
        "applications": "应用",
    }.get(cap, cap)


def _offset_minutes(key: str) -> int:
    """Deterministic but spread-out cron minute per (integration, cap)."""
    return int(hashlib.sha256(key.encode()).hexdigest(), 16) % 60


# ── Per-capability sync cadence tiers (2026-07-19) ────────────────────
# The original default scheduled EVERY capability of EVERY cloud account
# hourly. Prod telemetry (3-day window, 2026-07-19): 27 enabled
# resource_sync jobs × 24 runs/day = ~1 600 ok-runs with **100% zero
# changes written** (upserted=0 deleted=0 on every single run) — cloud
# inventory at a ~500-person company changes a few times a week, not
# hourly. That fleet was also the exact cron cluster that drained the
# DB pool in the 2026-07-19 incident. Tiered defaults:
#
#   * HOT (every 4h)  — resources that gate people/cost flows and are
#     the likeliest to change: servers, iam, applications.
#   * COLD (every 12h) — near-static registrar/network/edge objects:
#     domains, dns_records, certificates, cdn, object_storage,
#     load_balancers. (certificates additionally get probed live daily
#     by ``cmdb.cert_probe_sweep``, so 12h discovery lag is invisible
#     to expiry alerting.)
#
# The hour PHASE is hashed per job (like the minute) so tiers don't
# re-create a thundering herd at 00:00 / 12:00. Admins can still tune
# any row's cadence in /admin/scheduled-jobs (reconcile never touches
# cadence on existing rows — see sync_jobs_for_integration) and can
# always hit 立即执行 / the sync-now API for on-demand freshness.
# Migration 418 rewrites EXISTING rows that still carry the old hourly
# code default; admin-customised rows are left untouched.
_CAPABILITY_SYNC_PERIOD_HOURS: dict[str, int] = {
    "servers": 4,
    "iam": 4,
    "applications": 4,
    "domains": 12,
    "dns_records": 12,
    "certificates": 12,
    "cdn": 12,
    "object_storage": 12,
    "load_balancers": 12,
}
_DEFAULT_SYNC_PERIOD_HOURS = 4


def _capability_cron(name: str, capability: str) -> str:
    """Tiered default cron for one (integration × capability) job.

    ``{hashed_minute} {hashed_phase}-23/{period} * * *`` — deterministic
    per job name, spread across both the hour and the minute.
    """
    minute = _offset_minutes(name)
    period = _CAPABILITY_SYNC_PERIOD_HOURS.get(
        capability, _DEFAULT_SYNC_PERIOD_HOURS,
    )
    phase = int(hashlib.sha256(f"{name}|phase".encode()).hexdigest(), 16) % period
    return f"{minute} {phase}-23/{period} * * *"


# ``metrics`` is a capability of its own dedicated handler
# (``cmdb.server_metrics_sync``), NOT the generic ``cmdb.resource_sync``
# inventory engine — exclude it here so ``resolve_sync_capabilities``
# never schedules a resource_sync row that would try to fetch utilization
# as if it were an inventory list.
_SYNC_CAPABILITY_VALUES: frozenset[str] = frozenset(
    c.value for c in ProviderCapability if c != ProviderCapability.METRICS
)


def resolve_sync_capabilities(integration: Integration) -> list[str]:
    """Capabilities for this integration that the periodic sync engine
    actually knows how to fetch.

    Rules:
      * Start from the admin-configured ``capabilities`` list when set —
        admins explicitly opt out of certain resource types (e.g. a
        read-only AWS role that can't see S3).
      * Otherwise fall back to whatever the underlying provider class
        declares via ``supported_capabilities()``.
      * Filter to the ``ProviderCapability`` enum: action-only entries
        like ``user_revoke`` (used at offboarding-revoke time, not by
        the resource sync engine) are dropped here instead of being
        scheduled as cron rows that fire hourly only to land as
        "unknown capability"-skipped runs.
    """
    if integration.capabilities:
        candidates = [c for c in integration.capabilities if c]
    else:
        try:
            provider_cls = get_provider_class(integration.provider)
        except ValueError:
            return []
        candidates = [c.value for c in provider_cls.supported_capabilities()]
    return [c for c in candidates if c in _SYNC_CAPABILITY_VALUES]


def _build_row(
    integration: Integration,
    capability: str,
) -> dict[str, Any]:
    slug = integration.slug or str(integration.id)
    name = f"cmdb.sync.{slug}.{capability}"
    display_name = f"{integration.name} · {_capability_label(capability)}同步"
    period = _CAPABILITY_SYNC_PERIOD_HOURS.get(
        capability, _DEFAULT_SYNC_PERIOD_HOURS,
    )
    description = (
        f"按能力拆分的云资源同步:{integration.name}({integration.provider})"
        f" → {capability}。每个 (云账号 × 资源类型) 都是独立 scheduled_job,"
        f"默认每 {period} 小时一跑(分钟/相位按名称哈希铺开避峰),"
        "管理员可以在此页单独改 cadence / 暂停 / 立即执行 / 查历史;"
        "需要即时新鲜度用「立即执行」。"
    )
    return {
        "name": name,
        "display_name": display_name,
        "description": description,
        "handler_key": HANDLER_KEY,
        "trigger_type": "cron",
        "cron_expr": _capability_cron(name, capability),
        "interval_seconds": None,
        "timezone": "Asia/Shanghai",
        "enabled": True,
        "config": {
            "integration_id": str(integration.id),
            "capability": capability,
        },
    }


def _build_billing_row(integration: Integration) -> dict[str, Any]:
    slug = integration.slug or str(integration.id)
    name = f"cmdb.billing.{slug}"
    display_name = f"{integration.name} · 月度账单抓取"
    description = (
        "【账单四件套·① 月汇总】每月 2 号凌晨拉取【上月】账单汇总"
        "(各 provider 的 get_monthly_billing:AWS Cost Explorer 月维度 / "
        "Aliyun·Tencent·UCloud BSS 月账单 / Azure·GCP 月成本),写入 "
        "cmdb_billing_monthly —— 只存「账号 × 月」一行总额。供 "
        "/admin/cloud/overview?tab=billing 的多云月度聚合报表 + 环比卡片用。"
        "粒度最粗,不含资源/服务/人(那是 cmdb_cost_facts)。与 ② 日趋势、"
        "③ 明细、④ 月结重拉 各司其职:本任务只负责跨云的月度大盘。"
    )
    # Spread across day 2 hour 2:00-2:59 to avoid thundering herd.
    minute = _offset_minutes(name)
    return {
        "name": name,
        "display_name": display_name,
        "description": description,
        "handler_key": BILLING_HANDLER_KEY,
        "trigger_type": "cron",
        "cron_expr": f"{minute} 2 2 * *",
        "interval_seconds": None,
        "timezone": "Asia/Shanghai",
        "enabled": True,
        "config": {"integration_id": str(integration.id)},
    }


def _build_lifecycle_rows(integration: Integration) -> list[dict[str, Any]]:
    """Build the five lifecycle scheduled_jobs rows for this tenant.

    All five run per-integration — scopes (04:00), role catalog
    (04:30), reverse grants pull (05:00), daily scope-level billing
    rollup (06:00), daily line-item cost facts (04:xx). Cron minutes
    are hashed per row to avoid thundering herd across tenants.

    Whenever a new lifecycle handler ships, add it here AND to
    ``_LIFECYCLE_HANDLERS`` AND to the ``lifecycle_names`` list in
    ``sync_jobs_for_integration`` — those three lists must stay
    aligned, otherwise reconcile will treat the row as unrecognised
    and DELETE it on every boot.
    """
    slug = integration.slug or str(integration.id)
    out: list[dict[str, Any]] = []

    def _row(name: str, display: str, desc: str, handler: str, cron: str) -> dict[str, Any]:
        return {
            "name": name,
            "display_name": display,
            "description": desc,
            "handler_key": handler,
            "trigger_type": "cron",
            "cron_expr": cron,
            "interval_seconds": None,
            "timezone": "Asia/Shanghai",
            "enabled": True,
            "config": {"integration_id": str(integration.id)},
        }

    scopes_name = f"cmdb.scopes.{slug}"
    out.append(_row(
        scopes_name,
        f"{integration.name} · 子账号/订阅同步",
        (
            f"每日 04:00 抓取 {integration.name} ({integration.provider}) "
            "下的子账号列表,写入 cmdb_cloud_scopes。"
        ),
        SCOPES_HANDLER_KEY,
        f"{_offset_minutes(scopes_name)} 4 * * *",
    ))
    roles_name = f"cmdb.roles.{slug}"
    out.append(_row(
        roles_name,
        f"{integration.name} · 角色目录同步",
        (
            "每日 04:30 抓取每个 scope 下的 role 目录(预定义 + 自定义),"
            "写入 cmdb_cloud_roles;申请 flow 的 role 下拉直接读此表。"
        ),
        ROLES_HANDLER_KEY,
        f"{_offset_minutes(roles_name)} 4 * * *",
    ))
    grants_name = f"cmdb.grants.{slug}"
    out.append(_row(
        grants_name,
        f"{integration.name} · 云端权限同步",
        (
            "每日 05:00 反向拉取云上已有的 role/policy 绑定(Azure "
            "roleAssignments、GCP IAM bindings、AWS Identity Center "
            "assignments、Aliyun/Tencent RAM-CAM 用户策略),写入 "
            "cmdb_cloud_access_grants 并标注 source='discovered';"
            "scope 无 owner 且只有一人持有 owner 等价角色时,自动"
            "把该人填入 scope.owner_person_id。"
        ),
        GRANTS_HANDLER_KEY,
        f"{_offset_minutes(grants_name)} 5 * * *",
    ))
    billing_daily_name = f"cmdb.billing_daily.{slug}"
    out.append(_row(
        billing_daily_name,
        f"{integration.name} · 每日账单抓取",
        (
            "【账单四件套·② 日趋势 + 异常告警】每日 06:00 抓取【昨天】的 "
            "scope(子账号)级日总额,写入 cmdb_cloud_billing_daily,并对"
            "昨天 vs 过去 7 日均值做日环比异常检测、超阈值发飞书告警。"
            "粒度到「账号 × scope × 日」总额,用于花费趋势线 + 突增预警;"
            "不含资源/服务/人粒度(那是 ③ cmdb_cost_facts)。与 ① 月汇总"
            "互为粗细两档:① 给月度大盘,② 给每日趋势 + 实时告警。"
        ),
        BILLING_DAILY_HANDLER_KEY,
        f"{_offset_minutes(billing_daily_name)} 6 * * *",
    ))
    cost_facts_name = f"cmdb.cost_facts.{slug}"
    out.append(_row(
        cost_facts_name,
        f"{integration.name} · 成本明细抓取",
        (
            "【账单四件套·③ 明细(最细粒度)】每日 04:xx(分钟按 slug 哈希"
            "铺开避峰)拉取【昨天起滚动 3 天】窗口的 line-item 成本明细"
            "(Azure Cost Management ResourceId / Aliyun BSS "
            "DescribeInstanceBill / AWS Cost Explorer "
            "GetCostAndUsageWithResources / Tencent DescribeBillDetail / "
            "GCP BigQuery export),写入 cmdb_cost_facts —— 粒度到「日 × "
            "scope × 资源 × 服务 × 标签」。喂 FinOps 账单页的多维下钻 / "
            "归属人 / 资源排行 / 成本分摊 / 预测异常,是归属(谁花的)的唯一"
            "数据源。⚠️ 滚动 3 天窗口只能追平最近 3 天的厂商修订;月末 "
            "finalize 的晚到修订(日累积 S3/ECR、未打标签用量)靠 ④ 单例任务 "
            "cmdb.cost_facts_monthclose(每月 5 号整月重拉)对账到分。驱动未"
            "实现 list_cost_facts 时返回 not_supported 不报错。回填窗口 "
            "config={\"start_date\":\"YYYY-MM-DD\",\"end_date\":\"YYYY-MM-DD\"}。"
        ),
        COST_FACTS_HANDLER_KEY,
        f"{_offset_minutes(cost_facts_name)} 4 * * *",
    ))
    return out


def _provider_supports_metrics(provider_name: str) -> bool:
    """Metrics rows are provisioned only for provider classes that declare
    the ``metrics`` capability (AWS / Aliyun / Azure / GCP). Others (Tencent
    / UCloud / Ksyun) get no utilization jobs until an adapter ships."""
    try:
        provider_cls = get_provider_class(provider_name)
    except ValueError:
        return False
    return ProviderCapability.METRICS in provider_cls.supported_capabilities()


def _build_metrics_rows(integration: Integration) -> list[dict[str, Any]]:
    """The two per-integration cloud-efficiency rows: daily utilization
    pull (03:xx) then efficiency recompute (07:xx — after cost_facts at
    04:xx and billing_daily at 06:00 so savings can be attributed)."""
    slug = integration.slug or str(integration.id)
    metrics_name = f"cmdb.server_metrics.{slug}"
    eff_name = f"cmdb.efficiency.{slug}"

    def _row(
        name: str,
        display: str,
        desc: str,
        handler: str,
        cron: str,
        *,
        timeout_secs: int | None = None,
    ) -> dict[str, Any]:
        config: dict[str, Any] = {"integration_id": str(integration.id)}
        if timeout_secs is not None:
            config["timeout_secs"] = timeout_secs
        return {
            "name": name,
            "display_name": display,
            "description": desc,
            "handler_key": handler,
            "trigger_type": "cron",
            "cron_expr": cron,
            "interval_seconds": None,
            "timezone": "Asia/Shanghai",
            "enabled": True,
            "config": config,
        }

    return [
        _row(
            metrics_name,
            f"{integration.name} · 利用率采集",
            (
                f"【云成本优化·① 利用率采集】每日 03:xx 从 {integration.name} "
                f"({integration.provider}) 的云监控(AWS CloudWatch / 阿里云 "
                "CloudMonitor / Azure Monitor / GCP Cloud Monitoring)拉取"
                "过去 14 天各实例的 CPU/网络(内存/磁盘需装 agent,尽力而为)"
                "小时级利用率,按日聚合写入 cmdb_server_metrics_daily(保留 90 "
                "天)。只读监控 API,不动云资源。"
            ),
            SERVER_METRICS_HANDLER_KEY,
            f"{_offset_minutes(metrics_name)} 3 * * *",
            timeout_secs=_METRICS_TIMEOUT_SECS,
        ),
        _row(
            eff_name,
            f"{integration.name} · 效率评估",
            (
                "【云成本优化·② 效率评估】每日 07:xx 用 cmdb_server_metrics_daily "
                "的滚动窗口 + cmdb_cost_facts 的单资源成本 + 规则引擎,重算每台"
                "机器的效率画像(闲置/超配/健康 + 建议 + 预估月省)写入 "
                "cmdb_server_efficiency。排在利用率采集(03)、成本明细(04)、"
                "日账单(06)之后,确保口径齐备。纯只读洞察。"
            ),
            EFFICIENCY_HANDLER_KEY,
            f"{_offset_minutes(eff_name)} 7 * * *",
        ),
    ]


def _provider_supports_billing(provider_name: str) -> bool:
    """A monthly billing scheduled_job row is provisioned only for
    provider classes that ship a real native ``get_monthly_billing``.

    Every live cloud provider (aws / aliyun / alicloud_intl / azure /
    gcp / tencent / ucloud) implements it against the vendor's
    billing API. Stub-only providers like ``ksyun`` (frozen historical
    tenant with no live API) deliberately don't, so their billing row
    is skipped instead of running a job that would always 500.

    Validate the provider name against the registry so a typo doesn't
    provision an orphan row.
    """
    try:
        provider_cls = get_provider_class(provider_name)
    except ValueError:
        return False
    return callable(getattr(provider_cls, "get_monthly_billing", None))


# 为什么账单两条 cron(``billing_daily`` / ``cost_facts``)**不**按能力闸:
#
# 2026-08-15 试过,当天回退。动机是合理的:``tencent-dnspod`` 只声明
# ``dns_records``(0 scope / 0 facts),却挂着一条永远禁用的 cost_facts
# cron,而云成本页的健康横幅正是按它告警的。但按「库存能力里有没有会花钱
# 的资源」来闸,会把**库存能力**和**计费**这两件正交的事耦合起来:
#
#   ``sync_jobs_for_integration`` 是 diff-and-apply —— 不在 desired 里的行
#   会被 **DELETE**。于是管理员把某个集成的能力收窄(比如临时改成
#   ``["iam"]``)就会**静默删掉它的账单同步**。对 ``aws-primary``
#   (capabilities=None、全公司最大花费方)这类集成,后果是账单从此不再
#   摄取,而删掉的恰好又是那条本来会在横幅上告警的 cron —— 静默失效。
#   `test_sync_diffs_when_capabilities_change` 当场抓到(掉了 3 行而非 1 行)。
#
# 月度账单行走的是 ``_provider_supports_billing``(provider 类有没有实现
# ``get_monthly_billing``),**不看能力**,正是因为计费与库存能力无关。
# 而 tencent-dnspod 与 tencent-primary 同一个 provider 类,provider 维度
# 又区分不开它俩。结论:没有既安全又能挡住 dnspod 的判据,那就**不挡** ——
# 一条多余的禁用 cron 只是横幅上多一行字(文案已改为真实语义),而误删
# 一条账单 cron 是静默丢数据。宁可诚实地粗,不要精确地错。


async def sync_jobs_for_integration(
    db: AsyncSession, integration: Integration,
) -> dict[str, Any]:
    """Diff-and-apply scheduled_jobs rows for one cloud integration.

    Returns a summary dict so callers can log / audit. Safe to call on
    every create+update; it only issues INSERTs / DELETEs when the set
    actually differs. Never flips the ``enabled`` flag on an existing
    row — admins who disabled a row manually keep their override.
    """
    if integration.kind != "cloud":
        return {"skipped": "not cloud integration"}

    desired: dict[str, dict[str, Any]] = {}
    if _provider_supports_billing(integration.provider):
        billing_row = _build_billing_row(integration)
        desired[billing_row["name"]] = billing_row
    for cap in resolve_sync_capabilities(integration):
        row = _build_row(integration, cap)
        desired[row["name"]] = row
    # Lifecycle rows (scopes / roles / billing_daily). Always seeded —
    # drivers that can't supply a method raise NotImplementedError and
    # the handler returns a skipped status, which is cheap.
    for row in _build_lifecycle_rows(integration):
        desired[row["name"]] = row
    # Efficiency rows — only for providers with a metrics adapter.
    if _provider_supports_metrics(integration.provider):
        for row in _build_metrics_rows(integration):
            desired[row["name"]] = row

    # Pre-compute the exact row names we care about so the query only
    # fetches what we'd match in Python anyway — avoids pulling every
    # resource_sync row in the DB at fleet scale.
    candidate_names = list(desired.keys())
    slug = integration.slug or integration.id
    # Legacy names may exist from a prior capability that was removed;
    # scan with a name LIKE filter so we still pick up rows scheduled
    # for deletion on this diff.
    name_prefix = f"cmdb.sync.{slug}."
    billing_name = f"cmdb.billing.{slug}"
    lifecycle_names = [
        f"cmdb.scopes.{slug}",
        f"cmdb.roles.{slug}",
        f"cmdb.grants.{slug}",
        f"cmdb.billing_daily.{slug}",
        f"cmdb.cost_facts.{slug}",
    ]
    metrics_names = [
        f"cmdb.server_metrics.{slug}",
        f"cmdb.efficiency.{slug}",
    ]
    conditions = [
        ScheduledJob.name.like(f"{name_prefix}%"),
        ScheduledJob.name == billing_name,
        ScheduledJob.name.in_(lifecycle_names),
        ScheduledJob.name.in_(metrics_names),
    ]
    if candidate_names:
        conditions.insert(0, ScheduledJob.name.in_(candidate_names))
    from sqlalchemy import or_
    existing_rows = (
        await db.execute(
            select(ScheduledJob).where(
                ScheduledJob.handler_key.in_(
                    [
                        HANDLER_KEY, BILLING_HANDLER_KEY,
                        *_LIFECYCLE_HANDLERS, *_METRICS_HANDLERS,
                    ],
                ),
                or_(*conditions),
            )
        )
    ).scalars().all()
    current_integration_id = str(integration.id)
    foreign_owner_ids = {
        str(owner_id)
        for existing_job in existing_rows
        if (owner_id := (existing_job.config or {}).get("integration_id"))
        and str(owner_id) != current_integration_id
    }
    valid_foreign_owner_ids: list[uuid.UUID] = []
    for owner_id in foreign_owner_ids:
        try:
            valid_foreign_owner_ids.append(uuid.UUID(owner_id))
        except ValueError:
            # Malformed legacy config cannot identify a live owner and is
            # therefore treated like an orphan.  Do not let a bad JSON value
            # abort every connector's boot reconcile.
            continue
    live_foreign_owner_ids = (
        {
            str(owner_id)
            for owner_id in (
                await db.scalars(
                    select(Integration.id).where(
                        Integration.id.in_(valid_foreign_owner_ids)
                    )
                )
            ).all()
        }
        if valid_foreign_owner_ids
        else set()
    )

    existing_by_name: dict[str, ScheduledJob] = {}
    adopted_orphan_names: list[str] = []
    for existing_job in existing_rows:
        owner_id = str((existing_job.config or {}).get("integration_id") or "")
        if owner_id == current_integration_id:
            existing_by_name[existing_job.name] = existing_job
            continue
        if existing_job.name not in desired:
            continue
        if owner_id and owner_id in live_foreign_owner_ids:
            raise RuntimeError(
                f"scheduled job name {existing_job.name!r} is owned by live integration "
                f"{owner_id}"
            )
        # A cloud connector may be deleted/re-created with the same unique
        # slug while legacy scheduled rows survive an older deletion path.
        # Reuse that row instead of violating uq_scheduled_jobs_name.  Keep
        # cadence/enabled admin overrides, replacing only the dead owner id.
        existing_job.config = {
            **(
                existing_job.config
                or desired[existing_job.name].get("config")
                or {}
            ),
            "integration_id": current_integration_id,
        }
        existing_by_name[existing_job.name] = existing_job
        adopted_orphan_names.append(existing_job.name)

    to_add = [v for k, v in desired.items() if k not in existing_by_name]
    to_remove = [r for k, r in existing_by_name.items() if k not in desired]
    to_update_display: list[ScheduledJob] = []

    # Refresh display-name / description on existing rows so renaming
    # an integration propagates to the UI. Don't touch cadence or
    # enabled — those are admin-owned.
    for key, spec in desired.items():
        existing_row = existing_by_name.get(key)
        if existing_row is None:
            continue
        if (
            existing_row.display_name != spec["display_name"]
            or existing_row.description != spec["description"]
        ):
            existing_row.display_name = spec["display_name"]
            existing_row.description = spec["description"]
            to_update_display.append(existing_row)

    for spec in to_add:
        db.add(ScheduledJob(id=uuid.uuid4(), **spec))
    for removed_job in to_remove:
        await db.delete(removed_job)
    await db.flush()

    return {
        "integration_id": str(integration.id),
        "added": [s["name"] for s in to_add],
        "removed": [r.name for r in to_remove],
        "display_name_refreshed": [r.name for r in to_update_display],
        "adopted_orphan_names": adopted_orphan_names,
        "desired_count": len(desired),
    }


async def remove_jobs_for_integration(
    db: AsyncSession, integration_id: str,
) -> int:
    """Delete every scheduled_job tied to this integration id.

    Called from the delete hook BEFORE the Integration row itself is
    gone so we can still resolve the FK. Returns the count deleted.
    The name prefix filter limits the scan; we still post-filter by
    ``config.integration_id`` for paranoia (slug collision, legacy rows).
    """
    # Narrow the scan by name prefix — we can't derive the slug here
    # because the Integration row may already be detached, so we scan
    # all resource_sync rows and filter in Python. At the scale a single
    # tenant is about to delete, this is fine.
    rows = (
        await db.execute(
            select(ScheduledJob).where(
                ScheduledJob.handler_key.in_(
                    [
                        HANDLER_KEY, BILLING_HANDLER_KEY,
                        *_LIFECYCLE_HANDLERS, *_METRICS_HANDLERS,
                    ],
                ),
            )
        )
    ).scalars().all()
    deleted = 0
    for row in rows:
        if (row.config or {}).get("integration_id") == str(integration_id):
            await db.delete(row)
            deleted += 1
    await db.flush()
    return deleted


async def reconcile_all_cloud_integrations(
    db: AsyncSession,
) -> dict[str, Any]:
    """Walk every cloud integration and re-run ``sync_jobs_for_integration``.

    Called once at worker startup so a newly-added lifecycle handler
    (e.g. ``cmdb.grants_sync`` in PR #24) gets its per-integration
    ``scheduled_jobs`` rows auto-created without anyone having to
    click "save" on each integration in the admin UI. Idempotent: if
    every desired row already exists, this is a no-op.

    One integration failing (bad slug, provider class gone, …) must
    not block the others — we log and continue.
    """
    rows = (
        await db.execute(
            select(Integration).where(Integration.kind == "cloud")
        )
    ).scalars().all()

    added_total = 0
    removed_total = 0
    failed: list[str] = []
    for integ in rows:
        integration_id = str(integ.id)
        integration_label = str(integ.slug or integ.id)
        try:
            # A duplicate/corrupt row for one connector must not poison the
            # AsyncSession and abort every later connector.  SAVEPOINT gives
            # each integration an isolated rollback boundary while retaining
            # the single final commit for the successful reconciliations.
            async with db.begin_nested():
                summary = await sync_jobs_for_integration(db, integ)
            added_total += len(summary.get("added", []))
            removed_total += len(summary.get("removed", []))
            if (
                summary.get("added")
                or summary.get("removed")
                or summary.get("adopted_orphan_names")
            ):
                logger.info(
                    "scheduled_jobs reconcile: %s added=%s removed=%s adopted=%s",
                    integration_label,
                    summary.get("added"),
                    summary.get("removed"),
                    summary.get("adopted_orphan_names"),
                )
        except Exception:
            logger.exception(
                "scheduled_jobs reconcile failed for integration %s",
                integration_label,
            )
            failed.append(integration_id)
    await db.commit()

    return {
        "integrations_scanned": len(rows),
        "rows_added": added_total,
        "rows_removed": removed_total,
        "failed_integration_ids": failed,
    }


__all__ = [
    "BILLING_HANDLER_KEY",
    "HANDLER_KEY",
    "reconcile_all_cloud_integrations",
    "remove_jobs_for_integration",
    "sync_jobs_for_integration",
]