from __future__ import annotations from dataclasses import dataclass from datetime import datetime from decimal import Decimal from typing import Iterable from sqlalchemy import func, select from sqlalchemy.ext.asyncio import AsyncSession from app.enums.credit_balance import ( CREDIT_LEVEL_SORT, CreditAllocationAction, CreditBalanceSourceType, CreditBalanceStatus, CreditLevel, ) from app.enums.credit_record import ( CreditRecordAction, CreditRecordBillingScene, CreditRecordType, ) from app.enums.common import BillingBlockEventEnum from app.models.credit.allocation import CreditRecordAllocation from app.models.credit.balance import UserCreditBalance from app.models.credit_record import CreditRecord from app.models.user import User from app.services.credit.locking import acquire_user_credit_lock from app.services.credit.query_service import attach_credit_snapshot, get_available_credits 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 _RefundAllocationSource: original_allocation: CreditRecordAllocation credit_balance_id: str amount: Decimal @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 async def _load_user(db: AsyncSession, user_id: str) -> User: result = await db.execute(select(User).where(User.id == user_id).limit(1)) user = result.scalar_one_or_none() if user is None: raise ValueError("User not found") return user async def _record_meta_kwargs( db: AsyncSession, user: User, record_meta: CreditRecordMeta | dict | None, ) -> dict: if isinstance(record_meta, CreditRecordMeta): populated = await with_user_snapshot(db, record_meta, user.id, user=user) return populated.to_record_kwargs() if isinstance(record_meta, dict): return {key: value for key, value in record_meta.items() if value is not None} return {} 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, ) -> CreditMutationResult: grant_amount = to_credit_decimal(amount) checked_at = request_time or utc_now() starts_at = valid_from or checked_at ends_at = expires_at or add_natural_months(starts_at, 1) await acquire_user_credit_lock(db, user_id) user = await _load_user(db, user_id) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id=user_id, biz_key=biz_key) if existing is not None: attach_credit_snapshot(user, before) return CreditMutationResult( user=user, record=existing, created=False, amount=abs(to_float(existing.amount)), balance_before=to_float(before), balance_after=to_float(before), ) if grant_amount <= 0: attach_credit_snapshot(user, before) return CreditMutationResult(user, None, False, 0.0, to_float(before), to_float(before)) if ends_at <= starts_at: raise ValueError("积分过期时间必须晚于生效时间") meta_kwargs = await _record_meta_kwargs(db, user, record_meta) after = before + grant_amount if starts_at <= checked_at < ends_at else before record = CreditRecord( id=generate_id(), user_id=user_id, type=record_type, amount=grant_amount, balance_delta=(grant_amount if starts_at <= checked_at < ends_at else Decimal("0.00")), expired_amount=Decimal("0.00"), balance_after=after, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, credit_level_snapshot=credit_level, **meta_kwargs, ) db.add(record) await db.flush() balance = UserCreditBalance( id=generate_id(), user_id=user_id, credit_level=credit_level, credit_level_rank=CREDIT_LEVEL_SORT.get(credit_level, CREDIT_LEVEL_SORT[CreditLevel.GENERAL.value]), 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=grant_amount, unspent_amount=grant_amount, consumed_amount=Decimal("0.00"), expired_amount=Decimal("0.00"), revoked_amount=Decimal("0.00"), valid_from=starts_at, expires_at=ends_at, status=CreditBalanceStatus.ACTIVE.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=grant_amount, request_time=checked_at, credit_level_snapshot=credit_level, source_type_snapshot=source_type, source_id_snapshot=source_id, valid_from_snapshot=starts_at, expires_at_snapshot=ends_at, unspent_before=Decimal("0.00"), unspent_after=grant_amount, consumed_before=Decimal("0.00"), consumed_after=Decimal("0.00"), ) db.add(allocation) attach_credit_snapshot(user, after) await db.flush() log_operation_event( domain="billing", module="dynamic_credit", event_type="CREDIT_GRANTED", event_status="success", source="app.services.credit.ledger_service.grant_credits", user_id=user_id, task_id=related_id, message="动态积分发放完成", detail={ "record_id": record.id, "balance_id": balance.id, "amount": to_float(grant_amount), "source_type": source_type, "credit_level": credit_level, "valid_from": starts_at.isoformat(), "expires_at": ends_at.isoformat(), "biz_key": biz_key, }, ) return CreditMutationResult( user=user, record=record, created=True, amount=to_float(grant_amount), balance_before=to_float(before), balance_after=to_float(after), ) 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, ) -> CreditMutationResult: consume_amount = to_credit_decimal(amount) checked_at = request_time or utc_now() await acquire_user_credit_lock(db, user_id) user = await _load_user(db, user_id) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id=user_id, biz_key=biz_key) if existing is not None: attach_credit_snapshot(user, before) return CreditMutationResult( user=user, record=existing, created=False, amount=abs(to_float(existing.amount)), balance_before=to_float(before), balance_after=to_float(before), ) if consume_amount <= 0 and not create_zero_record: attach_credit_snapshot(user, before) return CreditMutationResult(user, None, False, 0.0, to_float(before), to_float(before)) if consume_amount > before: log_operation_event( domain="billing", module="dynamic_credit", event_type=BillingBlockEventEnum.INSUFFICIENT_CREDITS.value, event_status="failed", source="app.services.credit.ledger_service.deduct_credits", user_id=user_id, task_id=related_id, message="有效积分不足，已在创建业务任务前同步拦截", detail={ "required_credits": to_float(consume_amount), "available_credits": to_float(before), "biz_key": biz_key, "request_time": checked_at.isoformat(), "description": description, }, ) raise InsufficientCreditsError() result = await db.execute( select(UserCreditBalance) .where( UserCreditBalance.user_id == user_id, UserCreditBalance.valid_from <= checked_at, UserCreditBalance.expires_at > checked_at, 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() ) balances = list(result.scalars().all()) remaining = consume_amount allocations_data: list[tuple[UserCreditBalance, Decimal, Decimal, Decimal, Decimal, Decimal]] = [] for balance in balances: if remaining <= 0: break available = to_credit_decimal(balance.unspent_amount) if available <= 0: continue allocated = min(available, remaining) before_unspent = available before_consumed = to_credit_decimal(balance.consumed_amount) after_unspent = before_unspent - allocated after_consumed = before_consumed + allocated balance.unspent_amount = after_unspent balance.consumed_amount = after_consumed if after_unspent == 0: balance.status = CreditBalanceStatus.CONSUMED.value allocations_data.append( (balance, allocated, before_unspent, after_unspent, before_consumed, after_consumed) ) remaining -= allocated if remaining > 0: # 理论上用户级锁和前置汇总后不应发生；保留硬失败以避免部分扣除。 raise RuntimeError("积分分摊不足，事务将回滚") meta_kwargs = await _record_meta_kwargs(db, user, record_meta) after = before - consume_amount record = CreditRecord( id=generate_id(), user_id=user_id, type=record_type, amount=-consume_amount, balance_delta=-consume_amount, expired_amount=Decimal("0.00"), balance_after=after, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, refund_for_biz_key=refund_for_biz_key, **meta_kwargs, ) db.add(record) await db.flush() db.add_all( [ CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, allocation_action=CreditAllocationAction.CONSUME.value, amount=allocated, request_time=checked_at, credit_level_snapshot=balance.credit_level, source_type_snapshot=balance.source_type, source_id_snapshot=balance.source_id, valid_from_snapshot=balance.valid_from, expires_at_snapshot=balance.expires_at, unspent_before=before_unspent, unspent_after=after_unspent, consumed_before=before_consumed, consumed_after=after_consumed, ) for balance, allocated, before_unspent, after_unspent, before_consumed, after_consumed in allocations_data ] ) attach_credit_snapshot(user, after) await db.flush() log_operation_event( domain="billing", module="dynamic_credit", event_type="CREDIT_DEDUCTED", event_status="success", source="app.services.credit.ledger_service.deduct_credits", user_id=user_id, task_id=related_id, message="动态积分同步扣除完成", detail={ "record_id": record.id, "amount": to_float(consume_amount), "balance_before": to_float(before), "balance_after": to_float(after), "allocation_count": len(allocations_data), "allocation_balance_ids": [item[0].id for item in allocations_data[:20]], "biz_key": biz_key, "request_time": checked_at.isoformat(), }, ) return CreditMutationResult( user=user, record=record, created=True, amount=to_float(consume_amount), balance_before=to_float(before), balance_after=to_float(after), ) async def _load_original_consume_allocations( db: AsyncSession, *, original_record_id: str, ) -> list[_RefundAllocationSource]: result = await db.execute( select(CreditRecordAllocation) .where( CreditRecordAllocation.credit_record_id == original_record_id, CreditRecordAllocation.allocation_action == CreditAllocationAction.CONSUME.value, ) .order_by(CreditRecordAllocation.id.desc()) ) originals = list(result.scalars().all()) if not originals: return [] original_ids = [item.id for item in originals] transfer_result = await db.execute( select(CreditRecordAllocation) .where( CreditRecordAllocation.source_allocation_id.in_(original_ids), CreditRecordAllocation.allocation_action == CreditAllocationAction.UPGRADE_SOURCE_TRANSFER_IN.value, ) .order_by(CreditRecordAllocation.id.asc()) ) transfers_by_source: dict[str, list[CreditRecordAllocation]] = {} for item in transfer_result.scalars().all(): if item.source_allocation_id: transfers_by_source.setdefault(item.source_allocation_id, []).append(item) sources: list[_RefundAllocationSource] = [] for original in originals: transferred = Decimal("0.00") for transfer in transfers_by_source.get(original.id, []): amount = to_credit_decimal(transfer.amount) transferred += amount sources.append( _RefundAllocationSource( original_allocation=original, credit_balance_id=transfer.credit_balance_id, amount=amount, ) ) remaining = to_credit_decimal(original.amount) - transferred if remaining < 0: raise RuntimeError("升级积分来源迁移金额超过原消费分摊") if remaining > 0: sources.append( _RefundAllocationSource( original_allocation=original, credit_balance_id=original.credit_balance_id, amount=remaining, ) ) return sources 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() await acquire_user_credit_lock(db, user_id) user = await _load_user(db, user_id) 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.refund_for_biz_key == refund_for_biz_key, CreditRecord.refund_kind.in_(["available", "expired"]), ) ) existing_records = list(existing_result.scalars().all()) if existing_records: available = sum((to_credit_decimal(item.balance_delta) for item in existing_records), Decimal("0.00")) expired = sum((to_credit_decimal(item.expired_amount) for item in existing_records), Decimal("0.00")) attach_credit_snapshot(user, before) return CreditRefundResult( records=tuple(existing_records), created=False, total_amount=available + expired, available_amount=available, expired_amount=expired, balance_before=before, balance_after=before, ) original_result = await db.execute( select(CreditRecord) .where(CreditRecord.user_id == user_id, CreditRecord.biz_key == refund_for_biz_key) .limit(1) ) original = original_result.scalar_one_or_none() if original is None: raise ValueError("未找到原积分消费流水") allocations = await _load_original_consume_allocations(db, original_record_id=original.id) balance_ids = list(dict.fromkeys(item.credit_balance_id for item in allocations)) balances_result = await db.execute( select(UserCreditBalance) .where(UserCreditBalance.id.in_(balance_ids), UserCreditBalance.user_id == user_id) .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_allocations: list[tuple[CreditRecordAllocation, UserCreditBalance, Decimal, Decimal, Decimal, Decimal, Decimal]] = [] expired_allocations: list[tuple[CreditRecordAllocation, UserCreditBalance, Decimal, Decimal, Decimal, Decimal, Decimal]] = [] for allocation_source in allocations: allocation = allocation_source.original_allocation balance = balance_map.get(allocation_source.credit_balance_id) if balance is None: raise RuntimeError(f"原积分来源不存在: {allocation_source.credit_balance_id}") refund_amount = to_credit_decimal(allocation_source.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 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_allocations.append( (allocation, balance, refund_amount, before_unspent, after_unspent, before_consumed, after_consumed) ) else: before_expired = to_credit_decimal(balance.expired_amount) balance.expired_amount = before_expired + refund_amount after_unspent = before_unspent if balance.unspent_amount == 0 and balance.consumed_amount == 0: balance.status = CreditBalanceStatus.EXPIRED.value expired_total += refund_amount expired_allocations.append( (allocation, 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] = [] balance_after = before + available_total async def create_refund_record(kind: str, amount: Decimal, expired_amount: Decimal) -> CreditRecord: record = CreditRecord( id=generate_id(), user_id=user_id, type=CreditRecordType.REFUND.value, amount=amount, balance_delta=(amount if kind == "available" else Decimal("0.00")), expired_amount=expired_amount, balance_after=balance_after, 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"{biz_key or refund_for_biz_key + ':refund'}:{kind}", refund_for_biz_key=refund_for_biz_key, refund_kind=kind, **meta_kwargs, ) db.add(record) await db.flush() return record if available_total > 0: available_record = await create_refund_record("available", available_total, Decimal("0.00")) created_records.append(available_record) db.add_all( [ CreditRecordAllocation( id=generate_id(), credit_record_id=available_record.id, credit_balance_id=balance.id, user_id=user_id, source_allocation_id=allocation.id, allocation_action=CreditAllocationAction.REFUND_AVAILABLE.value, amount=amount, request_time=checked_at, credit_level_snapshot=balance.credit_level, source_type_snapshot=balance.source_type, source_id_snapshot=balance.source_id, valid_from_snapshot=balance.valid_from, expires_at_snapshot=balance.expires_at, unspent_before=before_unspent, unspent_after=after_unspent, consumed_before=before_consumed, consumed_after=after_consumed, ) for allocation, balance, amount, before_unspent, after_unspent, before_consumed, after_consumed in available_allocations ] ) if expired_total > 0: expired_record = await create_refund_record("expired", expired_total, expired_total) created_records.append(expired_record) db.add_all( [ CreditRecordAllocation( id=generate_id(), credit_record_id=expired_record.id, credit_balance_id=balance.id, user_id=user_id, source_allocation_id=allocation.id, allocation_action=CreditAllocationAction.REFUND_EXPIRED.value, amount=amount, request_time=checked_at, credit_level_snapshot=balance.credit_level, source_type_snapshot=balance.source_type, source_id_snapshot=balance.source_id, valid_from_snapshot=balance.valid_from, expires_at_snapshot=balance.expires_at, unspent_before=before_unspent, unspent_after=after_unspent, consumed_before=before_consumed, consumed_after=after_consumed, ) for allocation, balance, amount, before_unspent, after_unspent, before_consumed, after_consumed in expired_allocations ] ) attach_credit_snapshot(user, balance_after) await db.flush() log_operation_event( domain="billing", module="dynamic_credit", event_type="CREDIT_SOURCE_REFUNDED", event_status="success", source="app.services.credit.ledger_service.refund_consumption", user_id=user_id, task_id=related_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(allocations), "created_record_ids": [item.id for item in created_records], "refund_time": checked_at.isoformat(), }, ) 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=balance_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 await acquire_user_credit_lock(db, user_id) user = await _load_user(db, user_id) before = await get_available_credits(db, user_id, request_time=checked_at) existing = await _find_record_by_biz_key(db, user_id=user_id, biz_key=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)) total = Decimal("0.00") allocation_rows: list[CreditRecordAllocation] = [] record = CreditRecord( id=generate_id(), user_id=user_id, type=CreditRecordType.REVOKE.value, amount=Decimal("0.00"), balance_delta=Decimal("0.00"), expired_amount=Decimal("0.00"), balance_after=before, description=description, related_id=related_id, request_time=checked_at, biz_key=biz_key, billing_scene=CreditRecordBillingScene.CREDIT_REVOKE.value, charge_action=CreditRecordAction.REFUND.value, ) db.add(record) await db.flush() for balance in items: if balance.user_id != user_id: raise ValueError("不能跨用户撤销积分") amount = to_credit_decimal(balance.unspent_amount) if amount <= 0: continue before_unspent = amount before_consumed = to_credit_decimal(balance.consumed_amount) balance.unspent_amount = Decimal("0.00") balance.revoked_amount = to_credit_decimal(balance.revoked_amount) + amount balance.revoked_at = checked_at balance.status = CreditBalanceStatus.REVOKED.value total += amount allocation_rows.append( CreditRecordAllocation( id=generate_id(), credit_record_id=record.id, credit_balance_id=balance.id, user_id=user_id, allocation_action=CreditAllocationAction.REVOKE.value, amount=amount, request_time=checked_at, credit_level_snapshot=balance.credit_level, source_type_snapshot=balance.source_type, source_id_snapshot=balance.source_id, valid_from_snapshot=balance.valid_from, expires_at_snapshot=balance.expires_at, unspent_before=before_unspent, unspent_after=Decimal("0.00"), consumed_before=before_consumed, consumed_after=before_consumed, ) ) after = before - total record.amount = -total record.balance_delta = -total record.balance_after = after db.add_all(allocation_rows) attach_credit_snapshot(user, after) await db.flush() log_operation_event( domain="billing", module="dynamic_credit", event_type="CREDIT_REVOKED", event_status="success", source="app.services.credit.ledger_service.revoke_balances", user_id=user_id, task_id=related_id, message="积分批次撤销完成", detail={ "record_id": record.id, "amount": to_float(total), "balance_count": len(allocation_rows), "biz_key": biz_key, }, ) return CreditMutationResult(user, record, True, to_float(total), to_float(before), to_float(after))