272 lines
21 KiB
Java
272 lines
21 KiB
Java
package com.baiyue.finance.service;
|
|
|
|
import com.baiyue.common.event.BusinessFeeSourceChangedV1;
|
|
import com.baiyue.common.event.EventEnvelope;
|
|
import com.baiyue.common.event.DeliveryAck;
|
|
import com.baiyue.common.event.EventPayloadCanonicalizer;
|
|
import com.baiyue.common.event.EventPayloadHash;
|
|
import com.baiyue.common.event.EventTypes;
|
|
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 DeliveryAck 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) {
|
|
throw error("FIN-EVT-001", "跨区费用事件类型或版本无效");
|
|
}
|
|
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
|
|
throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致");
|
|
}
|
|
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 DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
|
|
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 DeliveryAck.of(envelope, DeliveryAck.Outcome.APPLIED);
|
|
}
|
|
|
|
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", "作废事件对应的费用来源版本不存在");
|
|
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);
|
|
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,review_status,status) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?, 'APPROVED','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()),
|
|
voucherJson(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());
|
|
linkSettledWalletHold(id, source);
|
|
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
|
|
|| !Set.of("CNY", "USD").contains(value.originalCurrency())
|
|
|| value.originalAmount() == null || value.originalAmount().signum() <= 0
|
|
|| value.exchangeRate() == null || value.exchangeRate().signum() <= 0
|
|
|| blank(value.exchangeRateMonth())
|
|
|| (value.periodStart() == null) != (value.periodEnd() == null)
|
|
|| value.periodStart() != null && value.periodStart().isAfter(value.periodEnd())
|
|
|| (value.chargeableQuantity() == null) != blank(value.quantityUnit())
|
|
|| !"APPROVED".equals(value.reviewStatus()) || blank(value.sourceHash())) {
|
|
throw error("FIN-EVT-012", "跨区费用来源快照字段不完整");
|
|
}
|
|
positive(value.sourceFeeId(), "sourceFeeId");
|
|
positive(value.customerId(), "customerId");
|
|
if (value.walletSettlement() != null) {
|
|
if (!"RECEIVABLE".equals(value.direction())) throw error("FIN-EVT-012", "钱包消费只能关联应收费用");
|
|
positive(value.walletSettlement().holdId(), "walletSettlement.holdId");
|
|
positive(value.walletSettlement().ledgerId(), "walletSettlement.ledgerId");
|
|
if (blank(value.walletSettlement().commandEventId())) throw error("FIN-EVT-012", "缺少原资金命令引用");
|
|
} else if ("OVERSEAS_PROCUREMENT_FEE".equals(value.sourceType()) && "RECEIVABLE".equals(value.direction())) {
|
|
throw error("FIN-EVT-012", "采购应收必须引用成功实扣资金，不能进入待收费池");
|
|
}
|
|
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", "费用来源尚未发布或已失效");
|
|
}
|
|
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.supplierCategory())
|
|
|| !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(null, 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.originalCurrency(), value.originalAmount(), value.exchangeRate(),
|
|
value.exchangeRateMonth(), value.chargeableQuantity(), value.quantityUnit(), value.unitPrice(),
|
|
value.periodStart(), value.periodEnd(), value.reviewStatus(), "PUBLISHED", "ACTIVE", null, value.walletSettlement());
|
|
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 (f.wallet_hold_id IS NOT NULL OR 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 (f.wallet_hold_id IS NOT NULL OR 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 linkSettledWalletHold(long feeSourceId, BusinessFeeSourceChangedV1 source) {
|
|
if (!"RECEIVABLE".equals(source.direction())) return;
|
|
if (source.walletSettlement() != null) {
|
|
var reference = source.walletSettlement();
|
|
long customerId = positive(source.customerId(), "customerId");
|
|
long holdId = positive(reference.holdId(), "holdId");
|
|
long ledgerId = positive(reference.ledgerId(), "ledgerId");
|
|
List<Map<String, Object>> rows = jdbc.queryForList(
|
|
"SELECT l.amount,c.result_json FROM fin_wallet_hold h JOIN fin_wallet_ledger l ON l.id=h.action_ledger_id AND l.customer_id=h.customer_id JOIN fin_wallet_command_inbox c ON c.event_id=? AND c.status='SUCCEEDED' WHERE h.id=? AND l.id=? AND h.customer_id=? AND h.business_type=? AND h.business_no=? AND h.status='SETTLED' AND l.change_type='CONSUMPTION' FOR UPDATE",
|
|
reference.commandEventId(), holdId, ledgerId, customerId, source.businessType(), source.businessNo());
|
|
if (rows.size() != 1 || !(rows.get(0).get("amount") instanceof java.math.BigDecimal amount)
|
|
|| amount.negate().compareTo(source.totalAmount()) != 0) {
|
|
throw error("FIN-EVT-017", "费用来源与成功扣款的客户、业务或金额不匹配");
|
|
}
|
|
com.fasterxml.jackson.databind.JsonNode result;
|
|
try { result = objectMapper.readTree(String.valueOf(rows.get(0).get("result_json"))); }
|
|
catch (JsonProcessingException exception) { throw error("FIN-EVT-017", "原资金结果无法核对"); }
|
|
if (!"SUCCEEDED".equals(result.path("result").asText())
|
|
|| !Set.of("SETTLE", "SETTLE_ACTUAL").contains(result.path("operation").asText())
|
|
|| !reference.commandEventId().equals(result.path("commandEventId").asText())
|
|
|| !reference.holdId().equals(result.path("walletHoldId").asText())
|
|
|| !reference.ledgerId().equals(result.path("walletLedgerId").asText())
|
|
|| result.path("customerId").asLong() != customerId
|
|
|| !source.businessType().equals(result.path("businessType").asText())
|
|
|| !source.businessNo().equals(result.path("businessNo").asText())
|
|
|| result.path("amount").decimalValue().compareTo(source.totalAmount()) != 0) {
|
|
throw error("FIN-EVT-017", "费用来源引用的资金结果不匹配");
|
|
}
|
|
List<Long> used = jdbc.queryForList("SELECT id FROM fin_fee_source WHERE wallet_hold_id=? AND id<>? FOR UPDATE",
|
|
Long.class, holdId, feeSourceId);
|
|
if (!used.isEmpty()) throw error("FIN-EVT-017", "同一成功扣款已关联其他费用来源");
|
|
jdbc.update("UPDATE fin_fee_source SET wallet_hold_id=?,status='SETTLED' WHERE id=? AND customer_id=? AND status='AVAILABLE'",
|
|
holdId, feeSourceId, customerId);
|
|
return;
|
|
}
|
|
jdbc.update("UPDATE fin_fee_source f JOIN fin_wallet_hold h ON h.customer_id=f.customer_id AND h.business_type=f.business_type AND h.business_no=f.business_no JOIN fin_wallet_ledger l ON l.id=h.action_ledger_id AND l.customer_id=h.customer_id AND l.change_type='CONSUMPTION' SET f.wallet_hold_id=h.id,f.status='SETTLED' WHERE f.id=? AND f.status='AVAILABLE' AND h.status='SETTLED' AND f.amount=-l.amount",
|
|
feeSourceId);
|
|
}
|
|
|
|
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,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.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());
|
|
}
|
|
|
|
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 String voucherJson(List<String> values) {
|
|
return values == null || values.isEmpty() ? null : json(values);
|
|
}
|
|
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); }
|
|
}
|