baiyue-firstleg/finance/src/test/java/com/baiyue/finance/BusinessFeeSourceReceptionPersistenceTest.java

package com.baiyue.finance;

import com.baiyue.common.event.DeliveryAck;
import com.baiyue.common.event.BusinessFeeSourceChangedV1;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.common.event.EventPayloadCanonicalizer;
import com.baiyue.common.event.EventPayloadHash;
import com.baiyue.common.event.EventTypes;
import com.baiyue.finance.client.WarehouseSupplierClient.SupplierSnapshot;
import com.baiyue.finance.mapper.BusinessFeeSourceMapper;
import com.baiyue.finance.repository.BusinessFeeSourceRepository;
import com.baiyue.finance.service.BusinessFeeSourceEventService;
import com.baiyue.finance.service.FinanceEventException;
import com.baiyue.finance.service.FinanceSupplierValidator;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.ibatis.session.Configuration;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import org.mybatis.spring.SqlSessionFactoryBean;
import org.mybatis.spring.SqlSessionTemplate;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.datasource.DataSourceTransactionManager;
import org.springframework.jdbc.datasource.DriverManagerDataSource;
import org.springframework.transaction.support.TransactionTemplate;

import java.math.BigDecimal;
import java.time.Instant;
import java.util.UUID;

import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.mockito.ArgumentMatchers.*;
import static org.mockito.Mockito.*;

/** 隔离数据库执行真实 Mapper 和接收事务，验证失败不留下收件箱或费用半成品。 */
class BusinessFeeSourceReceptionPersistenceTest {
    private final ObjectMapper json = new ObjectMapper().findAndRegisterModules();
    private final FinanceSupplierValidator suppliers = mock(FinanceSupplierValidator.class);
    private JdbcTemplate jdbc;
    private TransactionTemplate transaction;
    private BusinessFeeSourceEventService service;

    /** MyBatis 与既有接收写入共用 Spring 事务连接，不连接共享 TEST 数据库。 */
    @BeforeEach
    void setUp() throws Exception {
        var data = new DriverManagerDataSource("jdbc:h2:mem:fee-reception-" + UUID.randomUUID()
            + ";MODE=MySQL;DB_CLOSE_DELAY=-1", "sa", "");
        jdbc = new JdbcTemplate(data);
        jdbc.execute("""
            CREATE TABLE fin_fee_source(id BIGINT AUTO_INCREMENT PRIMARY KEY,
              source_system VARCHAR(32),source_type VARCHAR(32),source_fee_id BIGINT,source_fee_version INT,
              customer_id BIGINT,source_no VARCHAR(64),business_type VARCHAR(32),business_id BIGINT,
              business_no VARCHAR(128),inbound_code VARCHAR(64),warehouse_id BIGINT,direction VARCHAR(16),
              order_id BIGINT,inbound_id BIGINT,container_id BIGINT,container_booking_id BIGINT,container_no VARCHAR(64),
              fee_item_code VARCHAR(64),fee_item_name VARCHAR(200),amount DECIMAL(20,6),currency_code VARCHAR(3),
              settlement_mode VARCHAR(16),paid_at TIMESTAMP,payment_method VARCHAR(32),payment_reference_no VARCHAR(128),
              payment_voucher_file_ids VARCHAR(2000),source_hash VARCHAR(71),source_snapshot VARCHAR(10000),
              review_status VARCHAR(16),status VARCHAR(16),wallet_hold_id BIGINT,operation_version INT DEFAULT 0,
              UNIQUE(source_system,source_type,source_fee_id,source_fee_version))
            """);
        jdbc.execute("""
            CREATE TABLE fin_inbox_event(event_id VARCHAR(128),source_system VARCHAR(32),event_type VARCHAR(64),
              payload VARCHAR(15000),payload_hash VARCHAR(71),status VARCHAR(16),result_resource_id BIGINT,
              processed_at TIMESTAMP,error_code VARCHAR(64),PRIMARY KEY(source_system,event_id))
            """);
        jdbc.execute("""
            CREATE TABLE fin_cost(id BIGINT AUTO_INCREMENT PRIMARY KEY,tenant_id BIGINT,cost_no VARCHAR(200),
              cost_type VARCHAR(32),cost_origin VARCHAR(32),source_system VARCHAR(32),source_type VARCHAR(32),
              source_fee_id BIGINT,source_fee_version INT,source_hash VARCHAR(71),order_id BIGINT,inbound_id BIGINT,
              container_id BIGINT,warehouse_id BIGINT,supplier_id BIGINT,supplier_name_snapshot VARCHAR(128),
              fee_item_code VARCHAR(64),fee_item_name VARCHAR(200),currency_code VARCHAR(3),cost_amount DECIMAL(20,6),
              voucher_file_ids VARCHAR(2000),paid_at TIMESTAMP,payment_method VARCHAR(32),payment_reference_no VARCHAR(128),
              status VARCHAR(16),confirmed_at TIMESTAMP,created_by BIGINT,created_by_name VARCHAR(64),updated_by BIGINT,
              voided_at TIMESTAMP,void_reason VARCHAR(200),version INT DEFAULT 0,
              UNIQUE(source_system,source_type,source_fee_id,source_fee_version))
            """);
        jdbc.execute("CREATE TABLE fin_bill(id BIGINT,tenant_id BIGINT,status VARCHAR(16))");
        jdbc.execute("CREATE TABLE fin_bill_line(bill_id BIGINT,tenant_id BIGINT,source_id BIGINT,source_version INT)");
        jdbc.execute("CREATE TABLE fin_settlement(id BIGINT,tenant_id BIGINT,status VARCHAR(16))");
        jdbc.execute("CREATE TABLE fin_settlement_line(settlement_id BIGINT,tenant_id BIGINT,fee_source_id BIGINT)");
        var configuration = new Configuration();
        configuration.addMapper(BusinessFeeSourceMapper.class);
        var factory = new SqlSessionFactoryBean();
        factory.setDataSource(data);
        factory.setConfiguration(configuration);
        var mapper = new SqlSessionTemplate(factory.getObject()).getMapper(BusinessFeeSourceMapper.class);
        service = new BusinessFeeSourceEventService(jdbc, json, suppliers, new BusinessFeeSourceRepository(mapper));
        transaction = new TransactionTemplate(new DataSourceTransactionManager(data));
        when(suppliers.requireActive(71L, null)).thenReturn(new SupplierSnapshot(71L, "供应商", null, "ACTIVE"));
    }

    /** 释放每个用例独占的内存数据库。 */
    @AfterEach
    void tearDown() {
        if (jdbc != null) jdbc.execute("SHUTDOWN");
    }

    /** 已提交后模拟丢失回执，供应商停用或故障都不产生第二笔费用和成本。 */
    @ParameterizedTest
    @ValueSource(strings = {"FIN_409", "FIN_503"})
    void committedDirectPaidReplayKeepsExactlyOneSourceAndCost(String errorCode) {
        var event = event(701, 1, 10001, "DIRECT_PAID", "CNY", "12.00", "1", "12.00");
        assertThat(consume(event).outcome()).isEqualTo(DeliveryAck.Outcome.APPLIED);
        reset(suppliers);
        when(suppliers.requireActive(71L, null)).thenThrow(new FinanceEventException(errorCode, "供应商不可用"));

        assertThat(consume(event).outcome()).isEqualTo(DeliveryAck.Outcome.DUPLICATE);
        verifyNoInteractions(suppliers);
        assertCounts(1, 1, 1);
        assertThat(jdbc.queryForObject("SELECT status FROM fin_inbox_event", String.class)).isEqualTo("PROCESSED");
        assertThat(jdbc.queryForObject("SELECT cost_amount FROM fin_cost", BigDecimal.class)).isEqualByComparingTo("12.00");
    }

    /** 不同客户的升版失败时回滚已插入 Inbox，原费用与原直付成本保持有效。 */
    @Test
    void foreignCustomerRevisionRollsBackAndSameCustomerCanStillAdjust() {
        consume(event(701, 1, 10001, "DIRECT_PAID", "CNY", "12.00", "1", "12.00"));
        var wrong = event(701, 2, 10002, "DIRECT_PAID", "CNY", "18.00", "1", "18.00");
        assertThatThrownBy(() -> consume(wrong)).isInstanceOfSatisfying(FinanceEventException.class,
            error -> assertThat(error.code()).isEqualTo("FIN-EVT-018"));
        assertCounts(1, 1, 1);
        assertThat(jdbc.queryForObject("SELECT status FROM fin_fee_source", String.class)).isEqualTo("AVAILABLE");
        assertThat(jdbc.queryForObject("SELECT status FROM fin_cost", String.class)).isEqualTo("CONFIRMED");
        assertThat(jdbc.queryForObject("SELECT customer_id FROM fin_fee_source", Long.class)).isEqualTo(10001L);

        consume(event(701, 2, 10001, "DIRECT_PAID", "CNY", "18.00", "1", "18.00"));
        assertCounts(2, 2, 2);
        assertThat(jdbc.queryForList("SELECT status FROM fin_fee_source ORDER BY source_fee_version", String.class))
            .containsExactly("SUPERSEDED", "AVAILABLE");
        assertThat(jdbc.queryForList("SELECT status FROM fin_cost ORDER BY source_fee_version", String.class))
            .containsExactly("VOIDED", "CONFIRMED");
    }

    /** 错误换算拒收后不残留收件记录，已存在版本也不会被替代。 */
    @Test
    void invalidConversionLeavesOriginalAmountsAndReceiptIntact() {
        consume(event(701, 1, 10001, "ON_ACCOUNT", "CNY", "12.00", "1", "12.00"));
        assertThatThrownBy(() -> consume(event(701, 2, 10001, "ON_ACCOUNT", "USD", "1.00", "8", "120.00")))
            .isInstanceOfSatisfying(FinanceEventException.class,
                error -> assertThat(error.code()).isEqualTo("FIN-EVT-019"));
        assertCounts(1, 1, 0);
        assertThat(jdbc.queryForObject("SELECT amount FROM fin_fee_source", BigDecimal.class)).isEqualByComparingTo("12.00");
        assertThat(jdbc.queryForObject("SELECT status FROM fin_fee_source", String.class)).isEqualTo("AVAILABLE");
    }

    /** 只锁定相同系统、来源类型和费用号，其他客户的独立费用不会阻塞正常升版。 */
    @Test
    void independentSourcesKeepSeparateCustomerIdentity() {
        consume(event(701, 1, 10001, "ON_ACCOUNT", "CNY", "12.00", "1", "12.00"));
        consume(event(702, 1, 10002, "ON_ACCOUNT", "CNY", "9.00", "1", "9.00"));
        consume(event(701, 2, 10001, "ON_ACCOUNT", "CNY", "18.00", "1", "18.00"));
        assertCounts(3, 3, 0);
        assertThat(jdbc.queryForObject("SELECT customer_id FROM fin_fee_source WHERE source_fee_id=702", Long.class))
            .isEqualTo(10002L);
        assertThat(jdbc.queryForObject("SELECT status FROM fin_fee_source WHERE source_fee_id=702", String.class))
            .isEqualTo("AVAILABLE");
    }

    /** 提前恢复幂等结果不放宽事件冲突校验，异载荷仍须拒绝。 */
    @Test
    void changedPayloadCannotUseProcessedEventIdentity() {
        var first = event(701, 1, 10001, "ON_ACCOUNT", "CNY", "12.00", "1", "12.00");
        consume(first);
        var changed = event(701, 1, 10001, "ON_ACCOUNT", "CNY", "18.00", "1", "18.00");
        var conflict = new EventEnvelope(first.eventId(), changed.eventType(), changed.eventVersion(), changed.occurredAt(),
            changed.tenantId(), changed.aggregateType(), changed.aggregateId(), changed.aggregateVersion(), null,
            changed.traceId(), changed.correlationId(), null, changed.payload(), changed.payloadHash());
        reset(suppliers);
        assertThatThrownBy(() -> consume(conflict)).isInstanceOfSatisfying(FinanceEventException.class,
            error -> assertThat(error.code()).isEqualTo("FIN-EVT-002"));
        verifyNoInteractions(suppliers);
        assertCounts(1, 1, 0);
    }

    /** 即使失败发生在费用插入之后，收件箱、费用及成本也必须一起回滚。 */
    @Test
    void costWriteFailureRollsBackReceiptAndSource() {
        jdbc.execute("ALTER TABLE fin_cost ADD CONSTRAINT test_cost_limit CHECK(cost_amount <= 10)");
        assertThatThrownBy(() -> consume(event(701, 1, 10001, "DIRECT_PAID", "CNY", "12.00", "1", "12.00")))
            .isInstanceOf(org.springframework.dao.DataIntegrityViolationException.class);
        assertCounts(0, 0, 0);
    }

    /** 以正式服务事务边界执行，Mapper 与遗留 JDBC 写入必须共进退。 */
    private DeliveryAck consume(EventEnvelope event) {
        return transaction.execute(status -> service.consume(event));
    }

    /** 校验整笔接收事实数量，防止幂等通过但产生额外财务明细。 */
    private void assertCounts(int inbox, int sources, int costs) {
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fin_inbox_event", Integer.class)).isEqualTo(inbox);
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fin_fee_source", Integer.class)).isEqualTo(sources);
        assertThat(jdbc.queryForObject("SELECT COUNT(*) FROM fin_cost", Integer.class)).isEqualTo(costs);
    }

    /** 构造经过两个正式摘要计算的费用快照，金额错误用例不会被摘要错误提前拦截。 */
    private EventEnvelope event(long feeId, int version, long customer, String mode, String currency,
                                String original, String rate, String total) {
        ObjectNode payload = json.createObjectNode();
        payload.putNull("action");
        payload.put("sourceSystem", "LASTLEG").put("sourceType", "OVERSEAS_WAREHOUSE_FEE")
            .put("sourceFeeId", String.valueOf(feeId)).put("sourceVersion", version)
            .put("customerId", String.valueOf(customer)).put("customerName", "验收客户")
            .put("sourceNo", "ZY-RECEPTION").put("businessType", "WAREHOUSE_SERVICE")
            .put("businessId", "801").put("businessNo", "ZY-RECEPTION");
        for (String field : new String[]{"orderId", "inboundId", "inboundCode", "containerId", "containerBookingId", "containerNo"})
            payload.putNull(field);
        payload.put("warehouseId", "91").put("feeItemCode", "FEE_TYPE_003").put("feeItemName", "人工费")
            .put("direction", "PAYABLE").put("supplierId", "71").put("supplierName", "供应商");
        payload.putNull("supplierCategory"); payload.put("settlementMode", mode);
        if ("DIRECT_PAID".equals(mode)) {
            payload.set("paidAt", json.valueToTree(Instant.parse("2026-10-27T01:00:00Z")));
            payload.put("paymentMethod", "BANK_TRANSFER").put("paymentReferenceNo", "TEST-RECEPTION");
        } else {
            payload.putNull("paidAt"); payload.putNull("paymentMethod"); payload.putNull("paymentReferenceNo");
        }
        payload.putArray("paymentVoucherFileIds");
        payload.put("businessDate", "2026-10-27").put("currency", "CNY").put("totalAmount", new BigDecimal(total))
            .put("originalCurrency", currency).put("originalAmount", new BigDecimal(original))
            .put("exchangeRate", new BigDecimal(rate)).put("exchangeRateMonth", "2026-10");
        for (String field : new String[]{"chargeableQuantity", "quantityUnit", "unitPrice", "periodStart", "periodEnd"})
            payload.putNull(field);
        payload.put("reviewStatus", "APPROVED").put("publicationStatus", "PUBLISHED").put("sourceStatus", "ACTIVE");
        payload.putNull("sourceHash"); payload.putNull("walletSettlement");
        // 使用正式 DTO 规范化日期等类型，和生产者的序列化输入保持一致。
        payload = json.valueToTree(json.convertValue(payload, BusinessFeeSourceChangedV1.class));
        payload.put("sourceHash", EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload)));
        payload.put("action", version == 1 ? "PUBLISHED" : "ADJUSTED");
        return new EventEnvelope(UUID.randomUUID().toString(), EventTypes.BUSINESS_FEE_SOURCE_CHANGED, 1,
            Instant.parse("2026-10-27T01:00:00Z"), String.valueOf(customer), "BUSINESS_FEE", String.valueOf(feeId), version,
            null, UUID.randomUUID().toString(), "TEST:FEE:" + feeId + ':' + version, null, payload,
            EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload)));
    }
}