feat(统一支付): 下游回调/SSRF 防护/重试

java/src/main/java/com/payment/platform/module/payment/unified/config/UnifiedHttpConfig.java

@@ -0,0 +1,72 @@
+package com.payment.platform.module.payment.unified.config;
+
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.client.SimpleClientHttpRequestFactory;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.net.HttpURLConnection;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+/**
+ * 统一支付下游回调的专用出站资源（spec §7.6）。
+ *
+ * <p>三点刻意的设计：
+ *
+ * <ol>
+ *   <li><b>独立的 {@code RestTemplate}，不加 {@code @Primary}</b> —— 容器里已有
+ *       {@code WebMvcConfig.restTemplate()}（故意无超时，既有调用方依赖）。本 bean 按
+ *       {@code @Qualifier} 注入；给它加 {@code @Primary} 会让按类型注入的既有调用方
+ *       静默换 bean（{@code WebMvcConfig} 的注释里记录过三种布局的实测读数）。</li>
+ *   <li><b>不跟随 30x</b> —— 商户把回调地址指向自己的重定向端点时，跟随会让一次 push
+ *       变成对另一个地址的请求，绕过 {@code UnifiedCallbackService} 的 SSRF 检查
+ *       （检查的只是第一跳）。{@code HttpURLConnection} 默认跟随重定向，
+ *       这里在 {@code prepareConnection} 里显式关掉。</li>
+ *   <li><b>有限超时</b> —— 读 {@code unified.pay.callback-timeout-ms}。回调重试跑在
+ *       {@code @Scheduled} 线程上，而 {@code spring.task.scheduling.pool.size} 默认是 1：
+ *       一个只收包不响应的商户端点足以把整族定时任务一起饿死。</li>
+ * </ol>
+ */
+@Configuration
+public class UnifiedHttpConfig {
+
+    /** 回调专用 RestTemplate 的 bean 名（服务侧用 {@code @Qualifier} 按名注入）。 */
+    public static final String CALLBACK_REST_TEMPLATE = "unifiedCallbackRestTemplate";
+
+    /** 回调线程池的 bean 名。 */
+    public static final String CALLBACK_EXECUTOR = "unifiedCallbackExecutor";
+
+    @Bean(name = CALLBACK_REST_TEMPLATE)
+    public RestTemplate callbackRestTemplate(UnifiedPayProperties properties) {
+        SimpleClientHttpRequestFactory factory = new SimpleClientHttpRequestFactory() {
+            @Override
+            protected void prepareConnection(HttpURLConnection connection, String httpMethod)
+                    throws IOException {
+                super.prepareConnection(connection, httpMethod);
+                connection.setInstanceFollowRedirects(false);
+            }
+        };
+        factory.setConnectTimeout(properties.getCallbackTimeoutMs());
+        factory.setReadTimeout(properties.getCallbackTimeoutMs());
+        return new RestTemplate(factory);
+    }
+
+    /**
+     * 固定大小的 daemon 线程池。
+     *
+     * <p>不用 {@code @Async}：项目没有开 {@code @EnableAsync}，{@code @Async} 会静默地
+     * 同步执行，把上游通知的响应时间拖长（同 {@code HarCallbackService} 的注释）。
+     * daemon + {@code destroyMethod="shutdown"} 保证容器关闭时能收掉线程。
+     */
+    @Bean(name = CALLBACK_EXECUTOR, destroyMethod = "shutdown")
+    public ExecutorService unifiedCallbackExecutor(UnifiedPayProperties properties) {
+        int poolSize = Math.max(1, properties.getCallbackPoolSize());
+        return Executors.newFixedThreadPool(poolSize, runnable -> {
+            Thread thread = new Thread(runnable, "unified-callback");
+            thread.setDaemon(true);
+            return thread;
+        });
+    }
+}

java/src/main/java/com/payment/platform/module/payment/unified/scheduler/UnifiedCallbackRetryScheduler.java

@@ -0,0 +1,38 @@
+package com.payment.platform.module.payment.unified.scheduler;
+
+import com.payment.platform.module.payment.unified.service.UnifiedCallbackService;
+import com.payment.platform.module.payment.unified.util.UnifiedLogSanitizer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 下游回调失败重试（spec §7.6）。
+ *
+ * <p>每 60 秒扫一次 {@code notify_status='FAILED' AND notify_count < 3}（含首次，最多
+ * 投递 3 次），按 {@code last_notify_time} 升序饿最久的先推、每批 LIMIT 50。
+ *
+ * <p>异常必须在这里兜住：本任务跑在 {@code @Scheduled} 线程上，而
+ * {@code spring.task.scheduling.pool.size} 默认是 1 —— 一个逃逸的异常会把整族定时任务
+ * 一起饿死。扫描与逐行的异常处理在 {@link UnifiedCallbackService#retryFailed()} 内，
+ * 本层只兜「连服务调用本身都炸了」的情况。
+ */
+@Slf4j
+@Component
+public class UnifiedCallbackRetryScheduler {
+
+    private final UnifiedCallbackService callbackService;
+
+    public UnifiedCallbackRetryScheduler(UnifiedCallbackService callbackService) {
+        this.callbackService = callbackService;
+    }
+
+    @Scheduled(fixedDelayString = "${unified.pay.callback-retry-interval-ms:60000}")
+    public void retry() {
+        try {
+            callbackService.retryFailed();
+        } catch (Exception e) {
+            log.error("[统一支付回调重试] 执行失败: {}", UnifiedLogSanitizer.flatten(e.getMessage()), e);
+        }
+    }
+}

java/src/main/java/com/payment/platform/module/payment/unified/service/UnifiedCallbackService.java

@@ -0,0 +1,564 @@
+package com.payment.platform.module.payment.unified.service;
+
+import cn.hutool.core.util.StrUtil;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.payment.platform.module.payment.apikey.entity.TenantApiKeyEntity;
+import com.payment.platform.module.payment.apikey.mapper.TenantApiKeyMapper;
+import com.payment.platform.module.payment.openapi.entity.OpenConfEntity;
+import com.payment.platform.module.payment.openapi.mapper.OpenConfMapper;
+import com.payment.platform.module.payment.unified.config.UnifiedHttpConfig;
+import com.payment.platform.module.payment.unified.config.UnifiedPayProperties;
+import com.payment.platform.module.payment.unified.entity.UnifiedOrderEntity;
+import com.payment.platform.module.payment.unified.entity.UnifiedRefundEntity;
+import com.payment.platform.module.payment.unified.enums.UnifiedNotifyStatus;
+import com.payment.platform.module.payment.unified.mapper.UnifiedOrderMapper;
+import com.payment.platform.module.payment.unified.mapper.UnifiedRefundMapper;
+import com.payment.platform.module.payment.unified.util.UnifiedLogSanitizer;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.http.ResponseEntity;
+import org.springframework.stereotype.Service;
+import org.springframework.util.LinkedMultiValueMap;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.client.RestTemplate;
+
+import java.net.URI;
+import java.time.OffsetDateTime;
+import java.time.ZoneId;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.concurrent.Executor;
+
+/**
+ * 把钱事件（订单/退款）回调给下游商户（spec §7.6）。
+ *
+ * <p>报文沿用平台既有 form 形状（{@code notify_id} / {@code timestamp} / {@code content}），
+ * {@code content} 按统一支付自己的分层结构（{@code event/order/refund/upstream}）。
+ *
+ * <p>四条刻意的设计：
+ *
+ * <ol>
+ *   <li><b>{@code notify_id} 确定性</b>：{@code uo_<order_no>_<trade_status>} /
+ *       {@code ur_<refund_no>_<state>}。同一事件重试保持同一个 id，接收方可直接去重
+ *       （HAR 每次新生成 id，这里刻意改进）。</li>
+ *   <li><b>先写 FAILED 再异步推送，2xx 后写 SUCCESS</b>：进程在推送途中崩溃时，行仍然是
+ *       FAILED + 计数已加，重试调度器能捞到它；反过来先写 SUCCESS 再推送会永久丢事件。</li>
+ *   <li><b>{@code content.upstream} 是白名单投影</b>（R35）：上游通知混有我们上送的
+ *       {@code channelExtra}（付款码 / openid / buyerUserId）。整体塞原始 JsonNode 等于把
+ *       付款凭据转发给下游，命中本模块的硬约束。</li>
+ *   <li><b>SSRF 最小防线</b>：只允许 http/https，拒绝本机/环回/云元数据地址；<b>不解析</b>
+ *       私网 IP —— 企业内网回调是合法用法。30x 由专用 {@code RestTemplate} 负责不跟随。</li>
+ * </ol>
+ *
+ * <p>日志约束：url 与异常 message 先过 {@link UnifiedLogSanitizer#flatten}（CWE-117）；
+ * {@code content} 永不整体打印。
+ */
+@Slf4j
+@Service
+public class UnifiedCallbackService {
+
+    /**
+     * 回调报文的序列化器：**裸** {@code ObjectMapper}，不能注入全局 Jackson 的
+     * SNAKE_CASE 实例 —— map 的键已经手写成 snake_case，而 {@code upstream} 的白名单键
+     * 是上游原样的 camelCase（{@code payOrderId} 等），被再命名一次就不是契约了。
+     * 时间在放入 map 前已转成字符串，因此不需要 JavaTimeModule。
+     */
+    private static final ObjectMapper CONTENT_MAPPER = new ObjectMapper();
+
+    /** 契约里的时间一律按北京时间输出（{@code OffsetDateTime.toString()} 带偏移）。 */
+    private static final ZoneId SHANGHAI = ZoneId.of("Asia/Shanghai");
+
+    /** 每轮重试扫描上限。 */
+    private static final String RETRY_LIMIT = "LIMIT 50";
+
+    /**
+     * {@code content.upstream} 的白名单（R35，逐字取 spec §7.5）。只拷贝存在的字段；
+     * 绝不整体转发原始节点 —— 它可能带 {@code channelExtra} / {@code openid} / {@code sign}。
+     */
+    private static final List<String> UPSTREAM_WHITELIST = List.of(
+            "payOrderId", "mchOrderNo", "state", "orderState", "amount", "currency",
+            "wayCode", "ifCode", "channelOrderNo", "successTime", "createdAt",
+            "refundOrderId", "mchRefundNo", "refundAmount", "payAmount",
+            "extParam", "errCode", "errMsg");
+
+    private final UnifiedOrderMapper orderMapper;
+    private final UnifiedRefundMapper refundMapper;
+    private final TenantApiKeyMapper apiKeyMapper;
+    private final OpenConfMapper confMapper;
+    private final RestTemplate restTemplate;
+    private final Executor callbackExecutor;
+    private final UnifiedPayProperties properties;
+
+    public UnifiedCallbackService(UnifiedOrderMapper orderMapper,
+                                  UnifiedRefundMapper refundMapper,
+                                  TenantApiKeyMapper apiKeyMapper,
+                                  OpenConfMapper confMapper,
+                                  @Qualifier(UnifiedHttpConfig.CALLBACK_REST_TEMPLATE) RestTemplate restTemplate,
+                                  @Qualifier(UnifiedHttpConfig.CALLBACK_EXECUTOR) Executor callbackExecutor,
+                                  UnifiedPayProperties properties) {
+        this.orderMapper = orderMapper;
+        this.refundMapper = refundMapper;
+        this.apiKeyMapper = apiKeyMapper;
+        this.confMapper = confMapper;
+        this.restTemplate = restTemplate;
+        this.callbackExecutor = callbackExecutor;
+        this.properties = properties;
+    }
+
+    // ───────────────────────────── 对外入口 ─────────────────────────────
+
+    /**
+     * 订单事件回调。事件矩阵由调用方保证（只在 trade_status 2/3/4/5/6 调用）。
+     *
+     * @param upstream 上游通知/查单的 data 节点；重试时传 {@code null}
+     */
+    public void notifyOrder(UnifiedOrderEntity order, JsonNode upstream) {
+        if (order == null) {
+            return;
+        }
+        String notifyId = orderNotifyId(order.getOrderNo(), order.getTradeStatus());
+        String url = resolveOrderCallbackUrl(order);
+        if (!isAllowedCallbackUrl(url)) {
+            log.warn("[统一支付回调] 订单无可用的回调地址，记 FAILED 待重试: order_no={}, url={}",
+                    order.getOrderNo(), UnifiedLogSanitizer.flatten(url));
+            writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
+            return;
+        }
+
+        String content = buildOrderContent(order, upstream);
+        if (content == null) {
+            writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
+            return;
+        }
+
+        // 先落 FAILED + 计数：崩溃窗口里这一单仍会被重试扫描捞起
+        writeOrderNotifyStatus(order, UnifiedNotifyStatus.FAILED, true);
+        submit(() -> {
+            if (pushOnce(url, notifyId, content)) {
+                writeOrderNotifyStatus(order, UnifiedNotifyStatus.SUCCESS, false);
+            }
+        }, "order_no=" + order.getOrderNo());
+    }
+
+    /**
+     * 退款事件回调。事件矩阵由调用方保证（只在 state 2/3/4 调用）。
+     *
+     * @param order    退款所属订单（退款事件里要带订单当前快照）；重试时由本类按 order_no 查出
+     * @param upstream 上游通知/查单的 data 节点；重试时传 {@code null}
+     */
+    public void notifyRefund(UnifiedRefundEntity refund, UnifiedOrderEntity order, JsonNode upstream) {
+        if (refund == null) {
+            return;
+        }
+        String notifyId = refundNotifyId(refund.getRefundNo(), refund.getState());
+        String url = resolveRefundCallbackUrl(refund, order);
+        if (!isAllowedCallbackUrl(url)) {
+            log.warn("[统一支付回调] 退款无可用的回调地址，记 FAILED 待重试: refund_no={}, url={}",
+                    refund.getRefundNo(), UnifiedLogSanitizer.flatten(url));
+            writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
+            return;
+        }
+
+        String content = buildRefundContent(refund, order, upstream);
+        if (content == null) {
+            writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
+            return;
+        }
+
+        writeRefundNotifyStatus(refund, UnifiedNotifyStatus.FAILED, true);
+        submit(() -> {
+            if (pushOnce(url, notifyId, content)) {
+                writeRefundNotifyStatus(refund, UnifiedNotifyStatus.SUCCESS, false);
+            }
+        }, "refund_no=" + refund.getRefundNo());
+    }
+
+    /**
+     * 扫出失败且未达次数上限的订单/退款再推一遍 —— 供 60 秒调度器调用。
+     *
+     * <p>扫描与逐行推送各自捕获异常：调用方是调度线程，上面还挂着其他定时任务
+     * （{@code spring.task.scheduling.pool.size} 默认 1）。
+     */
+    public void retryFailed() {
+        retryFailedOrders();
+        retryFailedRefunds();
+    }
+
+    // ───────────────────────────── 重试扫描 ─────────────────────────────
+
+    private void retryFailedOrders() {
+        List<UnifiedOrderEntity> orders;
+        try {
+            orders = orderMapper.selectList(new LambdaQueryWrapper<UnifiedOrderEntity>()
+                    .eq(UnifiedOrderEntity::getNotifyStatus, UnifiedNotifyStatus.FAILED)
+                    .lt(UnifiedOrderEntity::getNotifyCount, properties.getCallbackMaxRetries())
+                    .orderByAsc(UnifiedOrderEntity::getLastNotifyTime)
+                    .last(RETRY_LIMIT));
+        } catch (Exception e) {
+            log.error("[统一支付回调] 扫描待重试订单失败: {}", UnifiedLogSanitizer.flatten(e.getMessage()), e);
+            return;
+        }
+        if (orders == null) {
+            return;
+        }
+        for (UnifiedOrderEntity order : orders) {
+            try {
+                notifyOrder(order, null);
+            } catch (Exception e) {
+                // 单行失败不能拖垮整批
+                log.error("[统一支付回调] 重试订单异常: order_no={}, error={}",
+                        order.getOrderNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
+            }
+        }
+    }
+
+    private void retryFailedRefunds() {
+        List<UnifiedRefundEntity> refunds;
+        try {
+            refunds = refundMapper.selectList(new LambdaQueryWrapper<UnifiedRefundEntity>()
+                    .eq(UnifiedRefundEntity::getNotifyStatus, UnifiedNotifyStatus.FAILED)
+                    .lt(UnifiedRefundEntity::getNotifyCount, properties.getCallbackMaxRetries())
+                    .orderByAsc(UnifiedRefundEntity::getLastNotifyTime)
+                    .last(RETRY_LIMIT));
+        } catch (Exception e) {
+            log.error("[统一支付回调] 扫描待重试退款失败: {}", UnifiedLogSanitizer.flatten(e.getMessage()), e);
+            return;
+        }
+        if (refunds == null) {
+            return;
+        }
+        for (UnifiedRefundEntity refund : refunds) {
+            try {
+                UnifiedOrderEntity order = orderMapper.selectOne(new LambdaQueryWrapper<UnifiedOrderEntity>()
+                        .eq(UnifiedOrderEntity::getOrderNo, refund.getOrderNo()));
+                if (order == null) {
+                    // R34：订单行不存在属于数据损坏。不改状态、不递增次数 —— 置 SUCCESS 会吞掉
+                    // 一条本该投递的事件，递增计数会在数据修复前白白烧掉重试次数。
+                    log.warn("[统一支付回调] 重试退款时找不到所属订单，跳过该行: refund_no={}",
+                            refund.getRefundNo());
+                    continue;
+                }
+                notifyRefund(refund, order, null);
+            } catch (Exception e) {
+                log.error("[统一支付回调] 重试退款异常: refund_no={}, error={}",
+                        refund.getRefundNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
+            }
+        }
+    }
+
+    // ───────────────────────────── 取址 ─────────────────────────────
+
+    /** 订单：快照 callback_url → Key → open_conf。 */
+    private String resolveOrderCallbackUrl(UnifiedOrderEntity order) {
+        if (StrUtil.isNotBlank(order.getCallbackUrl())) {
+            return order.getCallbackUrl();
+        }
+        return resolveTenantCallbackUrl(order.getTenantId(), order.getApiKey());
+    }
+
+    /**
+     * 退款：退款快照 → 订单快照 → Key → open_conf。
+     *
+     * <p>Key 一级优先 {@code refund.apiKey}（发起退款时记录的那把），为空再用订单的
+     * （R21）。
+     */
+    private String resolveRefundCallbackUrl(UnifiedRefundEntity refund, UnifiedOrderEntity order) {
+        if (StrUtil.isNotBlank(refund.getCallbackUrl())) {
+            return refund.getCallbackUrl();
+        }
+        if (order != null && StrUtil.isNotBlank(order.getCallbackUrl())) {
+            return order.getCallbackUrl();
+        }
+        String apiKey = StrUtil.isNotBlank(refund.getApiKey())
+                ? refund.getApiKey()
+                : (order == null ? null : order.getApiKey());
+        Long tenantId = refund.getTenantId() != null
+                ? refund.getTenantId()
+                : (order == null ? null : order.getTenantId());
+        return resolveTenantCallbackUrl(tenantId, apiKey);
+    }
+
+    /** Key 按 api_key 精确查（该列全局唯一）；open_conf 按租户取 updated_time 最新一行。 */
+    private String resolveTenantCallbackUrl(Long tenantId, String apiKey) {
+        if (StrUtil.isNotBlank(apiKey)) {
+            TenantApiKeyEntity key = apiKeyMapper.selectOne(new LambdaQueryWrapper<TenantApiKeyEntity>()
+                    .eq(TenantApiKeyEntity::getApiKey, apiKey)
+                    .last("LIMIT 1"));
+            if (key != null && StrUtil.isNotBlank(key.getReturnUrl())) {
+                return key.getReturnUrl();
+            }
+        }
+        if (tenantId == null) {
+            return null;
+        }
+        // open_conf.tenant_id 没有全局唯一约束 —— 不用 selectOne(... LIMIT 1) 静默挑行
+        List<OpenConfEntity> confs = confMapper.selectList(new LambdaQueryWrapper<OpenConfEntity>()
+                .eq(OpenConfEntity::getTenantId, tenantId)
+                .orderByDesc(OpenConfEntity::getUpdatedTime));
+        if (confs == null || confs.isEmpty()) {
+            return null;
+        }
+        if (confs.size() > 1) {
+            log.warn("[统一支付回调] 租户 open_conf 有多行，取 updated_time 最新一行: tenant_id={}, rows={}",
+                    tenantId, confs.size());
+        }
+        OpenConfEntity latest = confs.get(0);
+        return latest == null ? null : latest.getReturnUrl();
+    }
+
+    // ───────────────────────────── 报文内容 ─────────────────────────────
+
+    private String buildOrderContent(UnifiedOrderEntity order, JsonNode upstream) {
+        Map<String, Object> content = new LinkedHashMap<>();
+        content.put("event", "ORDER");
+        content.put("order", orderSnapshot(order));
+        content.put("upstream", projectUpstream(upstream));
+        return serialize(content, "order_no=" + order.getOrderNo());
+    }
+
+    private String buildRefundContent(UnifiedRefundEntity refund, UnifiedOrderEntity order, JsonNode upstream) {
+        Map<String, Object> content = new LinkedHashMap<>();
+        content.put("event", "REFUND");
+
+        Map<String, Object> refundNode = new LinkedHashMap<>();
+        refundNode.put("refund_no", refund.getRefundNo());
+        refundNode.put("out_refund_no", refund.getOutRefundNo());
+        refundNode.put("order_no", refund.getOrderNo());
+        refundNode.put("amount", refund.getAmount());
+        refundNode.put("state", refund.getState());
+        refundNode.put("upstream_refund_order_id", refund.getUpstreamRefundOrderId());
+        refundNode.put("success_time", shanghai(refund.getSuccessTime()));
+        refundNode.put("err_code", refund.getErrCode());
+        refundNode.put("err_msg", refund.getErrMsg());
+        content.put("refund", refundNode);
+
+        // 退款事件同时带订单的当前快照（trade_status/refunded_amount/refundable_amount）
+        if (order != null) {
+            content.put("order", orderSnapshot(order));
+        }
+        content.put("upstream", projectUpstream(upstream));
+        return serialize(content, "refund_no=" + refund.getRefundNo());
+    }
+
+    /**
+     * 订单当前快照。
+     *
+     * <p>{@code refundable_amount = amount - sumOccupiedAmount(order.id)}：occupied 已含
+     * state=2（成功退款），**不得**再减 {@code refunded_amount}（R21）—— 重复相减会少算，
+     * 商户预填一个接口必然拒绝的金额。
+     */
+    private Map<String, Object> orderSnapshot(UnifiedOrderEntity order) {
+        Map<String, Object> node = new LinkedHashMap<>();
+        node.put("order_no", order.getOrderNo());
+        node.put("out_trade_no", order.getOutTradeNo());
+        node.put("mch_no", order.getMchNo());
+        node.put("way_code", order.getWayCode());
+        node.put("amount", order.getAmount());
+        node.put("refunded_amount", order.getRefundedAmount());
+        node.put("refundable_amount", refundableAmount(order));
+        node.put("trade_status", order.getTradeStatus());
+        node.put("upstream_pay_order_id", order.getUpstreamPayOrderId());
+        node.put("upstream_channel_order_no", order.getUpstreamChannelOrderNo());
+        node.put("pay_time", shanghai(order.getPayTime()));
+        node.put("err_code", order.getErrCode());
+        node.put("err_msg", order.getErrMsg());
+        return node;
+    }
+
+    private long refundableAmount(UnifiedOrderEntity order) {
+        long amount = order.getAmount() == null ? 0L : order.getAmount();
+        long occupied = order.getId() == null ? 0L : refundMapper.sumOccupiedAmount(order.getId());
+        return amount - occupied;
+    }
+
+    /**
+     * {@code content.upstream} 的白名单投影（R35）：只拷贝存在的字段，绝不整体转发原始节点。
+     * 原始上游通知混有我们上送的 {@code channelExtra}（付款码 / openid / buyerUserId）。
+     */
+    private static Map<String, Object> projectUpstream(JsonNode upstream) {
+        Map<String, Object> projected = new LinkedHashMap<>();
+        if (upstream == null || !upstream.isObject()) {
+            return projected;
+        }
+        for (String field : UPSTREAM_WHITELIST) {
+            JsonNode value = upstream.get(field);
+            if (value != null && !value.isNull()) {
+                projected.put(field, value);
+            }
+        }
+        return projected;
+    }
+
+    private static String shanghai(OffsetDateTime time) {
+        return time == null ? null : time.atZoneSameInstant(SHANGHAI).toOffsetDateTime().toString();
+    }
+
+    private static String serialize(Map<String, Object> content, String identity) {
+        try {
+            return CONTENT_MAPPER.writeValueAsString(content);
+        } catch (Exception e) {
+            log.error("[统一支付回调] 报文序列化失败，本次不推送: {}, error={}",
+                    identity, UnifiedLogSanitizer.flatten(e.getMessage()));
+            return null;
+        }
+    }
+
+    // ───────────────────────────── 推送 ─────────────────────────────
+
+    /**
+     * 单次推送 —— 线程内不重试（重试交给 {@link #retryFailed}）。
+     *
+     * @return true = 下游返回 2xx；任何非 2xx / 异常都算一次失败尝试
+     */
+    boolean pushOnce(String url, String notifyId, String content) {
+        MultiValueMap<String, String> form = new LinkedMultiValueMap<>();
+        form.add("notify_id", notifyId);
+        form.add("timestamp", String.valueOf(System.currentTimeMillis()));
+        form.add("content", content);
+
+        HttpHeaders headers = new HttpHeaders();
+        headers.setContentType(MediaType.APPLICATION_FORM_URLENCODED);
+        try {
+            ResponseEntity<String> response = restTemplate.postForEntity(
+                    url, new HttpEntity<>(form, headers), String.class);
+            if (response.getStatusCode().is2xxSuccessful()) {
+                log.info("[统一支付回调] 推送成功: notify_id={}, url={}",
+                        notifyId, UnifiedLogSanitizer.flatten(url));
+                return true;
+            }
+            log.warn("[统一支付回调] 下游返回非 2xx，记 FAILED 待重试: notify_id={}, status={}, url={}",
+                    notifyId, response.getStatusCode(), UnifiedLogSanitizer.flatten(url));
+            return false;
+        } catch (Exception e) {
+            log.warn("[统一支付回调] 推送异常，记 FAILED 待重试: notify_id={}, url={}, error={}",
+                    notifyId, UnifiedLogSanitizer.flatten(url), UnifiedLogSanitizer.flatten(e.getMessage()));
+            return false;
+        }
+    }
+
+    private void submit(Runnable task, String identity) {
+        try {
+            callbackExecutor.execute(task);
+        } catch (Exception e) {
+            // 池已满/已关：行已落 FAILED，下一轮重试会捞起
+            log.warn("[统一支付回调] 异步任务提交失败，保持 FAILED 等重试: {}, error={}",
+                    identity, UnifiedLogSanitizer.flatten(e.getMessage()));
+        }
+    }
+
+    // ───────────────────────────── 状态回写 ─────────────────────────────
+
+    /**
+     * 回调状态回写。用 {@code LambdaUpdateWrapper} 而不是实体 + {@code updateById}：
+     * ① 实体带 {@code @Version}，回调线程可能在事务提交前读到旧版本，{@code updateById}
+     *    会被乐观锁静默挡掉（0 行）；② 计数必须落在 SQL 侧自增，并发重试下
+     *    read-modify-write 会丢计数、上限形同虚设。代价是 {@code updated_time} 要手写
+     *    （{@code update(null, wrapper)} 不经自动填充）。
+     */
+    private void writeOrderNotifyStatus(UnifiedOrderEntity order, String status, boolean countAttempt) {
+        try {
+            OffsetDateTime now = OffsetDateTime.now();
+            LambdaUpdateWrapper<UnifiedOrderEntity> update = new LambdaUpdateWrapper<UnifiedOrderEntity>()
+                    .eq(UnifiedOrderEntity::getOrderNo, order.getOrderNo())
+                    .set(UnifiedOrderEntity::getNotifyStatus, status)
+                    .set(UnifiedOrderEntity::getLastNotifyTime, now)
+                    .set(UnifiedOrderEntity::getUpdatedTime, now);
+            if (countAttempt) {
+                update.setSql("notify_count = notify_count + 1");
+            }
+            if (orderMapper.update(null, update) == 0) {
+                log.warn("[统一支付回调] 状态写入未命中订单: order_no={}", order.getOrderNo());
+            }
+        } catch (Exception e) {
+            log.warn("[统一支付回调] 回写订单 notify_status 失败: order_no={}, error={}",
+                    order.getOrderNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
+        }
+    }
+
+    private void writeRefundNotifyStatus(UnifiedRefundEntity refund, String status, boolean countAttempt) {
+        try {
+            OffsetDateTime now = OffsetDateTime.now();
+            LambdaUpdateWrapper<UnifiedRefundEntity> update = new LambdaUpdateWrapper<UnifiedRefundEntity>()
+                    .eq(UnifiedRefundEntity::getRefundNo, refund.getRefundNo())
+                    .set(UnifiedRefundEntity::getNotifyStatus, status)
+                    .set(UnifiedRefundEntity::getLastNotifyTime, now)
+                    .set(UnifiedRefundEntity::getUpdatedTime, now);
+            if (countAttempt) {
+                update.setSql("notify_count = notify_count + 1");
+            }
+            if (refundMapper.update(null, update) == 0) {
+                log.warn("[统一支付回调] 状态写入未命中退款: refund_no={}", refund.getRefundNo());
+            }
+        } catch (Exception e) {
+            log.warn("[统一支付回调] 回写退款 notify_status 失败: refund_no={}, error={}",
+                    refund.getRefundNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
+        }
+    }
+
+    // ───────────────────────────── 静态契约 ─────────────────────────────
+
+    /**
+     * 下游地址的最小 SSRF 防线（spec §7.6）：只允许 http/https 且 host 非空；拒绝
+     * localhost / 127.0.0.0/8 / ::1 / 0.0.0.0 / 云元数据地址。<b>不解析私网 IP</b> ——
+     * 企业内网回调是合法用法，全禁会误伤。DNS 解析也不做（见 spec §11 残余风险）。
+     */
+    static boolean isAllowedCallbackUrl(String url) {
+        if (StrUtil.isBlank(url)) {
+            return false;
+        }
+        URI uri;
+        try {
+            uri = URI.create(url.trim());
+        } catch (IllegalArgumentException e) {
+            return false;
+        }
+        String scheme = uri.getScheme();
+        if (scheme == null) {
+            return false;
+        }
+        scheme = scheme.toLowerCase(Locale.ROOT);
+        if (!"http".equals(scheme) && !"https".equals(scheme)) {
+            return false;
+        }
+        String host = uri.getHost();
+        if (StrUtil.isBlank(host)) {
+            return false;
+        }
+        // 字面 IPv6 的 getHost() 带方括号：[::1]
+        if (host.startsWith("[") && host.endsWith("]")) {
+            host = host.substring(1, host.length() - 1);
+        }
+        host = host.toLowerCase(Locale.ROOT);
+        if ("localhost".equals(host)) {
+            return false;
+        }
+        if (host.startsWith("127.")) {
+            return false;
+        }
+        if ("::1".equals(host)) {
+            return false;
+        }
+        if ("0.0.0.0".equals(host)) {
+            return false;
+        }
+        return !"169.254.169.254".equals(host);
+    }
+
+    /** 订单事件的确定性 notify_id：同一对象同一状态的重试保持同一个 id。 */
+    static String orderNotifyId(String orderNo, String tradeStatus) {
+        return "uo_" + orderNo + "_" + tradeStatus;
+    }
+
+    /** 退款事件的确定性 notify_id。 */
+    static String refundNotifyId(String refundNo, String state) {
+        return "ur_" + refundNo + "_" + state;
+    }
+}

java/src/test/java/com/payment/platform/module/payment/unified/config/UnifiedHttpConfigTest.java

@@ -0,0 +1,119 @@
+package com.payment.platform.module.payment.unified.config;
+
+import com.sun.net.httpserver.HttpServer;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Primary;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.client.RestTemplate;
+
+import java.io.IOException;
+import java.lang.reflect.Method;
+import java.net.InetSocketAddress;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * 下游回调专用出站配置（spec §7.6）。
+ *
+ * <p>两条主断言各自防一类回归：
+ * <ul>
+ *   <li>**不跟随 30x**：商户把回调地址指向自己的重定向端点时，跟随会让一次 push 变成
+ *       对另一个地址的请求 —— SSRF 的最小防线在这里（用真实 HttpServer 读 302 与目标命中数，
+ *       不读源码文本）。</li>
+ *   <li>**不是 {@code @Primary}**：容器里已有 {@code WebMvcConfig.restTemplate()}（故意无超时）。
+ *       给本 bean 加 {@code @Primary} 会让按类型注入的既有调用方静默拿到带超时的 bean
+ *       （{@code WebMvcConfig} 的注释里记录过三种布局的实测读数）。</li>
+ * </ul>
+ */
+class UnifiedHttpConfigTest {
+
+    private HttpServer server;
+
+    @AfterEach
+    void stopServer() {
+        if (server != null) {
+            server.stop(0);
+        }
+    }
+
+    @Test
+    void callbackRestTemplate_returns302WithoutFollowingRedirect() throws IOException {
+        AtomicInteger targetHits = new AtomicInteger();
+        server = HttpServer.create(new InetSocketAddress("127.0.0.1", 0), 0);
+        server.createContext("/start", exchange -> {
+            exchange.getResponseHeaders().add("Location", "/target");
+            exchange.sendResponseHeaders(302, -1);
+            exchange.close();
+        });
+        server.createContext("/target", exchange -> {
+            targetHits.incrementAndGet();
+            exchange.sendResponseHeaders(200, -1);
+            exchange.close();
+        });
+        server.start();
+
+        UnifiedPayProperties props = new UnifiedPayProperties();
+        props.setCallbackTimeoutMs(3000);
+        RestTemplate restTemplate = new UnifiedHttpConfig().callbackRestTemplate(props);
+
+        String base = "http://127.0.0.1:" + server.getAddress().getPort();
+        ResponseEntity<String> response = restTemplate.getForEntity(base + "/start", String.class);
+
+        assertEquals(HttpStatus.FOUND, response.getStatusCode(),
+                "回调客户端必须把 302 原样返回，不能自己转到下一跳");
+        assertEquals(0, targetHits.get(),
+                "Location 目标一次都不能被命中 —— 跟随 30x 等于把 SSRF 防护绕过去了");
+    }
+
+    @Test
+    void callbackRestTemplate_beanNameIsQualifierAndNotPrimary() throws Exception {
+        Method method = UnifiedHttpConfig.class.getMethod("callbackRestTemplate", UnifiedPayProperties.class);
+
+        assertNull(method.getAnnotation(Primary.class),
+                "不得给统一支付回调 bean 加 @Primary —— 会抢走 WebMvcConfig 的共享 bean");
+
+        Bean bean = method.getAnnotation(Bean.class);
+        assertNotNull(bean, "必须是 @Bean 方法");
+        assertEquals(1, bean.name().length, "@Bean 名只能是一个，实测: " + String.join(",", bean.name()));
+        assertEquals(UnifiedHttpConfig.CALLBACK_REST_TEMPLATE, bean.name()[0],
+                "服务用 @Qualifier 按名注入，bean 名必须逐字一致");
+    }
+
+    @Test
+    void callbackExecutor_isDaemonPoolWithShutdownDestroyMethod() throws Exception {
+        Method method = UnifiedHttpConfig.class.getMethod("unifiedCallbackExecutor", UnifiedPayProperties.class);
+        Bean bean = method.getAnnotation(Bean.class);
+        assertNotNull(bean, "必须是 @Bean 方法");
+        assertEquals(UnifiedHttpConfig.CALLBACK_EXECUTOR, bean.name()[0],
+                "服务用 @Qualifier 按名注入，bean 名必须逐字一致");
+        assertEquals("shutdown", bean.destroyMethod(), "容器关闭时必须收掉线程池");
+
+        UnifiedPayProperties props = new UnifiedPayProperties();
+        props.setCallbackPoolSize(2);
+        ExecutorService executor = new UnifiedHttpConfig().unifiedCallbackExecutor(props);
+        try {
+            AtomicReference<String> name = new AtomicReference<>();
+            AtomicReference<Boolean> daemon = new AtomicReference<>();
+            executor.submit(() -> {
+                name.set(Thread.currentThread().getName());
+                daemon.set(Thread.currentThread().isDaemon());
+            }).get();
+
+            assertTrue(name.get() != null && name.get().startsWith("unified-callback"),
+                    "线程名要可辨认，实测: " + name.get());
+            assertTrue(Boolean.TRUE.equals(daemon.get()),
+                    "回调线程必须是 daemon，否则推不动的地址会拖住 JVM 退出");
+        } finally {
+            executor.shutdownNow();
+        }
+    }
+}

java/src/test/java/com/payment/platform/module/payment/unified/scheduler/UnifiedCallbackRetrySchedulerTest.java

@@ -0,0 +1,49 @@
+package com.payment.platform.module.payment.unified.scheduler;
+
+import com.payment.platform.module.payment.unified.service.UnifiedCallbackService;
+import org.junit.jupiter.api.Test;
+import org.springframework.scheduling.annotation.Scheduled;
+
+import java.lang.reflect.Method;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.verify;
+
+/**
+ * 回调重试调度器：只做「定时 + 兜异常」两件事。
+ *
+ * <p>重试整批炸了不能掀翻调度线程 —— 上面还挂着其他 {@code @Scheduled} 任务
+ * （{@code spring.task.scheduling.pool.size} 默认是 1，异常逃逸会拖垮整个调度族）。
+ */
+class UnifiedCallbackRetrySchedulerTest {
+
+    @Test
+    void retry_delegatesToService() {
+        UnifiedCallbackService service = mock(UnifiedCallbackService.class);
+        new UnifiedCallbackRetryScheduler(service).retry();
+
+        verify(service).retryFailed();
+    }
+
+    @Test
+    void retry_serviceThrows_exceptionDoesNotEscape() {
+        UnifiedCallbackService service = mock(UnifiedCallbackService.class);
+        doThrow(new RuntimeException("db down")).when(service).retryFailed();
+
+        assertDoesNotThrow(() -> new UnifiedCallbackRetryScheduler(service).retry());
+    }
+
+    @Test
+    void retry_isScheduledWithConfigurableInterval() throws Exception {
+        Method method = UnifiedCallbackRetryScheduler.class.getMethod("retry");
+        Scheduled scheduled = method.getAnnotation(Scheduled.class);
+
+        assertNotNull(scheduled, "retry() 必须挂 @Scheduled");
+        assertEquals("${unified.pay.callback-retry-interval-ms:60000}", scheduled.fixedDelayString(),
+                "扫描节拍来自统一支付配置，默认 60 秒");
+    }
+}

java/src/test/java/com/payment/platform/module/payment/unified/service/UnifiedCallbackServiceTest.java

@@ -0,0 +1,714 @@
+package com.payment.platform.module.payment.unified.service;
+
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.conditions.update.LambdaUpdateWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.payment.platform.module.payment.apikey.entity.TenantApiKeyEntity;
+import com.payment.platform.module.payment.apikey.mapper.TenantApiKeyMapper;
+import com.payment.platform.module.payment.openapi.entity.OpenConfEntity;
+import com.payment.platform.module.payment.openapi.mapper.OpenConfMapper;
+import com.payment.platform.module.payment.unified.config.UnifiedPayProperties;
+import com.payment.platform.module.payment.unified.entity.UnifiedOrderEntity;
+import com.payment.platform.module.payment.unified.entity.UnifiedRefundEntity;
+import com.payment.platform.module.payment.unified.enums.UnifiedNotifyStatus;
+import com.payment.platform.module.payment.unified.mapper.UnifiedOrderMapper;
+import com.payment.platform.module.payment.unified.mapper.UnifiedRefundMapper;
+import org.apache.ibatis.builder.MapperBuilderAssistant;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.http.HttpEntity;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.MultiValueMap;
+import org.springframework.web.client.RestTemplate;
+
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.junit.jupiter.api.Assertions.fail;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+
+/**
+ * 下游回调的报文契约、取址序、SSRF 防线与重试形状（spec §7.6，R21/R34/R35）。
+ *
+ * <p>【为什么这些是主断言】
+ * <ul>
+ *   <li>**取址序**写反时其余测试全绿，只有地址断言会红 —— 而它决定钱事件推到哪个
+ *       商户服务器。</li>
+ *   <li>**{@code content.upstream} 白名单**：上游通知混有我们上送的 {@code channelExtra}
+ *       （付款码 / openid / buyerUserId）。整体转发 = 把付款凭据交给下游，是本模块的硬约束。
+ *       断言用**真实表单**解析，不读源码文本。</li>
+ *   <li>**先 FAILED 再异步推送**：进程在推送中途崩溃时，行必须已经是 FAILED + count+1，
+ *       否则重试扫描永远捞不到它（扫描入口就是 {@code notify_status='FAILED'}）。</li>
+ *   <li>**SSRF**：恶意地址必须**一次出站都不发**，但仍记一次失败尝试（不能静默卡在 PENDING）。</li>
+ * </ul>
+ */
+@ExtendWith(MockitoExtension.class)
+class UnifiedCallbackServiceTest {
+
+    private static final Long TENANT = 7L;
+    private static final Long ORDER_DB_ID = 5L;
+
+    @Mock private UnifiedOrderMapper orderMapper;
+    @Mock private UnifiedRefundMapper refundMapper;
+    @Mock private TenantApiKeyMapper apiKeyMapper;
+    @Mock private OpenConfMapper confMapper;
+    @Mock private RestTemplate restTemplate;
+
+    private UnifiedPayProperties props;
+    private UnifiedCallbackService service;
+
+    @BeforeEach
+    void setUp() {
+        props = new UnifiedPayProperties();
+        props.setCallbackMaxRetries(3);
+        // Executor 传 Runnable::run：推送在调用线程内同步完成，断言无需等待。
+        service = new UnifiedCallbackService(orderMapper, refundMapper, apiKeyMapper, confMapper,
+                restTemplate, Runnable::run, props);
+        // 无 Spring 上下文时让 LambdaWrapper 能解析列名（同 HarCallbackServiceTest 的引导）
+        initTable(UnifiedOrderEntity.class);
+        initTable(UnifiedRefundEntity.class);
+        initTable(TenantApiKeyEntity.class);
+        initTable(OpenConfEntity.class);
+    }
+
+    private static void initTable(Class<?> entityType) {
+        MybatisConfiguration configuration = new MybatisConfiguration();
+        TableInfoHelper.initTableInfo(new MapperBuilderAssistant(configuration, ""), entityType);
+    }
+
+    // ───────────────────────────── 夹具 ─────────────────────────────
+
+    private static UnifiedOrderEntity order(String orderNo, String tradeStatus) {
+        UnifiedOrderEntity o = new UnifiedOrderEntity();
+        o.setId(ORDER_DB_ID);
+        o.setTenantId(TENANT);
+        o.setOrderNo(orderNo);
+        o.setOutTradeNo("C1");
+        o.setMchNo("M1");
+        o.setWayCode("QR_CASHIER");
+        o.setAmount(100L);
+        o.setRefundedAmount(20L);
+        o.setCurrency("cny");
+        o.setTradeStatus(tradeStatus);
+        return o;
+    }
+
+    private static UnifiedRefundEntity refund() {
+        UnifiedRefundEntity r = new UnifiedRefundEntity();
+        r.setId(9L);
+        r.setTenantId(TENANT);
+        r.setOrderId(ORDER_DB_ID);
+        r.setOrderNo("UP1");
+        r.setRefundNo("UR1");
+        r.setOutRefundNo("CR1");
+        r.setAmount(40L);
+        r.setCurrency("cny");
+        r.setState("2");
+        return r;
+    }
+
+    private void stubPushOk() {
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenReturn(new ResponseEntity<>("ok", HttpStatus.OK));
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private List<LambdaUpdateWrapper<?>> captureOrderUpdates(int expectedCalls) {
+        ArgumentCaptor<LambdaUpdateWrapper> captor = ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
+        verify(orderMapper, times(expectedCalls)).update(isNull(), captor.capture());
+        List<LambdaUpdateWrapper<?>> captured = new ArrayList<>();
+        for (LambdaUpdateWrapper wrapper : captor.getAllValues()) {
+            captured.add(wrapper);
+        }
+        return captured;
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private List<LambdaUpdateWrapper<?>> captureRefundUpdates(int expectedCalls) {
+        ArgumentCaptor<LambdaUpdateWrapper> captor = ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
+        verify(refundMapper, times(expectedCalls)).update(isNull(), captor.capture());
+        List<LambdaUpdateWrapper<?>> captured = new ArrayList<>();
+        for (LambdaUpdateWrapper wrapper : captor.getAllValues()) {
+            captured.add(wrapper);
+        }
+        return captured;
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private static HttpEntity<?> capturePush(RestTemplate restTemplate, String url) {
+        ArgumentCaptor<HttpEntity> captor = ArgumentCaptor.forClass(HttpEntity.class);
+        verify(restTemplate).postForEntity(eq(url), captor.capture(), eq(String.class));
+        return captor.getValue();
+    }
+
+    @SuppressWarnings("unchecked")
+    private static MultiValueMap<String, String> formOf(HttpEntity<?> entity) {
+        return (MultiValueMap<String, String>) entity.getBody();
+    }
+
+    private static JsonNode contentOf(MultiValueMap<String, String> form) throws Exception {
+        return new ObjectMapper().readTree(form.getFirst("content"));
+    }
+
+    // ───────────────────────── 确定性 notify_id ─────────────────────────
+
+    @Test
+    void notifyIds_areDeterministic() {
+        assertEquals("uo_UP1_2", UnifiedCallbackService.orderNotifyId("UP1", "2"));
+        assertEquals("ur_UR1_2", UnifiedCallbackService.refundNotifyId("UR1", "2"));
+    }
+
+    // ───────────────────────────── 订单成功路径 ─────────────────────────────
+
+    @Test
+    void notifyOrder_success_sendsOrderEventAndWhitelistsUpstream() throws Exception {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://merchant.example.com/cb");
+        order.setPayTime(OffsetDateTime.parse("2026-10-20T04:00:05Z"));
+        order.setUpstreamPayOrderId("P1");
+        order.setUpstreamChannelOrderNo("CH1");
+        when(refundMapper.sumOccupiedAmount(ORDER_DB_ID)).thenReturn(30L);
+        stubPushOk();
+
+        JsonNode upstream = new ObjectMapper().readTree("""
+                {"payOrderId":"P1","mchOrderNo":"UP1","state":"2","orderState":"2","amount":100,
+                 "successTime":1760000000000,"channelExtra":{"auth_code":"284211","openid":"o-1"},
+                 "openid":"o-1","sign":"DEADBEEF"}
+                """);
+
+        service.notifyOrder(order, upstream);
+
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/cb"));
+        assertEquals("uo_UP1_2", form.getFirst("notify_id"), "同一事件重试必须同一个 id，接收方可去重");
+        assertTrue(form.getFirst("timestamp") != null && form.getFirst("timestamp").matches("\\d{13}"),
+                "timestamp 必须是 13 位毫秒值，实测: " + form.getFirst("timestamp"));
+
+        JsonNode content = contentOf(form);
+        assertEquals("ORDER", content.path("event").asText());
+
+        JsonNode orderNode = content.path("order");
+        assertEquals("UP1", orderNode.path("order_no").asText());
+        assertEquals("C1", orderNode.path("out_trade_no").asText());
+        assertEquals("M1", orderNode.path("mch_no").asText());
+        assertEquals("QR_CASHIER", orderNode.path("way_code").asText());
+        assertEquals(100, orderNode.path("amount").asInt());
+        assertEquals(20, orderNode.path("refunded_amount").asInt());
+        assertEquals(70, orderNode.path("refundable_amount").asInt(),
+                "refundable = amount - sumOccupied(30)；occupied 已含 state=2，不得再减 refunded_amount");
+        assertEquals("2", orderNode.path("trade_status").asText());
+        assertEquals("P1", orderNode.path("upstream_pay_order_id").asText());
+        assertEquals("CH1", orderNode.path("upstream_channel_order_no").asText());
+        assertEquals("2026-10-20T12:00:05+08:00", orderNode.path("pay_time").asText(),
+                "时间必须转成 Asia/Shanghai 的 OffsetDateTime 字符串");
+
+        JsonNode upstreamNode = content.path("upstream");
+        assertEquals("2", upstreamNode.path("state").asText());
+        assertEquals("P1", upstreamNode.path("payOrderId").asText(),
+                "白名单键名逐字；若用了全局 SNAKE_CASE mapper，这里会变成 pay_order_id");
+        assertEquals(1760000000000L, upstreamNode.path("successTime").asLong());
+        assertFalse(upstreamNode.has("channelExtra"), "channelExtra 绝不转发下游");
+        assertFalse(upstreamNode.has("openid"), "openid 绝不转发下游");
+        assertFalse(upstreamNode.has("sign"), "签名不转发下游");
+        assertFalse(upstreamNode.has("pay_order_id"), "白名单键名不得被改名");
+
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(2);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"),
+                "先写 FAILED 再推（进程崩溃时仍可被重试扫到）。实测 SET: " + updates.get(0).getSqlSet());
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"),
+                "首推也要计数（含首次）。实测 SET: " + updates.get(0).getSqlSet());
+        assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"),
+                "2xx 后写 SUCCESS。实测 SET: " + updates.get(1).getSqlSet());
+        assertFalse(updates.get(1).getSqlSet().contains("notify_count"),
+                "成功回写不得再加一次计数（否则一次推送烧掉两次上限）。实测 SET: " + updates.get(1).getSqlSet());
+    }
+
+    @Test
+    void notifyOrder_non2xx_leavesFailedAttempt() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://merchant.example.com/cb");
+        when(restTemplate.postForEntity(anyString(), any(HttpEntity.class), eq(String.class)))
+                .thenReturn(new ResponseEntity<>("boom", HttpStatus.INTERNAL_SERVER_ERROR));
+
+        service.notifyOrder(order, null);
+
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(1);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+    }
+
+    @Test
+    void notifyOrder_noAddress_recordsFailedAttemptWithoutOutbound() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl(null);
+        order.setApiKey(null);
+        when(confMapper.selectList(any())).thenReturn(List.of());
+
+        service.notifyOrder(order, null);
+
+        verifyNoInteractions(restTemplate);
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(1);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"),
+                "没地址也要落 FAILED（商户补配地址后调度器还能捞起）。实测 SET: " + updates.get(0).getSqlSet());
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+        assertNotNull(setValueOf(updates.get(0), "last_notify_time"), "排序键不能为空");
+        assertTrue(whereBoundValues(updates.get(0)).contains("UP1"),
+                "WHERE 必须钉住这一单。实测: " + updates.get(0).getSqlSegment());
+    }
+
+    // ───────────────────────────── 订单取址序 ─────────────────────────────
+
+    @Test
+    void notifyOrder_snapshotCallbackUrlWinsWithoutKeyLookup() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://snapshot.example.com/cb");
+        order.setApiKey("K1");
+        stubPushOk();
+
+        service.notifyOrder(order, null);
+
+        verify(restTemplate).postForEntity(eq("https://snapshot.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verifyNoInteractions(apiKeyMapper, confMapper);
+    }
+
+    @Test
+    void notifyOrder_fallsBackToApiKeyReturnUrlBeforeOpenConf() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl(null);
+        order.setApiKey("K1");
+        TenantApiKeyEntity key = new TenantApiKeyEntity();
+        key.setApiKey("K1");
+        key.setReturnUrl("https://key.example.com/cb");
+        when(apiKeyMapper.selectOne(any())).thenReturn(key);
+        stubPushOk();
+
+        service.notifyOrder(order, null);
+
+        verify(restTemplate).postForEntity(eq("https://key.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verifyNoInteractions(confMapper);
+
+        ArgumentCaptor<LambdaQueryWrapper> query = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(apiKeyMapper).selectOne(query.capture());
+        assertTrue(query.getValue().getSqlSegment().contains("api_key = #{"),
+                "Key 必须按 api_key 精确查（该列全局唯一）。实测: " + query.getValue().getSqlSegment());
+        assertEquals(Set.of("K1"), new HashSet<>(query.getValue().getParamNameValuePairs().values()));
+    }
+
+    @Test
+    void notifyOrder_fallsBackToLatestOpenConfRow() {
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl(null);
+        order.setApiKey(null);
+
+        OpenConfEntity older = new OpenConfEntity();
+        older.setTenantId(TENANT);
+        older.setReturnUrl("https://conf-old.example.com/cb");
+        older.setUpdatedTime(OffsetDateTime.parse("2026-10-01T00:00:00+08:00"));
+        OpenConfEntity newer = new OpenConfEntity();
+        newer.setTenantId(TENANT);
+        newer.setReturnUrl("https://conf-new.example.com/cb");
+        newer.setUpdatedTime(OffsetDateTime.parse("2026-10-19T00:00:00+08:00"));
+        // 模拟数据库已按 updated_time DESC 排好序的两行
+        when(confMapper.selectList(any())).thenReturn(List.of(newer, older));
+        stubPushOk();
+
+        service.notifyOrder(order, null);
+
+        verify(restTemplate).postForEntity(eq("https://conf-new.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verify(confMapper, never()).selectOne(any());
+
+        ArgumentCaptor<LambdaQueryWrapper> query = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(confMapper).selectList(query.capture());
+        assertTrue(query.getValue().getSqlSegment().contains("ORDER BY updated_time DESC"),
+                "open_conf 兜底必须取 updated_time 最新一行，实测: " + query.getValue().getSqlSegment());
+        assertTrue(query.getValue().getParamNameValuePairs().values().contains(TENANT),
+                "必须按租户过滤，实测: " + query.getValue().getParamNameValuePairs());
+    }
+
+    // ───────────────────────────── SSRF ─────────────────────────────
+
+    @Test
+    void hostileCallbackUrls_areRejectedWithoutOutboundCall() {
+        List<String> hostile = List.of(
+                "http://localhost/x",
+                "http://127.0.0.1/x",
+                "http://127.8.9.10/x",
+                "http://[::1]/x",
+                "http://0.0.0.0/x",
+                "http://169.254.169.254/latest/meta-data",
+                "file:///etc/passwd",
+                "javascript:alert(1)",
+                "");
+        for (String url : hostile) {
+            assertFalse(UnifiedCallbackService.isAllowedCallbackUrl(url), "必须拒绝: " + url);
+        }
+        // 正对照：合法地址不得被误伤（私网回调是企业内网的合法用法，spec 明确不解析私网 IP）
+        assertTrue(UnifiedCallbackService.isAllowedCallbackUrl("https://example.com/cb"));
+        assertTrue(UnifiedCallbackService.isAllowedCallbackUrl("http://10.1.2.3/cb"));
+
+        when(confMapper.selectList(any())).thenReturn(List.of());
+        for (String url : hostile) {
+            UnifiedOrderEntity order = order("UP1", "2");
+            order.setCallbackUrl(url);
+            service.notifyOrder(order, null);
+        }
+
+        verifyNoInteractions(restTemplate);
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(hostile.size());
+        for (LambdaUpdateWrapper<?> update : updates) {
+            assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(update, "notify_status"),
+                    "恶意/空地址也必须记一次失败尝试，不能卡在 PENDING");
+            assertTrue(update.getSqlSet().contains("notify_count = notify_count + 1"));
+        }
+    }
+
+    // ───────────────────────────── 退款事件 ─────────────────────────────
+
+    @Test
+    void notifyRefund_sendsRefundEventWithOrderSnapshot() throws Exception {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl("https://merchant.example.com/refund");
+        refund.setSuccessTime(OffsetDateTime.parse("2026-10-20T04:00:05Z"));
+        refund.setUpstreamRefundOrderId("R1");
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setRefundedAmount(40L);
+        when(refundMapper.sumOccupiedAmount(ORDER_DB_ID)).thenReturn(40L);
+        stubPushOk();
+
+        JsonNode upstream = new ObjectMapper().readTree("""
+                {"refundOrderId":"R1","mchRefundNo":"UR1","state":"2","refundAmount":40,"payAmount":100,
+                 "channelExtra":{"auth_code":"284211"},"openid":"o-1"}
+                """);
+
+        service.notifyRefund(refund, order, upstream);
+
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/refund"));
+        assertEquals("ur_UR1_2", form.getFirst("notify_id"));
+
+        JsonNode content = contentOf(form);
+        assertEquals("REFUND", content.path("event").asText());
+
+        JsonNode refundNode = content.path("refund");
+        assertEquals("UR1", refundNode.path("refund_no").asText());
+        assertEquals("CR1", refundNode.path("out_refund_no").asText());
+        assertEquals("UP1", refundNode.path("order_no").asText());
+        assertEquals(40, refundNode.path("amount").asInt());
+        assertEquals("2", refundNode.path("state").asText());
+        assertEquals("R1", refundNode.path("upstream_refund_order_id").asText());
+        assertEquals("2026-10-20T12:00:05+08:00", refundNode.path("success_time").asText());
+
+        JsonNode orderNode = content.path("order");
+        assertEquals("2", orderNode.path("trade_status").asText(), "退款事件必须带订单当前快照");
+        assertEquals(40, orderNode.path("refunded_amount").asInt());
+        assertEquals(60, orderNode.path("refundable_amount").asInt(),
+                "refundable = 100 - occupied(40)，不得再减 refunded");
+
+        JsonNode upstreamNode = content.path("upstream");
+        assertEquals("R1", upstreamNode.path("refundOrderId").asText());
+        assertFalse(upstreamNode.has("channelExtra"));
+        assertFalse(upstreamNode.has("openid"));
+
+        List<LambdaUpdateWrapper<?>> updates = captureRefundUpdates(2);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+        assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"));
+        assertTrue(whereBoundValues(updates.get(0)).contains("UR1"),
+                "退款状态按 refund_no 更新。实测: " + updates.get(0).getSqlSegment());
+    }
+
+    @Test
+    void notifyRefund_refundCallbackUrlWinsOverOrderAndKey() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl("https://refund-snapshot.example.com/cb");
+        refund.setApiKey("KR");
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://order-snapshot.example.com/cb");
+        stubPushOk();
+
+        service.notifyRefund(refund, order, null);
+
+        verify(restTemplate).postForEntity(eq("https://refund-snapshot.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verifyNoInteractions(apiKeyMapper, confMapper);
+    }
+
+    @Test
+    void notifyRefund_fallsBackToOrderCallbackUrl() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl(null);
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setCallbackUrl("https://order-snapshot.example.com/cb");
+        stubPushOk();
+
+        service.notifyRefund(refund, order, null);
+
+        verify(restTemplate).postForEntity(eq("https://order-snapshot.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verifyNoInteractions(apiKeyMapper, confMapper);
+    }
+
+    @Test
+    void notifyRefund_usesRefundApiKeyBeforeOrderApiKey() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl(null);
+        refund.setApiKey("KR");
+        UnifiedOrderEntity order = order("UP1", "2");
+        TenantApiKeyEntity key = new TenantApiKeyEntity();
+        key.setReturnUrl("https://key.example.com/cb");
+        when(apiKeyMapper.selectOne(any())).thenReturn(key);
+        stubPushOk();
+
+        service.notifyRefund(refund, order, null);
+
+        verify(restTemplate).postForEntity(eq("https://key.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+
+        ArgumentCaptor<LambdaQueryWrapper> query = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(apiKeyMapper).selectOne(query.capture());
+        // 先渲染 SQL 段：MyBatis-Plus 的绑定值是在 SQL 段求值时写进参数表的
+        query.getValue().getSqlSegment();
+        assertEquals(Set.of("KR"), new HashSet<>(query.getValue().getParamNameValuePairs().values()),
+                "退款行的 apiKey 优先（发起退款时记录的那把）");
+    }
+
+    @Test
+    void notifyRefund_fallsBackToOrderApiKeyWhenRefundKeyBlank() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl(null);
+        refund.setApiKey(null);
+        UnifiedOrderEntity order = order("UP1", "2");
+        order.setApiKey("KO");
+        TenantApiKeyEntity key = new TenantApiKeyEntity();
+        key.setReturnUrl("https://order-key.example.com/cb");
+        when(apiKeyMapper.selectOne(any())).thenReturn(key);
+        stubPushOk();
+
+        service.notifyRefund(refund, order, null);
+
+        verify(restTemplate).postForEntity(eq("https://order-key.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+
+        ArgumentCaptor<LambdaQueryWrapper> query = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(apiKeyMapper).selectOne(query.capture());
+        query.getValue().getSqlSegment();
+        assertEquals(Set.of("KO"), new HashSet<>(query.getValue().getParamNameValuePairs().values()),
+                "退款行 apiKey 为空时用订单 apiKey");
+    }
+
+    @Test
+    void notifyRefund_fallsBackToOpenConf() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl(null);
+        refund.setApiKey(null);
+        UnifiedOrderEntity order = order("UP1", "2");
+        OpenConfEntity conf = new OpenConfEntity();
+        conf.setReturnUrl("https://conf.example.com/cb");
+        when(confMapper.selectList(any())).thenReturn(List.of(conf));
+        stubPushOk();
+
+        service.notifyRefund(refund, order, null);
+
+        verify(restTemplate).postForEntity(eq("https://conf.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        verify(confMapper, never()).selectOne(any());
+    }
+
+    @Test
+    void notifyRefund_noAddress_recordsFailedWithoutOutbound() {
+        UnifiedRefundEntity refund = refund();
+        refund.setCallbackUrl(null);
+        refund.setApiKey(null);
+        UnifiedOrderEntity order = order("UP1", "2");
+        when(confMapper.selectList(any())).thenReturn(List.of());
+
+        service.notifyRefund(refund, order, null);
+
+        verifyNoInteractions(restTemplate);
+        List<LambdaUpdateWrapper<?>> updates = captureRefundUpdates(1);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"));
+    }
+
+    // ───────────────────────────── 重试 ─────────────────────────────
+
+    @Test
+    void retryFailed_pinsScanQuery() {
+        props.setCallbackMaxRetries(4);   // ≠ 默认值 3：钉住上界来自配置而不是写死
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+
+        service.retryFailed();
+
+        assertScanQuery(orderMapper);
+        assertScanQuery(refundMapper);
+    }
+
+    @SuppressWarnings({"rawtypes", "unchecked"})
+    private void assertScanQuery(Object mapper) {
+        ArgumentCaptor<LambdaQueryWrapper> captor = ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        if (mapper == orderMapper) {
+            verify(orderMapper).selectList(captor.capture());
+        } else {
+            verify(refundMapper).selectList(captor.capture());
+        }
+        LambdaQueryWrapper<?> query = captor.getValue();
+        // 先渲染 SQL 段：MyBatis-Plus 的绑定值是在 SQL 段求值时写进参数表的
+        String sql = query.getSqlSegment();
+        assertEquals(Set.of(UnifiedNotifyStatus.FAILED, 4),
+                new HashSet<>(query.getParamNameValuePairs().values()),
+                "取值必须恰好是 {FAILED, 上限}，实测: " + query.getParamNameValuePairs());
+        assertTrue(sql.contains("notify_status = #{"), "必须按 notify_status 等值过滤，实测: " + sql);
+        assertTrue(sql.contains("notify_count < #{"), "达到上限的行不再扫，实测: " + sql);
+        assertTrue(sql.contains("ORDER BY last_notify_time ASC"), "饿最久的先推，实测: " + sql);
+        assertTrue(sql.contains("LIMIT 50"), "必须带 LIMIT，实测: " + sql);
+    }
+
+    @Test
+    void retryFailed_failedOrderCountTwo_pushesAndMarksSuccess() throws Exception {
+        UnifiedOrderEntity failed = order("UP1", "2");
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(2);
+        when(orderMapper.selectList(any())).thenReturn(List.of(failed));
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+        stubPushOk();
+
+        service.retryFailed();
+
+        MultiValueMap<String, String> form = formOf(capturePush(restTemplate, "https://merchant.example.com/cb"));
+        assertEquals("uo_UP1_2", form.getFirst("notify_id"), "重试保持同一个 notify_id");
+        JsonNode content = contentOf(form);
+        assertEquals(0, content.path("upstream").size(), "重试没有上游快照 ⇒ upstream 是空对象");
+
+        List<LambdaUpdateWrapper<?>> updates = captureOrderUpdates(2);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertTrue(updates.get(0).getSqlSet().contains("notify_count = notify_count + 1"),
+                "这次尝试也要计数（含首次共 3 次）");
+        assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"));
+    }
+
+    @Test
+    void retryFailed_afterSuccess_secondScanDoesNotPushAgain() {
+        UnifiedOrderEntity failed = order("UP1", "2");
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(2);
+        // 第一轮扫到它；成功后状态变 SUCCESS，第二轮扫描（模拟数据库）为空
+        when(orderMapper.selectList(any())).thenReturn(List.of(failed)).thenReturn(List.of());
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+        stubPushOk();
+
+        service.retryFailed();
+        service.retryFailed();
+
+        verify(restTemplate, times(1)).postForEntity(anyString(), any(HttpEntity.class), eq(String.class));
+    }
+
+    @Test
+    void retryFailed_refundWithoutOrder_skipsWithoutTouchingRow() {
+        UnifiedRefundEntity failed = refund();
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(1);
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+        when(refundMapper.selectList(any())).thenReturn(List.of(failed));
+        when(orderMapper.selectOne(any())).thenReturn(null);
+
+        service.retryFailed();
+
+        verifyNoInteractions(restTemplate);
+        verify(orderMapper, never()).update(isNull(), any());
+        verify(refundMapper, never()).update(isNull(), any());
+        verify(orderMapper).selectOne(any());
+    }
+
+    @Test
+    void retryFailed_refundWithOrder_pushesAndMarksSuccess() {
+        UnifiedRefundEntity failed = refund();
+        failed.setCallbackUrl("https://merchant.example.com/cb");
+        failed.setNotifyStatus(UnifiedNotifyStatus.FAILED);
+        failed.setNotifyCount(1);
+        UnifiedOrderEntity order = order("UP1", "2");
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+        when(refundMapper.selectList(any())).thenReturn(List.of(failed));
+        when(orderMapper.selectOne(any())).thenReturn(order);
+        stubPushOk();
+
+        service.retryFailed();
+
+        verify(restTemplate).postForEntity(eq("https://merchant.example.com/cb"),
+                any(HttpEntity.class), eq(String.class));
+        List<LambdaUpdateWrapper<?>> updates = captureRefundUpdates(2);
+        assertEquals(UnifiedNotifyStatus.FAILED, setValueOf(updates.get(0), "notify_status"));
+        assertEquals(UnifiedNotifyStatus.SUCCESS, setValueOf(updates.get(1), "notify_status"));
+    }
+
+    @Test
+    void retryFailed_scanThrows_doesNotPropagate() {
+        when(orderMapper.selectList(any())).thenThrow(new RuntimeException("db down"));
+        when(refundMapper.selectList(any())).thenReturn(List.of());
+
+        assertDoesNotThrow(() -> service.retryFailed());
+
+        verify(orderMapper).selectList(any());
+        verify(refundMapper).selectList(any());
+    }
+
+    // ───────────────────────────── 断言工具 ─────────────────────────────
+
+    /** 只取 SET 段里某一列的绑定值（SET 与 WHERE 共用参数表，混着找会假绿）。 */
+    private static Object setValueOf(LambdaUpdateWrapper<?> w, String column) {
+        String sqlSet = w.getSqlSet();
+        assertNotNull(sqlSet, "SET 段为空");
+        for (String item : sqlSet.split(",")) {
+            if (!item.startsWith(column + "=")) {
+                continue;
+            }
+            Matcher m = Pattern.compile("paramNameValuePairs\\.(\\w+)").matcher(item);
+            assertTrue(m.find(), "这一列没有绑定值，形状不认识: " + item);
+            return w.getParamNameValuePairs().get(m.group(1));
+        }
+        fail("SET 子句里没有这一列: " + column + "，实测 SET: " + sqlSet);
+        return null;
+    }
+
+    /** 只取 WHERE 段引用到的绑定值，不倒参数表全集。 */
+    private static Set<Object> whereBoundValues(LambdaUpdateWrapper<?> w) {
+        Matcher m = Pattern.compile("paramNameValuePairs\\.([A-Za-z0-9_]+)")
+                .matcher(w.getSqlSegment());
+        Set<Object> out = new HashSet<>();
+        while (m.find()) {
+            out.add(w.getParamNameValuePairs().get(m.group(1)));
+        }
+        return out;
+    }
+}