feat(recommend): 数仓同步 Service（防回灌+去重+全量替换+历史增量+切批次+清旧）

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

mango-domain/src/main/java/com/mangoo/rating/recommend/service/RuleSyncService.java

@@ -0,0 +1,18 @@
+package com.mangoo.rating.recommend.service;
+
+import com.mangoo.rating.recommend.AjaxResult;
+import com.mangoo.rating.recommend.request.recommend.RuleSyncRequest;
+
+/**
+ * 数仓规则同步服务（全量批次替换 + 价格历史增量 + 生效批次切换）
+ *
+ * @author gengjintao
+ * @date 2026/06/25
+ */
+public interface RuleSyncService {
+
+    /**
+     * 接收数仓推送的全量规则，单事务落库并切换生效批次
+     */
+    AjaxResult sync(RuleSyncRequest request);
+}

mango-domain/src/main/java/com/mangoo/rating/recommend/service/impl/RuleSyncServiceImpl.java

@@ -0,0 +1,154 @@
+package com.mangoo.rating.recommend.service.impl;
+
+import com.mangoo.rating.recommend.AjaxResult;
+import com.mangoo.rating.recommend.manager.ActiveBatchManager;
+import com.mangoo.rating.recommend.manager.CardPopManager;
+import com.mangoo.rating.recommend.manager.CardValueHistoryManager;
+import com.mangoo.rating.recommend.manager.RatingRecommendRuleManager;
+import com.mangoo.rating.recommend.po.CardPopPO;
+import com.mangoo.rating.recommend.po.CardValueHistoryPO;
+import com.mangoo.rating.recommend.po.RatingRecommendRulePO;
+import com.mangoo.rating.recommend.request.recommend.RuleSyncCardDTO;
+import com.mangoo.rating.recommend.request.recommend.RuleSyncRequest;
+import com.mangoo.rating.recommend.request.recommend.SyncPopDTO;
+import com.mangoo.rating.recommend.service.RuleSyncService;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+
+import javax.annotation.Resource;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * 数仓规则同步服务实现
+ *
+ * 单事务流程：防回灌校验 → cardId 去重 → 全量替换规则/POP（删该批次残留+批量插）
+ *           → 价格历史增量 UPSERT → 切生效批次指针 → 清理更早批次（保留新批次+原生效批次）
+ *
+ * @author gengjintao
+ * @date 2026/06/25
+ */
+@Slf4j
+@Service
+public class RuleSyncServiceImpl implements RuleSyncService {
+
+    @Resource
+    private RatingRecommendRuleManager ratingRecommendRuleManager;
+    @Resource
+    private CardPopManager cardPopManager;
+    @Resource
+    private CardValueHistoryManager cardValueHistoryManager;
+    @Resource
+    private ActiveBatchManager activeBatchManager;
+
+    @Override
+    @Transactional(rollbackFor = Exception.class)
+    public AjaxResult sync(RuleSyncRequest request) {
+        if (request == null || request.getBatchNo() == null || request.getBatchNo().trim().isEmpty()) {
+            return AjaxResult.error("batchNo 不能为空");
+        }
+        if (request.getCards() == null || request.getCards().isEmpty()) {
+            // 空数据拒绝，防止全表清空导致推荐全走兜底
+            return AjaxResult.error("cards 不能为空（拒绝空批次以保护线上数据）");
+        }
+        String batchNo = request.getBatchNo().trim();
+
+        // 防回灌：新批次必须比当前生效批次新
+        String oldActive = activeBatchManager.selectActiveBatchNo();
+        if (oldActive != null && batchNo.compareTo(oldActive) <= 0) {
+            return AjaxResult.error("批次号不新于当前生效批次，拒绝回灌：batchNo=" + batchNo + ", active=" + oldActive);
+        }
+
+        // 按 cardId 去重（同批次内重复取最后一条）
+        Map<String, RuleSyncCardDTO> dedup = new LinkedHashMap<>();
+        for (RuleSyncCardDTO c : request.getCards()) {
+            if (c == null || c.getCardId() == null || c.getCardId().trim().isEmpty()) {
+                continue;
+            }
+            dedup.put(c.getCardId(), c);
+        }
+        if (dedup.isEmpty()) {
+            return AjaxResult.error("无有效卡片数据（cardId 均为空）");
+        }
+
+        // 转换 PO
+        List<RatingRecommendRulePO> rulePOs = new ArrayList<>(dedup.size());
+        List<CardPopPO> popPOs = new ArrayList<>();
+        List<CardValueHistoryPO> historyPOs = new ArrayList<>();
+        for (RuleSyncCardDTO c : dedup.values()) {
+            RatingRecommendRulePO r = new RatingRecommendRulePO();
+            r.setCardId(c.getCardId());
+            r.setCardNo(c.getCardNo());
+            r.setPlayer(c.getPlayer());
+            r.setYear(c.getYear());
+            r.setSeries(c.getSeries());
+            r.setCardSet(c.getCardSet());
+            r.setRecommendEfficiency(c.getRecommendEfficiency());
+            r.setCurrentValue(c.getCurrentValue());
+            r.setValueChangePct(c.getValueChangePct());
+            r.setValueMin(c.getValueMin());
+            r.setValueMax(c.getValueMax());
+            r.setBatchNo(batchNo);
+            rulePOs.add(r);
+
+            if (c.getPops() != null) {
+                for (SyncPopDTO p : c.getPops()) {
+                    if (p == null) {
+                        continue;
+                    }
+                    CardPopPO pop = new CardPopPO();
+                    pop.setCardId(c.getCardId());
+                    pop.setAgency(p.getAgency());
+                    pop.setGradeAll(p.getAll());
+                    pop.setGrade10(p.getG10());
+                    pop.setGrade9(p.getG9());
+                    pop.setGrade8(p.getG8());
+                    pop.setGrade7(p.getG7());
+                    pop.setBatchNo(batchNo);
+                    popPOs.add(pop);
+                }
+            }
+
+            // 价格历史增量：仅当日期与当日价都给了才追加
+            if (c.getRecordDate() != null && c.getTodayPrice() != null) {
+                CardValueHistoryPO h = new CardValueHistoryPO();
+                h.setCardId(c.getCardId());
+                h.setPrice(c.getTodayPrice());
+                h.setRecordDate(c.getRecordDate());
+                historyPOs.add(h);
+            }
+        }
+
+        // 全量替换：清该批次残留（幂等重入）后批量插
+        ratingRecommendRuleManager.deleteByBatchNo(batchNo);
+        cardPopManager.deleteByBatchNo(batchNo);
+        ratingRecommendRuleManager.batchInsert(rulePOs);
+        if (!popPOs.isEmpty()) {
+            cardPopManager.batchInsert(popPOs);
+        }
+        // 价格历史增量 UPSERT
+        if (!historyPOs.isEmpty()) {
+            cardValueHistoryManager.batchUpsert(historyPOs);
+        }
+
+        // 切生效批次指针（此后查询读新批次）
+        activeBatchManager.updateActiveBatchNo(batchNo);
+
+        // 清理更早批次：保留新批次 + 原生效批次（供回滚）
+        List<String> allBatches = ratingRecommendRuleManager.selectDistinctBatchNos();
+        for (String b : allBatches) {
+            if (b == null || b.equals(batchNo) || b.equals(oldActive)) {
+                continue;
+            }
+            ratingRecommendRuleManager.deleteByBatchNo(b);
+            cardPopManager.deleteByBatchNo(b);
+        }
+
+        log.info("数仓规则同步完成：batchNo={}, 卡片数={}, POP行={}, 历史增量={}",
+                batchNo, rulePOs.size(), popPOs.size(), historyPOs.size());
+        return AjaxResult.success("同步成功，生效批次=" + batchNo);
+    }
+}