from __future__ import annotations from dataclasses import dataclass from datetime import datetime from decimal import Decimal from typing import Iterable from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.enums.credit_balance import ( CREDIT_LEVEL_SORT, CREDIT_SCOPE_SORT, CreditAllocationAction, CreditBalanceSourceType, CreditBalanceStatus, CreditLevel, CreditScope, ) from app.enums.credit_record import CreditRecordBillingScene, CreditRecordType from app.models.credit.allocation import CreditRecordAllocation from app.models.credit.balance import UserCreditBalance from app.models.credit.subscription import UserCreditSubscription from app.models.credit.team_seat_usage import TeamSubscriptionSeatUsage from app.models.credit_record import CreditRecord from app.models.user import User from app.services.credit.locking import ( acquire_subscription_credit_locks, acquire_team_business_lock, acquire_user_credit_lock, ) from app.services.credit.query_service import attach_credit_snapshot, get_available_credits from app.services.credit.team_subscription_service import ( TeamCreditCandidate, get_or_create_usage_for_update, lock_user_team_credit_candidates, ) from app.services.credit.time_policy import add_natural_months from app.services.credit.utils import to_credit_decimal, to_float, utc_now from app.services.credit_record_meta_service import CreditRecordMeta, with_user_snapshot from app.services.operation_log_service import log_operation_event from app.utils.exceptions import InsufficientCreditsError from app.utils.id_gen import generate_id @dataclass(slots=True) class CreditMutationResult: user: User record: CreditRecord | None created: bool amount: float balance_before: float balance_after: float refund_available: float = 0.0 refund_expired: float = 0.0 @dataclass(slots=True, frozen=True) class CreditRefundResult: records: tuple[CreditRecord, ...] created: bool total_amount: Decimal available_amount: Decimal expired_amount: Decimal balance_before: Decimal balance_after: Decimal @dataclass(slots=True) class _SpendCandidate: scope: str balance: UserCreditBalance available: Decimal team: TeamCreditCandidate | None = None async def _load_user(db: AsyncSession, user_id: str, *, for_update: bool = False) -> User: stmt = select(User).where(User.id == user_id).limit(1) if for_update: stmt = stmt.with_for_update() result = await db.execute(stmt) user = result.scalar_one_or_none() if user is None: raise ValueError("用户不存在") return user async def _record_meta_kwargs( db: AsyncSession, user: User, record_meta: CreditRecordMeta | dict | None, ) -> dict: if record_meta is None: meta = CreditRecordMeta() elif isinstance(record_meta, CreditRecordMeta): meta = record_meta else: meta = CreditRecordMeta(**{k: v for k, v in record_meta.items() if k in CreditRecordMeta.__dataclass_fields__}) await with_user_snapshot(db, meta, user.id, user=user) return meta.to_record_kwargs() def _normalize_expiry( request_time: datetime, valid_from: datetime | None, expires_at: datetime | None, ) -> tuple[datetime, datetime]: start = valid_from or request_time end = expires_at or add_natural_months(start, 1) if end <= start: raise ValueError("积分有效期结束时间必须晚于开始时间") return start, end def _record( *, user_id: str, record_type: str, amount: Decimal, balance_delta: Decimal, balance_after: Decimal, description: str, related_id: str | None, request_time: datetime, biz_key: str | None, refund_for_biz_key: str | None = None, refund_kind: str | None = None, credit_level_snapshot: str | None = None, expired_amount: Decimal = Decimal("0.00"), meta_kwargs: dict | None = None, ) -> CreditRecord: return CreditRecord( id=generate_id(), user_id=user_id, type=record_type, amount=amount, balance_delta=balance_delta, expired_amount=expired_amount, balance_after=balance_after, description=description, related_id=related_id, request_time=request_time, biz_key=biz_key, refund_for_biz_key=refund_for_biz_key, refund_kind=refund_kind, credit_level_snapshot=credit_level_snapshot, **(meta_kwargs or {}), ) def _allocation_snapshot_kwargs( *, balance: UserCreditBalance, scope: str | None = None, team_id: str | None = None, team_manager_id: str | None = None, subscription_id: str | None = None, subscription_period_id: str | None = None, seat_id: str | None = None, ) -> dict: return { "credit_level_snapshot": balance.credit_level, "credit_scope_snapshot": scope or balance.credit_scope or CreditScope.PERSONAL.value, "source_type_snapshot": balance.source_type, "source_id_snapshot": balance.source_id, "team_id_snapshot": team_id if team_id is not None else balance.team_id, "team_manager_id_snapshot": team_manager_id, "subscription_id_snapshot": subscription_id if subscription_id is not None else balance.subscription_id, "subscription_period_id_snapshot": ( subscription_period_id if subscription_period_id is not None else balance.subscription_period_id ), "seat_id_snapshot": seat_id, "valid_from_snapshot": balance.valid_from, "expires_at_snapshot": balance.expires_at, } async def _find_record_by_biz_key(db: AsyncSession, user_id: str, biz_key: str | None) -> CreditRecord | None: if not biz_key: return None result = await db.execute( select(CreditRecord).where(CreditRecord.user_id == user_id, CreditRecord.biz_key == biz_key).limit(1) ) return result.scalar_one_or_none() async def grant_credits( db: AsyncSession, *, user_id: str, amount: Decimal | float | int, description: str, source_type: str, valid_from: datetime | None = None, expires_at: datetime | None = None, credit_level: str = CreditLevel.GENERAL.value, source_id: str | None = None, product_id: str | None = None, payment_order_id: str | None = None, subscription_id: str | None = None, subscription_period_id: str | None = None, related_id: str | None = None, record_type: str = CreditRecordType.RECHARGE.value, biz_key: str | None = None, record_meta: CreditRecordMeta | dict | None = None, metadata_json: dict | None = None, request_time: datetime | None = None, credit_scope: str = CreditScope.PERSONAL.value, team_id: str | None = None, team_manager_id_snapshot: str | None = None, ) -> CreditMutationResult: checked_at = request_time or utc_now() value = to_credit_decimal(amount) if value < 0: raise ValueError("发放积分不能小于0") if credit_scope not in (CreditScope.PERSONAL.value, CreditScope.TEAM.value): raise ValueError("未知积分资金域") if credit_scope == CreditScope.TEAM.value and (not team_id or not subscription_id or not subscription_period_id): raise ValueError("团队积分必须绑定团队、订阅和订阅周期") await acquire_user_credit_lock(db, user_id) if team_id: await acquire_team_business_lock(db, team_id) if subscription_id: await acquire_subscription_credit_locks(db, [subscription_id]) user = await _load_user(db, user_id, for_update=True) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id, biz_key) if existing: attach_credit_snapshot(user, before) return CreditMutationResult(user, existing, False, to_float(existing.amount), to_float(before), to_float(before)) start, end = _normalize_expiry(checked_at, valid_from, expires_at) level_rank = CREDIT_LEVEL_SORT.get(credit_level) if level_rank is None: raise ValueError("未知积分等级") record_after = before + value if credit_scope == CreditScope.PERSONAL.value else before meta_kwargs = await _record_meta_kwargs(db, user, record_meta) record = _record( user_id=user_id, record_type=record_type, amount=value, balance_delta=value if credit_scope == CreditScope.PERSONAL.value else Decimal("0.00"), balance_after=record_after, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, credit_level_snapshot=credit_level, meta_kwargs=meta_kwargs, ) db.add(record) await db.flush() balance = UserCreditBalance( id=generate_id(), user_id=user_id, credit_scope=credit_scope, team_id=team_id, credit_level=credit_level, credit_level_rank=level_rank, source_type=source_type, source_id=source_id, product_id=product_id, payment_order_id=payment_order_id, subscription_id=subscription_id, subscription_period_id=subscription_period_id, grant_record_id=record.id, grant_amount=value, unspent_amount=value, consumed_amount=Decimal("0.00"), expired_amount=Decimal("0.00"), revoked_amount=Decimal("0.00"), valid_from=start, expires_at=end, status=CreditBalanceStatus.ACTIVE.value if start <= checked_at else CreditBalanceStatus.SCHEDULED.value, biz_key=f"{biz_key or record.id}:balance", metadata_json=metadata_json, ) db.add(balance) await db.flush() allocation = CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, allocation_action=CreditAllocationAction.GRANT.value, amount=value, request_time=checked_at, **_allocation_snapshot_kwargs( balance=balance, scope=credit_scope, team_id=team_id, team_manager_id=team_manager_id_snapshot, subscription_id=subscription_id, subscription_period_id=subscription_period_id, ), unspent_before=Decimal("0.00"), unspent_after=value, consumed_before=Decimal("0.00"), consumed_after=Decimal("0.00"), ) db.add(allocation) after = await get_available_credits(db, user_id, request_time=checked_at) record.balance_after = after attach_credit_snapshot(user, after) log_operation_event( domain="credit", module="ledger", event_type="CREDIT_GRANT", user_id=user_id, message="积分发放成功", detail={ "record_id": record.id, "balance_id": balance.id, "amount": to_float(value), "credit_scope": credit_scope, "team_id": team_id, "subscription_id": subscription_id, "subscription_period_id": subscription_period_id, }, ) return CreditMutationResult(user, record, True, to_float(value), to_float(before), to_float(after)) async def _load_personal_spend_candidates( db: AsyncSession, user_id: str, request_time: datetime, ) -> list[_SpendCandidate]: result = await db.execute( select(UserCreditBalance) .where( UserCreditBalance.user_id == user_id, UserCreditBalance.credit_scope == CreditScope.PERSONAL.value, UserCreditBalance.valid_from <= request_time, UserCreditBalance.expires_at > request_time, UserCreditBalance.unspent_amount > 0, UserCreditBalance.revoked_at.is_(None), ) .order_by( UserCreditBalance.credit_level_rank.asc(), UserCreditBalance.expires_at.asc(), UserCreditBalance.valid_from.asc(), UserCreditBalance.id.asc(), ) .with_for_update() ) return [ _SpendCandidate(CreditScope.PERSONAL.value, balance, to_credit_decimal(balance.unspent_amount)) for balance in result.scalars().all() ] def _candidate_sort(candidate: _SpendCandidate) -> tuple: balance = candidate.balance return ( int(balance.credit_level_rank or 999), CREDIT_SCOPE_SORT.get(candidate.scope, 999), balance.expires_at, balance.valid_from, balance.id, ) async def deduct_credits( db: AsyncSession, *, user_id: str, amount: Decimal | float | int, description: str, related_id: str | None = None, biz_key: str | None = None, refund_for_biz_key: str | None = None, record_meta: CreditRecordMeta | dict | None = None, record_type: str = CreditRecordType.CONSUME.value, create_zero_record: bool = False, request_time: datetime | None = None, allowed_scopes: set[str] | None = None, ) -> CreditMutationResult: checked_at = request_time or utc_now() value = to_credit_decimal(amount) if value < 0: raise ValueError("扣减积分不能小于0") if value == 0 and not create_zero_record: user = await _load_user(db, user_id) available = await get_available_credits(db, user_id, request_time=checked_at) attach_credit_snapshot(user, available) return CreditMutationResult(user, None, False, 0.0, to_float(available), to_float(available)) await acquire_user_credit_lock(db, user_id) scope_filter = allowed_scopes or {CreditScope.PERSONAL.value, CreditScope.TEAM.value} invalid_scopes = scope_filter - {CreditScope.PERSONAL.value, CreditScope.TEAM.value} if invalid_scopes: raise ValueError(f"不支持的积分资金域: {','.join(sorted(invalid_scopes))}") # 固定锁顺序：user advisory -> team advisory -> user/balance/seat row locks。 if CreditScope.TEAM.value in scope_filter: current_team_id = (await db.execute(select(User.team_id).where(User.id == user_id).limit(1))).scalar_one_or_none() if current_team_id: await acquire_team_business_lock(db, current_team_id) user = await _load_user(db, user_id, for_update=True) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id, biz_key) if existing: attach_credit_snapshot(user, before) return CreditMutationResult(user, existing, False, to_float(abs(existing.amount)), to_float(before), to_float(before)) team_candidates = ( await lock_user_team_credit_candidates(db, user_id=user_id, request_time=checked_at) if CreditScope.TEAM.value in scope_filter else [] ) personal_candidates = ( await _load_personal_spend_candidates(db, user_id, checked_at) if CreditScope.PERSONAL.value in scope_filter else [] ) candidates = personal_candidates + [ _SpendCandidate(CreditScope.TEAM.value, item.balance, item.available, item) for item in team_candidates if item.available > 0 ] candidates.sort(key=_candidate_sort) if sum((c.available for c in candidates), Decimal("0.00")) < value: raise InsufficientCreditsError() meta_kwargs = await _record_meta_kwargs(db, user, record_meta) after_expected = before - value record = _record( user_id=user_id, record_type=record_type, amount=-value, balance_delta=-value, balance_after=after_expected, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, refund_for_biz_key=refund_for_biz_key, meta_kwargs=meta_kwargs, ) db.add(record) await db.flush() remaining = value team_amount = Decimal("0.00") personal_amount = Decimal("0.00") for candidate in candidates: if remaining <= 0: break take = min(candidate.available, remaining) if take <= 0: continue balance = candidate.balance unspent_before = to_credit_decimal(balance.unspent_amount) consumed_before = to_credit_decimal(balance.consumed_amount) if unspent_before < take: raise RuntimeError("积分并发结算异常：余额不足") balance.unspent_amount = unspent_before - take balance.consumed_amount = consumed_before + take if balance.unspent_amount <= 0: balance.status = CreditBalanceStatus.CONSUMED.value team_id = None manager_id = None subscription_id = balance.subscription_id period_id = balance.subscription_period_id seat_id = None if candidate.scope == CreditScope.TEAM.value: if not candidate.team: raise RuntimeError("团队积分候选缺少 Seat 上下文") team_ctx = candidate.team usage = team_ctx.usage if usage is None: usage = await get_or_create_usage_for_update( db, seat=team_ctx.seat, period=team_ctx.period, ) team_ctx.usage = usage usage_before = to_credit_decimal(usage.used_credits) if usage_before + take > to_credit_decimal(team_ctx.seat.monthly_allocated_credits): raise RuntimeError("团队席位额度发生并发变化，请重试") usage.used_credits = usage_before + take team_id = team_ctx.subscription.team_id manager_id = team_ctx.subscription.team_manager_id_snapshot subscription_id = team_ctx.subscription.id period_id = team_ctx.period.id seat_id = team_ctx.seat.id team_amount += take else: personal_amount += take db.add( CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, allocation_action=CreditAllocationAction.CONSUME.value, amount=take, request_time=checked_at, **_allocation_snapshot_kwargs( balance=balance, scope=candidate.scope, team_id=team_id, team_manager_id=manager_id, subscription_id=subscription_id, subscription_period_id=period_id, seat_id=seat_id, ), unspent_before=unspent_before, unspent_after=to_credit_decimal(balance.unspent_amount), consumed_before=consumed_before, consumed_after=to_credit_decimal(balance.consumed_amount), ) ) remaining -= take if remaining > 0: raise RuntimeError("积分结算候选不足，事务将回滚") await db.flush() after = await get_available_credits(db, user_id, request_time=checked_at) record.balance_after = after attach_credit_snapshot(user, after) log_operation_event( domain="credit", module="ledger", event_type="CREDIT_CONSUME", user_id=user_id, message="积分消费成功", detail={ "credit_record_id": record.id, "amount": to_float(value), "team_amount": to_float(team_amount), "personal_amount": to_float(personal_amount), "biz_key": biz_key, }, ) return CreditMutationResult(user, record, True, to_float(value), to_float(before), to_float(after)) async def _load_original_consume_allocations( db: AsyncSession, *, user_id: str, refund_for_biz_key: str, ) -> tuple[CreditRecord, list[CreditRecordAllocation]]: result = await db.execute( select(CreditRecord) .where( CreditRecord.user_id == user_id, CreditRecord.biz_key == refund_for_biz_key, CreditRecord.type == CreditRecordType.CONSUME.value, ) .limit(1) ) original_record = result.scalar_one_or_none() if not original_record: raise ValueError("未找到可退款的原消费记录") allocations_result = await db.execute( select(CreditRecordAllocation) .where( CreditRecordAllocation.credit_record_id == original_record.id, CreditRecordAllocation.allocation_action == CreditAllocationAction.CONSUME.value, CreditRecordAllocation.user_id == user_id, ) .order_by(CreditRecordAllocation.id.asc()) ) allocations = list(allocations_result.scalars().all()) if not allocations: raise ValueError("原消费记录缺少资金分配明细") return original_record, allocations async def refund_consumption( db: AsyncSession, *, user_id: str, refund_for_biz_key: str, description: str, related_id: str | None = None, biz_key: str | None = None, record_meta: CreditRecordMeta | dict | None = None, refund_time: datetime | None = None, ) -> CreditRefundResult: checked_at = refund_time or utc_now() refund_biz_key = biz_key or f"refund:{refund_for_biz_key}" await acquire_user_credit_lock(db, user_id) # 先在 user advisory lock 保护下定位原资金链，再按固定顺序获取 team/subscription 锁， # 最后才锁 User/Balance/SeatUsage 行，避免与席位/团队操作形成反向锁序。 _, originals = await _load_original_consume_allocations( db, user_id=user_id, refund_for_biz_key=refund_for_biz_key ) team_ids = sorted( { a.team_id_snapshot for a in originals if a.credit_scope_snapshot == CreditScope.TEAM.value and a.team_id_snapshot } ) for team_id in team_ids: await acquire_team_business_lock(db, team_id) await acquire_subscription_credit_locks( db, [a.subscription_id_snapshot for a in originals if a.subscription_id_snapshot], ) user = await _load_user(db, user_id, for_update=True) before = await get_available_credits(db, user_id, request_time=checked_at) existing_result = await db.execute( select(CreditRecord) .where( CreditRecord.user_id == user_id, CreditRecord.type == CreditRecordType.REFUND.value, CreditRecord.refund_for_biz_key == refund_for_biz_key, CreditRecord.biz_key.in_([ f"{refund_biz_key}:available", f"{refund_biz_key}:expired", ]), ) .order_by(CreditRecord.id.asc()) ) existing = list(existing_result.scalars().all()) if existing: available = sum((to_credit_decimal(r.amount) for r in existing if r.refund_kind == "available"), Decimal("0.00")) expired = sum((to_credit_decimal(r.expired_amount or r.amount) for r in existing if r.refund_kind == "expired"), Decimal("0.00")) attach_credit_snapshot(user, before) return CreditRefundResult(tuple(existing), False, available + expired, available, expired, before, before) balance_ids = list(dict.fromkeys(a.credit_balance_id for a in originals)) balances_result = await db.execute( select(UserCreditBalance) .where(UserCreditBalance.id.in_(balance_ids)) .order_by(UserCreditBalance.id.asc()) .with_for_update() ) balance_map = {item.id: item for item in balances_result.scalars().all()} available_total = Decimal("0.00") expired_total = Decimal("0.00") available_rows: list[tuple] = [] expired_rows: list[tuple] = [] for original in originals: balance = balance_map.get(original.credit_balance_id) if balance is None: raise RuntimeError(f"原积分来源不存在: {original.credit_balance_id}") if original.credit_scope_snapshot == CreditScope.PERSONAL.value and balance.user_id != user_id: raise RuntimeError("个人积分退款资金所有人与消费者不一致") refund_amount = to_credit_decimal(original.amount) before_unspent = to_credit_decimal(balance.unspent_amount) before_consumed = to_credit_decimal(balance.consumed_amount) if before_consumed < refund_amount: raise RuntimeError("原积分批次已消费金额不足以退款") after_consumed = before_consumed - refund_amount balance.consumed_amount = after_consumed if original.credit_scope_snapshot == CreditScope.TEAM.value and original.seat_id_snapshot: usage_result = await db.execute( select(TeamSubscriptionSeatUsage) .where( TeamSubscriptionSeatUsage.seat_id == original.seat_id_snapshot, TeamSubscriptionSeatUsage.subscription_period_id == original.subscription_period_id_snapshot, TeamSubscriptionSeatUsage.user_id == user_id, ) .limit(1) .with_for_update() ) usage = usage_result.scalar_one_or_none() if usage is None: raise RuntimeError("原团队积分消费缺少席位月度使用记录") used_before = to_credit_decimal(usage.used_credits) if used_before < refund_amount: raise RuntimeError("团队席位历史已用积分不足以回冲") usage.used_credits = used_before - refund_amount if checked_at < balance.expires_at and balance.revoked_at is None: after_unspent = before_unspent + refund_amount balance.unspent_amount = after_unspent balance.status = CreditBalanceStatus.ACTIVE.value available_total += refund_amount available_rows.append( (original, balance, refund_amount, before_unspent, after_unspent, before_consumed, after_consumed) ) else: balance.expired_amount = to_credit_decimal(balance.expired_amount) + refund_amount after_unspent = before_unspent if to_credit_decimal(balance.unspent_amount) <= 0 and to_credit_decimal(balance.consumed_amount) <= 0: balance.status = CreditBalanceStatus.EXPIRED.value expired_total += refund_amount expired_rows.append( (original, balance, refund_amount, before_unspent, after_unspent, before_consumed, after_consumed) ) meta_kwargs = await _record_meta_kwargs(db, user, record_meta) created_records: list[CreditRecord] = [] async def create_refund_record(kind: str, amount: Decimal, expired_amount: Decimal) -> CreditRecord: current_after = await get_available_credits(db, user_id, request_time=checked_at) record = _record( user_id=user_id, record_type=CreditRecordType.REFUND.value, amount=amount, balance_delta=amount if kind == "available" else Decimal("0.00"), balance_after=current_after, expired_amount=expired_amount, description=( f"{description}，退回有效积分：{to_float(amount)}" if kind == "available" else f"{description}，原积分已过期：{to_float(expired_amount)}" ), related_id=related_id, request_time=checked_at, biz_key=f"{refund_biz_key}:{kind}", refund_for_biz_key=refund_for_biz_key, refund_kind=kind, meta_kwargs=meta_kwargs, ) db.add(record) await db.flush() return record async def add_refund_allocations(record: CreditRecord, rows: list[tuple], action: str) -> None: for original, balance, amount, before_unspent, after_unspent, before_consumed, after_consumed in rows: db.add( CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, source_allocation_id=original.id, allocation_action=action, amount=amount, request_time=checked_at, credit_level_snapshot=original.credit_level_snapshot, credit_scope_snapshot=original.credit_scope_snapshot, source_type_snapshot=original.source_type_snapshot, source_id_snapshot=original.source_id_snapshot, team_id_snapshot=original.team_id_snapshot, team_manager_id_snapshot=original.team_manager_id_snapshot, subscription_id_snapshot=original.subscription_id_snapshot, subscription_period_id_snapshot=original.subscription_period_id_snapshot, seat_id_snapshot=original.seat_id_snapshot, valid_from_snapshot=original.valid_from_snapshot, expires_at_snapshot=original.expires_at_snapshot, unspent_before=before_unspent, unspent_after=after_unspent, consumed_before=before_consumed, consumed_after=after_consumed, ) ) if available_total > 0: record = await create_refund_record("available", available_total, Decimal("0.00")) created_records.append(record) await add_refund_allocations(record, available_rows, CreditAllocationAction.REFUND_AVAILABLE.value) if expired_total > 0: record = await create_refund_record("expired", expired_total, expired_total) created_records.append(record) await add_refund_allocations(record, expired_rows, CreditAllocationAction.REFUND_EXPIRED.value) await db.flush() after = await get_available_credits(db, user_id, request_time=checked_at) for record in created_records: record.balance_after = after attach_credit_snapshot(user, after) log_operation_event( domain="credit", module="ledger", event_type="CREDIT_SOURCE_REFUNDED", user_id=user_id, message="按原积分来源完成业务失败退款", detail={ "refund_for_biz_key": refund_for_biz_key, "available_refund": to_float(available_total), "expired_refund": to_float(expired_total), "source_count": len(originals), "created_record_ids": [item.id for item in created_records], }, ) return CreditRefundResult( records=tuple(created_records), created=True, total_amount=available_total + expired_total, available_amount=available_total, expired_amount=expired_total, balance_before=before, balance_after=after, ) async def revoke_balances( db: AsyncSession, *, balances: Iterable[UserCreditBalance], description: str, related_id: str | None, biz_key: str, request_time: datetime | None = None, ) -> CreditMutationResult: items = list(balances) if not items: raise ValueError("没有可撤销的积分") checked_at = request_time or utc_now() user_id = items[0].user_id if any(item.user_id != user_id for item in items): raise ValueError("一次撤销只能处理同一资金所有人的积分") await acquire_user_credit_lock(db, user_id) user = await _load_user(db, user_id, for_update=True) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id, biz_key) if existing: attach_credit_snapshot(user, before) return CreditMutationResult(user, existing, False, abs(to_float(existing.amount)), to_float(before), to_float(before)) record = _record( user_id=user_id, record_type=CreditRecordType.REVOKE.value, amount=Decimal("0.00"), balance_delta=Decimal("0.00"), balance_after=before, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, meta_kwargs={"billing_scene": CreditRecordBillingScene.CREDIT_REVOKE.value}, ) db.add(record) await db.flush() total = Decimal("0.00") for balance in items: if balance.revoked_at is not None: continue unspent_before = to_credit_decimal(balance.unspent_amount) consumed_before = to_credit_decimal(balance.consumed_amount) revoke_amount = unspent_before if revoke_amount <= 0: continue balance.unspent_amount = Decimal("0.00") balance.revoked_amount = to_credit_decimal(balance.revoked_amount) + revoke_amount balance.revoked_at = checked_at balance.status = CreditBalanceStatus.REVOKED.value total += revoke_amount db.add( CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, allocation_action=CreditAllocationAction.REVOKE.value, amount=revoke_amount, request_time=checked_at, **_allocation_snapshot_kwargs(balance=balance), unspent_before=unspent_before, unspent_after=Decimal("0.00"), consumed_before=consumed_before, consumed_after=consumed_before, ) ) record.amount = -total record.balance_delta = -total await db.flush() after = await get_available_credits(db, user_id, request_time=checked_at) record.balance_after = after attach_credit_snapshot(user, after) return CreditMutationResult(user, record, True, to_float(total), to_float(before), to_float(after))