import logging
from datetime import datetime, timedelta, timezone
from math import ceil
from typing import Any

from fastapi import HTTPException
from sqlalchemy import or_, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.crypto import decrypt_secret
from app.models.cloud_server import CloudServer
from app.models.instance import Instance, InstanceStatus, SuspendType
from app.models.notification import NotificationEvent
from app.models.product import Product, ProductApiType
from app.services.dcim_cloud_client import DcimCloudClient, DcimCloudCredentials, DcimCloudError
from app.services import stock_compensation_service, system_setting_service, upstream_action_service, zjmf_instance_adapter

log = logging.getLogger("qycw.instance.lifecycle")

EXPIRATION_REMINDER_DAYS = (7, 3, 1)
EXPIRATION_REMINDER_EVENT_TYPE = "instance_expiry_reminder"
CN_TZ = timezone(timedelta(hours=8))
NAT_ACL_ORDER_FIELDS = {"id", "name", "ext_port", "int_port", "protocol"}
NAT_ACL_SORT_VALUES = {"asc", "desc"}
NAT_ACL_LIST_TYPES = {"all", "page"}


def _enum_value(value: Any) -> str:
    return value.value if hasattr(value, "value") else str(value)


def _to_int_or_none(value: Any) -> int | None:
    if value in (None, ""):
        return None
    if isinstance(value, str):
        value = value.split("|", 1)[0].strip()
    try:
        return int(float(value))
    except (TypeError, ValueError):
        return None


def _normalize_nat_acl_item(item: Any) -> dict[str, Any] | None:
    if not isinstance(item, dict):
        return None
    return {
        "id": _to_int_or_none(item.get("id")),
        "name": item.get("name") or "",
        "description": item.get("description") or item.get("remarks") or item.get("remark") or item.get("desc") or "",
        "ext_port": _to_int_or_none(item.get("ext_port")),
        "int_port": _to_int_or_none(item.get("int_port")),
        "int_ip": item.get("int_ip") or item.get("internal_ip") or item.get("private_ip") or item.get("lan_ip") or "",
        "protocol": _to_int_or_none(item.get("protocol")),
    }


def _empty_nat_acl_result(page: int, per_page: int) -> dict[str, Any]:
    return {
        "list": [],
        "meta": {"total": 0, "total_page": 0, "page": page, "per_page": per_page},
        "nat_host_ip": None,
    }


def _normalize_nat_acl_response(body: Any, page: int, per_page: int) -> dict[str, Any]:
    if not isinstance(body, dict):
        return _empty_nat_acl_result(page, per_page)

    data = body.get("data")
    meta = body.get("meta")
    nat_host_ip = body.get("nat_host_ip")

    if isinstance(data, dict):
        nat_host_ip = nat_host_ip or data.get("nat_host_ip")
        meta = meta or data.get("meta")
        data = data.get("list") or data.get("items") or data.get("data") or []

    items = []
    if isinstance(data, list):
        for item in data:
            normalized = _normalize_nat_acl_item(item)
            if normalized is not None:
                items.append(normalized)

    meta = meta if isinstance(meta, dict) else {}
    return {
        "list": items,
        "meta": {
            "total": _to_int_or_none(meta.get("total")) or len(items),
            "total_page": _to_int_or_none(meta.get("total_page")) or (1 if items else 0),
            "page": _to_int_or_none(meta.get("page")) or page,
            "per_page": _to_int_or_none(meta.get("per_page")) or per_page,
        },
        "nat_host_ip": nat_host_ip or None,
    }


def _aware(value: datetime | None) -> datetime | None:
    if value is None:
        return None
    if value.tzinfo is None:
        return value.replace(tzinfo=timezone.utc)
    return value


def _days_until(expires_at: datetime, now: datetime) -> int:
    seconds = (expires_at - now).total_seconds()
    return ceil(seconds / 86400) if seconds > 0 else 0


def _format_expiry_time(value: datetime) -> str:
    return value.astimezone(CN_TZ).strftime("%Y/%m/%d %H:%M")


def _instance_name(instance: Instance) -> str:
    name = instance.alias or instance.hostname
    return f"{name}（实例 #{instance.id}）" if name else f"实例 #{instance.id}"


def _expiration_reminder_event_key(instance: Instance, expires_at: datetime, days_remaining: int) -> str:
    stamp = expires_at.astimezone(timezone.utc).strftime("%Y%m%d%H%M%S")
    return f"instance_expiry:{instance.id}:{stamp}:{days_remaining}"


async def get_dcim_client_for_instance(
    db: AsyncSession,
    instance: Instance,
) -> tuple[DcimCloudClient, CloudServer]:
    if not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例未绑定云节点")

    result = await db.execute(select(CloudServer).where(CloudServer.id == instance.cloud_server_id))
    server = result.scalar_one_or_none()
    if not server:
        raise HTTPException(status_code=500, detail="云节点不存在")

    if zjmf_instance_adapter.is_zjmf_server(server):
        raise HTTPException(status_code=400, detail="魔方财务实例暂不支持该 DCIM 功能")

    creds = DcimCloudCredentials(
        base_url=server.base_url,
        username=server.username,
        password=decrypt_secret(server.password_encrypted),
        account_type=server.account_type,
    )
    return DcimCloudClient(creds, server_id=server.id), server


async def call_instance_cloud(
    db: AsyncSession,
    instance: Instance,
    api_path: str,
    method: str,
    *,
    data: dict[str, Any] | None = None,
    timeout: float | None = None,
) -> dict[str, Any]:
    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法操作")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return await zjmf_instance_adapter.operate_instance(
            db,
            instance,
            api_path,
            method,
            data=data,
            timeout=timeout,
        )

    client, _ = await get_dcim_client_for_instance(db, instance)
    try:
        kwargs: dict[str, Any] = {"method": method, "data": data}
        if timeout is not None:
            kwargs["timeout"] = timeout
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}{api_path}",
            **kwargs,
        )
    except DcimCloudError as exc:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    return payload if isinstance(payload, dict) else {}


async def call_cloud_path(
    db: AsyncSession,
    instance: Instance,
    path: str,
    method: str,
    *,
    data: dict[str, Any] | None = None,
    timeout: float | None = None,
) -> dict[str, Any]:
    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return await zjmf_instance_adapter.call_path(
            db,
            instance,
            path,
            method,
            data=data,
            timeout=timeout,
        )

    client, _ = await get_dcim_client_for_instance(db, instance)
    try:
        kwargs: dict[str, Any] = {"method": method, "data": data}
        if timeout is not None:
            kwargs["timeout"] = timeout
        code, text, body = await client.request(path, **kwargs)
    except DcimCloudError as exc:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    payload = (body or {}).get("data") or body or {}
    return payload if isinstance(payload, dict) else {}


async def list_instance_nat_acl(
    db: AsyncSession,
    instance: Instance,
    *,
    page: int = 1,
    per_page: int = 20,
    orderby: str = "id",
    sort: str = "desc",
    search: str | None = None,
    list_type: str = "page",
) -> dict[str, Any]:
    if orderby not in NAT_ACL_ORDER_FIELDS:
        raise HTTPException(status_code=400, detail="排序字段无效")
    if sort not in NAT_ACL_SORT_VALUES:
        raise HTTPException(status_code=400, detail="排序方向无效")
    if list_type not in NAT_ACL_LIST_TYPES:
        raise HTTPException(status_code=400, detail="获取类型无效")
    if not instance.upstream_id or not instance.cloud_server_id:
        return _empty_nat_acl_result(page, per_page)

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return await zjmf_instance_adapter.list_nat_acl(
            db,
            instance,
            page=page,
            per_page=per_page,
            orderby=orderby,
            sort=sort,
            search=search,
            list_type=list_type,
        )

    params: dict[str, Any] = {
        "page": page,
        "per_page": per_page,
        "orderby": orderby,
        "sort": sort,
        "list_type": list_type,
    }
    if search:
        params["search"] = search.strip()

    client, _ = await get_dcim_client_for_instance(db, instance)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/nat_acl",
            method="GET",
            data=params,
            timeout=8.0,
        )
    except DcimCloudError as exc:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    return _normalize_nat_acl_response(body or {}, page, per_page)


def _validate_nat_acl_port(value: int, *, field_name: str) -> None:
    if value < 0 or value > 65535:
        raise HTTPException(status_code=400, detail=f"{field_name}必须在 0-65535 之间")


async def create_instance_nat_acl(
    db: AsyncSession,
    instance: Instance,
    *,
    name: str,
    int_port: int,
    protocol: int,
    ext_port: int | None = None,
) -> dict[str, Any]:
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法添加 NAT 转发")
    name = name.strip()
    if not name:
        raise HTTPException(status_code=400, detail="名称不能为空")
    _validate_nat_acl_port(int_port, field_name="内部端口")
    if ext_port is not None:
        _validate_nat_acl_port(ext_port, field_name="外部端口")
        if ext_port in (22, 80, 443):
            raise HTTPException(status_code=400, detail="外部端口不能使用 22、80、443")
    if protocol not in (1, 2, 3):
        raise HTTPException(status_code=400, detail="协议无效")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return await zjmf_instance_adapter.create_nat_acl(
            db,
            instance,
            name=name,
            int_port=int_port,
            protocol=protocol,
            ext_port=ext_port,
        )

    payload: dict[str, Any] = {
        "name": name,
        "int_port": int_port,
        "protocol": protocol,
    }
    if ext_port is not None:
        payload["ext_port"] = ext_port

    client, _ = await get_dcim_client_for_instance(db, instance)
    try:
        code, text, body = await client.request(
            f"/clouds/{instance.upstream_id}/nat_acl",
            method="POST",
            data=payload,
            timeout=8.0,
        )
    except DcimCloudError as exc:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    data = (body or {}).get("data") or body or {}
    normalized = _normalize_nat_acl_item(data) if isinstance(data, dict) else None
    return normalized or (data if isinstance(data, dict) else {})


async def delete_instance_nat_acl(
    db: AsyncSession,
    instance: Instance,
    *,
    nat_acl_id: int,
) -> dict[str, Any]:
    if not instance.upstream_id or not instance.cloud_server_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法删除 NAT 转发")
    if nat_acl_id <= 0:
        raise HTTPException(status_code=400, detail="NAT 转发 ID 无效")

    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        return await zjmf_instance_adapter.delete_nat_acl(
            db,
            instance,
            nat_acl_id=nat_acl_id,
        )

    client, _ = await get_dcim_client_for_instance(db, instance)
    try:
        code, text, body = await client.request(
            f"/nat_acl/{nat_acl_id}?hostid={instance.upstream_id}",
            method="DELETE",
            timeout=8.0,
        )
    except DcimCloudError as exc:
        raise HTTPException(status_code=502, detail=f"上游调用失败：{exc}") from exc

    if not (200 <= code < 300):
        raise HTTPException(status_code=502, detail=f"上游返回异常（HTTP {code}）：{text[:200]}")

    data = (body or {}).get("data") or body or {}
    return data if isinstance(data, dict) else {}


async def force_delete_instance(
    db: AsyncSession,
    instance: Instance,
    *,
    reason: str,
    source: str = "lifecycle",
    user_id: int | None = None,
    order_id: int | None = None,
) -> dict[str, Any]:
    if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
        raise HTTPException(status_code=400, detail=zjmf_instance_adapter.FINANCE_TICKET_MESSAGE)

    payload: dict[str, Any] = {}
    audit_log_id: int | None = None

    # 插件型产品：先调插件 terminate 删除上游资源
    product_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = product_res.scalar_one_or_none()
    if product and product.api_type == ProductApiType.PLUGIN.value and product.api_identifier:
        try:
            from app.routers.instances import build_plugin_manage_context
            from app.services import provision_plugin_service

            plugin_id = str(product.api_identifier)
            manifest = provision_plugin_service.get_plugin_manifest(plugin_id)
            if provision_plugin_service.plugin_supports_action(manifest, "terminate"):
                provider = provision_plugin_service.create_plugin_provider(plugin_id)
                manage = getattr(provider, "manage", None)
                if callable(manage):
                    from dataclasses import asdict, is_dataclass
                    from mode.provision.base import ProvisionPluginError
                    context, _snap, _order = await build_plugin_manage_context(instance, product, db)
                    try:
                        result = manage("terminate", context)
                        if hasattr(result, "__await__"):
                            result = await result
                        if is_dataclass(result):
                            result = asdict(result)
                        log.info("plugin terminate instance=%s plugin=%s result=%s", instance.id, plugin_id, result)
                    except ProvisionPluginError as exc:
                        log.warning("plugin terminate instance=%s plugin=%s error=%s", instance.id, plugin_id, exc)
                    except Exception as exc:
                        log.error("plugin terminate instance=%s plugin=%s unexpected error=%s", instance.id, plugin_id, exc)
        except Exception as exc:
            log.warning("plugin terminate setup failed for instance=%s: %s", instance.id, exc)

    if instance.upstream_id:
        payload, audit_log_id = await upstream_action_service.call_path_with_audit(
            db,
            instance,
            f"/clouds/{instance.upstream_id}",
            "DELETE",
            action_type="instance_delete",
            resource_type="instance",
            resource_id=instance.upstream_id,
            source=source,
            user_id=user_id,
            order_id=order_id,
            data={"force": 1, "delete_data_only": 0},
            timeout=30.0,
            remark=reason,
        )

    instance.status = InstanceStatus.TERMINATED
    instance.terminated_at = datetime.now(timezone.utc)
    instance.suspend_type = None
    instance.provision_log = reason
    await stock_compensation_service.restock_released_instances(
        db,
        [instance],
        source=source,
        reason=reason,
    )
    await db.flush()
    return {"taskid": payload.get("taskid"), "instance_id": instance.id, "audit_log_id": audit_log_id}


async def suspend_instance(
    db: AsyncSession,
    instance: Instance,
    *,
    suspend_type: str = SuspendType.DUE.value,
    free_gpu: int = 0,
    reason: str = "到期暂停",
    source: str = "lifecycle",
    user_id: int | None = None,
    order_id: int | None = None,
) -> dict[str, Any]:
    if _enum_value(instance.status) == InstanceStatus.TERMINATED.value:
        raise HTTPException(status_code=400, detail="实例已销毁，不能暂停")

    # 插件型产品：调插件 suspend
    product_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = product_res.scalar_one_or_none()
    if product and product.api_type == ProductApiType.PLUGIN.value and product.api_identifier:
        try:
            from app.routers.instances import build_plugin_manage_context
            from app.services import provision_plugin_service

            manifest = provision_plugin_service.get_plugin_manifest(str(product.api_identifier))
            if provision_plugin_service.plugin_supports_action(manifest, "suspend"):
                provider = provision_plugin_service.create_plugin_provider(str(product.api_identifier))
                manage = getattr(provider, "manage", None)
                if callable(manage):
                    from dataclasses import asdict, is_dataclass
                    from mode.provision.base import ProvisionPluginError
                    context, _snap, _order = await build_plugin_manage_context(instance, product, db)
                    try:
                        result = manage("suspend", context)
                        if hasattr(result, "__await__"):
                            result = await result
                        log.info("plugin suspend instance=%s result=%s", instance.id, result)
                    except ProvisionPluginError as exc:
                        log.warning("plugin suspend instance=%s error=%s", instance.id, exc)
                    except Exception as exc:
                        log.error("plugin suspend instance=%s unexpected error=%s", instance.id, exc)
        except Exception as exc:
            log.warning("plugin suspend setup failed instance=%s: %s", instance.id, exc)
        # 插件实例不需要调魔方云 suspend
        instance.status = InstanceStatus.SUSPENDED
        instance.suspend_type = SuspendType(suspend_type)
        instance.provision_log = reason
        await db.flush()
        from app.services import notification_service
        await notification_service.safe_notify(
            db,
            user_id=instance.user_id,
            category="instance",
            title="服务器已暂停",
            content=f"您的服务（实例 #{instance.id}）已被暂停：{reason}。如需恢复请及时续费或联系客服。",
            link=f"/control/server/{instance.id}",
        )
        return {"taskid": None, "instance_id": instance.id}

    if not instance.upstream_id:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法操作")

    payload, audit_log_id = await upstream_action_service.call_path_with_audit(
        db,
        instance,
        f"/clouds/{instance.upstream_id}/suspend",
        "POST",
        action_type="instance_suspend",
        resource_type="instance",
        resource_id=instance.upstream_id,
        source=source,
        user_id=user_id,
        order_id=order_id,
        data={"type": suspend_type, "free_gpu": 1 if free_gpu else 0},
        remark=reason,
    )
    instance.status = InstanceStatus.SUSPENDED
    instance.suspend_type = SuspendType(suspend_type)
    instance.provision_log = reason
    await db.flush()

    from app.services import notification_service

    await notification_service.safe_notify(
        db,
        user_id=instance.user_id,
        category="instance",
        title="服务器已暂停",
        content=f"您的服务器（实例 #{instance.id}）已被暂停：{reason}。如需恢复请及时续费或联系客服。",
        link=f"/control/server/{instance.id}",
    )
    return {"taskid": payload.get("taskid"), "instance_id": instance.id, "audit_log_id": audit_log_id}


async def unsuspend_instance(
    db: AsyncSession,
    instance: Instance,
    *,
    reason: str = "解除暂停",
    source: str = "manual",
    user_id: int | None = None,
    order_id: int | None = None,
) -> dict[str, Any]:
    expires_at = _aware(instance.expires_at)
    if expires_at and expires_at <= datetime.now(timezone.utc):
        raise HTTPException(status_code=400, detail="实例已到期，请先续费")

    # 插件型产品：调插件 unsuspend
    product_res = await db.execute(select(Product).where(Product.id == instance.product_id))
    product = product_res.scalar_one_or_none()
    is_plugin = product and product.api_type == ProductApiType.PLUGIN.value and product.api_identifier

    if not instance.upstream_id and not is_plugin:
        raise HTTPException(status_code=400, detail="实例尚未开通，无法操作")

    if is_plugin:
        try:
            from app.routers.instances import build_plugin_manage_context
            from app.services import provision_plugin_service

            manifest = provision_plugin_service.get_plugin_manifest(str(product.api_identifier))
            if provision_plugin_service.plugin_supports_action(manifest, "unsuspend"):
                provider = provision_plugin_service.create_plugin_provider(str(product.api_identifier))
                manage = getattr(provider, "manage", None)
                if callable(manage):
                    from dataclasses import asdict, is_dataclass
                    from mode.provision.base import ProvisionPluginError
                    context, _snap, _order = await build_plugin_manage_context(instance, product, db)
                    try:
                        result = manage("unsuspend", context)
                        if hasattr(result, "__await__"):
                            result = await result
                        log.info("plugin unsuspend instance=%s result=%s", instance.id, result)
                    except ProvisionPluginError as exc:
                        log.warning("plugin unsuspend instance=%s error=%s", instance.id, exc)
                    except Exception as exc:
                        log.error("plugin unsuspend instance=%s unexpected error=%s", instance.id, exc)
        except Exception as exc:
            log.warning("plugin unsuspend setup failed instance=%s: %s", instance.id, exc)
        instance.status = InstanceStatus.ACTIVE
        instance.suspend_type = None
        instance.provision_log = reason
        await db.flush()
        return {"taskid": None, "instance_id": instance.id}

    payload, audit_log_id = await upstream_action_service.call_path_with_audit(
        db,
        instance,
        f"/clouds/{instance.upstream_id}/unsuspend",
        "POST",
        action_type="instance_unsuspend",
        resource_type="instance",
        resource_id=instance.upstream_id,
        source=source,
        user_id=user_id,
        order_id=order_id,
        remark=reason,
    )
    instance.status = InstanceStatus.ACTIVE
    instance.suspend_type = None
    instance.provision_log = reason
    await db.flush()
    return {"taskid": payload.get("taskid"), "instance_id": instance.id, "audit_log_id": audit_log_id}


async def auto_unsuspend_after_renewal(db: AsyncSession, instance: Instance) -> dict[str, Any] | None:
    if (
        _enum_value(instance.status) == InstanceStatus.SUSPENDED.value
        and _enum_value(instance.suspend_type) == SuspendType.DUE.value
    ):
        return await unsuspend_instance(db, instance, reason="续费后自动解除到期暂停")
    return None


async def send_expiration_reminders(db: AsyncSession, *, limit: int = 200) -> dict[str, int]:
    now = datetime.now(timezone.utc)
    max_window = max(EXPIRATION_REMINDER_DAYS)

    result = await db.execute(
        select(Instance)
        .where(
            Instance.status == InstanceStatus.ACTIVE,
            Instance.expires_at.is_not(None),
            Instance.expires_at > now,
            Instance.expires_at <= now + timedelta(days=max_window),
        )
        .order_by(Instance.expires_at.asc(), Instance.id.asc())
        .limit(limit)
    )
    instances = list(result.scalars().all())

    sent = 0
    skipped = 0
    failed = 0

    from app.services import notification_service

    for instance in instances:
        try:
            expires_at = _aware(instance.expires_at)
            if not expires_at:
                continue

            days_remaining = _days_until(expires_at, now)
            if days_remaining not in EXPIRATION_REMINDER_DAYS:
                continue

            event_key = _expiration_reminder_event_key(instance, expires_at, days_remaining)
            existing = await db.execute(
                select(NotificationEvent.id).where(NotificationEvent.event_key == event_key).limit(1)
            )
            if existing.scalar_one_or_none():
                skipped += 1
                continue

            db.add(
                NotificationEvent(
                    event_key=event_key,
                    event_type=EXPIRATION_REMINDER_EVENT_TYPE,
                    user_id=instance.user_id,
                    resource_type="instance",
                    resource_id=instance.id,
                )
            )
            await db.flush()

            renew_url = f"/control/server/{instance.id}?tab=renewal"
            await notification_service.notify(
                db,
                user_id=instance.user_id,
                category="instance",
                level="urgent" if days_remaining == 1 else "warning",
                title="服务器到期提醒",
                content=(
                    f"您的服务器 {_instance_name(instance)} 将在 {days_remaining} 天内到期，"
                    f"到期时间 {_format_expiry_time(expires_at)}。请及时续费以避免服务暂停。"
                ),
                link=renew_url,
                action_label="立即续费",
                action_url=renew_url,
            )
            sent += 1
            await db.commit()
        except ValueError as exc:
            if "preference" in str(exc):
                skipped += 1
                await db.commit()
                continue
            failed += 1
            await db.rollback()
            log.exception("expiration reminder failed: instance=%s", instance.id)
        except IntegrityError:
            skipped += 1
            await db.rollback()
        except Exception:
            failed += 1
            await db.rollback()
            log.exception("expiration reminder failed: instance=%s", instance.id)

    return {"sent": sent, "skipped": skipped, "failed": failed}


async def process_expired_instances(db: AsyncSession, *, limit: int = 50) -> dict[str, int]:
    settings = await system_setting_service.get_business_settings(db)
    now = datetime.now(timezone.utc)
    suspend_cutoff = now - timedelta(days=settings.expire_suspend_after_days)
    delete_cutoff = now - timedelta(days=settings.expire_delete_after_days)

    result = await db.execute(
        select(Instance)
        .where(
            Instance.expires_at.is_not(None),
            or_(
                (Instance.status == InstanceStatus.ACTIVE) & (Instance.expires_at <= suspend_cutoff),
                (
                    (Instance.status == InstanceStatus.SUSPENDED)
                    & (Instance.suspend_type == SuspendType.DUE)
                    & (Instance.expires_at <= delete_cutoff)
                ),
            ),
        )
        .order_by(Instance.expires_at.asc(), Instance.id.asc())
        .limit(limit)
    )
    instances = list(result.scalars().all())

    suspended = 0
    deleted = 0
    failed = 0
    for instance in instances:
        try:
            expires_at = _aware(instance.expires_at)
            if not expires_at:
                continue

            if expires_at <= delete_cutoff:
                await force_delete_instance(db, instance, reason="到期超过 3 天自动强制删除")
                deleted += 1
                await db.commit()
                continue

            if _enum_value(instance.status) == InstanceStatus.ACTIVE.value and expires_at <= suspend_cutoff:
                await suspend_instance(db, instance, suspend_type=SuspendType.DUE.value, reason="到期自动暂停")
                suspended += 1
                await db.commit()
        except Exception:
            failed += 1
            await db.rollback()
            log.exception("expired instance lifecycle failed: instance=%s", instance.id)

    return {"suspended": suspended, "deleted": deleted, "failed": failed}
