功能：质押收益发放记录收益实际USDT价值并进行返佣推送

bex-cloud-staking-entity/src/main/java/com/bex/staking/entity/StakingProfitJournal.java

@@ -46,9 +46,12 @@ public class StakingProfitJournal extends BaseEntity {
     /** 用户 UID */
     private Long userId;
 
-    /** 本次派发收益金额 */
+    /** 本次派发收益金额（BEX） */
     private BigDecimal profitAmount;
 
+    /** 本次派发收益的USDT价值 */
+    private BigDecimal profitUsdtAmount;
+
     /** 结算业务日期 YYYYMMDD */
     private Integer settleDate;
 

bex-cloud-staking-service/src/main/java/com/bex/staking/engine/SettlementEngine.java

@@ -7,9 +7,11 @@ import com.bex.staking.entity.StakingProfitJournal;
 import com.bex.staking.enums.OrderStatus;
 import com.bex.staking.enums.ProductType;
 import com.bex.staking.enums.ProfitStatus;
+import com.bex.staking.grpc.MarketPriceClient;
 import com.bex.staking.mapper.StakingOrderMapper;
 import com.bex.staking.mapper.StakingProductMapper;
 import com.bex.staking.mapper.StakingProfitJournalMapper;
+import com.bex.staking.mq.TradeSettlementProducer;
 import java.math.BigDecimal;
 import java.math.RoundingMode;
 import java.time.LocalDateTime;
@@ -82,6 +84,8 @@ public class SettlementEngine {
     private final StakingProductMapper productMapper;
     private final StakingProfitJournalMapper profitJournalMapper;
     private final TransactionTemplate transactionTemplate;
+    private final TradeSettlementProducer tradeSettlementProducer;
+    private final MarketPriceClient marketPriceClient;
 
     // ==================== 纯逻辑：账本累加计划（供属性测试直接调用） ====================
 
@@ -238,11 +242,20 @@ public class SettlementEngine {
             }
 
             try {
-                Boolean applied = applyInTransaction(current, plan, bizDate);
-                if (Boolean.TRUE.equals(applied)) {
+                BigDecimal bexUSDTPrice = marketPriceClient.getBexUsdtPrice();
+                BigDecimal profitUSDTAmount = plan.profitAmount().multiply(bexUSDTPrice).setScale(SCALE, ROUNDING);
+
+                StakingProfitJournal journal = applyInTransaction(current, plan, bizDate, profitUSDTAmount);
+                if (journal != null) {
+                    tradeSettlementProducer.sendTradeSettledEvent(
+                            journal.getId(),
+                            current.getUserId(),
+                            profitUSDTAmount,
+                            "USDT",
+                            "STAKING");
                     return SettleResult.success(orderNo);
                 }
-                // applied == false：乐观锁版本冲突，重新查询后重试
+                // journal == null：乐观锁版本冲突，重新查询后重试
                 log.debug("订单结算乐观锁冲突, 第 {} 次重试, orderNo={}", attempt, orderNo);
             } catch (DuplicateKeyException e) {
                 // 并发下唯一约束 uk_order_date 兜底：视为已结算跳过（需求 9.7）
@@ -268,16 +281,17 @@ public class SettlementEngine {
      * 在单个本地事务内应用账本更新与流水插入（订单更新 → 流水插入，保证原子一致）。
      *
      * <p>先以乐观锁更新订单账本：{@code updateById} 携带 {@code version}，返回 0 表示版本冲突，此时事务内
-     * 未发生任何变更，直接返回 {@code false}，由调用方重读后重试；返回 1 后插入收益流水
+     * 未发生任何变更，直接返回 {@code null}，由调用方重读后重试；返回 1 后插入收益流水
      * （{@link ProfitStatus#UNCLAIMED}），若命中唯一约束 {@code uk_order_date} 抛 {@link DuplicateKeyException}，
      * 触发整个事务回滚（订单账本更新一并撤销），由调用方按"已结算"幂等跳过处理。
      *
      * @param current 当前订单快照（提供 id 与 version）
      * @param plan 账本累加计划
      * @param bizDate 业务日期 {@code YYYYMMDD}
-     * @return {@code true} 表示成功落库；{@code false} 表示乐观锁版本冲突需重试
+     * @param profitUsdtAmount 收益的USDT价值
+     * @return 创建的收益流水实体；{@code null} 表示乐观锁版本冲突需重试
      */
-    private Boolean applyInTransaction(StakingOrder current, SettlementPlan plan, int bizDate) {
+    private StakingProfitJournal applyInTransaction(StakingOrder current, SettlementPlan plan, int bizDate, BigDecimal profitUsdtAmount) {
         return transactionTemplate.execute(status -> {
             // 3.1 乐观锁更新订单账本（需求 9.5）
             StakingOrder update = new StakingOrder();
@@ -291,12 +305,13 @@ public class SettlementEngine {
             int rows = orderMapper.updateById(update);
             if (rows == 0) {
                 // 版本冲突：事务内无变更，回滚（实为空操作）后由调用方重试
-                return Boolean.FALSE;
+                return null;
             }
 
             // 3.2 插入收益流水（status=UNCLAIMED）；uk_order_date 冲突将抛 DuplicateKeyException 触发回滚
-            profitJournalMapper.insert(buildJournal(current, plan, bizDate));
-            return Boolean.TRUE;
+            StakingProfitJournal journal = buildJournal(current, plan, bizDate, profitUsdtAmount);
+            profitJournalMapper.insert(journal);
+            return journal;
         });
     }
 
@@ -309,14 +324,16 @@ public class SettlementEngine {
      * @param order 订单（提供 orderNo / userId）
      * @param plan 账本累加计划（提供本次派发金额）
      * @param bizDate 业务日期 {@code YYYYMMDD}
+     * @param profitUsdtAmount 收益的USDT价值
      * @return 待插入的收益流水实体
      */
-    private StakingProfitJournal buildJournal(StakingOrder order, SettlementPlan plan, int bizDate) {
+    private StakingProfitJournal buildJournal(StakingOrder order, SettlementPlan plan, int bizDate, BigDecimal profitUsdtAmount) {
         return StakingProfitJournal.builder()
                 .journalNo(JOURNAL_NO_PREFIX + IdUtil.getSnowflakeNextIdStr())
                 .orderNo(order.getOrderNo())
                 .userId(order.getUserId())
                 .profitAmount(plan.profitAmount())
+                .profitUsdtAmount(profitUsdtAmount)
                 .settleDate(bizDate)
                 .status(ProfitStatus.UNCLAIMED.getValue())
                 .createdAt(LocalDateTime.now())

bex-cloud-staking-service/src/main/java/com/bex/staking/mq/TradeSettlementProducer.java

@@ -0,0 +1,41 @@
+package com.bex.staking.mq;
+
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.rocketmq.spring.core.RocketMQTemplate;
+import org.springframework.stereotype.Component;
+
+import java.math.BigDecimal;
+
+/**
+ * 交易结算事件生产者
+ * <p>通过 RocketMQ 发送质押收益结算事件，供其他服务消费。</p>
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class TradeSettlementProducer {
+
+    private final RocketMQTemplate rocketMQTemplate;
+
+    private static final String STAKING_TOPIC = "staking_topic";
+
+    /**
+     * 发送交易结算事件
+     *
+     * @param tradeId      交易流水ID
+     * @param userId       用户ID
+     * @param fee          费用/收益金额
+     * @param feeCurrency  费用币种
+     * @param marketType   市场类型
+     */
+    public void sendTradeSettledEvent(Long tradeId, Long userId, BigDecimal fee, String feeCurrency, String marketType) {
+        String destination = STAKING_TOPIC + ":trade_settled";
+        String message = String.format(
+                "{\"tradeId\":%d,\"userId\":%d,\"fee\":\"%s\",\"feeCurrency\":\"%s\",\"marketType\":\"%s\"}",
+                tradeId, userId, fee, feeCurrency, marketType);
+        rocketMQTemplate.convertAndSend(destination, message);
+        log.info("发送交易结算事件: tradeId={}, userId={}, fee={}, feeCurrency={}, marketType={}",
+                tradeId, userId, fee, feeCurrency, marketType);
+    }
+}