feat(统一支付): 查单/订单轮询/手动同步

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

@@ -0,0 +1,56 @@
+package com.payment.platform.module.payment.unified.scheduler;
+
+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.service.UnifiedOrderService;
+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;
+
+import java.util.List;
+
+/**
+ * 订单轮询（spec §7.2）：每 {@code unified.pay.order-poll-interval-ms}（默认 10 秒）
+ * 扫一批到期非终态单，逐单向上游查单。
+ *
+ * <p>两道异常边界都在本层：扫描失败放弃本轮；单笔失败只记日志、继续下一笔 ——
+ * 本任务跑在 {@code @Scheduled} 线程上，而 {@code spring.task.scheduling.pool.size}
+ * 默认是 1，逃逸异常会把整族定时任务一起饿死。
+ */
+@Slf4j
+@Component
+public class UnifiedOrderPollScheduler {
+
+    private final UnifiedOrderService orderService;
+    private final UnifiedPayProperties properties;
+
+    public UnifiedOrderPollScheduler(UnifiedOrderService orderService,
+                                     UnifiedPayProperties properties) {
+        this.orderService = orderService;
+        this.properties = properties;
+    }
+
+    @Scheduled(fixedDelayString = "${unified.pay.order-poll-interval-ms:10000}")
+    public void poll() {
+        List<UnifiedOrderEntity> orders;
+        try {
+            orders = orderService.findPendingForPoll(properties.getPollBatch());
+        } catch (Exception e) {
+            log.error("[统一支付订单轮询] 扫描失败: {}", UnifiedLogSanitizer.flatten(e.getMessage()), e);
+            return;
+        }
+        if (orders == null || orders.isEmpty()) {
+            return;
+        }
+        for (UnifiedOrderEntity order : orders) {
+            try {
+                orderService.syncFromUpstream(order);
+            } catch (Exception e) {
+                // 单笔失败不能拖垮整批
+                log.warn("[统一支付订单轮询] 单笔查询失败: order_no={}, error={}",
+                        order.getOrderNo(), UnifiedLogSanitizer.flatten(e.getMessage()));
+            }
+        }
+    }
+}

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

@@ -629,6 +629,86 @@ public class UnifiedOrderService {
         return null;
     }
 
+    // ────────────────────────────── 主动轮询与手动同步 ──────────────────────────────
+
+    /**
+     * 向上游查单并收敛本地状态（10 秒轮询调度器与后台「同步上游」共用）。
+     *
+     * <p>五个刻意保持的顺序/口径（spec §7.2）：
+     *
+     * <ol>
+     *   <li><b>先判停止条件</b>：已满 {@link UnifiedPollPolicy#MAX_QUERY_COUNT} 次或创建
+     *       超过 {@link UnifiedPollPolicy#MAX_AGE_MINUTES} 分钟 → 不打上游，直接把
+     *       {@code next_query_time} 置空。</li>
+     *   <li><b>配置按 {@code merchant_id} 取、不看启停</b>：{@link UnifiedMerchantService#requireById}
+     *       不校验 {@code status} —— 停用商户的历史单还要继续收敛。</li>
+     *   <li><b>{@code recordQueryAttempt} 在解析状态之前</b>：上游回了响应就消耗一次查询，
+     *       未知状态/金额不符同样计数，否则这类单会无限轮询。</li>
+     *   <li><b>金额核对只在 2/5</b>（与通知腿同一规则）：不符只记错误码、不推进状态。</li>
+     *   <li><b>状态真的迁移时 {@link #applyOrderState} 自己写 {@code next_query_time}，
+     *       这里不得覆盖</b>；没有迁移才按当前 {@code queryCount} 退避续排。</li>
+     * </ol>
+     *
+     * <p>上游异常直接抛出：调度器逐单兜住，手动同步由控制器转成失败应答。
+     */
+    public void syncFromUpstream(UnifiedOrderEntity order) {
+        if (order == null || order.getOrderNo() == null) {
+            return;
+        }
+        if (shouldStopPolling(order)) {
+            scheduleNextQuery(order, null);
+            return;
+        }
+
+        // 无认证上下文：按订单行上的商户主键取配置（不看 status）
+        UnifiedMerchantEntity cfg = merchantService.requireById(order.getMerchantId());
+        JsonNode data = client.queryOrder(cfg, order.getOrderNo());
+
+        // 上游回了响应（无论状态是否可识别）就算一次查询；必须先于任何解析/分支
+        recordQueryAttempt(order);
+
+        String state = UnifiedResponseParser.pickState(data);
+        if (UnifiedOrderStatus.canTransition(order.getTradeStatus(), state)) {
+            boolean needsAmountCheck = UnifiedOrderStatus.SUCCESS.equals(state)
+                    || UnifiedOrderStatus.REFUNDED.equals(state);
+            if (needsAmountCheck && !amountMatches(order, data)) {
+                // 不推进状态：留在轮询主路径里，等上游/人工把金额修正一致
+                markAmountMismatch(order, upstreamAmountDetail(order, data));
+            } else if (applyOrderState(order, state, UnifiedUpstreamSnapshot.of(data), null, null)) {
+                // applyOrderState 已按新状态写好 next_query_time（非终态 = 退避，终态 = null）。
+                // 唯一例外：最后一次查询恰好把订单推进到非终态 —— 策略已判定停止，必须置空。
+                if (UnifiedOrderStatus.PENDING.contains(order.getTradeStatus())
+                        && shouldStopPolling(order)) {
+                    scheduleNextQuery(order, null);
+                }
+                return;
+            }
+        }
+
+        if (shouldStopPolling(order)) {
+            scheduleNextQuery(order, null);
+            return;
+        }
+        if (UnifiedOrderStatus.PENDING.contains(order.getTradeStatus())) {
+            int queryCount = order.getQueryCount() == null ? 0 : order.getQueryCount();
+            scheduleNextQuery(order, OffsetDateTime.now()
+                    .plusSeconds(UnifiedPollPolicy.nextDelaySeconds(queryCount)));
+        }
+    }
+
+    /**
+     * 轮询停止条件（spec §7.2）：满 60 次或创建超过 30 分钟。
+     *
+     * <p>{@code created_time} 缺失时按「刚创建」处理 —— 不能把数据损坏当成「很老」
+     * 而让这一单永久停轮。
+     */
+    private static boolean shouldStopPolling(UnifiedOrderEntity order) {
+        int queryCount = order.getQueryCount() == null ? 0 : order.getQueryCount();
+        OffsetDateTime createdAt = order.getCreatedTime() == null
+                ? OffsetDateTime.now() : order.getCreatedTime();
+        return UnifiedPollPolicy.shouldStop(queryCount, createdAt, OffsetDateTime.now());
+    }
+
     /**
      * 轮询扫描（Task 11 调度器消费）：非终态、到期、未超 30 分钟、未满 60 次，
      * 按到期时间升序，最多 {@code limit} 条。

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

@@ -0,0 +1,170 @@
+package com.payment.platform.module.payment.unified.scheduler;
+
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
+import com.payment.platform.module.payment.unified.client.UnifiedApiException;
+import com.payment.platform.module.payment.unified.client.UnifiedClient;
+import com.payment.platform.module.payment.unified.config.UnifiedPayProperties;
+import com.payment.platform.module.payment.unified.entity.UnifiedMerchantEntity;
+import com.payment.platform.module.payment.unified.entity.UnifiedOrderEntity;
+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.service.UnifiedCallbackService;
+import com.payment.platform.module.payment.unified.service.UnifiedMerchantService;
+import com.payment.platform.module.payment.unified.service.UnifiedOrderService;
+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.scheduling.annotation.Scheduled;
+
+import java.lang.reflect.Method;
+import java.time.OffsetDateTime;
+import java.util.List;
+
+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.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * 订单轮询调度器（spec §7.2）：扫描到期单、逐单查上游、单笔失败不拖垮整批。
+ *
+ * <p>这里用**真实** {@link UnifiedOrderService}（mapper/client 等仍是 mock）而不是 mock
+ * service：要证明的正是「上游异常从 {@code syncFromUpstream} 逃逸后，调度器仍然继续处理
+ * 下一单」这条跨层契约 —— mock 掉 service 就把被测行为本身 mock 掉了。
+ *
+ * <p>调度线程上还挂着其他 {@code @Scheduled} 任务（{@code spring.task.scheduling.pool.size}
+ * 默认是 1），任何逃逸异常都会把整族定时任务饿死。
+ */
+@ExtendWith(MockitoExtension.class)
+class UnifiedOrderPollSchedulerTest {
+
+    private static final ObjectMapper JSON = new ObjectMapper();
+    private static final Long TENANT = 2L;
+
+    @Mock private UnifiedOrderMapper orderMapper;
+    @Mock private UnifiedRefundMapper refundMapper;
+    @Mock private UnifiedClient client;
+    @Mock private UnifiedMerchantService merchantService;
+    @Mock private UnifiedCallbackService callbackService;
+    @Mock private OrderEventPublisher publisher;
+
+    private UnifiedPayProperties props;
+    private UnifiedOrderService orderService;
+    private UnifiedOrderPollScheduler scheduler;
+
+    @BeforeEach
+    void setUp() {
+        props = new UnifiedPayProperties();
+        orderService = new UnifiedOrderService(orderMapper, refundMapper, client,
+                merchantService, callbackService, publisher, props);
+        scheduler = new UnifiedOrderPollScheduler(orderService, props);
+        MybatisConfiguration configuration = new MybatisConfiguration();
+        MapperBuilderAssistant assistant = new MapperBuilderAssistant(configuration, "");
+        TableInfoHelper.initTableInfo(assistant, UnifiedOrderEntity.class);
+    }
+
+    private static JsonNode json(String text) {
+        try {
+            return JSON.readTree(text);
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("测试 JSON 写错了", e);
+        }
+    }
+
+    private static UnifiedMerchantEntity merchant() {
+        UnifiedMerchantEntity cfg = new UnifiedMerchantEntity();
+        cfg.setId(11L);
+        cfg.setTenantId(TENANT);
+        cfg.setMchNo("M001");
+        cfg.setAppId("app-1");
+        cfg.setAppSecret("SECRET");
+        cfg.setStatus(UnifiedMerchantService.STATUS_ENABLED);
+        return cfg;
+    }
+
+    private static UnifiedOrderEntity order(String orderNo) {
+        UnifiedOrderEntity o = new UnifiedOrderEntity();
+        o.setId(5L);
+        o.setTenantId(TENANT);
+        o.setMerchantId(11L);
+        o.setMchNo("M001");
+        o.setOrderNo(orderNo);
+        o.setAmount(100L);
+        o.setCurrency("cny");
+        o.setTradeStatus("0");
+        o.setQueryCount(0);
+        o.setCreatedTime(OffsetDateTime.now());
+        return o;
+    }
+
+    @Test
+    void poll_keepsProcessingAfterOneOrderFails() {
+        UnifiedOrderEntity first = order("UP1");
+        UnifiedOrderEntity second = order("UP2");
+        when(orderMapper.selectList(any())).thenReturn(List.of(first, second));
+
+        UnifiedMerchantEntity cfg = merchant();
+        when(merchantService.requireById(11L)).thenReturn(cfg);
+        // 第一单上游炸了：异常必须被逐单 try/catch 吃掉，第二单照常查
+        when(client.queryOrder(same(cfg), eq("UP1"))).thenThrow(
+                new UnifiedApiException(UnifiedApiException.CODE_TRANSPORT, "connect timeout"));
+        when(client.queryOrder(same(cfg), eq("UP2"))).thenReturn(json("{\"state\":0}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        scheduler.poll();
+
+        verify(client).queryOrder(same(cfg), eq("UP1"));
+        verify(client).queryOrder(same(cfg), eq("UP2"));
+        assertEquals(0, first.getQueryCount(), "上游失败不算查询次数");
+        assertEquals(1, second.getQueryCount(), "第二单必须仍被处理");
+        assertNotNull(second.getNextQueryTime(), "第二单应被安排下一次查询");
+    }
+
+    @Test
+    void poll_scansWithConfiguredBatch() {
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+
+        scheduler.poll();
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<LambdaQueryWrapper<UnifiedOrderEntity>> w =
+                ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(orderMapper).selectList(w.capture());
+        assertTrue(w.getValue().getSqlSegment().contains("LIMIT " + props.getPollBatch()),
+                "扫描批量必须来自 unified.pay.poll-batch，实测: "
+                        + w.getValue().getSqlSegment());
+    }
+
+    @Test
+    void poll_scanFailureDoesNotEscape() {
+        when(orderMapper.selectList(any())).thenThrow(new RuntimeException("db down"));
+
+        assertDoesNotThrow(() -> scheduler.poll());
+    }
+
+    @Test
+    void poll_isScheduledWithConfigurableInterval() throws Exception {
+        Method method = UnifiedOrderPollScheduler.class.getMethod("poll");
+        Scheduled scheduled = method.getAnnotation(Scheduled.class);
+
+        assertNotNull(scheduled, "poll() 必须挂 @Scheduled");
+        assertEquals("${unified.pay.order-poll-interval-ms:10000}", scheduled.fixedDelayString(),
+                "订单轮询节拍来自统一支付配置，默认 10 秒");
+    }
+}

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

@@ -0,0 +1,418 @@
+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.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
+import com.payment.platform.module.payment.unified.client.UnifiedClient;
+import com.payment.platform.module.payment.unified.config.UnifiedPayProperties;
+import com.payment.platform.module.payment.unified.entity.UnifiedMerchantEntity;
+import com.payment.platform.module.payment.unified.entity.UnifiedOrderEntity;
+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.UnifiedPollPolicy;
+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 java.time.Duration;
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Set;
+import java.util.TreeSet;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+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;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.isNull;
+import static org.mockito.ArgumentMatchers.same;
+import static org.mockito.Mockito.atLeastOnce;
+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;
+
+/**
+ * 查单同步与轮询扫描（spec §7.2）。
+ *
+ * <p>本类盯的是「不会报错但会错钱 / 会让轮询静默停摆」的地方：
+ * <ol>
+ *   <li>按 {@code merchant_id} 取配置（不看启停）—— 停用商户的历史单还要继续收敛，
+ *       用 {@code requireByMchNo} 或带启停校验会让存量单永远停在非终态；</li>
+ *   <li>未知/缺失状态不得落库（R11）—— 一个 {@code state=99} 就能把订单推到不存在的状态；</li>
+ *   <li>上游金额不符仍标成功 —— 少付的单被标成已支付；</li>
+ *   <li>状态未迁移时把 {@code next_query_time} 清空或重复计数（R38）——
+ *       清空 = 该单永远不再被 {@code findPendingForPoll} 捞到；</li>
+ *   <li>超过 30 分钟 / 60 次仍打上游 —— 违反轮询停止条件。</li>
+ * </ol>
+ */
+@ExtendWith(MockitoExtension.class)
+class UnifiedOrderSyncTest {
+
+    private static final ObjectMapper JSON = new ObjectMapper();
+    private static final Long TENANT = 2L;
+
+    @Mock private UnifiedOrderMapper orderMapper;
+    @Mock private UnifiedRefundMapper refundMapper;
+    @Mock private UnifiedClient client;
+    @Mock private UnifiedMerchantService merchantService;
+    @Mock private UnifiedCallbackService callbackService;
+    @Mock private OrderEventPublisher publisher;
+
+    private UnifiedOrderService service;
+
+    @BeforeEach
+    void setUp() {
+        service = new UnifiedOrderService(orderMapper, refundMapper, client,
+                merchantService, callbackService, publisher, new UnifiedPayProperties());
+        // 初始化 MyBatis-Plus lambda 元数据缓存，使 getSqlSegment()/getSqlSet() 在无 Spring 上下文的单测中工作
+        MybatisConfiguration configuration = new MybatisConfiguration();
+        MapperBuilderAssistant assistant = new MapperBuilderAssistant(configuration, "");
+        TableInfoHelper.initTableInfo(assistant, UnifiedOrderEntity.class);
+    }
+
+    // ────────────────────────────── 工具 ──────────────────────────────
+
+    private static JsonNode json(String text) {
+        try {
+            return JSON.readTree(text);
+        } catch (JsonProcessingException e) {
+            throw new IllegalStateException("测试 JSON 写错了", e);
+        }
+    }
+
+    private static UnifiedMerchantEntity merchant() {
+        UnifiedMerchantEntity cfg = new UnifiedMerchantEntity();
+        cfg.setId(11L);
+        cfg.setTenantId(TENANT);
+        cfg.setMchNo("M001");
+        cfg.setAppId("app-1");
+        cfg.setAppSecret("SECRET");
+        cfg.setStatus(UnifiedMerchantService.STATUS_ENABLED);
+        return cfg;
+    }
+
+    private static UnifiedOrderEntity order(String tradeStatus) {
+        UnifiedOrderEntity o = new UnifiedOrderEntity();
+        o.setId(5L);
+        o.setTenantId(TENANT);
+        o.setMerchantId(11L);
+        o.setMchNo("M001");
+        o.setAppId("app-1");
+        o.setOrderNo("UP5");
+        o.setOutTradeNo("T001");
+        o.setApiKey("KEY-1");
+        o.setSource("OPENAPI");
+        o.setWayCode("QR_CASHIER");
+        o.setAmount(100L);
+        o.setRefundedAmount(0L);
+        o.setCurrency("cny");
+        o.setSubject("测试商品");
+        o.setTradeStatus(tradeStatus);
+        o.setQueryCount(0);
+        o.setCreatedTime(OffsetDateTime.now());
+        return o;
+    }
+
+    /** 断言安排在下一次查询的时间 ≈ now + {@code expectedSeconds}（服务内部取的 now 略早于此刻）。 */
+    private static void assertScheduledIn(UnifiedOrderEntity order, long expectedSeconds) {
+        assertNotNull(order.getNextQueryTime(), "非终态订单必须安排下一次查询");
+        long actual = Duration.between(OffsetDateTime.now(), order.getNextQueryTime()).toSeconds();
+        assertTrue(actual >= expectedSeconds - 1 && actual <= expectedSeconds + 2,
+                "期望约 " + expectedSeconds + "s 后查询，实测 " + actual + "s");
+    }
+
+    @SuppressWarnings("unchecked")
+    private static ArgumentCaptor<LambdaQueryWrapper<UnifiedOrderEntity>> queryCaptor() {
+        return ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+    }
+
+    @SuppressWarnings("unchecked")
+    private static ArgumentCaptor<LambdaUpdateWrapper<UnifiedOrderEntity>> updateCaptor() {
+        return ArgumentCaptor.forClass(LambdaUpdateWrapper.class);
+    }
+
+    /** 只取 {@code trade_status IN (...)} 谓词引用到的绑定值（同款口径见 UnifiedOrderServiceTest）。 */
+    private static Set<String> inValues(LambdaQueryWrapper<UnifiedOrderEntity> w) {
+        Matcher inPredicate = Pattern.compile("trade_status IN \\(([^)]*)\\)")
+                .matcher(w.getSqlSegment());
+        assertTrue(inPredicate.find(), "WHERE 里应有 trade_status IN (...) 谓词，实测: "
+                + w.getSqlSegment());
+        Matcher placeholder = Pattern.compile("paramNameValuePairs\\.([A-Za-z0-9_]+)")
+                .matcher(inPredicate.group(1));
+        Set<String> values = new TreeSet<>();
+        while (placeholder.find()) {
+            values.add(String.valueOf(w.getParamNameValuePairs().get(placeholder.group(1))));
+        }
+        return values;
+    }
+
+    // ────────────────────────────── 配置取用 ──────────────────────────────
+
+    @Test
+    void sync_usesMerchantIdConfigAndIgnoresStatus() {
+        UnifiedMerchantEntity cfg = merchant();
+        cfg.setStatus("1"); // 停用：历史单还要继续收敛，查单腿不看 status
+        when(merchantService.requireById(11L)).thenReturn(cfg);
+        when(client.queryOrder(same(cfg), eq("UP5"))).thenReturn(json("{\"state\":0}"));
+
+        UnifiedOrderEntity order = order("0");
+        service.syncFromUpstream(order);
+
+        verify(merchantService).requireById(11L);
+        verify(merchantService, never()).requireByMchNo(any());
+        verify(merchantService, never()).resolveForTenant(any(), any());
+        verify(merchantService, never()).requireEnabled(any());
+        verify(client).queryOrder(same(cfg), eq("UP5"));
+        assertEquals("0", order.getTradeStatus());
+    }
+
+    // ────────────────────────────── 状态收敛 ──────────────────────────────
+
+    @Test
+    void stateTwoWithMatchingAmount_appliesPublishesAndNotifies() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        // openid 属于 channelExtra 类敏感字段：只允许进白名单投影，绝不转发下游
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json(
+                "{\"payOrderId\":\"P1\",\"state\":2,\"amount\":100,\"currency\":\"cny\","
+                        + "\"successTime\":1760000000000,\"openid\":\"SECRET-OPENID\"}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("0");
+        service.syncFromUpstream(order);
+
+        assertEquals("2", order.getTradeStatus());
+        assertNotNull(order.getPayTime());
+        assertNull(order.getNextQueryTime(), "终态不再轮询");
+        assertEquals(1, order.getQueryCount(), "上游响应算一次查询");
+
+        verify(publisher).changed(OrderEventPublisher.CHANNEL_UNIFIED_ORDER, TENANT);
+        ArgumentCaptor<JsonNode> upstream = ArgumentCaptor.forClass(JsonNode.class);
+        verify(callbackService).notifyOrder(same(order), upstream.capture());
+        assertEquals("P1", upstream.getValue().get("payOrderId").asText());
+        assertNull(upstream.getValue().get("openid"),
+                "敏感字段不得进入下游回调的 upstream 投影");
+    }
+
+    @Test
+    void amountMismatch_doesNotAdvance() {
+        JsonNode[] responses = {
+                json("{\"state\":2,\"amount\":99,\"currency\":\"cny\"}"),
+                json("{\"state\":2,\"amount\":0,\"currency\":\"cny\"}"),
+                json("{\"state\":2,\"amount\":-1,\"currency\":\"cny\"}"),
+                json("{\"state\":2,\"currency\":\"cny\"}")
+        };
+        for (JsonNode data : responses) {
+            when(merchantService.requireById(11L)).thenReturn(merchant());
+            when(client.queryOrder(any(), eq("UP5"))).thenReturn(data);
+            when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+            UnifiedOrderEntity order = order("0");
+            service.syncFromUpstream(order);
+
+            assertEquals("0", order.getTradeStatus(), "金额不符不得推进状态: " + data);
+            assertEquals("AMOUNT_MISMATCH", order.getErrCode(), "实测: " + data);
+            assertTrue(order.getErrMsg().contains("本地=100"), "实测: " + order.getErrMsg());
+            assertEquals(1, order.getQueryCount(), "金额不符也只计一次查询");
+            assertNotNull(order.getNextQueryTime(), "金额不符仍留在轮询主路径");
+            verifyNoInteractions(publisher, callbackService);
+        }
+    }
+
+    @Test
+    void unknownState_doesNotAdvance() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":99}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("0");
+        service.syncFromUpstream(order);
+
+        assertEquals("0", order.getTradeStatus(), "R11：未知状态不得落库");
+        assertEquals(1, order.getQueryCount(), "未知状态也算一次查询");
+        assertNotNull(order.getNextQueryTime(), "未知状态仍要安排下一次查询");
+        verifyNoInteractions(publisher, callbackService);
+    }
+
+    @Test
+    void terminalState_isNotOverwrittenAndNotRescheduled() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(
+                json("{\"state\":2,\"amount\":100,\"currency\":\"cny\"}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("3");
+        service.syncFromUpstream(order);
+
+        assertEquals("3", order.getTradeStatus(), "终态不得被上游旧状态覆盖");
+        assertNull(order.getNextQueryTime(), "终态不排下一次查询");
+        verifyNoInteractions(publisher, callbackService);
+    }
+
+    @Test
+    void conditionalUpdateMiss_doesNotPublish() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(
+                json("{\"state\":2,\"amount\":100,\"currency\":\"cny\"}"));
+        // 并发下另一条腿（通知/另一次查单）已经写过：条件 UPDATE 落空
+        when(orderMapper.update(isNull(), any())).thenReturn(0);
+
+        UnifiedOrderEntity order = order("0");
+        service.syncFromUpstream(order);
+
+        assertEquals("0", order.getTradeStatus(), "0 行时内存也不能提前改状态");
+        assertNotNull(order.getNextQueryTime(), "未迁移仍要保留轮询主路径");
+        verifyNoInteractions(publisher, callbackService);
+    }
+
+    @Test
+    void generatedToPaying_doesNotPublish() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":1}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("0");
+        service.syncFromUpstream(order);
+
+        assertEquals("1", order.getTradeStatus());
+        assertNotNull(order.getNextQueryTime(), "非终态继续轮询");
+        verifyNoInteractions(publisher, callbackService);
+    }
+
+    // ────────────────────────────── 计数与退避 ──────────────────────────────
+
+    @Test
+    void eachSyncCountsOneQuery_andBackoffFollowsPolicy() {
+        // 服务在任何解析之前先 recordQueryAttempt：查询结束后的退避按「第 n 次查询之后」取值。
+        // 因此初始 queryCount=0..6 对应 nextDelaySeconds(1..7)，即 10/20/30/60/120/300/300。
+        long[] expectedSeconds = {10, 20, 30, 60, 120, 300, 300};
+        for (int initial = 0; initial <= 6; initial++) {
+            when(merchantService.requireById(11L)).thenReturn(merchant());
+            when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":0}"));
+            when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+            UnifiedOrderEntity order = order("0");
+            order.setQueryCount(initial);
+            service.syncFromUpstream(order);
+
+            assertEquals(initial + 1, order.getQueryCount(),
+                    "每次同步必须 +1（初始 " + initial + "）");
+            assertScheduledIn(order, expectedSeconds[initial]);
+        }
+    }
+
+    @Test
+    void recordQueryAttempt_usesAtomicSqlIncrement() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":0}"));
+
+        service.syncFromUpstream(order("0"));
+
+        ArgumentCaptor<LambdaUpdateWrapper<UnifiedOrderEntity>> w = updateCaptor();
+        verify(orderMapper, atLeastOnce()).update(isNull(), w.capture());
+        String set = w.getAllValues().stream()
+                .map(LambdaUpdateWrapper::getSqlSet)
+                .filter(s -> s != null && s.replace(" ", "").contains("query_count=query_count+1"))
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("没有找到 SQL 侧自增的 query_count 写: "
+                        + w.getAllValues().stream().map(LambdaUpdateWrapper::getSqlSet).toList()));
+    }
+
+    @Test
+    void queryCountAtLimit_clearsNextQueryTimeAfterFinalQuery() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":0}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("0");
+        order.setQueryCount(UnifiedPollPolicy.MAX_QUERY_COUNT - 1);
+        service.syncFromUpstream(order);
+
+        assertEquals(UnifiedPollPolicy.MAX_QUERY_COUNT, order.getQueryCount());
+        assertNull(order.getNextQueryTime(), "达到 60 次上限后置空 next_query_time");
+    }
+
+    @Test
+    void finalQueryThatTransitionsToPaying_stillClearsNextQueryTime() {
+        when(merchantService.requireById(11L)).thenReturn(merchant());
+        when(client.queryOrder(any(), eq("UP5"))).thenReturn(json("{\"state\":1}"));
+        when(orderMapper.update(isNull(), any())).thenReturn(1);
+
+        UnifiedOrderEntity order = order("0");
+        order.setQueryCount(UnifiedPollPolicy.MAX_QUERY_COUNT - 1);
+        service.syncFromUpstream(order);
+
+        assertEquals("1", order.getTradeStatus());
+        assertEquals(UnifiedPollPolicy.MAX_QUERY_COUNT, order.getQueryCount());
+        assertNull(order.getNextQueryTime(),
+                "第 60 次查询即使推进到非终态也必须停止轮询（applyOrderState 的排期要被停止条件覆盖）");
+    }
+
+    @Test
+    void createdBeforeThirtyMinutes_skipsUpstreamAndClearsSchedule() {
+        UnifiedOrderEntity order = order("0");
+        order.setCreatedTime(OffsetDateTime.now().minusMinutes(UnifiedPollPolicy.MAX_AGE_MINUTES + 1));
+        order.setNextQueryTime(OffsetDateTime.now().minusSeconds(5));
+
+        service.syncFromUpstream(order);
+
+        assertNull(order.getNextQueryTime(), "超过 30 分钟置空 next_query_time");
+        verifyNoInteractions(client, merchantService);
+    }
+
+    // ────────────────────────────── 轮询扫描 ──────────────────────────────
+
+    @Test
+    void findPendingForPoll_filtersDuePendingRows() {
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+
+        service.findPendingForPoll(50);
+
+        ArgumentCaptor<LambdaQueryWrapper<UnifiedOrderEntity>> w = queryCaptor();
+        verify(orderMapper).selectList(w.capture());
+        String sql = w.getValue().getSqlSegment();
+        for (String fragment : List.of("trade_status IN", "next_query_time <=",
+                "created_time >", "query_count <", "ORDER BY next_query_time ASC", "LIMIT 50")) {
+            assertTrue(sql.contains(fragment), "轮询查询缺片段 " + fragment + "，实测: " + sql);
+        }
+        assertEquals(Set.of("0", "1"), inValues(w.getValue()),
+                "只扫非终态 GENERATED/PAYING，实测: " + sql);
+
+        List<Object> params = List.copyOf(w.getValue().getParamNameValuePairs().values());
+        assertTrue(params.contains(UnifiedPollPolicy.MAX_QUERY_COUNT),
+                "query_count 上限必须是 60，实测参数: " + params);
+        assertTrue(params.stream().anyMatch(v -> v instanceof OffsetDateTime t
+                        && Duration.between(t, OffsetDateTime.now().minusMinutes(
+                        UnifiedPollPolicy.MAX_AGE_MINUTES)).abs().getSeconds() < 5),
+                "created_time 下界必须是 now-30min，实测参数: " + params);
+    }
+
+    @Test
+    void findPendingForPoll_clampsLimit() {
+        when(orderMapper.selectList(any())).thenReturn(List.of());
+
+        service.findPendingForPoll(0);
+        service.findPendingForPoll(9999);
+
+        ArgumentCaptor<LambdaQueryWrapper<UnifiedOrderEntity>> w = queryCaptor();
+        verify(orderMapper, times(2)).selectList(w.capture());
+        String lower = w.getAllValues().get(0).getSqlSegment();
+        String upper = w.getAllValues().get(1).getSqlSegment();
+        assertTrue(lower.contains("LIMIT 1"), "limit=0 必须夹到 1，实测: " + lower);
+        assertTrue(upper.contains("LIMIT 500"), "limit 必须封顶 500，实测: " + upper);
+    }
+}