「在线联机原型全集」第三章：异步与平台化层（#21–#30）

# 幂等结算示例：带版本号校验的乐观锁
class EconomyService:
    def settle_order(self, order_id: str, expected_version: int):
        with db.transaction():
            # 先读取当前版本
            order = db.query("SELECT * FROM orders WHERE id = %s", order_id)
            if order.status != 'PENDING':
                return  # 已处理，幂等返回
            if order.version != expected_version:
                raise ConcurrencyError("版本冲突，请重试")

            # 执行结算
            self.transfer(order.buyer_id, order.seller_id, order.amount)
            db.execute(
                "UPDATE orders SET status='SETTLED', version=version+1 WHERE id=%s AND version=%s",
                (order_id, expected_version)
            )

            # 写入结算日志（用于对账和审计）
            db.execute(
                "INSERT INTO settlement_logs (order_id, amount, settled_at) VALUES (%s, %s, NOW())",
                (order_id, order.amount)
            )

class GuildLevelingSystem:
    """公会等级经验曲线：指数增长，抑制快速满级"""

    @staticmethod
    def exp_for_level(level: int) -> int:
        """升到下一级所需经验"""
        return int(1000 * (1.5 ** (level - 1)))

    @staticmethod
    def calculate_contribution(member_actions):
        """综合计算成员贡献度"""
        weights = {
            'daily_login': 10,
            'guild_donation': 50,
            'guild_quest_completed': 100,
            'guild_war_participation': 200,
            'recruit_new_member': 150,
        }
        return sum(weights.get(action['type'], 0) * action['count'] for action in member_actions)

class OrderBook:
    """限价订单簿：买单按价格降序，卖单按价格升序"""

    def __init__(self):
        self.bids = []  # 最大堆：价格高优先
        self.asks = []  # 最小堆：价格低优先

    def place_buy(self, player_id, item_id, price, quantity):
        heapq.heappush(self.bids, (-price, time.time(), player_id, item_id, quantity))
        self._match()

    def place_sell(self, player_id, item_id, price, quantity):
        heapq.heappush(self.asks, (price, time.time(), player_id, item_id, quantity))
        self._match()

    def _match(self):
        """撮合引擎：每次新订单入库后尝试匹配"""
        while self.bids and self.asks and -self.bids[0][0] >= self.asks[0][0]:
            bid_price, _, bid_player, bid_item, bid_qty = self.bids[0]
            ask_price, _, ask_player, ask_item, ask_qty = self.asks[0]

            if bid_item != ask_item:
                break

            trade_qty = min(bid_qty, ask_qty)
            trade_price = (ask_price + (-bid_price)) / 2  # 中间价成交

            self._execute_trade(bid_player, ask_player, bid_item, trade_qty, trade_price)

            # 更新或移除已完成的订单
            if bid_qty == trade_qty:
                heapq.heappop(self.bids)
            else:
                self.bids[0] = (-bid_price, self.bids[0][1], bid_player, bid_item, bid_qty - trade_qty)

            if ask_qty == trade_qty:
                heapq.heappop(self.asks)
            else:
                self.asks[0] = (ask_price, self.asks[0][1], ask_player, ask_item, ask_qty - trade_qty)

class UGCModerationService:
    def moderate_upload(self, content: bytes, content_type: str, user_id: str) -> ModerationResult:
        # 1. 内容指纹去重（重复上传秒拒）
        fingerprint = self._compute_fingerprint(content)
        if self._is_duplicate(fingerprint):
            return ModerationResult(rejected=True, reason="DUPLICATE")

        # 2. AI 多模态预审
        ml_result = self.ml_models.predict(content, content_type)
        if ml_result.severity == 'CRITICAL':
            self._flag_user(user_id, 'critical_upload')
            return ModerationResult(rejected=True, reason=ml_result.category)

        if ml_result.severity == 'SUSPICIOUS':
            # 进入人工复审队列
            self.human_review_queue.enqueue(content, user_id, ml_result.confidence)
            return ModerationResult(rejected=True, reason="PENDING_REVIEW")

        # 3. 放行并通过 CDN 预热
        return ModerationResult(rejected=False, cdn_url=self._publish_to_cdn(content))

# 跨服交易 Saga 事务示例
class CrossShardTradeSaga:
    def execute(self, buyer_shard, seller_shard, item_id, price):
        txn_id = generate_uuid()

        try:
            # Phase 1: 买家扣款（本地事务）
            buyer_shard.deduct_currency(txn_id, buyer_id, price)

            # Phase 2: 卖家发货（远程调用）
            seller_shard.transfer_item(txn_id, seller_id, buyer_id, item_id)

            # Phase 3: 结算完成确认
            self._commit(txn_id)

        except Exception as e:
            # 补偿：回滚已执行的操作
            buyer_shard.refund(txn_id, buyer_id, price)
            seller_shard.recall_item(txn_id, seller_id, item_id)
            raise TradeFailedException(e)