package com.talentcard.server.business.payment;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;

import java.util.List;
import java.util.Map;

import com.talentcard.server.business.payment.PaymentReconciliationService.PendingPaymentConfirmation;
import com.talentcard.server.business.payment.huifu.HuifuMiniProgramPaymentGateway;
import com.talentcard.server.business.payment.huifu.HuifuPaymentCloseCommand;
import com.talentcard.server.business.payment.huifu.HuifuPaymentGatewayException;
import com.talentcard.server.business.portal.PendingOrderPortalService;
import com.talentcard.server.common.BusinessException;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.test.context.ActiveProfiles;
import org.springframework.test.context.bean.override.mockito.MockitoBean;

/**
 * Seam: the local state of a pending order can be behind the provider, because a payment
 * notification that never arrived leaves the order looking unpaid while the money was collected.
 *
 * <p>Cancelling such an order must therefore ask the provider first. A payment the provider
 * confirms settles the order and blocks the cancellation; a provider that says the money may still
 * be moving blocks it too, because a question about money is never answered by guessing.
 */
@SpringBootTest
@ActiveProfiles("test")
class PendingPaymentConfirmationTests {

    @Autowired
    private JdbcTemplate jdbc;

    @Autowired
    private PendingOrderPortalService pendingOrderService;

    @Autowired
    private PaymentReconciliationService reconciliationService;

    @Autowired
    private HuifuPaymentService paymentService;

    @MockitoBean
    private HuifuMiniProgramPaymentGateway paymentGateway;

    private Long talentId;
    private Long merchantId;
    private String huifuId;
    private String suffix;
    private long orderSequence;

    @BeforeEach
    void prepareData() {
        suffix = String.valueOf(System.nanoTime());
        // 共享测试库里可能留着历史遗留的挂起流水，只有本用例自己创建的才能落进对账窗口。
        jdbc.update("update payment_transaction set created_at = now(), last_reconciled_at = null where status in ('CREATED', 'PREPAY_CREATED', 'CREATE_UNKNOWN', 'CLOSING', 'CANCEL_UNCONFIRMED')");
        jdbc.update("insert into talent_user(openid, name, mobile, status, deleted) values (?, ?, ?, 1, 0)",
                "pending-payment-openid-" + suffix, "Pending payment user", "13900000003");
        talentId = jdbc.queryForObject("select max(id) from talent_user", Long.class);
        jdbc.update("insert into merchant(merchant_name, contact_mobile, address, status, deleted) values (?, ?, ?, 1, 0)",
                "Pending payment merchant " + suffix, "08120000002", "Test address");
        merchantId = jdbc.queryForObject("select max(id) from merchant", Long.class);
        huifuId = "H" + suffix.substring(Math.max(0, suffix.length() - 8));
        orderSequence = 0;
    }

    @Test
    void cancellingALocallyPendingOrderTheProviderCollectedIsRefusedAndTheOrderIsMarkedPaid() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "trans_amt", "20.00",
                "huifu_id", huifuId, "org_hf_seq_id", "HF-CANCEL-" + suffix)));

        assertThatThrownBy(() -> pendingOrderService.cancel(orderNo, talentId))
                .isInstanceOf(BusinessException.class)
                .hasMessageContaining("已支付");

        assertThat(jdbc.queryForMap("select order_status, verify_status from benefit_order where order_no = ?", orderNo))
                .containsEntry("order_status", 1)
                .containsEntry("verify_status", 1);
        assertThat(jdbc.queryForMap("select status, paid_at from payment_transaction where payment_no = ?", paymentNo))
                .containsEntry("status", "SUCCEEDED");
        assertThat(count("select count(*) from payment_transaction where payment_no = ? and paid_at is not null", paymentNo)).isEqualTo(1);
    }

    @Test
    void cancellingAnOrderTheProviderHasNotCollectedAsksTheProviderAndClosesIt() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "resp_desc", "交易成功", "org_trans_stat", "P")));

        Map<String, Object> result = pendingOrderService.cancel(orderNo, talentId);

        assertThat(((Number) result.get("orderStatus")).intValue()).isEqualTo(2);
        verify(paymentGateway).queryMiniProgramPayment(any());
        ArgumentCaptor<HuifuPaymentCloseCommand> close = ArgumentCaptor.forClass(HuifuPaymentCloseCommand.class);
        verify(paymentGateway).closeMiniProgramPayment(close.capture());
        assertThat(close.getValue().paymentNo()).isEqualTo(paymentNo);
        assertThat(count("select count(*) from payment_transaction where payment_no = ? and status = 'CLOSED'", paymentNo)).isEqualTo(1);
    }

    @Test
    void cancellingProceedsButKeepsThePrepayWatchableWhenTheProviderRefusesToCloseIt() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "20000001", "resp_desc", "不允许关闭一分钟以内的订单", "huifu_id", huifuId)));

        pendingOrderService.cancel(orderNo, talentId);

        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isEqualTo(2);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("关不掉的预支付单还可能被支付，本地不能记成已关闭，要留给对账任务继续盯着")
                .isEqualTo("CANCEL_UNCONFIRMED");
        assertThat(openIssueCount("PAYMENT_CLOSE_UNCONFIRMED", paymentNo)).isEqualTo(1L);
    }

    @Test
    void cancellingClosesThePrepayEvenWhenTheQueryOnlySaysItIsProcessing() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        // 查单只答「处理中」：旧实现把它当成关单的前置门，于是关单一次都不会被调用。
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "P", "resp_code", "00000100", "resp_desc", "交易处理中", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "20000001", "resp_desc", "不允许关闭一分钟以内的订单", "huifu_id", huifuId)));

        Map<String, Object> result = pendingOrderService.cancel(orderNo, talentId);

        verify(paymentGateway, times(1)).closeMiniProgramPayment(any());
        assertThat(((Number) result.get("orderStatus")).intValue())
                .as("关单回执是明确拒绝，不是「还在处理中」，因此允许取消并继续盯防")
                .isEqualTo(2);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CANCEL_UNCONFIRMED");
    }

    @Test
    void closingReportsTheOriginalTradeSucceededAndTheQueryThenLandsTheOrderAsPaid() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        // 第一次查单说交易不存在（所以会去关单），关单回执说原交易已经成功，再查一次拿到了金额。
        when(paymentGateway.queryMiniProgramPayment(any()))
                .thenReturn(providerResponse(Map.of("resp_code", "23000001", "resp_desc", "交易不存在", "huifu_id", huifuId)))
                .thenReturn(providerResponse(Map.of(
                        "trans_stat", "S", "resp_code", "00000000", "trans_amt", "20.00",
                        "huifu_id", huifuId, "org_hf_seq_id", "HF-CLOSE-PAID-" + suffix)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "23000000", "resp_desc", "原订单已为终态,请发起查询交易获取",
                "org_trans_stat", "S", "huifu_id", huifuId)));

        assertThatThrownBy(() -> pendingOrderService.cancel(orderNo, talentId))
                .isInstanceOf(BusinessException.class)
                .hasMessageContaining("已支付");

        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                .as("关单回执说原交易已成功，就必须落成已支付，而不是把一笔真钱留在待支付里")
                .isEqualTo(1);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("SUCCEEDED");
    }

    @Test
    void cancellingWaitsForTheCloseConclusionWhenClosingRevealsTheTradeWasAlreadyPaid() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        // 查单说「交易不存在」，关单却回「原订单已为终态」且原交易已成功：钱可能真的在，不能当成没收过款。
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "resp_code", "23000001", "resp_desc", "交易不存在", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "23000000", "resp_desc", "原订单已为终态,请发起查询交易获取",
                "org_trans_stat", "S", "huifu_id", huifuId)));

        // 查单拿不到金额，无法把订单落成已支付，所以不能谎称已支付；但订单也绝不能被取消。
        Map<String, Object> result = pendingOrderService.cancel(orderNo, talentId);

        assertThat(result)
                .as("关单没有结论时，取消的意图被受理但要等结论，而不是给一个永久报错")
                .containsEntry("cancellationPending", true);
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isZero();
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("关单中：既不能当已关闭（通道还可能收钱），也不该是终点")
                .isEqualTo("CLOSING");
        assertThat(count("select count(*) from payment_transaction where order_no = ?", orderNo))
                .as("关单中期间只能复用，不允许新建第二张可被支付的预支付单")
                .isEqualTo(1);
        assertThat(openIssueCount("PAYMENT_CLOSE_CONFLICT", paymentNo)).isEqualTo(1L);
    }

    @Test
    void cancellingWaitsForTheCloseConclusionWhileTheProviderIsStillClosingThePrepay() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "P", "resp_code", "00000100", "resp_desc", "交易处理中", "huifu_id", huifuId)));

        Map<String, Object> result = pendingOrderService.cancel(orderNo, talentId);

        assertThat(result).containsEntry("cancellationPending", true);
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isZero();
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CLOSING");
        assertThat(jdbc.queryForObject("select close_attempts from payment_transaction where payment_no = ?", Integer.class, paymentNo))
                .as("关单重试次数要落库，否则重试上限永远到不了")
                .isEqualTo(1);
    }

    @Test
    void cancellingWaitsForTheProviderWhenItCannotSayWhetherTheMoneyWasCollected() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any()))
                .thenThrow(new HuifuPaymentGatewayException("Huifu payment query failed", true, null));

        Map<String, Object> result = pendingOrderService.cancel(orderNo, talentId);

        assertThat(result).containsEntry("cancellationPending", true);
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isZero();
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("an unanswerable payment question must not destroy the only record of the money")
                .isEqualTo("PREPAY_CREATED");
        assertThat(openIssueCount("PAYMENT_QUERY_FAILED", paymentNo)).isEqualTo(1L);
    }

    @Test
    void retryingAPaymentTheProviderCollectedReportsItAsPaidInsteadOfAskingTheUserToPayAgain() {
        String orderNo = pendingOrder();
        pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "trans_amt", "20.00",
                "huifu_id", huifuId, "org_hf_seq_id", "HF-RETRY-" + suffix)));
        when(paymentGateway.createMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "P", "resp_code", "00000000", "pay_info", "{\"appId\":\"wx-test\"}")));

        Map<String, Object> result = paymentService.beginMiniProgramPayment(orderNo, talentId, 1, "retry-key-" + suffix);

        assertThat(result).containsEntry("paymentStatus", "PAID");
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isEqualTo(1);
        verify(paymentGateway, never()).createMiniProgramPayment(any());
    }

    @Test
    void retryingAPaymentReportsWaitingWhenTheProviderCannotSayWhetherTheMoneyWasCollected() {
        String orderNo = pendingOrder();
        pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any()))
                .thenThrow(new HuifuPaymentGatewayException("Huifu payment query failed", true, null));

        Map<String, Object> result = paymentService.beginMiniProgramPayment(orderNo, talentId, 1, "retry-key-" + suffix);

        assertThat(result)
                .as("问不出来的不确定只属于这一次支付尝试，不该把用户堵在一句报错上")
                .containsEntry("paymentStatus", "PAYMENT_PROCESSING");
        verify(paymentGateway, never()).createMiniProgramPayment(any());
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isZero();
    }

    @Test
    void askingToPayAgainReusesTheStoredPrepayWithoutAskingTheProvider() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L, "PREPAY_CREATED", "{\"appId\":\"wx-test\",\"timeStamp\":\"1\"}");

        Map<String, Object> result = paymentService.beginMiniProgramPayment(orderNo, talentId, 1, "reuse-key-" + suffix);

        assertThat(result)
                .as("本地已经持有可用的支付参数时直接把这张预支付单交回去——查通道只会得到「处理中」这种没有结论的答复")
                .containsEntry("paymentStatus", "PAYMENT_PENDING")
                .containsEntry("paymentNo", paymentNo);
        assertThat(result.get("paymentParams")).isNotNull();
        verify(paymentGateway, never()).queryMiniProgramPayment(any());
        verify(paymentGateway, never()).createMiniProgramPayment(any());
        assertThat(count("select count(*) from payment_transaction where order_no = ?", orderNo))
                .as("复用不新建支付流水")
                .isEqualTo(1);
    }

    @Test
    void askingToPayAgainIsRefusedWhenAnUndecidedPrepayCannotBeReused() {
        String orderNo = pendingOrder();
        // 先取消，通道答「处理中」→ 关单没有结论，这次尝试停在 CLOSING 且订单仍待支付；
        // 它没有留下可复用的支付参数，因此既不能复用、也不许新建。
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "P", "resp_code", "00000100", "resp_desc", "交易处理中", "huifu_id", huifuId)));
        pendingOrderService.cancel(orderNo, talentId);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo)).isEqualTo("CLOSING");
        clearInvocations(paymentGateway);

        Map<String, Object> result = paymentService.beginMiniProgramPayment(orderNo, talentId, 1, "blocked-key-" + suffix);

        assertThat(result).containsEntry("paymentStatus", "PAYMENT_PROCESSING");
        verify(paymentGateway, never()).createMiniProgramPayment(any());
        assertThat(count("select count(*) from payment_transaction where order_no = ?", orderNo))
                .as("上一张预支付单没有被证明不可能再被支付之前，不允许给同一笔订单再建一张")
                .isEqualTo(1);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CLOSING");
    }

    @Test
    void anExpiredOrderCannotBePaidEvenWhenAPrepayIsStillStoredLocally() {
        String orderNo = pendingOrder();
        pendingPayment(orderNo, 2000L, "PREPAY_CREATED", "{\"appId\":\"wx-test\",\"timeStamp\":\"1\"}");
        expireNow(orderNo);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "resp_code", "23000001", "resp_desc", "交易不存在", "huifu_id", huifuId)));

        assertThatThrownBy(() -> paymentService.beginMiniProgramPayment(orderNo, talentId, 1, "expired-key-" + suffix))
                .isInstanceOf(BusinessException.class)
                .hasMessageContaining("超时");

        verify(paymentGateway, never()).createMiniProgramPayment(any());
    }

    @Test
    void abandoningAPaymentOnlyClosesItAndNeverJudgesWhetherMoneyArrived() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        // 客户端说「我放弃了」，而关单回执说原交易已经成功——即使如此也不能据客户端的话去查单或结算。
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "23000000", "resp_desc", "原订单已为终态,请发起查询交易获取",
                "org_trans_stat", "S", "huifu_id", huifuId)));

        PendingPaymentConfirmation confirmation = reconciliationService.abandonPendingPayments(orderNo, paymentNo);

        verify(paymentGateway, never()).queryMiniProgramPayment(any());
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("放弃只触发关单：关单没拿到「关掉」的结论就停在关单中，绝不做任何收款判定")
                .isEqualTo("CLOSING");
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                .as("放弃支付不是取消订单")
                .isZero();
        assertThat(confirmation.collected()).isFalse();
        assertThat(openIssueCount("PAYMENT_CLOSE_CONFLICT", paymentNo)).isEqualTo(1L);
    }

    @Test
    void abandoningAPaymentClosesThePrepaySoTheOrderCannotBePaidTwice() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "resp_desc", "交易成功", "org_trans_stat", "P", "huifu_id", huifuId)));

        reconciliationService.abandonPendingPayments(orderNo, paymentNo);

        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CLOSED");
    }

    @Test
    void thePrepayCloseStopsRetryingAtTheCapAndEndsTheOrderWithAWatcher() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        // 通道一直答「还在处理中」：关单永远没有结论。
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "P", "resp_code", "00000100", "resp_desc", "交易处理中", "huifu_id", huifuId)));

        assertThat(pendingOrderService.cancel(orderNo, talentId)).containsEntry("cancellationPending", true);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CLOSING");

        // 快车道每分钟一轮。每轮都把行推回窗口内（共享库里的残留行不影响本用例：断言只按自己的流水号过滤）。
        for (int round = 0; round < 3; round++) reconcileOnce(paymentNo);

        assertThat(jdbc.queryForObject("select close_attempts from payment_transaction where payment_no = ?", Integer.class, paymentNo))
                .as("关单重试有明确上限")
                .isEqualTo(3);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("到上限后落成「取消未确认」，预支付单继续被盯着")
                .isEqualTo("CANCEL_UNCONFIRMED");
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                .as("关单中不能永远挂着：到上限时订单落成已取消，不留一笔既付不了也退不出的僵尸订单")
                .isEqualTo(2);
        assertThat(openIssueCount("PAYMENT_CLOSE_UNCONFIRMED", paymentNo))
                .as("放弃关单要留下问题记录给人工")
                .isEqualTo(1L);

        // 到上限后不再重试：再跑一轮，关单重试计数不再增长。
        reconcileOnce(paymentNo);
        assertThat(jdbc.queryForObject("select close_attempts from payment_transaction where payment_no = ?", Integer.class, paymentNo))
                .isEqualTo(3);
    }

    @Test
    void anExpiredOrderWithAClosingOrUnconfirmedAttemptIsNotCancelledByTheDeadline() {
        for (String status : List.of("CLOSING", "CANCEL_UNCONFIRMED")) {
            String orderNo = pendingOrder();
            String paymentNo = pendingPayment(orderNo, 2000L, status);
            expireNow(orderNo);

            pendingOrderService.expireDueOrders(talentId);
            assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                    .as("%s 的预支付单仍可能被支付，自动取消会把已收下的钱变成没有出口的悬账", status)
                    .isZero();

            pendingOrderService.detail(orderNo, talentId);
            assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                    .as("单个订单的到期路径同样不能取消它（%s）", status)
                    .isZero();
            assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                    .isEqualTo(status);
        }
    }

    @Test
    void theReconciliationRotatesInsteadOfAlwaysTakingTheOldestHundredUndecidedPayments() {
        String backlogOrderNo = pendingOrder();
        Long backlogOrderId = jdbc.queryForObject("select id from benefit_order where order_no = ?", Long.class, backlogOrderNo);
        // 100 条刚被处理过的历史未决流水：它们都在退避窗口内，不该再被选中。
        for (int index = 0; index < 100; index++) {
            jdbc.update("""
                    insert into payment_transaction(payment_no, provider, order_id, order_no, talent_id, merchant_id,
                                                    provider_huifu_id, amount, status, idempotency_key, created_at, last_reconciled_at)
                    values (?, 'HUIFU', ?, ?, ?, ?, ?, 2000, 'PREPAY_CREATED', ?, now() - interval 30 minute, now())
                    """, "P-BACKLOG-" + suffix + "-" + index, backlogOrderId, backlogOrderNo, talentId, merchantId, huifuId,
                    "backlog-" + suffix + "-" + index);
        }
        assertThat(count("select count(*) from payment_transaction where order_no = ?", backlogOrderNo)).isEqualTo(100);

        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        jdbc.update("update payment_transaction set created_at = now() - interval 5 minute where payment_no = ?", paymentNo);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "resp_code", "23000001", "resp_desc", "交易不存在", "huifu_id", huifuId)));

        reconciliationService.reconcile(5);

        assertThat(openIssueCount("PAYMENT_PENDING_TIMEOUT", paymentNo))
                .as("刚出现的未决流水不能被历史积压挤出处理窗口——它正是死锁唯一的自愈通道")
                .isEqualTo(1L);
        assertThat(count("select count(*) from payment_transaction where payment_no = ? and last_reconciled_at is not null", paymentNo))
                .isEqualTo(1);
    }

    @Test
    void askingForThePaymentStatusConfirmsWithTheProviderAndReportsTheCollectedPayment() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "trans_amt", "20.00",
                "huifu_id", huifuId, "org_hf_seq_id", "HF-STATUS-" + suffix)));

        Map<String, Object> result = paymentService.paymentStatus(orderNo, talentId, paymentNo);

        assertThat(result).containsEntry("paymentStatus", "PAID");
        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isEqualTo(1);
    }

    /** A pending order bound to a collecting sub-merchant, created directly so the payment seam is the only subject. */
    @Test
    void anExpiredOrderWithAnUndecidedPaymentIsNotCancelledByTheDeadlineSweep() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        expireNow(orderNo);

        pendingOrderService.expireDueOrders(talentId);

        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo))
                .as("到期不等于没收到钱：未定局的支付尝试只能交给通道确认后再处理")
                .isZero();
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("PREPAY_CREATED");
    }

    @Test
    void anExpiredOrderWhosePaymentNeverSucceededIsCancelledByTheDeadlineSweep() {
        String orderNo = expiredOrder();
        pendingPayment(orderNo, 2000L, "CREATE_FAILED");

        pendingOrderService.expireDueOrders(talentId);

        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isEqualTo(2);
    }

    @Test
    void anExpiredOrderWithoutAnyPaymentIsCancelledByTheDeadlineSweep() {
        String orderNo = expiredOrder();

        pendingOrderService.expireDueOrders(talentId);

        assertThat(jdbc.queryForObject("select order_status from benefit_order where order_no = ?", Integer.class, orderNo)).isEqualTo(2);
    }

    @Test
    void theReconciliationClosesAPrepayThatCouldNotBeClosedWhenTheOrderWasCancelled() {
        String paymentNo = cancelledOrderWithUnconfirmedClose();
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "resp_desc", "交易成功", "huifu_id", huifuId)));

        reconciliationService.reconcile(5);

        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .as("过了关单限制之后必须把预支付单关掉，否则用户手上那张单一直能付")
                .isEqualTo("CLOSED");
        assertThat(openIssueCount("PAYMENT_CLOSE_UNCONFIRMED", paymentNo)).isZero();
    }

    @Test
    void theReconciliationNeverClosesAPrepayTheProviderAlreadyCollected() {
        String paymentNo = cancelledOrderWithUnconfirmedClose();
        clearInvocations(paymentGateway);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "S", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId,
                "org_hf_seq_id", "HF-LATE-" + suffix)));

        reconciliationService.reconcile(5);

        verify(paymentGateway, never()).closeMiniProgramPayment(any());
        assertThat(openIssueCount("PAYMENT_ORDER_CANCELLED", paymentNo))
                .as("取消之后才收到的钱要留在账目上让人处理，而不是当成没收过")
                .isEqualTo(1L);
        assertThat(openIssueCount("PAYMENT_CLOSE_UNCONFIRMED", paymentNo)).isZero();
    }

    /** 订单刚创建就取消：通道不允许关 1 分钟以内的订单，取消成立，但预支付单留着等下一轮关。 */
    private String cancelledOrderWithUnconfirmedClose() {
        String orderNo = pendingOrder();
        String paymentNo = pendingPayment(orderNo, 2000L);
        when(paymentGateway.queryMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "I", "resp_code", "00000000", "trans_amt", "20.00", "huifu_id", huifuId)));
        when(paymentGateway.closeMiniProgramPayment(any())).thenReturn(providerResponse(Map.of(
                "trans_stat", "F", "resp_code", "20000001", "resp_desc", "不允许关闭一分钟以内的订单", "huifu_id", huifuId)));

        pendingOrderService.cancel(orderNo, talentId);
        assertThat(jdbc.queryForObject("select status from payment_transaction where payment_no = ?", String.class, paymentNo))
                .isEqualTo("CANCEL_UNCONFIRMED");
        // 对账只处理挂起超过 N 分钟的流水，把它推到窗口内。
        jdbc.update("update payment_transaction set created_at = now() - interval 10 minute where payment_no = ?", paymentNo);
        return paymentNo;
    }

    private String pendingOrder() {

        return pendingOrder("date_add(now(), interval 30 minute)");
    }

    /** The same order, already past its payment deadline. */
    private String expiredOrder() {
        return pendingOrder("date_sub(now(), interval 1 minute)");
    }

    private String pendingOrder(String expiresAtSql) {
        String orderNo = "PENDING-" + suffix + "-" + (++orderSequence);
        jdbc.update("""
                insert into benefit_order(order_no, talent_id, merchant_id, order_type, order_amount, base_pay_amount,
                                          pay_amount, order_status, verify_status, expires_at, price_version, collect_huifu_id,
                                          product_name_snapshot)
                values (?, ?, ?, 'STORE_CONSUME', ?, ?, ?, 0, 0, %s, 1, ?, ?)
                """.formatted(expiresAtSql), orderNo, talentId, merchantId, 2000L, 2000L, 2000L, huifuId, "待支付测试订单");
        return orderNo;
    }

    private String pendingPayment(String orderNo, long amount) {
        return pendingPayment(orderNo, amount, "PREPAY_CREATED");
    }

    private String pendingPayment(String orderNo, long amount, String status) {
        return pendingPayment(orderNo, amount, status, null);
    }

    private String pendingPayment(String orderNo, long amount, String status, String clientPaymentPayload) {
        String paymentNo = "P-PENDING-" + suffix + "-" + orderSequence;
        Long orderId = jdbc.queryForObject("select id from benefit_order where order_no = ?", Long.class, orderNo);
        jdbc.update("""
                insert into payment_transaction(payment_no, provider, order_id, order_no, talent_id, merchant_id,
                                                provider_huifu_id, amount, status, idempotency_key, created_at,
                                                client_payment_payload)
                values (?, 'HUIFU', ?, ?, ?, ?, ?, ?, ?, ?, now(), ?)
                """, paymentNo, orderId, orderNo, talentId, merchantId, huifuId, amount, status,
                "pending-payment-" + paymentNo, clientPaymentPayload);
        return paymentNo;
    }

    /** 把这一行推回对账窗口再跑一轮；断言只按本用例自己的流水号过滤，共享库里的残留行不影响结论。 */
    private void reconcileOnce(String paymentNo) {
        jdbc.update("""
                update payment_transaction set created_at = now() - interval 5 minute, last_reconciled_at = null
                where payment_no = ?
                """, paymentNo);
        reconciliationService.reconcile(5);
    }

    /**
     * 让订单现在就算过期。
     *
     * <p>必须建完订单与支付尝试之后再推时间：到期清理任务（{@code PendingOrderExpiryScheduler}，测试里
     * 也开着）会在「订单已过期、但还没有任何支付流水」的那一瞬间把它当成普通超时订单取消掉。
     */
    private void expireNow(String orderNo) {
        jdbc.update("update benefit_order set expires_at = now() - interval 1 minute where order_no = ?", orderNo);
    }

    private Map<String, Object> providerResponse(Map<String, Object> data) {
        return Map.of("data", data);
    }

    private Long openIssueCount(String type, String businessNo) {
        return jdbc.queryForObject("""
                select count(*) from payment_reconciliation_issue
                where issue_type = ? and business_no = ? and status = 'open'
                """, Long.class, type, businessNo);
    }

    private long count(String sql, Object... args) {
        Long value = jdbc.queryForObject(sql, Long.class, args);
        return value == null ? 0 : value;
    }
}