fix(finance): preserve container identity and source version order

2026-09-09 10:14:42 +08:00
parent 4379b721bc
commit 7e2626513e
4 changed files with 113 additions and 2 deletions
@@ -81,6 +81,11 @@ public class BusinessFeeSourceEventService {
        }
        if ("VOIDED".equals(source.action()))
            throw error("FIN-EVT-005", "作废事件对应的费用来源版本不存在");
+        List<Map<String, Object>> versions = jdbc.queryForList(
+            "SELECT source_fee_version FROM fin_fee_source WHERE source_system=? AND source_type=? AND source_fee_id=? FOR UPDATE",
+            system, source.sourceType(), feeId);
+        if (versions.stream().anyMatch(row -> ((Number) row.get("source_fee_version")).intValue() > source.sourceVersion()))
+            throw error("FIN-EVT-008", "费用来源版本已过期");
        requireSourceNotFrozen(system, source.sourceType(), feeId);
        jdbc.update("UPDATE fin_fee_source SET status='SUPERSEDED' WHERE source_system=? AND source_type=? AND source_fee_id=? AND status='AVAILABLE'",
            system, source.sourceType(), feeId);
@@ -115,6 +120,11 @@ public class BusinessFeeSourceEventService {
        }
        positive(value.sourceFeeId(), "sourceFeeId");
        positive(value.customerId(), "customerId");
+        if ("OVERSEAS_CONTAINER_FEE".equals(value.sourceType())) {
+            if (!"LASTLEG".equals(value.sourceSystem()) || !"OVERSEAS_CONTAINER".equals(value.businessType()))
+                throw error("FIN-EVT-012", "海外柜费来源不一致");
+            positive(value.containerBookingId(), "containerBookingId");
+        }
        if (!"VOIDED".equals(value.action()) && (!"PUBLISHED".equals(value.publicationStatus())
            || !"ACTIVE".equals(value.sourceStatus()))) {
            throw error("FIN-EVT-015", "费用来源尚未发布或已失效");
@@ -125,7 +135,7 @@ public class BusinessFeeSourceEventService {
            if (blank(value.supplierName()) || !value.supplierName().equals(supplier.name()))
                throw error("FIN-EVT-012", "应付来源供应商快照不一致");
            validateSettlement(value);
-        } else if (!blank(value.supplierId()) || !blank(value.supplierName())
+        } else if (!blank(value.supplierId()) || !blank(value.supplierName()) || !blank(value.supplierCategory())
            || !blank(value.settlementMode()) || value.paidAt() != null
            || !blank(value.paymentMethod()) || !blank(value.paymentReferenceNo())
            || (value.paymentVoucherFileIds() != null && !value.paymentVoucherFileIds().isEmpty())) {
@@ -169,7 +179,7 @@ public class BusinessFeeSourceEventService {
        jdbc.update("INSERT INTO fin_cost(tenant_id,cost_no,cost_type,cost_origin,source_system,source_type,source_fee_id,source_fee_version,source_hash,order_id,inbound_id,container_id,warehouse_id,supplier_id,supplier_name_snapshot,fee_item_code,fee_item_name,currency_code,cost_amount,voucher_file_ids,paid_at,payment_method,payment_reference_no,status,confirmed_at,created_by,created_by_name,updated_by) VALUES(?,?,'SOURCE_DIRECT_PAID','BUSINESS_DIRECT_PAID',?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'CONFIRMED',CURRENT_TIMESTAMP(3),0,'系统',0)",
            positive(value.customerId(), "customerId"), "DIRECT-" + system + '-' + value.sourceType() + '-' + feeId + "-V" + value.sourceVersion(),
            system, value.sourceType(), feeId, value.sourceVersion(), value.sourceHash(), number(value.orderId()),
-            number(value.inboundId()), number(value.containerId()), number(value.warehouseId()), positive(value.supplierId(), "supplierId"),
+            number(value.inboundId()), number(value.containerBookingId()), number(value.warehouseId()), positive(value.supplierId(), "supplierId"),
            value.supplierName(), value.feeItemCode(), value.feeItemName(), value.currency(), value.totalAmount(),
            json(value.paymentVoucherFileIds()), value.paidAt(), value.paymentMethod(), value.paymentReferenceNo());
    }
@@ -81,6 +81,12 @@ public class FinanceFeeSourceEventService {
        if ("FIRSTLEG_FEE_VOIDED".equals(event.eventType())) {
            throw error("FIN-EVT-005", "作废事件对应的费用来源版本不存在");
        }
+        List<Map<String, Object>> versions = jdbc.queryForList(
+            "SELECT source_fee_version FROM fin_fee_source WHERE source_system='FIRSTLEG' AND source_type=? AND source_fee_id=? FOR UPDATE",
+            snapshot.sourceType(), snapshot.feeId());
+        if (versions.stream().anyMatch(row -> ((Number) row.get("source_fee_version")).intValue() > snapshot.feeVersion())) {
+            throw error("FIN-EVT-008", "费用来源版本已过期");
+        }
        requireSourceNotFrozen(snapshot.sourceType(), snapshot.feeId());
        jdbc.update("UPDATE fin_fee_source SET status='SUPERSEDED' WHERE source_system='FIRSTLEG' AND source_type=? AND source_fee_id=? AND status='AVAILABLE'",
            snapshot.sourceType(), snapshot.feeId());
@@ -18,6 +18,10 @@ import java.math.BigDecimal;
 import java.time.Instant;
 import java.time.LocalDate;
 import java.util.List;
+import java.util.Map;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.mockito.Mockito.never;

 import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.ArgumentMatchers.any;
@@ -82,4 +86,82 @@ class BusinessFeeSourceEventServiceTest {
            null, "123e4567-e89b-12d3-a456-426614174702", "LASTLEG:FEE:701:1", null,
            node, payloadHash);
    }
+
+    @Test
+    void overseasDirectPaidCostUsesPhysicalContainerAndKeepsSourceIdentity() {
+        EventEnvelope event = containerEvent("CNY", "61", "PUBLISHED", 1);
+        when(suppliers.requireActive(71L, null)).thenReturn(
+            new com.baiyue.finance.client.WarehouseSupplierClient.SupplierSnapshot(71L, "供应商", null, "ACTIVE"));
+        when(jdbc.queryForObject(startsWith("SELECT id FROM fin_fee_source"), eq(Long.class),
+            eq("LASTLEG"), eq("OVERSEAS_CONTAINER_FEE"), eq(701L), eq(1))).thenReturn(901L);
+        assertThat(service.consume(event)).isEqualTo(901L);
+        ArgumentCaptor<Object[]> source = ArgumentCaptor.forClass(Object[].class);
+        verify(jdbc).update(startsWith("INSERT INTO fin_fee_source"), source.capture());
+        assertThat(source.getValue()[0]).isEqualTo("LASTLEG");
+        assertThat(source.getValue()[4]).isEqualTo("OVERSEAS_CONTAINER_FEE");
+        assertThat(source.getValue()[14]).isEqualTo(41L);
+        assertThat(source.getValue()[15]).isEqualTo(61L);
+        ArgumentCaptor<Object[]> cost = ArgumentCaptor.forClass(Object[].class);
+        verify(jdbc).update(startsWith("INSERT INTO fin_cost"), cost.capture());
+        assertThat(cost.getValue()[9]).isEqualTo(61L);
+    }
+
+    @Test
+    void rejectsForeignCurrencyAndMissingPhysicalContainerBeforeWriting() {
+        assertThatThrownBy(() -> service.consume(containerEvent("USD", "61", "PUBLISHED", 1)))
+            .isInstanceOf(com.baiyue.finance.service.FinanceEventException.class);
+        assertThatThrownBy(() -> service.consume(containerEvent("CNY", null, "PUBLISHED", 1)))
+            .isInstanceOf(com.baiyue.finance.service.FinanceEventException.class);
+        verify(jdbc, never()).update(anyString(), any(Object[].class));
+    }
+
+    @Test
+    void lateOlderVersionCannotSupersedeNewerSourceOrCost() {
+        when(suppliers.requireActive(71L, null)).thenReturn(
+            new com.baiyue.finance.client.WarehouseSupplierClient.SupplierSnapshot(71L, "供应商", null, "ACTIVE"));
+        when(jdbc.queryForList(startsWith("SELECT source_fee_version"), eq("LASTLEG"),
+            eq("OVERSEAS_CONTAINER_FEE"), eq(701L)))
+            .thenReturn(List.of(Map.of("source_fee_version", 2)));
+        assertThatThrownBy(() -> service.consume(containerEvent("CNY", "61", "PUBLISHED", 1)))
+            .hasMessageContaining("版本已过期");
+        verify(jdbc, never()).update(startsWith("UPDATE fin_fee_source"), any(Object[].class));
+        verify(jdbc, never()).update(startsWith("UPDATE fin_cost"), any(Object[].class));
+    }
+
+    @Test
+    void replayReturnsOriginalSourceWithoutDuplicatingCost() {
+        EventEnvelope event = containerEvent("CNY", "61", "PUBLISHED", 1);
+        when(suppliers.requireActive(71L, null)).thenReturn(
+            new com.baiyue.finance.client.WarehouseSupplierClient.SupplierSnapshot(71L, "供应商", null, "ACTIVE"));
+        when(jdbc.queryForList(startsWith("SELECT payload_hash,status"), eq("LASTLEG"), eq(event.eventId())))
+            .thenReturn(List.of(Map.of("payload_hash", event.payloadHash(), "status", "PROCESSED", "result_resource_id", 901L)));
+        assertThat(service.consume(event)).isEqualTo(901L);
+        verify(jdbc, never()).update(anyString(), any(Object[].class));
+    }
+
+    private EventEnvelope containerEvent(String currency, String containerBookingId, String action, int version) {
+        EventEnvelope base = eventWithoutInbound();
+        ObjectNode payload = base.payload().deepCopy();
+        payload.put("sourceType", "OVERSEAS_CONTAINER_FEE");
+        payload.put("businessType", "OVERSEAS_CONTAINER");
+        payload.put("containerId", "41");
+        payload.put("containerBookingId", containerBookingId);
+        payload.put("currency", currency);
+        payload.put("direction", "PAYABLE");
+        payload.put("supplierId", "71");
+        payload.put("supplierName", "供应商");
+        payload.put("settlementMode", "DIRECT_PAID");
+        payload.set("paidAt", json.valueToTree(Instant.parse("2026-09-02T08:00:00Z")));
+        payload.put("paymentMethod", "BANK_TRANSFER");
+        payload.put("paymentReferenceNo", "PAY-1");
+        payload.put("sourceVersion", version);
+        payload.putNull("action");
+        payload.putNull("sourceHash");
+        payload.put("sourceHash", EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload)));
+        payload.put("action", action);
+        return new EventEnvelope(base.eventId(), base.eventType(), 1, base.occurredAt(),
+            RegionCode.OVERSEAS, RegionCode.CN, "10001", "BUSINESS_FEE", "701", version,
+            null, base.traceId(), "LASTLEG:FEE:701:" + version, null, payload,
+            EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload)));
+    }
 }
@@ -284,6 +284,19 @@ class FinanceFeeSourceEventServiceTest {
        when(jdbc.queryForObject(startsWith("SELECT id FROM fin_fee_source"), eq(Long.class), anyString(), eq(31L), anyInt())).thenReturn(101L);
    }

+    @Test
+    void lateOlderVersionDoesNotReplaceNewerSource() {
+        givenNewEvent();
+        when(firstleg.container("test-token", 41, 31, 1)).thenReturn(
+            ApiResponse.success(snapshot("CONTAINER_FEE", 7, 1, null, 41L)));
+        when(jdbc.queryForList(startsWith("SELECT source_fee_version"), eq("CONTAINER_FEE"), eq(31L)))
+            .thenReturn(List.of(Map.of("source_fee_version", 2)));
+        assertThatThrownBy(() -> service.consume(containerEvent("late-version", "FIRSTLEG_FEE_PUBLISHED", 1, 7)))
+            .hasMessageContaining("版本已过期");
+        verify(jdbc, never()).update(startsWith("UPDATE fin_fee_source"), any(Object[].class));
+        verify(jdbc, never()).update(startsWith("UPDATE fin_cost"), any(Object[].class));
+    }
+
    private FinanceFeeSourceEvent inboundEvent(String id, String type, int version, long tenant) {
        return event(id, type, version, tenant, new FinanceFeeSourceEvent.Payload("21", null, "31", version));
    }