This commit is contained in:
2026-09-02 16:17:41 +08:00
parent 12fbf2aa67
commit 03d04431f2
33 changed files with 1367 additions and 22 deletions
@@ -0,0 +1,38 @@
package com.baiyue.finance.controller;
import com.baiyue.common.core.api.ApiResponse;
import com.baiyue.common.event.EventEnvelope;
import com.baiyue.finance.service.BusinessFeeSourceEventService;
import com.baiyue.finance.service.FinanceEventException;
import lombok.RequiredArgsConstructor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestHeader;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import java.nio.charset.StandardCharsets;
import java.security.MessageDigest;
import java.util.Map;
/** Receives full immutable fee snapshots from cross-region business producers. */
@RestController
@RequiredArgsConstructor
@RequestMapping("/internal/finance/v1/business-fee-source-events")
public class InternalBusinessFeeSourceController {
private final BusinessFeeSourceEventService service;
@Value("${baiyue.security.internal-service-token}") private String expectedToken;
@PostMapping
public ApiResponse<Map<String, Long>> consume(
@RequestHeader(value = "X-Internal-Service-Token", required = false) String token,
@RequestBody EventEnvelope event) {
if (token == null || expectedToken == null || expectedToken.isBlank()
|| !MessageDigest.isEqual(token.getBytes(StandardCharsets.UTF_8),
expectedToken.getBytes(StandardCharsets.UTF_8))) {
throw new FinanceEventException("FIN-EVT-401", "内部服务令牌无效");
}
return FinanceResponses.success(Map.of("feeSourceId", service.consume(event)));
}
}
@@ -0,0 +1,199 @@
package com.baiyue.finance.service;
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.common.event.RegionCode;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.RequiredArgsConstructor;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import java.util.List;
import java.util.Map;
import java.util.Set;
/** Consumes complete overseas business fee snapshots without a cross-region synchronous callback. */
@Service
@RequiredArgsConstructor
public class BusinessFeeSourceEventService {
private static final Set<String> ACTIONS = Set.of("PUBLISHED", "ADJUSTED", "VOIDED");
private static final Set<String> DIRECTIONS = Set.of("RECEIVABLE", "PAYABLE");
private final JdbcTemplate jdbc;
private final ObjectMapper objectMapper;
private final FinanceSupplierValidator supplierValidator;
@Transactional
public long consume(EventEnvelope envelope) {
try {
envelope.requireValid();
} catch (RuntimeException exception) {
throw error("FIN-EVT-001", "跨区费用事件信封无效");
}
if (!EventTypes.BUSINESS_FEE_SOURCE_CHANGED.equals(envelope.eventType())
|| envelope.eventVersion() != 1 || envelope.sourceRegion() != RegionCode.OVERSEAS
|| envelope.targetRegion() != RegionCode.CN) {
throw error("FIN-EVT-001", "跨区费用事件类型、版本或区域无效");
}
BusinessFeeSourceChangedV1 source = payload(envelope);
validate(envelope, source);
String sourceSystem = source.sourceSystem().trim().toUpperCase();
String eventJson = json(envelope);
List<Map<String, Object>> inbox = jdbc.queryForList(
"SELECT payload_hash,status,result_resource_id FROM fin_inbox_event WHERE source_system=? AND event_id=? FOR UPDATE",
sourceSystem, envelope.eventId());
if (!inbox.isEmpty()) {
Map<String, Object> row = inbox.get(0);
if (!envelope.payloadHash().equals(row.get("payload_hash")))
throw error("FIN-EVT-002", "同一事件ID对应不同费用快照");
if ("PROCESSED".equals(row.get("status")) && row.get("result_resource_id") instanceof Number id)
return id.longValue();
throw error("FIN-EVT-003", "相同事件正在处理或上次处理失败");
}
jdbc.update("INSERT INTO fin_inbox_event(event_id,source_system,event_type,payload,payload_hash,status) VALUES (?,?,?,?,?,'RECEIVED')",
envelope.eventId(), sourceSystem, envelope.eventType(), eventJson, envelope.payloadHash());
long result = apply(sourceSystem, source);
jdbc.update("UPDATE fin_inbox_event SET status='PROCESSED',result_resource_id=?,processed_at=CURRENT_TIMESTAMP(3),error_code=NULL WHERE source_system=? AND event_id=?",
result, sourceSystem, envelope.eventId());
return result;
}
private long apply(String system, BusinessFeeSourceChangedV1 source) {
long feeId = positive(source.sourceFeeId(), "sourceFeeId");
List<Map<String, Object>> same = jdbc.queryForList(
"SELECT id,source_hash,status FROM fin_fee_source WHERE source_system=? AND source_type=? AND source_fee_id=? AND source_fee_version=? FOR UPDATE",
system, source.sourceType(), feeId, source.sourceVersion());
if (!same.isEmpty()) {
Map<String, Object> row = same.get(0);
if (!source.sourceHash().equals(row.get("source_hash")))
throw error("FIN-EVT-004", "同一来源版本快照不一致");
long id = ((Number) row.get("id")).longValue();
if ("VOIDED".equals(source.action())) {
requireNotFrozen(id);
jdbc.update("UPDATE fin_fee_source SET status='VOIDED',operation_version=operation_version+1 WHERE id=? AND status<>'VOIDED'", id);
voidDirectPaidCost(system, source);
}
return id;
}
if ("VOIDED".equals(source.action()))
throw error("FIN-EVT-005", "作废事件对应的费用来源版本不存在");
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);
jdbc.update("UPDATE fin_cost SET status='VOIDED',voided_at=CURRENT_TIMESTAMP(3),void_reason='来源版本已替代',version=version+1,updated_by=0 WHERE cost_origin='BUSINESS_DIRECT_PAID' AND source_system=? AND source_type=? AND source_fee_id=? AND status='CONFIRMED'",
system, source.sourceType(), feeId);
jdbc.update("INSERT INTO fin_fee_source(source_system,source_fee_id,source_fee_version,customer_id,source_type,source_no,business_type,business_id,business_no,inbound_code,warehouse_id,direction,order_id,inbound_id,container_id,container_booking_id,container_no,fee_item_code,fee_item_name,amount,currency_code,settlement_mode,paid_at,payment_method,payment_reference_no,payment_voucher_file_ids,source_hash,source_snapshot,status) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, 'AVAILABLE')",
system, feeId, source.sourceVersion(), positive(source.customerId(), "customerId"),
source.sourceType(), trim(source.sourceNo()), trim(source.businessType()), number(source.businessId()),
trim(source.businessNo()), trim(source.inboundCode()), number(source.warehouseId()), source.direction(),
number(source.orderId()), number(source.inboundId()), number(source.containerId()),
number(source.containerBookingId()), trim(source.containerNo()), source.feeItemCode().trim(),
source.feeItemName().trim(), source.totalAmount(), source.currency(), trim(source.settlementMode()),
source.paidAt(), trim(source.paymentMethod()), trim(source.paymentReferenceNo()),
json(source.paymentVoucherFileIds()), source.sourceHash(), json(source));
long id = jdbc.queryForObject("SELECT id FROM fin_fee_source WHERE source_system=? AND source_type=? AND source_fee_id=? AND source_fee_version=?",
Long.class, system, source.sourceType(), feeId, source.sourceVersion());
if ("PAYABLE".equals(source.direction()) && "DIRECT_PAID".equals(source.settlementMode()))
materializeDirectPaidCost(system, feeId, source);
return id;
}
private void validate(EventEnvelope envelope, BusinessFeeSourceChangedV1 value) {
if (value == null || !ACTIONS.contains(value.action()) || blank(value.sourceSystem())
|| "FIRSTLEG".equalsIgnoreCase(value.sourceSystem()) || blank(value.sourceType())
|| value.sourceVersion() < 1 || !envelope.tenantId().equals(value.customerId())
|| blank(value.businessType()) || blank(value.businessNo()) || blank(value.feeItemCode())
|| blank(value.feeItemName()) || !DIRECTIONS.contains(value.direction())
|| value.businessDate() == null || !"CNY".equals(value.currency())
|| value.totalAmount() == null || value.totalAmount().signum() <= 0
|| !"APPROVED".equals(value.reviewStatus()) || blank(value.sourceHash())) {
throw error("FIN-EVT-012", "跨区费用来源快照字段不完整");
}
positive(value.sourceFeeId(), "sourceFeeId");
positive(value.customerId(), "customerId");
if (!"VOIDED".equals(value.action()) && (!"PUBLISHED".equals(value.publicationStatus())
|| !"ACTIVE".equals(value.sourceStatus()))) {
throw error("FIN-EVT-015", "费用来源尚未发布或已失效");
}
if ("PAYABLE".equals(value.direction())) {
long supplierId = positive(value.supplierId(), "supplierId");
var supplier = supplierValidator.requireActive(supplierId, trim(value.supplierCategory()));
if (blank(value.supplierName()) || !value.supplierName().equals(supplier.name()))
throw error("FIN-EVT-012", "应付来源供应商快照不一致");
validateSettlement(value);
} else if (!blank(value.supplierId()) || !blank(value.supplierName())
|| !blank(value.settlementMode()) || value.paidAt() != null
|| !blank(value.paymentMethod()) || !blank(value.paymentReferenceNo())
|| (value.paymentVoucherFileIds() != null && !value.paymentVoucherFileIds().isEmpty())) {
throw error("FIN-EVT-012", "应收来源不得携带供应商或付款事实");
}
BusinessFeeSourceChangedV1 unsigned = new BusinessFeeSourceChangedV1(value.action(), value.sourceSystem(),
value.sourceType(), value.sourceFeeId(), value.sourceVersion(), value.customerId(), value.customerName(),
value.sourceNo(), value.businessType(), value.businessId(), value.businessNo(), value.orderId(),
value.inboundId(), value.inboundCode(), value.containerId(), value.containerBookingId(), value.containerNo(),
value.warehouseId(), value.feeItemCode(), value.feeItemName(), value.direction(), value.supplierId(),
value.supplierName(), value.supplierCategory(), value.settlementMode(), value.paidAt(), value.paymentMethod(),
value.paymentReferenceNo(), value.paymentVoucherFileIds(), value.businessDate(), value.currency(),
value.totalAmount(), value.reviewStatus(), value.publicationStatus(), value.sourceStatus(), null);
String expected = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(objectMapper.valueToTree(unsigned)));
if (!expected.equals(value.sourceHash())) throw error("FIN-EVT-004", "费用来源快照哈希不一致");
}
private void validateSettlement(BusinessFeeSourceChangedV1 value) {
boolean files = value.paymentVoucherFileIds() != null && !value.paymentVoucherFileIds().isEmpty();
if ("ON_ACCOUNT".equals(value.settlementMode())) {
if (value.paidAt() != null || !blank(value.paymentMethod()) || !blank(value.paymentReferenceNo()) || files)
throw error("FIN-EVT-012", "账期应付不得携带已付事实");
} else if (!"DIRECT_PAID".equals(value.settlementMode()) || value.paidAt() == null
|| !Set.of("ONLINE", "BANK_TRANSFER", "CASH", "OTHER").contains(value.paymentMethod())
|| (blank(value.paymentReferenceNo()) && !files)) {
throw error("FIN-EVT-012", "直接已付来源快照不完整");
}
}
private void requireSourceNotFrozen(String system, String type, long feeId) {
Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM fin_fee_source f WHERE f.source_system=? AND f.source_type=? AND f.source_fee_id=? AND (EXISTS (SELECT 1 FROM fin_bill_line bl JOIN fin_bill b ON b.id=bl.bill_id AND b.tenant_id=bl.tenant_id WHERE bl.tenant_id=f.customer_id AND bl.source_id=f.id AND bl.source_version=f.source_fee_version AND b.status<>'VOIDED') OR EXISTS (SELECT 1 FROM fin_settlement_line sl JOIN fin_settlement s ON s.id=sl.settlement_id AND s.tenant_id=sl.tenant_id WHERE sl.tenant_id=f.customer_id AND sl.fee_source_id=f.id AND s.status<>'VOIDED'))", Integer.class, system, type, feeId);
if (count != null && count > 0) throw error("FIN-EVT-016", "费用来源已进入有效单据,禁止调整或替代");
}
private void requireNotFrozen(long id) {
Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM fin_fee_source f WHERE f.id=? AND (EXISTS (SELECT 1 FROM fin_bill_line bl JOIN fin_bill b ON b.id=bl.bill_id AND b.tenant_id=bl.tenant_id WHERE bl.tenant_id=f.customer_id AND bl.source_id=f.id AND bl.source_version=f.source_fee_version AND b.status<>'VOIDED') OR EXISTS (SELECT 1 FROM fin_settlement_line sl JOIN fin_settlement s ON s.id=sl.settlement_id AND s.tenant_id=sl.tenant_id WHERE sl.tenant_id=f.customer_id AND sl.fee_source_id=f.id AND s.status<>'VOIDED'))", Integer.class, id);
if (count != null && count > 0) throw error("FIN-EVT-016", "费用来源已进入有效单据,禁止作废");
}
private void materializeDirectPaidCost(String system, long feeId, BusinessFeeSourceChangedV1 value) {
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,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()), positive(value.supplierId(), "supplierId"),
value.supplierName(), value.feeItemCode(), value.feeItemName(), value.currency(), value.totalAmount(),
json(value.paymentVoucherFileIds()), value.paidAt(), value.paymentMethod(), value.paymentReferenceNo());
}
private void voidDirectPaidCost(String system, BusinessFeeSourceChangedV1 value) {
jdbc.update("UPDATE fin_cost SET status='VOIDED',voided_at=CURRENT_TIMESTAMP(3),void_reason='来源费用已作废',version=version+1,updated_by=0 WHERE cost_origin='BUSINESS_DIRECT_PAID' AND source_system=? AND source_type=? AND source_fee_id=? AND source_fee_version=? AND status='CONFIRMED'",
system, value.sourceType(), positive(value.sourceFeeId(), "sourceFeeId"), value.sourceVersion());
}
private BusinessFeeSourceChangedV1 payload(EventEnvelope envelope) {
try { return objectMapper.treeToValue(envelope.payload(), BusinessFeeSourceChangedV1.class); }
catch (JsonProcessingException exception) { throw error("FIN-EVT-001", "跨区费用事件载荷无法解析"); }
}
private String json(Object value) {
try { return objectMapper.writeValueAsString(value == null ? List.of() : value); }
catch (JsonProcessingException exception) { throw error("FIN-EVT-001", "费用事件序列化失败"); }
}
private static Long number(String value) { return blank(value) ? null : positive(value, "business identifier"); }
private static long positive(String value, String name) {
try { long parsed = Long.parseLong(value); if (parsed > 0) return parsed; }
catch (RuntimeException ignored) { }
throw error("FIN-EVT-001", name + "无效");
}
private static boolean blank(String value) { return value == null || value.isBlank(); }
private static String trim(String value) { return blank(value) ? null : value.trim(); }
private static FinanceEventException error(String code, String message) { return new FinanceEventException(code, message); }
}
@@ -63,8 +63,8 @@ public class FinanceFeeSourceEventService {
private long applyEvent(FinanceFeeSourceEvent event, FirstlegFeeSourceClient.Snapshot snapshot) {
List<Map<String, Object>> sameVersion = jdbc.queryForList(
"SELECT id,source_hash,status FROM fin_fee_source WHERE source_system='FIRSTLEG' AND source_fee_id=? AND source_fee_version=? FOR UPDATE",
snapshot.feeId(), snapshot.feeVersion());
"SELECT id,source_hash,status FROM fin_fee_source WHERE source_system='FIRSTLEG' AND source_type=? AND source_fee_id=? AND source_fee_version=? FOR UPDATE",
snapshot.sourceType(), snapshot.feeId(), snapshot.feeVersion());
if (!sameVersion.isEmpty()) {
Map<String, Object> row = sameVersion.get(0);
if (!snapshot.sourceHash().equals(row.get("source_hash"))) {
@@ -81,29 +81,29 @@ public class FinanceFeeSourceEventService {
if ("FIRSTLEG_FEE_VOIDED".equals(event.eventType())) {
throw error("FIN-EVT-005", "作废事件对应的费用来源版本不存在");
}
requireSourceNotFrozen(snapshot.feeId());
jdbc.update("UPDATE fin_fee_source SET status='SUPERSEDED' WHERE source_system='FIRSTLEG' AND source_fee_id=? AND status='AVAILABLE'",
snapshot.feeId());
jdbc.update("UPDATE fin_cost SET status='VOIDED',voided_at=CURRENT_TIMESTAMP(3),void_reason='来源版本已替代',version=version+1,updated_by=0 WHERE cost_origin='FIRSTLEG_DIRECT_PAID' AND source_system='FIRSTLEG' AND source_fee_id=? AND status='CONFIRMED'",
snapshot.feeId());
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());
jdbc.update("UPDATE fin_cost SET status='VOIDED',voided_at=CURRENT_TIMESTAMP(3),void_reason='来源版本已替代',version=version+1,updated_by=0 WHERE cost_origin='FIRSTLEG_DIRECT_PAID' AND source_system='FIRSTLEG' AND source_type=? AND source_fee_id=? AND status='CONFIRMED'",
snapshot.sourceType(), snapshot.feeId());
jdbc.update("INSERT INTO fin_fee_source(source_system,source_fee_id,source_fee_version,customer_id,source_type,source_no,direction,order_id,inbound_id,container_id,fee_item_code,fee_item_name,amount,currency_code,settlement_mode,paid_at,payment_method,payment_reference_no,payment_voucher_file_ids,source_hash,source_snapshot,status) VALUES ('FIRSTLEG',?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,'AVAILABLE')",
snapshot.feeId(), snapshot.feeVersion(), snapshot.customerId(), snapshot.sourceType(),
snapshot.sourceType().equals("INBOUND_FEE") ? snapshot.inboundCode() : snapshot.containerNo(),
snapshot.inboundCode() == null || snapshot.inboundCode().isBlank() ? snapshot.containerNo() : snapshot.inboundCode(),
snapshot.direction(), snapshot.orderId(), snapshot.inboundId(), snapshot.containerId(),
snapshot.feeItemCode(), snapshot.feeItemName(), snapshot.totalAmount(), snapshot.currency(),
snapshot.settlementMode(), snapshot.paidAt(), snapshot.paymentMethod(),
snapshot.paymentReferenceNo(), voucherJson(snapshot.paymentVoucherFileIds()),
snapshot.sourceHash(), json(snapshot));
long sourceId = jdbc.queryForObject("SELECT id FROM fin_fee_source WHERE source_system='FIRSTLEG' AND source_fee_id=? AND source_fee_version=?",
Long.class, snapshot.feeId(), snapshot.feeVersion());
long sourceId = jdbc.queryForObject("SELECT id FROM fin_fee_source WHERE source_system='FIRSTLEG' AND source_type=? AND source_fee_id=? AND source_fee_version=?",
Long.class, snapshot.sourceType(), snapshot.feeId(), snapshot.feeVersion());
if ("PAYABLE".equals(snapshot.direction()) && "DIRECT_PAID".equals(snapshot.settlementMode())) {
materializeDirectPaidCost(snapshot);
}
return sourceId;
}
private void requireSourceNotFrozen(long sourceFeeId) {
Integer frozen = jdbc.queryForObject("SELECT COUNT(*) FROM fin_fee_source f WHERE f.source_system='FIRSTLEG' AND f.source_fee_id=? AND (EXISTS (SELECT 1 FROM fin_bill_line bl JOIN fin_bill b ON b.id=bl.bill_id AND b.tenant_id=bl.tenant_id WHERE bl.tenant_id=f.customer_id AND bl.source_id=f.id AND bl.source_version=f.source_fee_version AND b.status<>'VOIDED') OR EXISTS (SELECT 1 FROM fin_settlement_line sl JOIN fin_settlement s ON s.id=sl.settlement_id AND s.tenant_id=sl.tenant_id WHERE sl.tenant_id=f.customer_id AND sl.fee_source_id=f.id AND s.status<>'VOIDED'))", Integer.class, sourceFeeId);
private void requireSourceNotFrozen(String sourceType, long sourceFeeId) {
Integer frozen = jdbc.queryForObject("SELECT COUNT(*) FROM fin_fee_source f WHERE f.source_system='FIRSTLEG' AND f.source_type=? AND f.source_fee_id=? AND (EXISTS (SELECT 1 FROM fin_bill_line bl JOIN fin_bill b ON b.id=bl.bill_id AND b.tenant_id=bl.tenant_id WHERE bl.tenant_id=f.customer_id AND bl.source_id=f.id AND bl.source_version=f.source_fee_version AND b.status<>'VOIDED') OR EXISTS (SELECT 1 FROM fin_settlement_line sl JOIN fin_settlement s ON s.id=sl.settlement_id AND s.tenant_id=sl.tenant_id WHERE sl.tenant_id=f.customer_id AND sl.fee_source_id=f.id AND s.status<>'VOIDED'))", Integer.class, sourceType, sourceFeeId);
if (frozen != null && frozen > 0) throw error("FIN-EVT-016", "费用来源已进入有效单据,禁止调整或替代");
}