feat(订单实时): har 订单表八处写点发布信号（含回滚与回调状态）

java/src/main/java/com/payment/platform/module/payment/har/service/HarCallbackService.java

@@ -11,6 +11,7 @@ import com.payment.platform.module.payment.har.enums.HarNotifyStatus;
 import com.payment.platform.module.payment.har.mapper.HarOrderMapper;
 import com.payment.platform.module.payment.openapi.entity.OpenConfEntity;
 import com.payment.platform.module.payment.openapi.mapper.OpenConfMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.fasterxml.jackson.databind.SerializationFeature;
 import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
@@ -68,17 +69,21 @@ public class HarCallbackService {
     private final OpenConfMapper confMapper;
     private final HarOrderMapper orderMapper;
     private final RestTemplate restTemplate;
+    /** 订单信号发布（订单实时推送）—— 回调状态（notify_status）落库后发。 */
+    private final OrderEventPublisher publisher;
     private final ExecutorService callbackPool;
 
     public HarCallbackService(TenantApiKeyMapper apiKeyMapper,
                               OpenConfMapper confMapper,
                               HarOrderMapper orderMapper,
                               RestTemplate restTemplate,
-                              HarProperties harProperties) {
+                              HarProperties harProperties,
+                              OrderEventPublisher publisher) {
         this.apiKeyMapper = apiKeyMapper;
         this.confMapper = confMapper;
         this.orderMapper = orderMapper;
         this.restTemplate = restTemplate;
+        this.publisher = publisher;
         int poolSize = Math.max(1, harProperties.getCallbackPoolSize());
         this.callbackPool = Executors.newFixedThreadPool(poolSize, r -> {
             Thread t = new Thread(r, "har-callback");
@@ -148,6 +153,9 @@ public class HarCallbackService {
             }
             latest.setNotifyStatus(ok ? HarNotifyStatus.SUCCESS : HarNotifyStatus.FAILED);
             orderMapper.updateById(latest);
+            // 回调状态也是页面上的一格（HAR 详情弹窗），成功失败都要发。
+            // 本方法 private + 跑在回调线程池里（Task 7 唯一没被单测钉住的写点，Task 14 E2E 覆盖）。
+            publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, latest.getTenantId());
         } catch (Exception e) {
             log.warn("[汇安融回调] 回写 notify_status 失败: order_no={}, error={}", orderNo, e.getMessage());
         }

java/src/main/java/com/payment/platform/module/payment/har/service/HarNotifyService.java

@@ -6,6 +6,7 @@ import com.payment.platform.module.payment.har.config.HarProperties;
 import com.payment.platform.module.payment.har.entity.HarOrderEntity;
 import com.payment.platform.module.payment.har.enums.HarOrderStatus;
 import com.payment.platform.module.payment.har.mapper.HarOrderMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.stereotype.Service;
 
@@ -43,15 +44,19 @@ public class HarNotifyService {
     private final HarOrderService orderService;
     private final HarOrderMapper orderMapper;
     private final HarCallbackService callbackService;
+    /** 订单信号发布（订单实时推送）—— 通知真的改了行之后发。 */
+    private final OrderEventPublisher publisher;
 
     public HarNotifyService(HarProperties harProperties,
                             HarOrderService orderService,
                             HarOrderMapper orderMapper,
-                            HarCallbackService callbackService) {
+                            HarCallbackService callbackService,
+                            OrderEventPublisher publisher) {
         this.harProperties = harProperties;
         this.orderService = orderService;
         this.orderMapper = orderMapper;
         this.callbackService = callbackService;
+        this.publisher = publisher;
     }
 
     /**
@@ -151,6 +156,9 @@ public class HarNotifyService {
         }
         orderService.reschedulePoll(order);
         orderMapper.updateById(order);
+        // 通知真的改了行（状态、上游单号、平台单号……）才发 —— 这是他方付款成功后
+        // HAR 订单页自己翻新的那一帧；上面「状态未变化，幂等跳过」的分支到不了这里。
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
 
         callbackService.notifyEnterprise(order);
         log.info("[汇安融通知] 已处理: order_no={}, status={}", orderNo, newStatus);

java/src/main/java/com/payment/platform/module/payment/har/service/HarOrderService.java

@@ -27,6 +27,7 @@ import com.payment.platform.module.payment.har.entity.HarMerchantEntity;
 import com.payment.platform.module.payment.har.entity.HarOrderEntity;
 import com.payment.platform.module.payment.har.enums.HarOrderStatus;
 import com.payment.platform.module.payment.har.mapper.HarOrderMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 import org.springframework.security.core.context.SecurityContextHolder;
@@ -77,6 +78,18 @@ public class HarOrderService {
     private final HarOrderMapper orderMapper;
     private final HarMerchantService merchantService;
     private final HarClient harClient;
+    /**
+     * 订单信号发布（订单实时推送）。
+     *
+     * <p>【发布纪律】只在**真的写了行**之后发 —— 守卫没命中（{@link #doRefund} 抢占、
+     * {@link #rollbackRefundPreempt} 回退）与只推查询游标的分支（{@link #syncFromUpstream}
+     * 的查询失败腿）一个字节都没改，发了就是让订单页白拉一次。
+     *
+     * <p>【挂在哪几处】{@link #create} 的落库与补单号、{@link #closeRejectedOrder}、
+     * {@link #syncFromUpstream} 的成功尾、{@link #doRefund} 的抢占、{@link #rollbackRefundPreempt}
+     * 的回退 —— 清册以本文件里的发布调用点为准（按方法名找，别按行号）。
+     */
+    private final OrderEventPublisher publisher;
 
     public PageResult<HarOrderVO> list(int pageNo, int pageSize, HarOrderQueryDTO q) {
         LambdaQueryWrapper<HarOrderEntity> w = new LambdaQueryWrapper<>();
@@ -256,6 +269,8 @@ public class HarOrderService {
                     order.getOrderNo(), order.getTradeStatus());
             throw new BusinessException("订单当前状态不可退款: " + order.getTradeStatus());
         }
+        // 抢占真的改了行（守卫命中）才发 —— 没命中的分支上面已经抛了，页面什么都没有变
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
 
         HarRefundRequest req = new HarRefundRequest();
         // 优先用上游单号；拿不到时用我方单号走 third_order_no（同查询口径，
@@ -315,6 +330,9 @@ public class HarOrderService {
         order.setTradeStatus(HarOrderStatus.PAID);
         order.setNextQueryTime(null);
         order.setRefundReason(null);
+        // 回滚也要发：页面刚从「退款中」被点亮，现在要翻回「已支付」——
+        // 不发这一帧，退款页/订单页会一直停在那个从没发生过的退款中。
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
     }
 
     /**
@@ -415,6 +433,9 @@ public class HarOrderService {
         entity.setNextQueryTime(OffsetDateTime.now().plusSeconds(FIRST_POLL_DELAY_SECONDS));
         entity.setDescription(dto.getTitle());
         orderMapper.insert(entity);
+        // 新行出现 —— HAR 订单页的第一帧。tenantId 由 AutoFillMetaObjectHandler
+        // 在 insert 时填进这个对象，所以这里读得到（不是 null）。
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, entity.getTenantId());
 
         // 本地单已落库后才发起上游调用：上游超时/失败时本地仍有完整记录，
         // 且 next_query_time 已排入轮询，不会因网络问题凭空丢单。
@@ -438,6 +459,9 @@ public class HarOrderService {
             entity.setUpstreamOrderNo(detail.getOrderNo());
         }
         orderMapper.updateById(entity);
+        // 同一次请求里的第二帧：补上了 pay_url / 上游单号（前一帧这些格还是空的）。
+        // 前端对同一渠道的连续信号做了合并（见 Task 9 useLiveRefetch），500ms 内只拉一次。
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, entity.getTenantId());
         return HarOrderVO.of(entity);
     }
 
@@ -462,6 +486,8 @@ public class HarOrderService {
                 .set("updated_time", OffsetDateTime.now())
                 .eq("id", order.getId());
         orderMapper.update(null, w);
+        // 上游拒绝 ⇒ 本地从「等待支付」翻成「交易关闭」，页面上这一行必须跟着关掉
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
     }
 
     /**
@@ -518,6 +544,10 @@ public class HarOrderService {
                     .plusSeconds(nextDelaySeconds(order.getQueryCount(), false)));
         }
         orderMapper.updateById(order);
+        // 查询成功这条尾（状态/金额/上游单号都可能变了）才发。
+        // **上面那条查询失败的分支不发** —— 它只推 next_query_time 一个游标，
+        // 订单列表页看不到这一格，每轮轮询发一帧就是纯噪音。
+        publisher.changed(OrderEventPublisher.CHANNEL_HAR_ORDER, order.getTenantId());
         return HarOrderVO.of(order);
     }
 

java/src/test/java/com/payment/platform/module/payment/har/service/HarNotifyServiceTest.java

@@ -5,6 +5,7 @@ import com.payment.platform.module.payment.har.config.HarProperties;
 import com.payment.platform.module.payment.har.entity.HarOrderEntity;
 import com.payment.platform.module.payment.har.enums.HarOrderStatus;
 import com.payment.platform.module.payment.har.mapper.HarOrderMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -46,9 +47,13 @@ class HarNotifyServiceTest {
         }
     }
 
+    /** 夹具租户 —— 理由同 HarOrderServiceTest 的同名常量：非 null 才能钉住「发的是这一行的租户」。 */
+    private static final Long TENANT = 7L;
+
     @Mock private HarOrderService orderService;
     @Mock private HarOrderMapper orderMapper;
     @Mock private HarCallbackService callbackService;
+    @Mock private OrderEventPublisher publisher;
 
     private HarProperties properties;
     private HarNotifyService service;
@@ -57,13 +62,14 @@ class HarNotifyServiceTest {
     void setUp() {
         properties = new HarProperties();
         properties.setPlatformPublicKey(PUBLIC_KEY);
-        service = new HarNotifyService(properties, orderService, orderMapper, callbackService);
+        service = new HarNotifyService(properties, orderService, orderMapper, callbackService, publisher);
     }
 
     private HarOrderEntity order(String orderNo, String tradeStatus) {
         HarOrderEntity o = new HarOrderEntity();
         o.setOrderNo(orderNo);
         o.setEnterpriseId("E1");
+        o.setTenantId(TENANT);
         o.setTradeStatus(tradeStatus);
         o.setPrice(new BigDecimal("12.34"));
         o.setApiKey("KEY1");
@@ -111,6 +117,8 @@ class HarNotifyServiceTest {
         verify(orderService, never()).getByOrderNo(any());
         verify(orderMapper, never()).updateById(any());
         verify(callbackService, never()).notifyEnterprise(any());
+        // 订单实时信号（负例）：伪造报文一个字节都没改 ⇒ 不发
+        verifyNoInteractions(publisher);
     }
 
     @Test
@@ -201,6 +209,9 @@ class HarNotifyServiceTest {
         assertEquals("PT-1", o.getPlatformTradeNo());
         // 排期逻辑自身在 HarOrderServiceTest 覆盖，这里只验证确实委托了
         verify(orderService).reschedulePoll(o);
+        // 订单实时信号：通知真的改了行（状态、上游单号、平台单号都落库）⇒ 一条。
+        //   这是 HAR 订单页最主要的一帧 —— 「收到回调后页面自己变」就是它。
+        verify(publisher).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     @Test

java/src/test/java/com/payment/platform/module/payment/har/service/HarOrderServiceTest.java

@@ -21,6 +21,7 @@ import com.payment.platform.module.payment.har.entity.HarMerchantEntity;
 import com.payment.platform.module.payment.har.entity.HarOrderEntity;
 import com.payment.platform.module.payment.har.enums.HarOrderStatus;
 import com.payment.platform.module.payment.har.mapper.HarOrderMapper;
+import com.payment.platform.module.payment.stream.OrderEventPublisher;
 import org.apache.ibatis.builder.MapperBuilderAssistant;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -43,15 +44,28 @@ import static org.mockito.Mockito.*;
 @ExtendWith(MockitoExtension.class)
 class HarOrderServiceTest {
 
+    /**
+     * 夹具租户。
+     *
+     * <p>★ 为什么夹具必须自己带一个：生产上 {@code har_order.tenant_id} 是
+     * {@code AutoFillMetaObjectHandler.insertFill} 在 insert 时**写进实体对象**的
+     * （common/config/AutoFillMetaObjectHandler.java），本类的 mapper 是 mock、不跑自动填充
+     * ⇒ 实体上这一格恒为 null。断言里若跟着写 null，「发布的是这一行自己的租户」就退化成
+     * 「恒发 null 也绿」—— 而恒发 null 在生产上意味着**一页都不会刷新**。
+     * 所以给一个非 null 且唯一的取值，需要模拟自动填充的两条用例见 insert 的 doAnswer。
+     */
+    private static final Long TENANT = 7L;
+
     @Mock private HarOrderMapper orderMapper;
     @Mock private HarMerchantService merchantService;
     @Mock private HarClient harClient;
+    @Mock private OrderEventPublisher publisher;
 
     private HarOrderService service;
 
     @BeforeEach
     void setUp() {
-        service = new HarOrderService(orderMapper, merchantService, harClient);
+        service = new HarOrderService(orderMapper, merchantService, harClient, publisher);
         // 初始化 MyBatis-Plus lambda 元数据缓存，使 LambdaQueryWrapper.getSqlSegment() 可在无 Spring 上下文的单测中工作
         // （否则 getSqlSegment() 抛 MybatisPlusException: can not find lambda cache for this entity。
         //   同款引导见 LsyCallbackServiceTest / AlipayBatchPayServiceTest。）
@@ -84,6 +98,7 @@ class HarOrderServiceTest {
         HarOrderEntity o = new HarOrderEntity();
         o.setOrderNo(orderNo);
         o.setEnterpriseId("E1");
+        o.setTenantId(TENANT);
         o.setTradeStatus(tradeStatus);
         o.setQueryCount(0);
         return o;
@@ -105,12 +120,22 @@ class HarOrderServiceTest {
     void create_persistsLocalRecordBeforeCallingUpstream() {
         when(merchantService.requireByEnterpriseId("E1")).thenReturn(merchant());
         when(harClient.precreate(any())).thenReturn(response("10000", precreateDetail()));
+        // 模拟生产上的自动填充：insertFill 会把 tenantId 写进**这个实体对象**
+        // （common/config/AutoFillMetaObjectHandler:24-31），发布点读的正是它。
+        doAnswer(inv -> {
+            ((HarOrderEntity) inv.getArgument(0)).setTenantId(TENANT);
+            return 1;
+        }).when(orderMapper).insert(any());
 
         service.create(dto(), "E1", "KEY1");
 
         InOrder inOrder = inOrder(orderMapper, harClient);
         inOrder.verify(orderMapper).insert(any());
         inOrder.verify(harClient).precreate(any());
+
+        // 订单实时信号：本路径两个写点 —— 落库 insert（新行出现）、补 pay_url/上游单号那次
+        //   updateById ⇒ 恰好 2 条。同一次请求里的两帧由前端合并（500ms 窗口）。
+        verify(publisher, times(2)).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     @Test
@@ -253,6 +278,11 @@ class HarOrderServiceTest {
     void create_upstreamNonSuccessCode_throwsAndClosesTheLocalRow() {
         when(merchantService.requireByEnterpriseId("E1")).thenReturn(merchant());
         when(harClient.precreate(any())).thenReturn(response("40004", null));
+        // 同 create_persistsLocalRecordBeforeCallingUpstream：模拟 insertFill 写回 tenantId
+        doAnswer(inv -> {
+            ((HarOrderEntity) inv.getArgument(0)).setTenantId(TENANT);
+            return 1;
+        }).when(orderMapper).insert(any());
 
         BusinessException e = assertThrows(BusinessException.class,
                 () -> service.create(dto(), "E1", "KEY1"));
@@ -280,6 +310,10 @@ class HarOrderServiceTest {
         // 漏掉 WHERE 就是**更新全表**，所以这一条必须留着。
         assertTrue(closing.getSqlSegment().contains("id"),
                 "少了 WHERE 就是更新全表。实测: " + closing.getSqlSegment());
+
+        // 订单实时信号：本路径也是 2 条 —— insert（新行出现）+ 收尾那次 update(null,w)
+        //   （状态落到「8 交易关闭」，页面上那一行要变）。中间没有 updateById：上游拒绝时不补 pay_url。
+        verify(publisher, times(2)).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     @Test
@@ -349,6 +383,8 @@ class HarOrderServiceTest {
         assertNotNull(vo.getUpstreamCreateTime());
         // 终态必须清空，否则轮询任务永远捞起已结束的订单
         assertNull(o.getNextQueryTime());
+        // 订单实时信号：查询成功那条尾巴写了一次库 ⇒ 一条。本方法只此一个发布点。
+        verify(publisher).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     @Test
@@ -422,6 +458,9 @@ class HarOrderServiceTest {
         assertNotNull(o.getNextQueryTime());
         assertEquals(1, o.getQueryCount());
         verify(orderMapper).updateById(any());
+        // 订单实时信号（负例）：这一支只推查询游标（query_count / next_query_time），
+        //   订单列表页不展示这两格 ⇒ 发了就是每轮轮询推一帧，纯噪音。
+        verifyNoInteractions(publisher);
     }
 
     @Test
@@ -824,6 +863,10 @@ class HarOrderServiceTest {
         InOrder inOrder = inOrder(orderMapper, harClient);
         inOrder.verify(orderMapper).update(any(), any());
         inOrder.verify(harClient).refund(any());
+
+        // 订单实时信号：抢占那一次条件更新真的写了行（状态已置「退款中」）⇒ 一条。
+        //   页面必须立刻从「已支付」翻成「退款中」，不能等上游来回。
+        verify(publisher).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     /**
@@ -869,6 +912,9 @@ class HarOrderServiceTest {
 
         assertTrue(e.getMessage().contains("不可退款"), "实测: " + e.getMessage());
         verify(harClient, never()).refund(any());
+        // 订单实时信号（负例）：条件更新命中 0 行 = 一个字节都没改（并发已把状态改掉了）
+        //   ⇒ 不发。发了就是让订单页白拉一次。
+        verifyNoInteractions(publisher);
     }
 
     /** 非「支付成功」的订单一律不受理 —— 尤其 2 退款成功：再退一次就是重复退款 */
@@ -947,6 +993,11 @@ class HarOrderServiceTest {
         // 原因也要一起清掉：这笔退款从没发生过，留着一句原因是页面对着一笔已支付的单说假话
         assertTrue(rollback.getSqlSet().contains("refund_reason"),
                 "回退要把 reason 清成 null（写 null 只能走 SET 子句）。实测 SET: " + rollback.getSqlSet());
+
+        // 订单实时信号：本路径两次写库 —— 抢占（置退款中）、回退（翻回已支付）⇒ 两条。
+        //   ★ 回退那一条不能少：页面刚从「已支付」翻成「退款中」，上游拒绝后必须翻回来，
+        //     否则它会一直显示一个**根本没有发生**的退款中。
+        verify(publisher, times(2)).changed(OrderEventPublisher.CHANNEL_HAR_ORDER, TENANT);
     }
 
     /** 优先用上游单号退款；拿不到时走 third_order_no（同查询口径） */