"""积分钱包服务:统一余额变更入口(行锁 + 流水 + 乐观锁,对齐《02-技术文档》3.4)""" from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core import errors from app.core.response import BizException from app.models.billing import PointTransaction from app.models.user import UserWallet # 流水业务类型 BIZ_RECHARGE = 1 # 充值 BIZ_WATCH = 2 # 观看消耗 BIZ_COMMENT = 3 # 评论奖励 BIZ_MEMBERSHIP = 4 # 开通会员 BIZ_GIFT = 5 # 活动赠送 BIZ_ADJUST = 6 # 后台调整 BIZ_CLAWBACK = 7 # 奖励扣回/退款扣回 BIZ_REFERRAL = 8 # 推荐奖励 async def get_or_create_wallet(db: AsyncSession, user_id: int, for_update: bool = False) -> UserWallet: stmt = select(UserWallet).where(UserWallet.user_id == user_id) if for_update: stmt = stmt.with_for_update() wallet = (await db.execute(stmt)).scalar_one_or_none() if wallet is None: wallet = UserWallet(user_id=user_id) db.add(wallet) await db.flush() if for_update: # 新建后再加行锁,保证后续并发读到同一行 wallet = (await db.execute(stmt)).scalar_one() return wallet async def change_points( db: AsyncSession, user_id: int, amount: int, biz_type: int, biz_id: str = "", remark: str = "", allow_negative: bool = False, ) -> int: """变更积分余额,返回变动后余额。 - amount 正增负减;扣减默认不允许透支(退款扣回场景传 allow_negative=True,余额可为负) - 同事务内:行锁 → 校验/更新余额与累计 → 写流水 """ wallet = await get_or_create_wallet(db, user_id, for_update=True) new_balance = wallet.points + amount if amount < 0 and new_balance < 0 and not allow_negative: raise BizException(errors.POINTS_NOT_ENOUGH) wallet.points = new_balance wallet.version += 1 if amount > 0: if biz_type == BIZ_RECHARGE: wallet.total_recharge += amount else: wallet.total_earned += amount else: wallet.total_consumed += -amount db.add( PointTransaction( user_id=user_id, change_amount=amount, balance_after=new_balance, biz_type=biz_type, biz_id=biz_id, remark=remark, ) ) await db.flush() return new_balance async def get_balance(db: AsyncSession, user_id: int) -> int: wallet = await get_or_create_wallet(db, user_id) return wallet.points