import asyncio
import logging
import random
import string
from datetime import datetime, timedelta, timezone
from typing import Any

from fastapi import APIRouter, Depends, HTTPException, Query
from pydantic import BaseModel
from sqlalchemy import or_, select, update
from sqlalchemy.ext.asyncio import AsyncSession

from app.config import settings
from app.core.deps import get_current_user
from app.database import get_db
from app.models.balance_log import BalanceChangeType, BalanceLog
from app.models.instance import Instance, InstanceStatus
from app.models.instance_disk_addon import InstanceDiskAddon, InstanceDiskAddonStatus
from app.models.order import Order, OrderStatus
from app.models.order_refund import OrderRefund
from app.models.point_reward import PointReward, PointRewardStatus
from app.models.product import Product
from app.models.promo_code import PromoCode
from app.models.coupon_mall_claim import CouponMallClaim, CouponMallClaimStatus
from app.models.user_agent import UserAgent, AgentStatus
from app.models.product_pricing import ProductPricing
from app.models.upstream_action_log import UpstreamActionLog
from app.models.user import User
from app.schemas.common import ApiResponse
from app.services import (
    agent_service,
    email_service,
    instance_lifecycle_service,
    membership_service,
    notification_service,
    order_service,
    provision_service,
    spend_service,
    system_setting_service,
    upstream_action_service,
    zjmf_instance_adapter,
)
from app.utils.specs import parse_memory_to_mb

log = logging.getLogger("qycw.orders")

router = APIRouter()

_DEPENDENCY_ORDER = ["package_upgrade", "disk_resize", "disk_addon", "server_purchase"]

class CheckoutRequest(BaseModel):
    cart_item_ids: list[int]
    use_points: int = 0
    coupon_code: str | None = None
    coupon_id: int | None = None
    coupon_target_cart_item_id: int | None = None


class CouponPreviewRequest(BaseModel):
    cart_item_ids: list[int]
    coupon_code: str | None = None
    coupon_id: int | None = None
    coupon_target_cart_item_id: int | None = None
    use_points: int = 0


class ClaimActivityCouponRequest(BaseModel):
    promo_code_id: int
    activity_id: int | None = None


class OrderItemOut(BaseModel):
    product_id: int
    product_name: str
    sku: str
    cycle: str
    quantity: int
    unit_price_cents: int
    item_total_cents: int


async def _resolve_coupon_code(db: AsyncSession, coupon_id: int | None, coupon_code: str | None, user_id: int) -> str | None:
    """优先使用 coupon_id 解析优惠码，避免把真实 code 暴露给前端。"""
    if coupon_id:
        coupon = await db.get(PromoCode, coupon_id)
        if not coupon:
            raise HTTPException(status_code=404, detail="优惠券不存在")
        return coupon.code
    return coupon_code

class OrderOut(BaseModel):
    id: int
    out_order_no: str
    status: str
    total_amount: int
    discount_amount: int
    instance_count: int

    class Config:
        from_attributes = True


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


def _as_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 _gen_refund_no() -> str:
    ts = datetime.now(timezone.utc).strftime("%Y%m%d%H%M%S")
    suffix = "".join(random.choices(string.digits, k=6))
    return f"RFD{ts}{suffix}"


def _cycle_to_timedelta(cycle: str | None) -> timedelta:
    cycle = (cycle or "monthly").lower()
    mapping = {
        "hourly": timedelta(hours=1),
        "daily": timedelta(days=1),
        "weekly": timedelta(days=7),
        "monthly": timedelta(days=30),
        "quarterly": timedelta(days=90),
        "semi_annually": timedelta(days=180),
        "semiannually": timedelta(days=180),
        "annually": timedelta(days=365),
        "yearly": timedelta(days=365),
        "biennially": timedelta(days=730),
        "triennially": timedelta(days=1095),
    }
    return mapping.get(cycle, timedelta(days=30))


def _split_amount(amount: int, parts: int) -> list[int]:
    if parts <= 0:
        return []
    base = amount // parts
    remainder = amount % parts
    return [base + (1 if idx < remainder else 0) for idx in range(parts)]


def _paid_amount(order: Order) -> int:
    return max(0, int(order.total_amount or 0) - int(order.discount_amount or 0))


def _extract_instance_ids(items: list | None) -> set[int]:
    ids: set[int] = set()
    for item in items or []:
        if not isinstance(item, dict):
            continue
        value = item.get("instance_id")
        if value in (None, ""):
            continue
        try:
            ids.add(int(value))
        except (TypeError, ValueError):
            continue
    return ids


def _item_summary(item: dict) -> str:
    item_type = item.get("type")
    if item_type == "package_upgrade":
        return f"套餐升级：{item.get('from_product_name') or '-'} -> {item.get('to_product_name') or '-'}"
    if item_type == "disk_addon":
        return f"新增云盘：{item.get('size_gb') or '-'} GB"
    if item_type == "disk_resize":
        return f"磁盘扩容：{item.get('source_size_gb') or '-'} GB -> {item.get('target_size_gb') or '-'} GB"
    product_name = item.get("product_name") or item.get("sku") or item.get("type") or "服务订单"
    quantity = int(item.get("quantity") or 1)
    cycle = item.get("cycle") or "-"
    return f"{product_name} x{quantity} / {cycle}"


async def _load_order_instances(
    db: AsyncSession,
    order: Order,
    *,
    for_update: bool = False,
) -> list[Instance]:
    instance_ids = _extract_instance_ids(order.items_snapshot)
    conditions = [Instance.order_id == order.id]
    if instance_ids:
        conditions.append(Instance.id.in_(instance_ids))
    stmt = select(Instance).where(Instance.user_id == order.user_id, or_(*conditions))
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)
    instances: dict[int, Instance] = {}
    for instance in result.scalars().all():
        instances[int(instance.id)] = instance
    return list(instances.values())


async def _load_order_point_rewards(
    db: AsyncSession,
    order_id: int,
    *,
    for_update: bool = False,
) -> list[PointReward]:
    stmt = select(PointReward).where(PointReward.order_id == order_id)
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)
    return list(result.scalars().all())


async def _load_order_actions(db: AsyncSession, order_id: int) -> list[UpstreamActionLog]:
    result = await db.execute(
        select(UpstreamActionLog)
        .where(UpstreamActionLog.order_id == order_id)
        .order_by(UpstreamActionLog.id.desc())
    )
    return list(result.scalars().all())


async def _latest_refund(db: AsyncSession, order_id: int) -> OrderRefund | None:
    result = await db.execute(
        select(OrderRefund)
        .where(OrderRefund.order_id == order_id)
        .order_by(OrderRefund.id.desc())
        .limit(1)
    )
    return result.scalar_one_or_none()


def _is_server_purchase_order(order: Order) -> bool:
    items = order.items_snapshot or []
    return any(
        isinstance(item, dict) and (item.get("type") or "server_purchase") == "server_purchase"
        for item in items
    )


async def _full_refund_count(
    db: AsyncSession,
    user_id: int,
    reset_at: datetime | None = None,
) -> int:
    """Count successful full refunds that consume the user's full-refund quota.

    Only server_purchase orders count toward the quota; full refunds for related
    orders (upgrades / disk addons) do not consume it.
    """
    conditions = [
        OrderRefund.user_id == user_id,
        OrderRefund.refund_type == "full",
        OrderRefund.status == "success",
        OrderRefund.refund_amount > 0,
    ]
    if reset_at is not None:
        conditions.append(OrderRefund.created_at >= reset_at)
    result = await db.execute(select(OrderRefund.order_id).where(*conditions))
    order_ids = [row[0] for row in result.all() if row[0] is not None]
    if not order_ids:
        return 0
    orders_res = await db.execute(select(Order).where(Order.id.in_(order_ids)))
    return sum(1 for o in orders_res.scalars().all() if _is_server_purchase_order(o))


async def _is_active_agent(db: AsyncSession, user_id: int) -> bool:
    result = await db.execute(
        select(UserAgent.id)
        .where(
            UserAgent.user_id == user_id,
            UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]),
        )
        .limit(1)
    )
    return result.scalar_one_or_none() is not None


def _full_refund_limit_for_user(settings, user: User) -> int:
    extra = max(0, int(getattr(user, "full_refund_quota_extra", 0) or 0))
    return int(settings.full_refund_limit) + extra


def _sort_orders_by_dependency(orders: list[Order]) -> list[Order]:
    """Sort orders in reverse dependency: upgrades first, then disks, then purchase last."""
    def sort_key(o: Order) -> int:
        items = o.items_snapshot or []
        types = {item.get("type") for item in items if isinstance(item, dict)}
        for idx, t in enumerate(_DEPENDENCY_ORDER):
            if t in types:
                return idx
        return len(_DEPENDENCY_ORDER)
    return sorted(orders, key=sort_key)


async def _load_related_orders(
    db: AsyncSession,
    order: Order,
    instances: list[Instance],
    *,
    for_update: bool = False,
) -> list[Order]:
    """Find related orders for the same instance(s) — upgrades, disk addons, disk resizes.

    For server_purchase orders, finds all other PAID orders that reference the
    same instance_ids in their items_snapshot, so they can be cascaded in
    dependency order (upgrades first, then disks, then the purchase itself).
    """
    if not instances:
        return []

    instance_ids = {i.id for i in instances}
    stmt = (
        select(Order)
        .where(
            Order.user_id == order.user_id,
            Order.id != order.id,
            Order.status == OrderStatus.PAID,
        )
        .order_by(Order.id.asc())
    )
    if for_update:
        stmt = stmt.with_for_update()
    result = await db.execute(stmt)

    # status==PAID already excludes refunded/cancelled orders, so no separate
    # refund check is needed. Match any candidate that references our instance(s).
    related: list[Order] = []
    for candidate in result.scalars().all():
        if instance_ids & _extract_instance_ids(candidate.items_snapshot):
            related.append(candidate)
    return related


def _has_claimed_points(order: Order, rewards: list[PointReward]) -> bool:
    return bool(order.points_claimed) or any(
        reward.status == PointRewardStatus.CLAIMED.value for reward in rewards
    )


def _calculate_prorated_refund(order: Order, instances: list[Instance], now: datetime) -> int:
    paid_amount = _paid_amount(order)
    if paid_amount <= 0 or not instances:
        return 0

    allocations = _split_amount(paid_amount, len(instances))
    refund_amount = 0
    for instance, amount in zip(instances, allocations):
        expires_at = _as_aware(instance.expires_at)
        if not expires_at or expires_at <= now:
            continue
        cycle_seconds = max(1, int(_cycle_to_timedelta(instance.cycle).total_seconds()))
        remaining_seconds = max(0, int((expires_at - now).total_seconds()))
        remaining_seconds = min(remaining_seconds, cycle_seconds)
        refund_amount += int(amount * remaining_seconds / cycle_seconds)
    return min(paid_amount, max(0, refund_amount))


def _build_refund_policy(
    order: Order,
    instances: list[Instance],
    rewards: list[PointReward],
    full_refund_used: int,
    latest_refund: OrderRefund | None,
    full_refund_days: int,
    full_refund_limit: int,
    *,
    allow_full_refund: bool = True,
    enforce_full_limit: bool = True,
    full_refund_unlimited: bool = False,
) -> dict:
    status = _enum_value(order.status)
    paid_amount = _paid_amount(order)
    now = datetime.now(timezone.utc)
    order_created_at = _as_aware(order.created_at) or now
    within_full_window = now - order_created_at <= timedelta(days=full_refund_days)
    claimed_points = _has_claimed_points(order, rewards)

    base = {
        "action_allowed": False,
        "refund_type": "unavailable",
        "refund_amount": 0,
        "paid_amount": paid_amount,
        "reason": "",
        "points_claimed": claimed_points,
        "full_refund_used": full_refund_used,
        "full_refund_limit": full_refund_limit,
        "full_refund_days": full_refund_days,
        "full_refund_unlimited": full_refund_unlimited,
        "within_full_refund_window": within_full_window,
    }

    if latest_refund:
        return {
            **base,
            "reason": "订单已退订，不能重复操作",
            "latest_refund": _refund_to_dict(latest_refund),
        }

    if status != OrderStatus.PAID.value:
        return {**base, "reason": "当前订单状态不支持退订"}

    if claimed_points:
        return {
            **base,
            "action_allowed": True,
            "refund_type": "no_refund",
            "reason": "已领取消费积分，退订不退款",
        }

    if (
        allow_full_refund
        and within_full_window
        and (full_refund_unlimited or not enforce_full_limit or full_refund_used < full_refund_limit)
    ):
        return {
            **base,
            "action_allowed": True,
            "refund_type": "full",
            "refund_amount": paid_amount,
            "reason": "符合三天内全额退订条件",
        }

    prorated_amount = _calculate_prorated_refund(order, instances, now)
    return {
        **base,
        "action_allowed": True,
        "refund_type": "prorated",
        "refund_amount": prorated_amount,
        "reason": "按服务剩余时间计算实额退订",
    }


def _refund_to_dict(refund: OrderRefund | None) -> dict | None:
    if not refund:
        return None
    return {
        "id": refund.id,
        "refund_no": refund.refund_no,
        "refund_type": refund.refund_type,
        "refund_amount": refund.refund_amount,
        "status": refund.status,
        "resource_status": getattr(refund, "resource_status", "none"),
        "resource_error": getattr(refund, "resource_error", None),
        "resource_completed_at": (
            refund.resource_completed_at.isoformat()
            if getattr(refund, "resource_completed_at", None)
            else None
        ),
        "reason": refund.reason,
        "created_at": refund.created_at.isoformat() if refund.created_at else None,
    }


def _action_to_dict(action: UpstreamActionLog) -> dict:
    return upstream_action_service.serialize_action_log(action)


def _instance_to_dict(instance: Instance) -> dict:
    return {
        "id": instance.id,
        "status": _enum_value(instance.status),
        "main_ip": instance.main_ip,
        "hostname": instance.hostname,
        "alias": instance.alias,
        "cycle": instance.cycle,
        "unit_price": instance.unit_price,
        "created_at": instance.created_at.isoformat() if instance.created_at else None,
        "expires_at": instance.expires_at.isoformat() if instance.expires_at else None,
        "terminated_at": instance.terminated_at.isoformat() if instance.terminated_at else None,
        "cpu": instance.cpu,
        "memory": instance.memory,
        "system_disk_size": instance.system_disk_size,
        "data_disk_size": instance.data_disk_size,
        "in_bw": instance.in_bw,
        "out_bw": instance.out_bw,
        "ip_num": instance.ip_num,
        "os_image": instance.os_image,
    }


def _order_to_service_dict(
    order: Order,
    instances: list[Instance],
    rewards: list[PointReward],
    refund: OrderRefund | None,
    policy: dict,
    actions: list[UpstreamActionLog] | None = None,
) -> dict:
    items = order.items_snapshot or []
    action_items = actions or []
    return {
        "id": order.id,
        "out_order_no": order.out_order_no,
        "status": _enum_value(order.status),
        "total_amount": order.total_amount,
        "discount_amount": order.discount_amount,
        "paid_amount": _paid_amount(order),
        "created_at": order.created_at.isoformat() if order.created_at else None,
        "remark": order.remark,
        "items_snapshot": items,
        "content": [_item_summary(item) for item in items if isinstance(item, dict)],
        "instances": [_instance_to_dict(instance) for instance in instances],
        "point_rewards": [membership_service.point_reward_to_dict(reward) for reward in rewards],
        "upstream_actions": [_action_to_dict(action) for action in action_items],
        "timeline": _service_order_timeline(order, instances, rewards, refund, action_items),
        "points_claimed": policy["points_claimed"],
        "refund_policy": policy,
        "refund": _refund_to_dict(refund),
    }


def _timeline_at(value: Any) -> str | None:
    if value is None:
        return None
    if hasattr(value, "isoformat"):
        return value.isoformat()
    return str(value)


def _service_order_timeline(
    order: Order,
    instances: list[Instance],
    rewards: list[PointReward],
    refund: OrderRefund | None,
    actions: list[UpstreamActionLog],
) -> list[dict]:
    rows: list[dict] = [
        {
            "key": f"order:{order.id}:created",
            "kind": "order",
            "title": "Order created",
            "description": order.out_order_no,
            "status": _enum_value(order.status),
            "at": _timeline_at(order.created_at),
            "ref_id": order.id,
        }
    ]
    for instance in instances:
        rows.append(
            {
                "key": f"instance:{instance.id}:created",
                "kind": "instance",
                "title": "Instance created",
                "description": instance.main_ip or instance.hostname or f"Instance #{instance.id}",
                "status": _enum_value(instance.status),
                "at": _timeline_at(instance.created_at),
                "ref_id": instance.id,
            }
        )
        if instance.terminated_at:
            rows.append(
                {
                    "key": f"instance:{instance.id}:terminated",
                    "kind": "instance",
                    "title": "Instance terminated",
                    "description": instance.main_ip or instance.hostname or f"Instance #{instance.id}",
                    "status": "terminated",
                    "at": _timeline_at(instance.terminated_at),
                    "ref_id": instance.id,
                }
            )
    for reward in rewards:
        rows.append(
            {
                "key": f"point_reward:{reward.id}",
                "kind": "points",
                "title": "Point reward",
                "description": getattr(reward, "remark", None) or f"{getattr(reward, 'points', 0)} points",
                "status": getattr(reward, "status", None),
                "at": _timeline_at(getattr(reward, "created_at", None)),
                "ref_id": reward.id,
            }
        )
    if refund:
        rows.append(
            {
                "key": f"refund:{refund.id}:created",
                "kind": "refund",
                "title": "Refund created",
                "description": refund.refund_no,
                "status": refund.resource_status or refund.status,
                "at": _timeline_at(refund.created_at),
                "ref_id": refund.id,
            }
        )
        if refund.resource_completed_at:
            rows.append(
                {
                    "key": f"refund:{refund.id}:resource_completed",
                    "kind": "resource",
                    "title": "Resource handling completed",
                    "description": refund.resource_error or refund.reason,
                    "status": refund.resource_status,
                    "at": _timeline_at(refund.resource_completed_at),
                    "ref_id": refund.id,
                }
            )
    for action in actions:
        rows.append(
            {
                "key": f"upstream_action:{action.id}",
                "kind": "resource",
                "title": action.action_type,
                "description": action.error_message or action.remark or action.request_path,
                "status": action.status,
                "at": _timeline_at(action.completed_at or action.updated_at or action.created_at),
                "ref_id": action.id,
            }
        )
    return sorted(rows, key=lambda item: item.get("at") or "")


def _collect_audit_log_ids(value: Any) -> list[int]:
    ids: list[int] = []
    if isinstance(value, dict):
        raw = value.get("audit_log_id")
        if raw not in (None, ""):
            try:
                ids.append(int(raw))
            except (TypeError, ValueError):
                pass
        for key in ("audit_log_ids", "resource_actions", "actions"):
            nested = value.get(key)
            if nested:
                ids.extend(_collect_audit_log_ids(nested))
    elif isinstance(value, list):
        for item in value:
            ids.extend(_collect_audit_log_ids(item))
    return list(dict.fromkeys(ids))


async def _mark_related_disk_addons(
    db: AsyncSession,
    *,
    order_id: int,
    refund_amount: int,
    reason: str,
) -> int:
    result = await db.execute(
        select(InstanceDiskAddon)
        .where(InstanceDiskAddon.order_id == order_id)
        .with_for_update()
    )
    addons = list(result.scalars().all())
    now = datetime.now(timezone.utc)
    for addon in addons:
        if _enum_value(addon.status) in (
            InstanceDiskAddonStatus.REFUNDED.value,
            InstanceDiskAddonStatus.DELETED.value,
        ):
            continue
        addon.status = InstanceDiskAddonStatus.REFUNDED if refund_amount > 0 else InstanceDiskAddonStatus.DELETED
        addon.refunded_at = now if refund_amount > 0 else addon.refunded_at
        addon.deleted_at = now if refund_amount <= 0 else addon.deleted_at
        addon.remark = reason
    return len(addons)


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 _config_value(product: Product | None, *keys: str) -> Any:
    if not product or not isinstance(product.provision_config, list):
        return None
    key_set = set(keys)
    for field in product.provision_config:
        if not isinstance(field, dict):
            continue
        field_key = str(field.get("def", "")).split("|", 1)[0].strip()
        if field_key in key_set and field.get("type", "fixed") == "fixed":
            value = field.get("value")
            return None if value in (None, "") else value
    return None


def _spec_value(product: Product | None, *keys: str) -> int | None:
    return _to_int_or_none(_config_value(product, *keys))


def _product_specs(product: Product) -> dict[str, int | None]:
    in_bw = _spec_value(product, "in_bw")
    out_bw = _spec_value(product, "out_bw")
    return {
        "cpu": product.vcpu if product.vcpu is not None else _spec_value(product, "cpu"),
        "memory": parse_memory_to_mb(product.memory) or _spec_value(product, "memory"),
        "in_bw": in_bw,
        "out_bw": out_bw,
        "traffic_quota": _spec_value(product, "traffic_quota"),
        "traffic_type": _spec_value(product, "traffic_type"),
        "nat_acl_limit": _spec_value(product, "nat_acl_limit"),
        "nat_web_limit": _spec_value(product, "nat_web_limit"),
    }


async def _pricing_cents(db: AsyncSession, product_id: int, cycle: str) -> int:
    candidates = [cycle]
    if cycle == "yearly":
        candidates.append("annually")
    if cycle == "annually":
        candidates.append("yearly")
    if cycle == "semiannually":
        candidates.append("semi_annually")
    if cycle == "semi_annually":
        candidates.append("semiannually")

    result = await db.execute(
        select(ProductPricing).where(
            ProductPricing.product_id == product_id,
            ProductPricing.cycle.in_(candidates),
            ProductPricing.is_enabled.is_(True),
        )
    )
    pricing = result.scalars().first()
    if not pricing:
        raise HTTPException(status_code=400, detail="原套餐当前周期定价不存在，无法自动回退配置")
    return int(round(float(pricing.price) * 100))


async def _active_disk_total(db: AsyncSession, instance_id: int) -> int:
    result = await db.execute(
        select(InstanceDiskAddon).where(
            InstanceDiskAddon.instance_id == instance_id,
            InstanceDiskAddon.status == InstanceDiskAddonStatus.ACTIVE,
        )
    )
    return sum(int(addon.recurring_amount or 0) for addon in result.scalars().all())


async def _fetch_upstream_detail(db: AsyncSession, instance: Instance) -> dict[str, Any]:
    payload = await instance_lifecycle_service.call_cloud_path(
        db,
        instance,
        f"/clouds/{instance.upstream_id}",
        "GET",
        timeout=8.0,
    )
    return payload


def _is_system_disk(disk: dict[str, Any]) -> bool:
    return str(disk.get("type") or "").lower() == "system" or str(disk.get("dev") or "").lower() == "vda"


def _find_disk(detail: dict[str, Any], disk_id: int) -> dict[str, Any]:
    disks = detail.get("disk") if isinstance(detail, dict) else []
    if not isinstance(disks, list):
        raise HTTPException(status_code=400, detail="上游未返回磁盘信息")
    for disk in disks:
        if isinstance(disk, dict) and str(disk.get("id") or "") == str(disk_id):
            return disk
    raise HTTPException(status_code=404, detail="磁盘不存在")


def _disk_resize_payload(target_size: int, disk: dict[str, Any]) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "size": target_size,
        "driver": str(disk.get("driver") or "virtio"),
        "cache": str(disk.get("cache") or "none"),
        "io": str(disk.get("io") or "native"),
    }
    for field in (
        "read_bytes_sec",
        "write_bytes_sec",
        "read_iops_sec",
        "write_iops_sec",
        "read_bytes_sec_max",
        "write_bytes_sec_max",
        "read_iops_sec_max",
        "write_iops_sec_max",
        "iops_min",
        "iops_max",
    ):
        value = _to_int_or_none(disk.get(field))
        if value is not None:
            payload[field] = value
    if disk.get("remarks") not in (None, ""):
        payload["remarks"] = str(disk.get("remarks"))
    return payload


async def _mark_addons_for_unsubscribe(
    db: AsyncSession,
    *,
    order_id: int,
    refunded: bool,
    reason: str,
) -> list[InstanceDiskAddon]:
    result = await db.execute(
        select(InstanceDiskAddon)
        .where(InstanceDiskAddon.order_id == order_id)
        .with_for_update()
    )
    addons = list(result.scalars().all())
    now = datetime.now(timezone.utc)
    for addon in addons:
        addon.status = InstanceDiskAddonStatus.REFUNDED if refunded else InstanceDiskAddonStatus.DELETED
        if refunded:
            addon.refunded_at = now
        else:
            addon.deleted_at = now
        addon.remark = reason
    return addons


async def _rollback_disk_resize(
    db: AsyncSession,
    *,
    order: Order,
    instance: Instance,
    item: dict,
    refunded: bool,
    reason: str,
) -> dict:
    disk_id = _to_int_or_none(item.get("diskid"))
    source_size = _to_int_or_none(item.get("source_size_gb"))
    if not disk_id or not source_size:
        raise HTTPException(status_code=400, detail="订单缺少磁盘扩容回退快照")

    detail = await _fetch_upstream_detail(db, instance)
    disk = _find_disk(detail, disk_id)
    if _is_system_disk(disk):
        raise HTTPException(status_code=400, detail="系统盘不支持退订回退")

    payload = _disk_resize_payload(source_size, disk)
    response, audit_log_id = await upstream_action_service.call_path_with_audit(
        db,
        instance,
        f"/disks/{disk_id}",
        "PUT",
        action_type="disk_resize_rollback",
        resource_type="disk",
        resource_id=disk_id,
        source="unsubscribe",
        user_id=order.user_id,
        order_id=order.id,
        data=payload,
        timeout=30.0,
        remark=reason,
    )
    await _mark_addons_for_unsubscribe(db, order_id=order.id, refunded=refunded, reason=reason)
    instance.next_renew_amount = int(instance.unit_price or 0) + await _active_disk_total(db, instance.id)
    return {
        "action": "disk_resize_rollback",
        "disk_id": disk_id,
        "target_size": source_size,
        "taskid": response.get("taskid"),
        "audit_log_id": audit_log_id,
    }


async def _wait_disk_unmounted(
    db: AsyncSession,
    instance: Instance,
    disk_id: int,
    *,
    attempts: int = 15,
    interval: float = 2.0,
) -> bool:
    """轮询上游直到磁盘卸载完成（status 变 '0'）或磁盘已不存在。返回是否已卸载。

    上游卸载是异步任务，必须等卸载完成才能删除，否则删除接口会报
    『开机状态下磁盘需要先卸载后删除』。
    """
    for _ in range(attempts):
        await asyncio.sleep(interval)
        try:
            detail = await _fetch_upstream_detail(db, instance)
            disk = _find_disk(detail, disk_id)
        except HTTPException:
            return True  # 磁盘已不在上游 == 已卸载/已删除
        if str(disk.get("status") or "") == "0":
            return True
    return False


async def _delete_added_disk(
    db: AsyncSession,
    *,
    order: Order,
    instance: Instance,
    item: dict,
    refunded: bool,
    reason: str,
) -> dict:
    disk_id = _to_int_or_none(item.get("diskid"))
    if not disk_id:
        result = await db.execute(
            select(InstanceDiskAddon).where(InstanceDiskAddon.order_id == order.id)
        )
        addon = result.scalars().first()
        disk_id = _to_int_or_none(addon.upstream_disk_id if addon else None)
    if not disk_id:
        raise HTTPException(status_code=400, detail="订单缺少新增磁盘快照，无法删除磁盘")

    audit_log_ids: list[int] = []
    detach_taskid = None

    # 检查磁盘是否还在上游、是否处于挂载状态
    disk_present = True
    needs_detach = False
    try:
        detail = await _fetch_upstream_detail(db, instance)
        disk = _find_disk(detail, disk_id)
        needs_detach = str(disk.get("status") or "") != "0"
    except HTTPException as exc:
        if exc.status_code == 404:
            disk_present = False  # 磁盘上游已不存在，视为已删除（幂等）
        else:
            raise

    # 开机挂载状态下需先卸载，再删除
    if disk_present and needs_detach:
        detach_resp, detach_log_id = await upstream_action_service.call_path_with_audit(
            db,
            instance,
            f"/clouds/{instance.upstream_id}/disks/{disk_id}",
            "DELETE",
            action_type="disk_detach",
            resource_type="disk",
            resource_id=disk_id,
            source="unsubscribe",
            user_id=order.user_id,
            order_id=order.id,
            data={"force": 1},
            timeout=30.0,
            remark=f"{reason}（删除前自动卸载）",
        )
        detach_taskid = detach_resp.get("taskid")
        audit_log_ids.append(detach_log_id)
        await _wait_disk_unmounted(db, instance, disk_id)

    delete_taskid = None
    if disk_present:
        response, audit_log_id = await upstream_action_service.call_path_with_audit(
            db,
            instance,
            f"/disks/{disk_id}",
            "DELETE",
            action_type="disk_delete",
            resource_type="disk",
            resource_id=disk_id,
            source="unsubscribe",
            user_id=order.user_id,
            order_id=order.id,
            data={"force": 1},
            timeout=30.0,
            remark=reason,
        )
        delete_taskid = response.get("taskid")
        audit_log_ids.append(audit_log_id)

    await _mark_addons_for_unsubscribe(db, order_id=order.id, refunded=refunded, reason=reason)
    instance.next_renew_amount = int(instance.unit_price or 0) + await _active_disk_total(db, instance.id)
    return {
        "action": "disk_delete",
        "disk_id": disk_id,
        "taskid": delete_taskid,
        "detach_taskid": detach_taskid,
        "audit_log_ids": audit_log_ids,
    }


async def _rollback_package_upgrade(
    db: AsyncSession,
    *,
    order: Order,
    instance: Instance,
    item: dict,
) -> dict:
    from_product_id = _to_int_or_none(item.get("from_product_id"))
    if not from_product_id:
        raise HTTPException(status_code=400, detail="订单缺少原套餐快照，无法回退配置")

    result = await db.execute(select(Product).where(Product.id == from_product_id))
    from_product = result.scalar_one_or_none()
    if not from_product:
        raise HTTPException(status_code=400, detail="原套餐不存在，无法回退配置")

    specs = _product_specs(from_product)
    instance_payload: dict[str, Any] = {}
    for source_key, upstream_key in (
        ("cpu", "cpu"),
        ("memory", "memory"),
        ("traffic_quota", "traffic_quota"),
        ("traffic_type", "traffic_type"),
        ("nat_acl_limit", "nat_acl_limit"),
        ("nat_web_limit", "nat_web_limit"),
    ):
        value = specs.get(source_key)
        if value is not None:
            instance_payload[upstream_key] = value

    bw_payload: dict[str, Any] = {}
    if specs.get("in_bw") is not None:
        bw_payload["in_bw"] = specs["in_bw"]
    if specs.get("out_bw") is not None:
        bw_payload["out_bw"] = specs["out_bw"]

    taskids: list[Any] = []
    audit_log_ids: list[int] = []
    if instance_payload:
        response, audit_log_id = await upstream_action_service.call_path_with_audit(
            db,
            instance,
            f"/clouds/{instance.upstream_id}",
            "PUT",
            action_type="package_instance_rollback",
            resource_type="package",
            resource_id=from_product.id,
            source="unsubscribe",
            user_id=order.user_id,
            order_id=order.id,
            data=instance_payload,
            timeout=30.0,
            remark=f"退订套餐升级，回退到 {from_product.name}",
        )
        taskids.append(response.get("taskid"))
        audit_log_ids.append(audit_log_id)
    if bw_payload:
        response, audit_log_id = await upstream_action_service.call_path_with_audit(
            db,
            instance,
            f"/clouds/{instance.upstream_id}/bw",
            "PUT",
            action_type="package_bandwidth_rollback",
            resource_type="bandwidth",
            resource_id=from_product.id,
            source="unsubscribe",
            user_id=order.user_id,
            order_id=order.id,
            data=bw_payload,
            timeout=30.0,
            remark=f"退订套餐升级，回退到 {from_product.name}",
        )
        taskids.append(response.get("taskid"))
        audit_log_ids.append(audit_log_id)

    unit_price = await _pricing_cents(db, from_product.id, instance.cycle)
    instance.product_id = from_product.id
    instance.unit_price = unit_price
    instance.next_renew_amount = unit_price + await _active_disk_total(db, instance.id)
    for key in ("cpu", "memory", "in_bw", "out_bw", "traffic_quota", "traffic_type", "nat_acl_limit"):
        value = specs.get(key)
        if value is not None and hasattr(instance, key):
            setattr(instance, key, value)
    return {
        "action": "package_upgrade_rollback",
        "from_product_id": from_product.id,
        "taskids": taskids,
        "audit_log_ids": audit_log_ids,
    }


async def _apply_unsubscribe_resource_actions(
    db: AsyncSession,
    *,
    order: Order,
    instances: list[Instance],
    refunded: bool,
    reason: str,
) -> list[dict]:
    actions: list[dict] = []
    items = [item for item in (order.items_snapshot or []) if isinstance(item, dict)]
    item_types = {str(item.get("type") or "server_purchase") for item in items}
    primary_instance = instances[0] if instances else None
    for instance in instances:
        if await zjmf_instance_adapter.get_zjmf_server_for_instance(db, instance):
            raise HTTPException(status_code=400, detail=zjmf_instance_adapter.FINANCE_TICKET_MESSAGE)

    if "disk_resize" in item_types:
        if not primary_instance:
            raise HTTPException(status_code=400, detail="找不到关联实例，无法回退磁盘扩容")
        for item in items:
            if item.get("type") == "disk_resize":
                actions.append(await _rollback_disk_resize(
                    db,
                    order=order,
                    instance=primary_instance,
                    item=item,
                    refunded=refunded,
                    reason=reason,
                ))
        return actions

    if "disk_addon" in item_types:
        if not primary_instance:
            raise HTTPException(status_code=400, detail="找不到关联实例，无法删除新增磁盘")
        for item in items:
            if item.get("type") == "disk_addon":
                actions.append(await _delete_added_disk(
                    db,
                    order=order,
                    instance=primary_instance,
                    item=item,
                    refunded=refunded,
                    reason=reason,
                ))
        return actions

    if "package_upgrade" in item_types:
        if not primary_instance:
            raise HTTPException(status_code=400, detail="找不到关联实例，无法回退套餐")
        for item in items:
            if item.get("type") == "package_upgrade":
                actions.append(await _rollback_package_upgrade(db, order=order, instance=primary_instance, item=item))
        return actions

    for instance in instances:
        if _enum_value(instance.status) in (InstanceStatus.TERMINATED.value, InstanceStatus.FAILED.value):
            continue
        actions.append(await instance_lifecycle_service.force_delete_instance(
            db,
            instance,
            reason=reason,
            source="unsubscribe",
            user_id=order.user_id,
            order_id=order.id,
        ))
    return actions


@router.get("/coupons/available", response_model=ApiResponse)
async def list_available_coupons(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """列出当前用户可用的优惠券（不含 code），供购物车直接选择。

    已激活代理身份的用户返回空列表并标记 agent_disabled，前端可据此提示。
    """
    from app.services import coupon_service

    is_agent = await db.scalar(
        select(UserAgent.id).where(
            UserAgent.user_id == current_user.id,
            UserAgent.status.in_([AgentStatus.ACTIVE, AgentStatus.OBSERVING]),
        )
    )

    if is_agent:
        return ApiResponse(data={"items": [], "agent_disabled": True})

    coupons = await coupon_service.list_available_coupons(db, user_id=current_user.id)
    return ApiResponse(data={
        "items": [
            {
                "id": c.id,
                "description": c.description,
                "type": c.type.value if hasattr(c.type, "value") else str(c.type),
                "value": c.value,
                "threshold_amount": c.min_order_amount,
                "max_discount_amount": c.max_discount_amount,
                "discount_level": c.discount_level.value if hasattr(c.discount_level, "value") else str(c.discount_level),
                "scope": c.scope.value if hasattr(c.scope, "value") else str(c.scope),
                "expires_at": c.expires_at.isoformat() if c.expires_at else None,
            }
            for c in coupons
        ],
        "agent_disabled": False,
    })


@router.post("/coupons/claim-activity", response_model=ApiResponse)
async def claim_activity_coupon(
    body: ClaimActivityCouponRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """从活动页面免费领取优惠券，绑定到当前用户账户。

    幂等：同一用户对同一张券已有 claimed 记录时直接返回成功。
    限制：按券自身的 max_uses_per_user 控制每人可领次数。
    """
    from app.services import coupon_service

    claim = await coupon_service.claim_activity_coupon(
        db,
        promo_code_id=body.promo_code_id,
        user_id=current_user.id,
        activity_id=body.activity_id,
    )
    await db.commit()
    return ApiResponse(data={"claim_id": claim.id, "promo_code_id": claim.promo_code_id})


@router.post("/coupons/preview", response_model=ApiResponse)
async def preview_coupon(
    body: CouponPreviewRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    coupon_code = await _resolve_coupon_code(db, body.coupon_id, body.coupon_code, current_user.id)
    data = await order_service.preview_coupon(
        current_user,
        body.cart_item_ids,
        db,
        coupon_code=coupon_code,
        coupon_target_cart_item_id=body.coupon_target_cart_item_id,
        use_points=body.use_points,
    )
    return ApiResponse(data=data)


@router.post("/checkout", response_model=ApiResponse)
async def checkout(
    body: CheckoutRequest,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    coupon_code = await _resolve_coupon_code(db, body.coupon_id, body.coupon_code, current_user.id)

    # 实名认证检查
    if settings.BAIDU_FACE_VERIFY_REQUIRED and settings.BAIDU_FACE_VERIFY_FOR_PURCHASE:
        from app.services.verification_service import require_verified
        require_verified(current_user)

    order = await order_service.checkout(
        current_user,
        body.cart_item_ids,
        db,
        use_points=body.use_points,
        coupon_code=coupon_code,
        coupon_target_cart_item_id=body.coupon_target_cart_item_id,
    )
    business_settings = await system_setting_service.get_business_settings(db)

    instance_snaps: list[tuple] = getattr(order, "_instance_snaps", [])
    coupon_application = getattr(order, "_coupon_application", None)
    paid_amount = int(getattr(order, "_paid_amount", max(0, order.total_amount - order.discount_amount)) or 0)
    coupon_discount_amount = int(getattr(order, "_coupon_discount_amount", 0) or 0)
    points_discount_amount = int(getattr(order, "_points_discount_amount", 0) or 0)
    await db.commit()

    if current_user.email and order.items_snapshot:
        asyncio.create_task(
            email_service.send_order_placed(
                to=current_user.email,
                username=current_user.username,
                out_order_no=order.out_order_no,
                items=order.items_snapshot,
                total_amount_cents=order.total_amount,
                frontend_url=business_settings.frontend_url,
            )
        )

    for instance, snap in instance_snaps:
        asyncio.create_task(_provision_with_new_session(instance.id, snap))

    await notification_service.safe_notify(
        db,
        user_id=current_user.id,
        category="order",
        title="订单提交成功",
        content=f"订单 {order.out_order_no} 已提交，实付 ¥{paid_amount / 100:.2f}，正在开通中。",
        link="/orders",
    )

    return ApiResponse(data={
        "order_id": order.id,
        "out_order_no": order.out_order_no,
        "total_amount": order.total_amount,
        "discount_amount": order.discount_amount,
        "coupon_discount_amount": coupon_discount_amount,
        "points_discount_amount": points_discount_amount,
        "paid_amount": paid_amount,
        "coupon": (
            order_service.coupon_service.coupon_application_to_dict(coupon_application)
            if coupon_application
            else None
        ),
        "instance_count": len(instance_snaps),
    })


async def _provision_with_new_session(instance_id: int, snap: dict) -> None:
    from app.database import async_session

    try:
        async with async_session() as db:
            try:
                await provision_service.provision_instance(instance_id, snap, db)
                await db.commit()
            except Exception:
                await db.rollback()
                log.exception("provision instance %s failed", instance_id)
    except Exception:
        log.exception("provision session error for instance %s", instance_id)


@router.get("/service-orders", response_model=ApiResponse)
async def list_service_orders(
    limit: int = Query(100, ge=1, le=200),
    offset: int = Query(0, ge=0),
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Order)
        .where(Order.user_id == current_user.id)
        .order_by(Order.id.desc())
        .offset(offset)
        .limit(limit)
    )
    orders = list(result.scalars().all())
    if not orders:
        return ApiResponse(data=[])

    full_refund_unlimited = await _is_active_agent(db, current_user.id)
    settings = await system_setting_service.get_business_settings(db)
    full_count = await _full_refund_count(
        db,
        current_user.id,
        getattr(current_user, "full_refund_quota_reset_at", None),
    )
    full_refund_limit = _full_refund_limit_for_user(settings, current_user)

    order_ids = [int(order.id) for order in orders]
    order_id_set = set(order_ids)
    snapshot_instance_ids: dict[int, set[int]] = {
        int(order.id): _extract_instance_ids(order.items_snapshot)
        for order in orders
    }
    instance_to_order_ids: dict[int, set[int]] = {}
    for order_id, instance_ids in snapshot_instance_ids.items():
        for instance_id in instance_ids:
            instance_to_order_ids.setdefault(instance_id, set()).add(order_id)

    instance_conditions = [Instance.order_id.in_(order_ids)]
    all_snapshot_instance_ids = set(instance_to_order_ids)
    if all_snapshot_instance_ids:
        instance_conditions.append(Instance.id.in_(all_snapshot_instance_ids))
    instance_res = await db.execute(
        select(Instance).where(
            Instance.user_id == current_user.id,
            or_(*instance_conditions),
        )
    )
    instances_by_order: dict[int, dict[int, Instance]] = {order_id: {} for order_id in order_ids}
    for instance in instance_res.scalars().all():
        target_order_ids: set[int] = set()
        if instance.order_id in order_id_set:
            target_order_ids.add(int(instance.order_id))
        target_order_ids.update(instance_to_order_ids.get(int(instance.id), set()))
        for order_id in target_order_ids:
            instances_by_order.setdefault(order_id, {})[int(instance.id)] = instance

    reward_res = await db.execute(
        select(PointReward).where(PointReward.order_id.in_(order_ids))
    )
    rewards_by_order: dict[int, list[PointReward]] = {order_id: [] for order_id in order_ids}
    for reward in reward_res.scalars().all():
        rewards_by_order.setdefault(int(reward.order_id), []).append(reward)

    refund_res = await db.execute(
        select(OrderRefund)
        .where(OrderRefund.order_id.in_(order_ids))
        .order_by(OrderRefund.order_id.asc(), OrderRefund.id.desc())
    )
    latest_refund_by_order: dict[int, OrderRefund] = {}
    for refund in refund_res.scalars().all():
        latest_refund_by_order.setdefault(int(refund.order_id), refund)

    action_res = await db.execute(
        select(UpstreamActionLog)
        .where(UpstreamActionLog.order_id.in_(order_ids))
        .order_by(UpstreamActionLog.id.desc())
    )
    actions_by_order: dict[int, list[UpstreamActionLog]] = {order_id: [] for order_id in order_ids}
    for action in action_res.scalars().all():
        actions_by_order.setdefault(int(action.order_id), []).append(action)

    data = []
    for order in orders:
        order_id = int(order.id)
        instances = list(instances_by_order.get(order_id, {}).values())
        rewards = rewards_by_order.get(order_id, [])
        refund = latest_refund_by_order.get(order_id)
        actions = actions_by_order.get(order_id, [])
        policy = _build_refund_policy(
            order,
            instances,
            rewards,
            full_count,
            refund,
            settings.full_refund_days,
            full_refund_limit,
            allow_full_refund=_is_server_purchase_order(order),
            enforce_full_limit=_is_server_purchase_order(order),
            full_refund_unlimited=full_refund_unlimited,
        )
        data.append(_order_to_service_dict(order, instances, rewards, refund, policy, actions))
    return ApiResponse(data=data)


async def _process_single_refund(
    db: AsyncSession,
    *,
    order: Order,
    instances: list[Instance],
    user: User,
    policy: dict,
) -> OrderRefund:
    """Process refund for a single order: create OrderRefund record,
    update user balance, reverse spend/points/commissions.
    Does NOT commit — caller owns the transaction.
    """
    refund_no = _gen_refund_no()
    refund_type = str(policy["refund_type"])
    refund_amount = int(policy["refund_amount"] or 0)
    paid_amount = int(policy["paid_amount"] or 0)
    reason = str(policy["reason"] or "订单退订")

    refund = OrderRefund(
        refund_no=refund_no,
        user_id=user.id,
        order_id=order.id,
        refund_type=refund_type,
        refund_amount=refund_amount,
        status="success",
        reason=reason,
        payload_snapshot={
            "policy": policy,
            "instance_ids": [inst.id for inst in instances],
            "cascade": policy.get("cascade", False),
        },
    )
    db.add(refund)
    await db.flush()

    if refund_amount > 0:
        user.balance += refund_amount
        db.add(BalanceLog(
            user_id=user.id,
            change_type=BalanceChangeType.ORDER_REFUND,
            delta=refund_amount,
            balance_after=user.balance,
            ref_no=refund_no,
            remark=reason,
        ))
        ratio_bps = 10000 if paid_amount <= 0 else min(10000, int(refund_amount * 10000 / paid_amount))
        await spend_service.reverse_spend_events_for_order(
            db, user_id=user.id, order_id=order.id,
            refund_no=refund_no, reason=reason, ratio_bps=ratio_bps,
        )
        await membership_service.refund_used_points_for_order(
            db, user_id=user.id, order_id=order.id,
            order_no=order.out_order_no, ratio_bps=ratio_bps, reason=reason,
        )
        await membership_service.reverse_order_point_rewards(
            db, order_id=order.id, reason=reason,
        )
        await agent_service.reverse_order_settlements(db, order, reason=reason)
        order.status = OrderStatus.REFUNDED
    else:
        await membership_service.cancel_unclaimed_order_point_rewards(
            db, order_id=order.id, reason=reason,
        )
        order.status = OrderStatus.CANCELLED

    order.remark = reason
    await db.flush()
    return refund


@router.post("/{order_id}/unsubscribe", response_model=ApiResponse)
async def unsubscribe_order(
    order_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    """Unsubscribe a service order. If it's a server_purchase,
    all related orders (upgrade, disk) on the same instance are
    also cascaded in reverse dependency order.
    """
    # lock user
    user_result = await db.execute(
        select(User).where(User.id == current_user.id).with_for_update()
    )
    user = user_result.scalar_one()
    if not user:
        raise HTTPException(status_code=404, detail="用户不存在")

    # lock order
    order_result = await db.execute(
        select(Order)
        .where(Order.id == order_id, Order.user_id == current_user.id)
        .with_for_update()
    )
    order = order_result.scalar_one_or_none()
    if not order:
        raise HTTPException(status_code=404, detail="订单不存在")

    # load context
    latest_refund = await _latest_refund(db, order.id)
    instances = await _load_order_instances(db, order, for_update=True)
    rewards = await _load_order_point_rewards(db, order.id, for_update=True)
    full_refund_unlimited = await _is_active_agent(db, current_user.id)
    settings = await system_setting_service.get_business_settings(db)
    full_count = await _full_refund_count(
        db,
        current_user.id,
        getattr(user, "full_refund_quota_reset_at", None),
    )
    full_refund_limit = _full_refund_limit_for_user(settings, user)

    # Determine order type
    is_server_purchase = _is_server_purchase_order(order)

    # ---- cascade: find related orders ----
    related_orders: list[Order] = []
    if is_server_purchase:
        related_orders = await _load_related_orders(
            db, order, instances, for_update=True,
        )
        related_orders = _sort_orders_by_dependency(related_orders)

    all_orders = [*related_orders, order]  # dependencies first, purchase last

    # ---- validate all orders can be unsubscribed ----
    all_policies: list[dict] = []
    for o in all_orders:
        o_instances = await _load_order_instances(db, o, for_update=True)
        o_rewards = await _load_order_point_rewards(db, o.id, for_update=True)
        lr = await _latest_refund(db, o.id)
        policy = _build_refund_policy(
            o, o_instances, o_rewards, full_count, lr,
            settings.full_refund_days, full_refund_limit,
            allow_full_refund=_is_server_purchase_order(o),
            enforce_full_limit=_is_server_purchase_order(o),
            full_refund_unlimited=full_refund_unlimited,
        )
        if not policy["action_allowed"]:
            raise HTTPException(
                status_code=400,
                detail=f"订单 {o.out_order_no} {policy['reason']}",
            )
        all_policies.append(policy)

    # ---- process resource actions & refunds in order ----
    all_refunds: list[dict] = []
    total_refund = 0
    instance_deleted = False

    for idx, (o, pol) in enumerate(zip(all_orders, all_policies)):
        o_instances = await _load_order_instances(db, o, for_update=True)
        refunded = int(pol["refund_amount"] or 0) > 0
        o_is_purchase = o is order and is_server_purchase

        if is_server_purchase and not o_is_purchase:
            # 级联删服务器：实例将被删除并连带清掉所有磁盘，因此关联订单
            # （套餐升级回退 / 磁盘卸载删除）的上游操作已无意义，跳过以省去
            # 卸载轮询并减少失败面。仅标记本地磁盘记录，财务退款照常。
            await _mark_addons_for_unsubscribe(
                db, order_id=o.id, refunded=refunded, reason=str(pol["reason"]),
            )
            resource_actions: list = []
        else:
            resource_actions = await _apply_unsubscribe_resource_actions(
                db, order=o, instances=o_instances,
                refunded=refunded, reason=str(pol["reason"]),
            )
            if o_is_purchase:
                instance_deleted = True

        # financial
        refund = await _process_single_refund(
            db, order=o, instances=o_instances, user=user, policy=pol,
        )

        # link audit logs
        audit_log_ids = _collect_audit_log_ids(resource_actions)
        if audit_log_ids:
            await db.execute(
                update(UpstreamActionLog)
                .where(UpstreamActionLog.id.in_(audit_log_ids))
                .values(order_refund_id=refund.id)
            )
        await upstream_action_service.refresh_refund_resource_status(db, refund.id)

        all_refunds.append({
            "refund": _refund_to_dict(refund),
            "order": _order_to_service_dict(
                o, o_instances,
                await _load_order_point_rewards(db, o.id),
                refund,
                pol,
                await _load_order_actions(db, o.id),
            ),
        })
        total_refund += int(pol["refund_amount"] or 0)

    await notification_service.safe_notify(
        db,
        user_id=current_user.id,
        category="finance",
        title="退订已处理",
        content=(
            f"已退订 {len(all_refunds)} 笔订单"
            + (f"，共退款 ¥{total_refund / 100:.2f} 至账户余额。" if total_refund > 0 else "。")
        ),
        link="/finance",
    )

    return ApiResponse(data={
        "refunds": all_refunds,
        "total_refund": total_refund,
        "instance_deleted": instance_deleted,
    }, message="退订已处理")


@router.get("", response_model=ApiResponse)
async def list_orders(
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Order)
        .where(Order.user_id == current_user.id)
        .order_by(Order.id.desc())
    )
    orders = result.scalars().all()
    return ApiResponse(data=[
        {
            "id": order.id,
            "out_order_no": order.out_order_no,
            "status": _enum_value(order.status),
            "total_amount": order.total_amount,
            "discount_amount": order.discount_amount,
            "created_at": order.created_at.isoformat(),
        }
        for order in orders
    ])


@router.get("/{order_id}", response_model=ApiResponse)
async def get_order(
    order_id: int,
    current_user: User = Depends(get_current_user),
    db: AsyncSession = Depends(get_db),
):
    result = await db.execute(
        select(Order).where(Order.id == order_id, Order.user_id == current_user.id)
    )
    order = result.scalar_one_or_none()
    if not order:
        raise HTTPException(status_code=404, detail="订单不存在")

    instance_result = await db.execute(
        select(Instance).where(Instance.order_id == order_id)
    )
    instances = instance_result.scalars().all()

    return ApiResponse(data={
        "id": order.id,
        "out_order_no": order.out_order_no,
        "status": _enum_value(order.status),
        "total_amount": order.total_amount,
        "discount_amount": order.discount_amount,
        "items_snapshot": order.items_snapshot,
        "created_at": order.created_at.isoformat(),
        "instances": [
            {
                "id": instance.id,
                "status": _enum_value(instance.status),
                "main_ip": instance.main_ip,
                "hostname": instance.hostname,
                "expires_at": instance.expires_at.isoformat() if instance.expires_at else None,
            }
            for instance in instances
        ],
    })
