Kaynağa Gözat

feat(转账): 补上到卡转账的自动反查(next_retry_at 此前只写不读)

`pay_transfer.next_retry_at` 在此之前是一列死字段:下单时写入它、四条通知/更新
路径(AccountHandler ×2、BillHandler、AccountService.updateTransferStatusByOrderNo)
又都清掉它,但全仓没有任何读取点。于是一笔收不到支付宝异步通知的转账会永远停在
DEALING,除非有人手点 POST /transfer/sync-all(该端点还挂着权限串
module_payment:account:transfer:update)。本提交补上消费侧,对齐 Python
AccountService.retry_dealing_transfers。

判定(逐字移植 Python):
- 谓词 status=DEALING AND retry_count<4 AND next_retry_at IS NOT NULL AND
  next_retry_at<=now(),按 next_retry_at 升序;
- 终态 SUCCESS/FAIL/REFUND ⇒ 落 status 与 fund_order_id、retry_count=4、
  next_retry_at=null,然后走对外回调出口 notifyTransferResult;
- 仍 DEALING 时对卡转账推进 3 档退避(累计 5/15/60/120 分钟,基准是 created_time
  而不是 now),支付宝账户转账不给二次机会,直接判 FAIL;
- 两条判失败文案沿用 Python 原文。

两处**故意偏离**(注释里写了理由):
1. 查不通(网络失败/上游报错/响应里没有 status)本轮不写任何字段。Python 把这种
   情形当成「仍 DEALING」继续推进退避,退避耗尽后会把一笔可能已经成功的转账盖成
   FAIL,那是假的资金结论。
2. 卡转账判定同时收 BANKCARD_ACCOUNT 与 BANK_CARD。本类 isBankCard 写的是支付宝的
   identity_type(前者),AccountService 的展示映射两种都列;只认一个会把按另一种
   写法落库的数据判错档,退避一档不推就判 FAIL。

定时器抄 F2T 轮询那支(@Scheduled(fixedDelay=60s) + 委派 + try/catch)。
租户插件不过滤:pay_transfer 在 TenantInnerInterceptor.CONDITIONAL_TABLES 里,
与 Python 的 check_data_scope=False 同义。mvn test 期间不会真跑:
PaymentApplicationContextTest 把 TaskScheduler 换成了不触发的实现。

新增 TransferRetryTest 13 条(account 包的首个测试)。两向变异实测:
- 退避算式改错        ⇒ 3 红(恰为三条 backoff 行)
- 只认 BANKCARD_ACCOUNT ⇒ 1 红(expected 1 but was 4)
- 摘掉 UNKNOWN 守闸    ⇒ 1 红(expected 0 but was 1)
- 只改日志文案        ⇒ 13 全绿(无害改动实测仍绿)
每次变异后还原到字节级一致(sha256 == 基线 60eb02a5…)。
全量 mvn clean test = 658 绿(改前 645)。
alphaH 3 hafta önce
ebeveyn
işleme
fbd7707548

+ 39 - 0
java/src/main/java/com/payment/platform/module/payment/account/scheduler/TransferRetryScheduler.java

@@ -0,0 +1,39 @@
+package com.payment.platform.module.payment.account.scheduler;
+
+import com.payment.platform.module.payment.account.service.AlipayTransferService;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Component;
+
+/**
+ * 转账反查定时任务 —— 兜底终结停在 DEALING 的转账。
+ *
+ * <p>到卡转账下单后支付宝会异步通知结果,但通知可能丢(回调地址不可达、应用重启期间错过)。
+ * 本任务每分钟捞一次「已到 next_retry_at 且还在处理中」的转账去反查,对应 Python
+ * {@code AccountService.retry_dealing_transfers}。
+ *
+ * <p>无认证上下文 ⇒ 查询不会被租户插件过滤({@code pay_transfer} 在
+ * {@code TenantInnerInterceptor.CONDITIONAL_TABLES} 里),与 Python 的
+ * {@code check_data_scope=False} 同义。写入的归属取自各行自己的 tenant_id。
+ *
+ * <p>★ 这个 {@code @Scheduled} 在 {@code mvn test} 期间**不会**真跑:本仓唯一的
+ * {@code @SpringBootTest}({@code PaymentApplicationContextTest})把 {@code TaskScheduler}
+ * 换成了不触发任何调度的实现。少了那层,每次跑测试都会真打上游查询接口。
+ */
+@Slf4j
+@Component
+@RequiredArgsConstructor
+public class TransferRetryScheduler {
+
+    private final AlipayTransferService alipayTransferService;
+
+    @Scheduled(fixedDelay = 60 * 1000) // 每 1 分钟
+    public void retryDealingTransfers() {
+        try {
+            alipayTransferService.retryDealingTransfers();
+        } catch (Exception e) {
+            log.error("[转账反查] 定时任务异常", e);
+        }
+    }
+}

+ 131 - 0
java/src/main/java/com/payment/platform/module/payment/account/service/AlipayTransferService.java

@@ -904,6 +904,137 @@ public class AlipayTransferService {
         return result;
     }
 
+    // ==================== 转账反查(定时兜底,对应 Python retry_dealing_transfers) ====================
+
+    /**
+     * 退避档位(分钟)。第 n 次反查之后,下次反查时间 = created_time + sum(RETRY_INTERVALS[0..n+1])。
+     * <p>逐字对应 Python {@code _RETRY_INTERVALS = [5, 10, 45, 60]} —— 累计 5 / 15 / 60 / 120 分钟。
+     */
+    static final int[] RETRY_INTERVALS = {5, 10, 45, 60};
+
+    /** retry_count 达到该值即「不再反查」。写终态与判失败**都**写这个值。 */
+    static final int RETRY_TERMINAL = 4;
+
+    /** 对卡转账最多推进 3 档退避(第 4 次反查仍 DEALING 才判失败)。 */
+    private static final int MAX_RETRIES_CARD = 3;
+
+    /** 支付宝账户转账不给二次反查机会(没有资金在途风险,超时即判失败)。 */
+    private static final int MAX_RETRIES_ALIPAY = 0;
+
+    private static final String FAIL_REASON_CARD_TIMEOUT = "反查超时判定为失败";
+    private static final String FAIL_REASON_ALIPAY_ONCE = "支付宝转账仅重试1次仍DEALING";
+
+    /**
+     * payee_type 是不是「对卡转账」。
+     *
+     * <p>★ 两种写法都要认,因为这一列的**写入口径换过**:本类 {@code transfer} 写的是支付宝的
+     * {@code identity_type}(对卡 = {@code "BANKCARD_ACCOUNT"},见下方 isBankCard 那处判断),
+     * 而 {@code AccountService} 的展示映射里同时列了 {@code "BANK_CARD"} 与 {@code "BANKCARD_ACCOUNT"}。
+     * 老数据(Python 时期写入)用后者,新数据用前者 —— 只认一个会把另一批判错档:
+     * 对卡被当成支付宝账户 ⇒ 退避一档都不推就判 FAIL。
+     */
+    private static boolean isCardTransfer(String payeeType) {
+        return "BANKCARD_ACCOUNT".equals(payeeType) || "BANK_CARD".equals(payeeType);
+    }
+
+    /**
+     * 反查「到点该查」的 DEALING 转账。
+     *
+     * <p>本方法之前不存在,{@code next_retry_at} 是一列**只写不读**的死字段:下单时写了它、
+     * 四条通知/更新路径又都清掉它,但没有任何消费点。于是一笔收不到支付宝异步通知的转账
+     * 会永远停在 DEALING,除非有人手点 {@code POST /transfer/sync-all}。
+     *
+     * <p>与 {@link #transferSyncAllService()} 的分工:那个是人工全量按钮(捞所有 DEALING、
+     * 不看退避、不走对外回调);这个是每分钟的兜底(只看到点的、按档位退避、终结时回调)。
+     *
+     * @return {@code {scanned, processed}}
+     */
+    public Map<String, Object> retryDealingTransfers() {
+        List<TransferEntity> pending = transferMapper.selectList(
+                new com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper<TransferEntity>()
+                        .eq(TransferEntity::getStatus, "DEALING")
+                        .lt(TransferEntity::getRetryCount, RETRY_TERMINAL)
+                        .isNotNull(TransferEntity::getNextRetryAt)
+                        .le(TransferEntity::getNextRetryAt, OffsetDateTime.now())
+                        .orderByAsc(TransferEntity::getNextRetryAt));
+
+        int processed = 0;
+        for (TransferEntity t : pending) {
+            try {
+                if (retryOneTransfer(t)) processed++;
+            } catch (Exception e) {
+                log.warn("[转账反查] out_biz_no={} 反查异常: {}", t.getOutBizNo(), e.getMessage());
+            }
+        }
+        if (!pending.isEmpty()) {
+            log.info("[转账反查] 扫描 {} 条,写入 {} 条", pending.size(), processed);
+        }
+        Map<String, Object> result = new LinkedHashMap<>();
+        result.put("scanned", pending.size());
+        result.put("processed", processed);
+        return result;
+    }
+
+    /** 反查单条。返回 true 表示本轮对它做了写入。 */
+    private boolean retryOneTransfer(TransferEntity t) {
+        String orderNo = (t.getOrderNo() != null && !t.getOrderNo().isBlank()) ? t.getOrderNo() : null;
+        Map<String, String> q = queryTransferStatus(orderNo, t.getOutBizNo(), t.getEnterpriseId());
+        String s = q.get("status");
+
+        if ("SUCCESS".equals(s) || "FAIL".equals(s) || "REFUND".equals(s)) {
+            t.setStatus(s);
+            if (q.get("pay_fund_order_id") != null) t.setFundOrderId(q.get("pay_fund_order_id"));
+            t.setRetryCount(RETRY_TERMINAL);
+            t.setNextRetryAt(null);
+            transferMapper.updateById(t);
+            notifyTransferToOpenApi(t);
+            return true;
+        }
+
+        if (s == null || "UNKNOWN".equals(s)) {
+            // 查不通(网络失败 / 上游报错 / 响应里没有 status)**不是**「还没出结果」的证据。
+            // Python 把这种情形也当成「仍 DEALING」继续推进退避,退避耗尽后就会把一笔
+            // 可能已经成功的转账盖成 FAIL —— 那是假的资金结论。这里改成本轮不动任何字段。
+            log.warn("[转账反查] out_biz_no={} 查询未取得结论(status={}),本轮跳过", t.getOutBizNo(), s);
+            return false;
+        }
+
+        // 仍是 DEALING —— 推进退避,或退避耗尽后判失败
+        boolean isCard = isCardTransfer(t.getPayeeType());
+        int retryCount = t.getRetryCount() != null ? t.getRetryCount() : 0;
+
+        if (retryCount < (isCard ? MAX_RETRIES_CARD : MAX_RETRIES_ALIPAY)) {
+            int accumulated = 0;
+            for (int i = 0; i <= retryCount + 1 && i < RETRY_INTERVALS.length; i++) {
+                accumulated += RETRY_INTERVALS[i];
+            }
+            OffsetDateTime base = t.getCreatedTime() != null ? t.getCreatedTime() : OffsetDateTime.now();
+            t.setRetryCount(retryCount + 1);
+            t.setNextRetryAt(base.plusMinutes(accumulated));
+            transferMapper.updateById(t);
+            return true;
+        }
+
+        t.setStatus("FAIL");
+        t.setRetryCount(RETRY_TERMINAL);
+        t.setNextRetryAt(null);
+        t.setErrorMsg(isCard ? FAIL_REASON_CARD_TIMEOUT : FAIL_REASON_ALIPAY_ONCE);
+        transferMapper.updateById(t);
+        notifyTransferToOpenApi(t);
+        return true;
+    }
+
+    /** 反查得到终态后对外回调(与通知到达路径同款出口)。回调失败不影响落库结果。 */
+    private void notifyTransferToOpenApi(TransferEntity t) {
+        try {
+            String outBizNo = t.getOutBizNo();
+            String key = (t.getOrderNo() != null && !t.getOrderNo().isBlank()) ? t.getOrderNo() : outBizNo;
+            openapiService.notifyTransferResult(key, outBizNo, t.getTenantId());
+        } catch (Exception e) {
+            log.warn("[转账反查] 对外回调失败: out_biz_no={}, err={}", t.getOutBizNo(), e.getMessage());
+        }
+    }
+
     // ==================== util ====================
 
     private String toJson(Object obj) {

+ 247 - 0
java/src/test/java/com/payment/platform/module/payment/account/service/TransferRetryTest.java

@@ -0,0 +1,247 @@
+package com.payment.platform.module.payment.account.service;
+
+import com.baomidou.mybatisplus.core.MybatisConfiguration;
+import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
+import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
+import com.payment.platform.module.payment.account.entity.TransferEntity;
+import com.payment.platform.module.payment.account.mapper.TransferMapper;
+import com.payment.platform.module.payment.openapi.service.OpenapiService;
+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.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.ArgumentCaptor;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.time.OffsetDateTime;
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.*;
+
+/**
+ * 转账自动反查(定时兜底)—— 对应 Python {@code AccountService.retry_dealing_transfers}。
+ *
+ * <p>这个功能在本批之前**完全不存在**:{@code next_retry_at} 只被三处写入、零处读取,
+ * 一笔收不到支付宝异步通知的 DEALING 转账会永远停在 DEALING,除非有人手点
+ * {@code POST /transfer/sync-all}。本测试钉住补上的那条腿。
+ *
+ * <p>量具说明:{@code queryTransferStatus} 走的是支付宝网关,测试里用 {@code spy} 换掉
+ * (**不是**换掉整个容器)——被测的是它外面的判定与写库逻辑。
+ */
+@ExtendWith(MockitoExtension.class)
+class TransferRetryTest {
+
+    private static final OffsetDateTime CREATED = OffsetDateTime.parse("2026-09-01T10:00:00+08:00");
+
+    @Mock private TransferMapper transferMapper;
+    @Mock private OpenapiService openapiService;
+
+    private AlipayTransferService service;
+
+    @BeforeEach
+    void setUp() {
+        service = spy(new AlipayTransferService(
+                null, null, transferMapper, null, null, null, null));
+        // openapiService 在生产类是 @Lazy @Autowired 字段注入(不在构造器参数里)
+        ReflectionTestUtils.setField(service, "openapiService", openapiService);
+        // 初始化 MyBatis-Plus lambda 元数据缓存,使 LambdaQueryWrapper.getSqlSegment() 可在无 Spring 上下文的单测中工作
+        TableInfoHelper.initTableInfo(new MapperBuilderAssistant(new MybatisConfiguration(), ""), TransferEntity.class);
+    }
+
+    // ==================== 夹具 ====================
+
+    private TransferEntity dealing(String payeeType, Integer retryCount) {
+        TransferEntity t = new TransferEntity();
+        t.setId(1L);
+        t.setStatus("DEALING");
+        t.setOutBizNo("OB1");
+        t.setOrderNo("ON1");
+        t.setEnterpriseId("E1");
+        t.setTenantId(1L);
+        t.setPayeeType(payeeType);
+        t.setRetryCount(retryCount);
+        t.setNextRetryAt(CREATED.plusMinutes(5));
+        t.setCreatedTime(CREATED);
+        return t;
+    }
+
+    private void givenPending(TransferEntity... rows) {
+        when(transferMapper.selectList(any())).thenReturn(List.of(rows));
+    }
+
+    private void givenQueryReturns(String status) {
+        doReturn(Map.of("status", status, "pay_fund_order_id", "F1"))
+                .when(service).queryTransferStatus(any(), any(), any());
+    }
+
+    // ==================== 1. 终态:写终态 + 清退避 + 对外回调 ====================
+
+    @Test
+    void terminalStatus_writesIt_clearsRetryFields_andNotifiesOpenApi() {
+        TransferEntity t = dealing("BANKCARD_ACCOUNT", 1);
+        givenPending(t);
+        givenQueryReturns("SUCCESS");
+
+        service.retryDealingTransfers();
+
+        assertEquals("SUCCESS", t.getStatus());
+        assertEquals(4, t.getRetryCount(), "终态后 retry_count 必须是 4(= 不再反查)");
+        assertNull(t.getNextRetryAt(), "终态后必须清掉 next_retry_at");
+        assertEquals("F1", t.getFundOrderId(), "终态要落支付宝资金流水号");
+        verify(transferMapper).updateById(t);
+        verify(openapiService).notifyTransferResult("ON1", "OB1", 1L);
+    }
+
+    @Test
+    void terminalStatus_withoutOrderNo_notifiesByOutBizNo() {
+        TransferEntity t = dealing("BANKCARD_ACCOUNT", 1);
+        t.setOrderNo(null);
+        givenPending(t);
+        givenQueryReturns("FAIL");
+
+        service.retryDealingTransfers();
+
+        assertEquals("FAIL", t.getStatus());
+        verify(openapiService).notifyTransferResult("OB1", "OB1", 1L);
+    }
+
+    // ==================== 2. 仍 DEALING:对卡转账按档位退避 ====================
+
+    @ParameterizedTest(name = "retryCount={0} ⇒ next_retry_at = created_time + {1} 分钟")
+    @CsvSource({"0,15", "1,60", "2,120"})
+    void cardStillDealing_advancesBackoffFromCreatedTime(int retryCount, int expectedMinutes) {
+        TransferEntity t = dealing("BANKCARD_ACCOUNT", retryCount);
+        givenPending(t);
+        givenQueryReturns("DEALING");
+
+        service.retryDealingTransfers();
+
+        assertEquals("DEALING", t.getStatus(), "还在处理中就不该动 status");
+        assertEquals(retryCount + 1, t.getRetryCount());
+        assertEquals(CREATED.plusMinutes(expectedMinutes), t.getNextRetryAt(),
+                "退避以 created_time 为基准(不是 now),档位累计 5/15/60/120 分钟");
+        verify(transferMapper).updateById(t);
+        verify(openapiService, never()).notifyTransferResult(any(), any(), any());
+    }
+
+    /** 两种写法都要走对卡分支 —— 老数据写 BANK_CARD、本仓写入口写 BANKCARD_ACCOUNT。 */
+    @ParameterizedTest
+    @ValueSource(strings = {"BANKCARD_ACCOUNT", "BANK_CARD"})
+    void bothCardSpellings_routeToCardBranch(String payeeType) {
+        TransferEntity t = dealing(payeeType, 0);
+        givenPending(t);
+        givenQueryReturns("DEALING");
+
+        service.retryDealingTransfers();
+
+        assertEquals(1, t.getRetryCount(), payeeType + " 必须按对卡转账退避,而不是立刻判失败");
+    }
+
+    @Test
+    void cardStillDealing_atLastRetry_stampsFailWithTimeoutReason() {
+        TransferEntity t = dealing("BANKCARD_ACCOUNT", 3);
+        givenPending(t);
+        givenQueryReturns("DEALING");
+
+        service.retryDealingTransfers();
+
+        assertEquals("FAIL", t.getStatus());
+        assertEquals(4, t.getRetryCount());
+        assertNull(t.getNextRetryAt());
+        assertEquals("反查超时判定为失败", t.getErrorMsg());
+        verify(openapiService).notifyTransferResult("ON1", "OB1", 1L);
+    }
+
+    // ==================== 3. 仍 DEALING:支付宝账户转账只反查一次 ====================
+
+    @Test
+    void alipayAccountStillDealing_afterFirstQuery_stampsFail() {
+        TransferEntity t = dealing("ALIPAY_USER_ID", 0);
+        givenPending(t);
+        givenQueryReturns("DEALING");
+
+        service.retryDealingTransfers();
+
+        assertEquals("FAIL", t.getStatus(), "非对卡转账没有二次反查机会");
+        assertEquals(4, t.getRetryCount());
+        assertNull(t.getNextRetryAt());
+        assertEquals("支付宝转账仅重试1次仍DEALING", t.getErrorMsg());
+        verify(openapiService).notifyTransferResult("ON1", "OB1", 1L);
+    }
+
+    // ==================== 4. 查不通:本轮不动任何字段 ====================
+
+    @Test
+    void queryWithoutConclusion_skipsRow_advancesNothing() {
+        TransferEntity t = dealing("BANKCARD_ACCOUNT", 0);
+        givenPending(t);
+        givenQueryReturns("UNKNOWN"); // 网络失败/上游报错 —— 不是「还没出结果」的证据
+
+        service.retryDealingTransfers();
+
+        assertEquals("DEALING", t.getStatus());
+        assertEquals(0, t.getRetryCount(), "查不通不该消耗退避档位");
+        assertEquals(CREATED.plusMinutes(5), t.getNextRetryAt());
+        verify(transferMapper, never()).updateById(any());
+        verify(openapiService, never()).notifyTransferResult(any(), any(), any());
+    }
+
+    // ==================== 5. 隔离与空集 ====================
+
+    @Test
+    void oneRowThrowing_doesNotStopTheRest() {
+        TransferEntity bad = dealing("BANKCARD_ACCOUNT", 0);
+        bad.setId(1L);
+        TransferEntity good = dealing("BANKCARD_ACCOUNT", 0);
+        good.setId(2L);
+        good.setOutBizNo("OB2");
+        givenPending(bad, good);
+
+        doThrow(new RuntimeException("boom")).doReturn(Map.of("status", "SUCCESS"))
+                .when(service).queryTransferStatus(any(), any(), any());
+
+        assertDoesNotThrow(() -> service.retryDealingTransfers());
+
+        assertEquals("DEALING", bad.getStatus());
+        assertEquals("SUCCESS", good.getStatus(), "前一行的异常不能吃掉后一行");
+        verify(transferMapper).updateById(good);
+    }
+
+    @Test
+    void noPendingRows_touchesNothing() {
+        when(transferMapper.selectList(any())).thenReturn(List.of());
+
+        assertDoesNotThrow(() -> service.retryDealingTransfers());
+
+        verify(transferMapper, never()).updateById(any());
+        verifyNoInteractions(openapiService);
+    }
+
+    // ==================== 6. 查询谓词(四个条件都在 SQL 段里) ====================
+
+    @Test
+    void pendingQuery_filtersOnAllFourConditions_inNextRetryOrder() {
+        when(transferMapper.selectList(any())).thenReturn(List.of());
+
+        service.retryDealingTransfers();
+
+        @SuppressWarnings("unchecked")
+        ArgumentCaptor<LambdaQueryWrapper<TransferEntity>> cap =
+                ArgumentCaptor.forClass(LambdaQueryWrapper.class);
+        verify(transferMapper).selectList(cap.capture());
+        String sql = cap.getValue().getSqlSegment();
+        assertAll(
+                () -> assertTrue(sql.contains("status"), "缺 status: " + sql),
+                () -> assertTrue(sql.contains("retry_count"), "缺 retry_count(否则会把已终态的行反复捞出来): " + sql),
+                () -> assertTrue(sql.contains("next_retry_at"), "缺 next_retry_at: " + sql),
+                () -> assertTrue(sql.contains("ORDER BY"), "缺排序: " + sql));
+    }
+}