video-gen/video-gen-api/app/services/credit/subscription_service.py

from __future__ import annotations

from datetime import datetime
from decimal import Decimal

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

from app.enums.credit_balance import (
    CreditAllocationAction,
    CreditBalanceSourceType,
    CreditBalanceStatus,
)
from app.enums.credit_product import CreditProductType
from app.enums.credit_record import CreditRecordBillingScene, CreditRecordType
from app.enums.credit_subscription import (
    CreditSubscriptionPeriodStatus,
    CreditSubscriptionStatus,
)
from app.models.credit.allocation import CreditRecordAllocation
from app.models.credit.balance import UserCreditBalance
from app.models.credit.product import CreditProduct
from app.models.credit.subscription import UserCreditSubscription
from app.models.credit.subscription_period import UserCreditSubscriptionPeriod
from app.models.credit_record import CreditRecord
from app.models.payment_order import PaymentOrder
from app.models.user import User
from app.services.credit.ledger_service import grant_credits
from app.services.credit.locking import acquire_user_credit_lock
from app.services.credit.product_service import grant_count_for_cycle
from app.services.credit.query_service import get_available_credits
from app.services.credit.time_policy import add_natural_months, natural_month_period
from app.services.credit.utils import to_credit_decimal, utc_now
from app.services.operation_log_service import log_operation_event
from app.utils.id_gen import generate_id


def _allocate_amount_by_period(total: Decimal, count: int) -> list[Decimal]:
    cents = int((to_credit_decimal(total) * 100).to_integral_value())
    base, remainder = divmod(cents, count)
    values = [Decimal(base) / 100 for _ in range(count)]
    values[-1] += Decimal(remainder) / 100
    return [to_credit_decimal(item) for item in values]


def _product_snapshot(product: CreditProduct) -> dict:
    return {
        "id": product.id,
        "product_code": product.product_code,
        "product_type": product.product_type,
        "name": product.name,
        "tier_code": product.tier_code,
        "tier_rank": product.tier_rank,
        "billing_cycle": product.billing_cycle,
        "monthly_grant_credits": float(product.monthly_grant_credits or 0),
        "first_purchase_price": float(product.first_purchase_price or 0),
        "regular_price": float(product.regular_price or 0),
        "activity_price": float(product.activity_price) if product.activity_price is not None else None,
        "grant_credits": float(product.grant_credits or 0),
        "credit_level": product.credit_level,
        "features": product.features_json or [],
    }


def _resolve_order_product_snapshot(
    order: PaymentOrder,
    product: CreditProduct | None,
) -> dict:
    """返回支付创建时冻结的商品快照。

    支付履约不得读取后台后来修改后的价格、积分或套餐周期。旧订单若尚未
    保存快照，才允许使用当前商品生成一次兼容快照。
    """
    snapshot = dict(order.product_snapshot_json or {})
    if not snapshot and product is not None:
        snapshot = _product_snapshot(product)
    if not snapshot:
        raise ValueError("订单缺少商品快照，无法履约")
    return snapshot


def _snapshot_decimal(snapshot: dict, key: str) -> Decimal:
    return to_credit_decimal(snapshot.get(key) or 0)


async def _create_subscription(
    db: AsyncSession,
    *,
    order: PaymentOrder,
    product: CreditProduct | None,
    paid_at: datetime,
) -> tuple[UserCreditSubscription, list[UserCreditSubscriptionPeriod]]:
    snapshot = _resolve_order_product_snapshot(order, product)
    billing_cycle = str(snapshot.get("billing_cycle") or "")
    count = grant_count_for_cycle(billing_cycle)
    expires_at = add_natural_months(paid_at, count)
    # 升级订单的现金实付额已扣除旧套餐未来周期价值；新订阅后续再升级时，
    # 周期价值必须按目标套餐完整价格快照分摊，不能只按本次现金实付额分摊。
    pricing_basis_amount = to_credit_decimal(
        order.target_price_snapshot
        if order.purchase_scene == "upgrade" and order.target_price_snapshot is not None
        else order.amount
    )
    subscription = UserCreditSubscription(
        id=generate_id(),
        user_id=order.user_id,
        product_id=order.product_id,
        payment_order_id=order.id,
        status=CreditSubscriptionStatus.ACTIVE.value,
        purchase_scene=str(order.purchase_scene or "renewal"),
        tier_code=str(snapshot.get("tier_code") or ""),
        tier_rank=int(snapshot.get("tier_rank") or 0),
        billing_cycle=billing_cycle,
        anchor_at=paid_at,
        start_at=paid_at,
        expires_at=expires_at,
        next_grant_at=(add_natural_months(paid_at, 1) if count > 1 else None),
        monthly_grant_credits_snapshot=_snapshot_decimal(snapshot, "monthly_grant_credits"),
        grant_count=count,
        granted_count=0,
        paid_amount_snapshot=pricing_basis_amount,
        product_snapshot_json=snapshot,
        source_subscription_id=order.source_subscription_id,
        upgrade_order_id=(order.id if order.purchase_scene == "upgrade" else None),
    )
    db.add(subscription)
    await db.flush()

    allocations = _allocate_amount_by_period(pricing_basis_amount, count)
    periods: list[UserCreditSubscriptionPeriod] = []
    for sequence in range(count):
        start, end = natural_month_period(paid_at, sequence)
        period = UserCreditSubscriptionPeriod(
            id=generate_id(),
            subscription_id=subscription.id,
            sequence=sequence + 1,
            scheduled_at=start,
            valid_from=start,
            expires_at=end,
            grant_credits=_snapshot_decimal(snapshot, "monthly_grant_credits"),
            allocated_paid_amount=allocations[sequence],
            status=CreditSubscriptionPeriodStatus.SCHEDULED.value,
        )
        db.add(period)
        periods.append(period)
    await db.flush()
    return subscription, periods


async def grant_subscription_period(
    db: AsyncSession,
    *,
    subscription: UserCreditSubscription,
    period: UserCreditSubscriptionPeriod,
    request_time: datetime | None = None,
) -> UserCreditBalance:
    checked_at = request_time or utc_now()
    if period.status == CreditSubscriptionPeriodStatus.GRANTED.value and period.issued_balance_id:
        result = await db.execute(
            select(UserCreditBalance).where(UserCreditBalance.id == period.issued_balance_id).limit(1)
        )
        existing = result.scalar_one_or_none()
        if existing:
            return existing
    if period.status not in {
        CreditSubscriptionPeriodStatus.SCHEDULED.value,
        CreditSubscriptionPeriodStatus.GRANTED.value,
    }:
        raise ValueError("当前订阅周期不能发放积分")
    mutation = await grant_credits(
        db,
        user_id=subscription.user_id,
        amount=period.grant_credits,
        description=f"订阅套餐第{period.sequence}个月积分发放",
        source_type=CreditBalanceSourceType.SUBSCRIPTION_GRANT.value,
        valid_from=period.valid_from,
        expires_at=period.expires_at,
        credit_level=str(subscription.product_snapshot_json.get("credit_level") or "general"),
        source_id=period.id,
        product_id=subscription.product_id,
        payment_order_id=subscription.payment_order_id,
        subscription_id=subscription.id,
        subscription_period_id=period.id,
        related_id=subscription.id,
        record_type=CreditRecordType.RECHARGE.value,
        biz_key=f"subscription:{subscription.id}:period:{period.sequence}:grant",
        metadata_json={"period_sequence": period.sequence},
        request_time=checked_at,
    )
    result = await db.execute(
        select(UserCreditBalance)
        .where(UserCreditBalance.grant_record_id == mutation.record.id)
        .limit(1)
    )
    balance = result.scalar_one()
    period.status = CreditSubscriptionPeriodStatus.GRANTED.value
    period.issued_balance_id = balance.id
    period.issued_at = checked_at
    subscription.granted_count = max(subscription.granted_count, period.sequence)
    subscription.next_grant_at = (
        add_natural_months(subscription.anchor_at, period.sequence)
        if period.sequence < subscription.grant_count
        else None
    )
    await db.flush()
    return balance


async def _reconcile_upgrade_period_mismatch(
    db: AsyncSession,
    *,
    order: PaymentOrder,
    new_first_balance: UserCreditBalance,
    checked_at: datetime,
) -> None:
    """核对升级订单创建时的抵扣周期与支付结算时状态。

    正常情况下抵扣周期一直处于 upgrade_reserved，仅取消未来发放。
    若历史竞态导致周期已发放，则撤销未消费积分，并把每一笔尚未退款的
    消费分摊通过 source_allocation_id 迁移到升级套餐首期积分，保证后续
    业务失败退款仍能准确退回当前实际承担成本的积分来源。
    """
    period_ids = list(order.upgrade_period_ids_json or [])
    if not period_ids:
        return
    result = await db.execute(
        select(UserCreditSubscriptionPeriod)
        .where(UserCreditSubscriptionPeriod.id.in_(period_ids))
        .order_by(UserCreditSubscriptionPeriod.id.asc())
        .with_for_update()
    )
    periods = list(result.scalars().all())
    expected_ids = [str(item) for item in period_ids]
    if len(expected_ids) != len(set(expected_ids)):
        raise RuntimeError("升级订单抵扣周期快照存在重复ID")
    actual_ids = {str(period.id) for period in periods}
    if actual_ids != set(expected_ids):
        raise RuntimeError("升级订单抵扣周期快照与实际周期集合不一致")
    if not order.source_subscription_id:
        raise RuntimeError("升级订单缺少原订阅ID")
    allowed_statuses = {
        CreditSubscriptionPeriodStatus.UPGRADE_RESERVED.value,
        CreditSubscriptionPeriodStatus.GRANTED.value,
    }
    for period in periods:
        if str(period.subscription_id) != str(order.source_subscription_id):
            raise RuntimeError("升级订单抵扣周期不属于原订阅")
        if str(period.upgrade_order_id or "") != str(order.id):
            raise RuntimeError("升级订单抵扣周期未绑定当前订单")
        if period.status not in allowed_statuses:
            raise RuntimeError(f"升级订单抵扣周期状态不允许结算：{period.status}")
        if period.status == CreditSubscriptionPeriodStatus.GRANTED.value and not period.issued_balance_id:
            raise RuntimeError("升级订单抵扣周期已发放但缺少积分来源记录")
        if period.status == CreditSubscriptionPeriodStatus.UPGRADE_RESERVED.value and period.issued_balance_id:
            raise RuntimeError("升级订单抵扣周期仍为预留状态但已存在积分来源记录")

    reserved = [
        period for period in periods
        if period.status == CreditSubscriptionPeriodStatus.UPGRADE_RESERVED.value
    ]
    inconsistent = [
        period
        for period in periods
        if period.status == CreditSubscriptionPeriodStatus.GRANTED.value
        and period.issued_balance_id
    ]
    for period in reserved:
        period.status = CreditSubscriptionPeriodStatus.CANCELLED_BY_UPGRADE.value
        period.cancelled_at = checked_at
    if not inconsistent:
        return

    balance_ids = [str(period.issued_balance_id) for period in inconsistent if period.issued_balance_id]
    balance_result = await db.execute(
        select(UserCreditBalance)
        .where(UserCreditBalance.id.in_(balance_ids))
        .order_by(UserCreditBalance.id.asc())
        .with_for_update()
    )
    old_balances = list(balance_result.scalars().all())
    if {str(item.id) for item in old_balances} != set(balance_ids):
        raise RuntimeError("升级边界补偿缺少原积分批次")
    period_by_balance = {str(period.issued_balance_id): period for period in inconsistent}
    for balance in old_balances:
        period = period_by_balance.get(str(balance.id))
        if period is None:
            raise RuntimeError("升级边界补偿积分批次与周期无法对应")
        if str(balance.user_id) != str(order.user_id):
            raise RuntimeError("升级边界补偿积分批次不属于当前用户")
        if str(balance.subscription_id or "") != str(order.source_subscription_id):
            raise RuntimeError("升级边界补偿积分批次不属于原订阅")
        if str(balance.subscription_period_id or "") != str(period.id):
            raise RuntimeError("升级边界补偿积分批次与订阅周期不一致")
    total_transfer = sum(
        (to_credit_decimal(item.consumed_amount) for item in old_balances),
        Decimal("0.00"),
    )
    if to_credit_decimal(new_first_balance.unspent_amount) < total_transfer:
        raise RuntimeError("升级套餐首期积分不足以承接边界消费来源迁移")

    before_available = await get_available_credits(db, order.user_id, request_time=checked_at)
    record = CreditRecord(
        id=generate_id(),
        user_id=order.user_id,
        type=CreditRecordType.REVOKE.value,
        amount=Decimal("0.00"),
        balance_delta=Decimal("0.00"),
        expired_amount=Decimal("0.00"),
        balance_after=before_available,
        description="套餐升级边界核对：废弃重复发放积分并迁移已消费来源",
        related_id=order.id,
        request_time=checked_at,
        biz_key=f"payment-order:{order.id}:upgrade-reconcile",
        billing_scene=CreditRecordBillingScene.CREDIT_REVOKE.value,
    )
    db.add(record)
    await db.flush()

    new_unspent_cursor = to_credit_decimal(new_first_balance.unspent_amount)
    new_consumed_cursor = to_credit_decimal(new_first_balance.consumed_amount)
    total_removed = Decimal("0.00")

    for old in old_balances:
        old_unspent = to_credit_decimal(old.unspent_amount)
        old_consumed = to_credit_decimal(old.consumed_amount)
        total_removed += old_unspent + old_consumed

        if old_unspent > 0:
            db.add(
                CreditRecordAllocation(
                    id=generate_id(),
                    credit_record_id=record.id,
                    credit_balance_id=old.id,
                    user_id=order.user_id,
                    allocation_action=CreditAllocationAction.REVOKE.value,
                    amount=old_unspent,
                    request_time=checked_at,
                    credit_level_snapshot=old.credit_level,
                    source_type_snapshot=old.source_type,
                    source_id_snapshot=old.source_id,
                    valid_from_snapshot=old.valid_from,
                    expires_at_snapshot=old.expires_at,
                    unspent_before=old_unspent,
                    unspent_after=Decimal("0.00"),
                    consumed_before=old_consumed,
                    consumed_after=old_consumed,
                )
            )

        if old_consumed > 0:
            consume_result = await db.execute(
                select(CreditRecordAllocation)
                .where(
                    CreditRecordAllocation.credit_balance_id == old.id,
                    CreditRecordAllocation.allocation_action
                    == CreditAllocationAction.CONSUME.value,
                )
                .order_by(CreditRecordAllocation.created_at.asc(), CreditRecordAllocation.id.asc())
            )
            original_allocations = list(consume_result.scalars().all())
            original_ids = [item.id for item in original_allocations]
            adjusted_by_source: dict[str, Decimal] = {}
            if original_ids:
                adjusted_result = await db.execute(
                    select(CreditRecordAllocation).where(
                        CreditRecordAllocation.source_allocation_id.in_(original_ids),
                        CreditRecordAllocation.allocation_action.in_(
                            [
                                CreditAllocationAction.REFUND_AVAILABLE.value,
                                CreditAllocationAction.REFUND_EXPIRED.value,
                                CreditAllocationAction.UPGRADE_SOURCE_TRANSFER_OUT.value,
                            ]
                        ),
                    )
                )
                for adjusted in adjusted_result.scalars().all():
                    if adjusted.source_allocation_id:
                        adjusted_by_source[adjusted.source_allocation_id] = (
                            adjusted_by_source.get(adjusted.source_allocation_id, Decimal("0.00"))
                            + to_credit_decimal(adjusted.amount)
                        )
            remaining = old_consumed
            old_consumed_cursor = old_consumed
            for original_allocation in original_allocations:
                if remaining <= 0:
                    break
                original_amount = to_credit_decimal(original_allocation.amount)
                already_adjusted = adjusted_by_source.get(original_allocation.id, Decimal("0.00"))
                available_to_transfer = original_amount - already_adjusted
                if available_to_transfer < 0:
                    raise RuntimeError("消费分摊的退款或迁移金额超过原消费金额")
                transfer_amount = min(available_to_transfer, remaining)
                if transfer_amount <= 0:
                    continue
                db.add(
                    CreditRecordAllocation(
                        id=generate_id(),
                        credit_record_id=record.id,
                        credit_balance_id=old.id,
                        user_id=order.user_id,
                        source_allocation_id=original_allocation.id,
                        allocation_action=CreditAllocationAction.UPGRADE_SOURCE_TRANSFER_OUT.value,
                        amount=transfer_amount,
                        request_time=checked_at,
                        credit_level_snapshot=old.credit_level,
                        source_type_snapshot=old.source_type,
                        source_id_snapshot=old.source_id,
                        valid_from_snapshot=old.valid_from,
                        expires_at_snapshot=old.expires_at,
                        unspent_before=Decimal("0.00"),
                        unspent_after=Decimal("0.00"),
                        consumed_before=old_consumed_cursor,
                        consumed_after=old_consumed_cursor - transfer_amount,
                    )
                )
                db.add(
                    CreditRecordAllocation(
                        id=generate_id(),
                        credit_record_id=record.id,
                        credit_balance_id=new_first_balance.id,
                        user_id=order.user_id,
                        source_allocation_id=original_allocation.id,
                        allocation_action=CreditAllocationAction.UPGRADE_SOURCE_TRANSFER_IN.value,
                        amount=transfer_amount,
                        request_time=checked_at,
                        credit_level_snapshot=new_first_balance.credit_level,
                        source_type_snapshot=new_first_balance.source_type,
                        source_id_snapshot=new_first_balance.source_id,
                        valid_from_snapshot=new_first_balance.valid_from,
                        expires_at_snapshot=new_first_balance.expires_at,
                        unspent_before=new_unspent_cursor,
                        unspent_after=new_unspent_cursor - transfer_amount,
                        consumed_before=new_consumed_cursor,
                        consumed_after=new_consumed_cursor + transfer_amount,
                    )
                )
                old_consumed_cursor -= transfer_amount
                new_unspent_cursor -= transfer_amount
                new_consumed_cursor += transfer_amount
                remaining -= transfer_amount
            if remaining > 0:
                raise RuntimeError("无法定位全部已消费积分的原始业务分摊，升级结算已中止")

        old.unspent_amount = Decimal("0.00")
        old.consumed_amount = Decimal("0.00")
        old.revoked_amount = to_credit_decimal(old.revoked_amount) + old_unspent + old_consumed
        old.revoked_at = checked_at
        old.status = CreditBalanceStatus.REVOKED.value
        for period in inconsistent:
            if period.issued_balance_id == old.id:
                period.status = CreditSubscriptionPeriodStatus.REVOKED_BY_UPGRADE.value
                period.revoked_at = checked_at

    new_first_balance.unspent_amount = new_unspent_cursor
    new_first_balance.consumed_amount = new_consumed_cursor
    if new_unspent_cursor == 0:
        new_first_balance.status = CreditBalanceStatus.CONSUMED.value

    record.amount = -total_removed
    record.balance_delta = -total_removed
    record.balance_after = before_available - total_removed
    await db.flush()


async def fulfill_payment_product(
    db: AsyncSession,
    *,
    order: PaymentOrder,
    fulfilled_at: datetime | None = None,
) -> None:
    if order.fulfillment_status == "fulfilled":
        return
    checked_at = fulfilled_at or order.paid_at or utc_now()
    await acquire_user_credit_lock(db, order.user_id)
    user_result = await db.execute(select(User).where(User.id == order.user_id).with_for_update().limit(1))
    user = user_result.scalar_one()
    product_result = await db.execute(
        select(CreditProduct).where(CreditProduct.id == order.product_id).limit(1)
    )
    product = product_result.scalar_one_or_none()
    snapshot = _resolve_order_product_snapshot(order, product)
    product_type = str(order.product_type or snapshot.get("product_type") or "")

    if product_type == CreditProductType.CREDIT_ADDON.value:
        await grant_credits(
            db,
            user_id=order.user_id,
            amount=_snapshot_decimal(snapshot, "grant_credits"),
            description=f"购买积分增值包：{order.product_name_snapshot or snapshot.get('name') or '积分增值包'}",
            source_type=CreditBalanceSourceType.CREDIT_ADDON.value,
            valid_from=checked_at,
            expires_at=add_natural_months(checked_at, 1),
            credit_level=str(snapshot.get("credit_level") or "general"),
            source_id=order.id,
            product_id=order.product_id,
            payment_order_id=order.id,
            related_id=order.id,
            biz_key=f"payment-order:{order.id}:credit-addon:grant",
            request_time=checked_at,
        )
        order.fulfillment_status = "fulfilled"
        order.fulfilled_at = checked_at
        return

    old_subscription: UserCreditSubscription | None = None
    if order.purchase_scene == "upgrade":
        if not order.source_subscription_id:
            order.fulfillment_status = CreditSubscriptionStatus.UPGRADE_RECONCILE_FAILED.value
            await db.flush()
            return
        old_result = await db.execute(
            select(UserCreditSubscription)
            .where(UserCreditSubscription.id == order.source_subscription_id)
            .with_for_update()
            .limit(1)
        )
        old_subscription = old_result.scalar_one_or_none()
        if old_subscription is None or str(old_subscription.user_id) != str(order.user_id):
            order.fulfillment_status = CreditSubscriptionStatus.UPGRADE_RECONCILE_FAILED.value
            await db.flush()
            return

    if order.purchase_scene != "upgrade":
        # 首充/续费维持原有支付履约语义：创建或发放异常继续向外抛出，
        # 不能被升级专用的核对失败状态吞掉。
        subscription, periods = await _create_subscription(
            db,
            order=order,
            product=product,
            paid_at=checked_at,
        )
        await grant_subscription_period(
            db,
            subscription=subscription,
            period=periods[0],
            request_time=checked_at,
        )
    else:
        try:
            async with db.begin_nested():
                subscription, periods = await _create_subscription(
                    db,
                    order=order,
                    product=product,
                    paid_at=checked_at,
                )
                first_balance = await grant_subscription_period(
                    db,
                    subscription=subscription,
                    period=periods[0],
                    request_time=checked_at,
                )
                await _reconcile_upgrade_period_mismatch(
                    db,
                    order=order,
                    new_first_balance=first_balance,
                    checked_at=checked_at,
                )
        except Exception as exc:
            order.fulfillment_status = CreditSubscriptionStatus.UPGRADE_RECONCILE_FAILED.value
            await db.flush()
            log_operation_event(
                domain="billing",
                module="credit_upgrade",
                event_type="CREDIT_UPGRADE_RECONCILE_FAILED",
                event_status="failed",
                source="app.services.credit.subscription_service.fulfill_payment_product",
                user_id=str(order.user_id),
                task_id=str(order.id),
                message="套餐升级支付履约周期核对失败",
                error=str(exc),
                detail={
                    "order_id": str(order.id),
                    "order_no": str(order.order_no),
                    "source_subscription_id": str(order.source_subscription_id or ""),
                    "upgrade_period_ids": list(order.upgrade_period_ids_json or []),
                },
            )
            return

    if old_subscription is not None:
        old_subscription.status = CreditSubscriptionStatus.UPGRADED.value
        old_subscription.upgrade_order_id = order.id
    order.subscription_id = subscription.id
    order.fulfillment_status = "fulfilled"
    order.fulfilled_at = checked_at
    if user.first_membership_paid_at is None:
        user.first_membership_paid_at = checked_at
    await db.flush()


async def list_due_subscription_period_candidates(
    db: AsyncSession,
    *,
    request_time: datetime,
    batch_size: int = 100,
) -> list[tuple[str, str, str]]:
    result = await db.execute(
        select(
            UserCreditSubscriptionPeriod.id,
            UserCreditSubscriptionPeriod.subscription_id,
            UserCreditSubscription.user_id,
        )
        .join(
            UserCreditSubscription,
            UserCreditSubscription.id == UserCreditSubscriptionPeriod.subscription_id,
        )
        .where(
            UserCreditSubscription.status == CreditSubscriptionStatus.ACTIVE.value,
            UserCreditSubscriptionPeriod.status == CreditSubscriptionPeriodStatus.SCHEDULED.value,
            UserCreditSubscriptionPeriod.scheduled_at <= request_time,
        )
        .order_by(UserCreditSubscriptionPeriod.scheduled_at.asc(), UserCreditSubscriptionPeriod.id.asc())
        .limit(max(1, batch_size))
    )
    return [(str(row.id), str(row.subscription_id), str(row.user_id)) for row in result.all()]


async def grant_due_subscription_period_by_id(
    db: AsyncSession,
    *,
    period_id: str,
    subscription_id: str,
    user_id: str,
    request_time: datetime,
) -> bool:
    await acquire_user_credit_lock(db, user_id)
    subscription_result = await db.execute(
        select(UserCreditSubscription)
        .where(UserCreditSubscription.id == subscription_id)
        .with_for_update()
        .limit(1)
    )
    subscription = subscription_result.scalar_one_or_none()
    if subscription is None or subscription.status != CreditSubscriptionStatus.ACTIVE.value:
        return False
    period_result = await db.execute(
        select(UserCreditSubscriptionPeriod)
        .where(UserCreditSubscriptionPeriod.id == period_id)
        .with_for_update()
        .limit(1)
    )
    period = period_result.scalar_one_or_none()
    if (
        period is None
        or period.subscription_id != subscription.id
        or period.status != CreditSubscriptionPeriodStatus.SCHEDULED.value
        or period.scheduled_at > request_time
    ):
        return False
    await grant_subscription_period(
        db,
        subscription=subscription,
        period=period,
        request_time=request_time,
    )
    return True


async def grant_due_subscription_periods(
    db: AsyncSession,
    *,
    request_time: datetime | None = None,
    batch_size: int = 100,
) -> int:
    checked_at = request_time or utc_now()
    candidates = await list_due_subscription_period_candidates(
        db, request_time=checked_at, batch_size=batch_size
    )
    count = 0
    for period_id, subscription_id, user_id in candidates:
        if await grant_due_subscription_period_by_id(
            db,
            period_id=period_id,
            subscription_id=subscription_id,
            user_id=user_id,
            request_time=checked_at,
        ):
            count += 1
    return count


async def list_due_subscription_ids(
    db: AsyncSession,
    *,
    request_time: datetime,
    batch_size: int = 500,
) -> list[str]:
    result = await db.execute(
        select(UserCreditSubscription.id)
        .where(
            UserCreditSubscription.status == CreditSubscriptionStatus.ACTIVE.value,
            UserCreditSubscription.expires_at <= request_time,
        )
        .order_by(UserCreditSubscription.expires_at.asc(), UserCreditSubscription.id.asc())
        .limit(max(1, batch_size))
    )
    return [str(item) for item in result.scalars().all()]


async def expire_subscription_by_id(
    db: AsyncSession,
    *,
    subscription_id: str,
    request_time: datetime,
) -> bool:
    result = await db.execute(
        select(UserCreditSubscription)
        .where(UserCreditSubscription.id == subscription_id)
        .with_for_update()
        .limit(1)
    )
    subscription = result.scalar_one_or_none()
    if (
        subscription is None
        or subscription.status != CreditSubscriptionStatus.ACTIVE.value
        or subscription.expires_at > request_time
    ):
        return False
    subscription.status = CreditSubscriptionStatus.EXPIRED.value
    subscription.next_grant_at = None
    await db.flush()
    return True


async def expire_due_subscriptions(
    db: AsyncSession,
    *,
    request_time: datetime | None = None,
    batch_size: int = 500,
) -> int:
    checked_at = request_time or utc_now()
    ids = await list_due_subscription_ids(db, request_time=checked_at, batch_size=batch_size)
    count = 0
    for subscription_id in ids:
        if await expire_subscription_by_id(
            db, subscription_id=subscription_id, request_time=checked_at
        ):
            count += 1
    return count


async def revoke_payment_order_credits(
    db: AsyncSession,
    *,
    order: PaymentOrder,
    reason: str,
    request_time: datetime | None = None,
) -> float:
    """支付现有退款流程的积分账本适配：只撤销该订单当前尚未消费的积分。"""
    from app.services.credit.ledger_service import revoke_balances

    checked_at = request_time or utc_now()
    await acquire_user_credit_lock(db, order.user_id)
    balance_result = await db.execute(
        select(UserCreditBalance)
        .where(
            UserCreditBalance.payment_order_id == order.id,
            UserCreditBalance.user_id == order.user_id,
            UserCreditBalance.unspent_amount > 0,
            UserCreditBalance.valid_from <= checked_at,
            UserCreditBalance.expires_at > checked_at,
            UserCreditBalance.revoked_at.is_(None),
        )
        .order_by(UserCreditBalance.id.asc())
        .with_for_update()
    )
    balances = list(balance_result.scalars().all())
    revoked = sum((to_credit_decimal(item.unspent_amount) for item in balances), Decimal("0.00"))
    if balances:
        await revoke_balances(
            db,
            balances=balances,
            description=f"{reason}：撤销订单未消费积分",
            related_id=order.id,
            biz_key=f"payment-order:{order.id}:refund-revoke",
            request_time=checked_at,
        )

    if order.subscription_id:
        subscription_result = await db.execute(
            select(UserCreditSubscription)
            .where(UserCreditSubscription.id == order.subscription_id)
            .limit(1)
            .with_for_update()
        )
        subscription = subscription_result.scalar_one_or_none()
        if subscription:
            subscription.status = CreditSubscriptionStatus.REFUNDED.value
            periods_result = await db.execute(
                select(UserCreditSubscriptionPeriod)
                .where(
                    UserCreditSubscriptionPeriod.subscription_id == subscription.id,
                    UserCreditSubscriptionPeriod.status.in_([
                        CreditSubscriptionPeriodStatus.SCHEDULED.value,
                        CreditSubscriptionPeriodStatus.UPGRADE_RESERVED.value,
                    ]),
                )
                .order_by(UserCreditSubscriptionPeriod.id.asc())
                .with_for_update()
            )
            for period in periods_result.scalars().all():
                period.status = CreditSubscriptionPeriodStatus.CANCELLED.value
                period.cancelled_at = checked_at
                period.upgrade_order_id = None
                period.reserved_at = None
    await db.flush()
    return float(revoked)