193 lines
13 KiB
Java
193 lines
13 KiB
Java
package com.baiyue.finance.service;
|
|
|
|
import com.baiyue.common.event.EventEnvelope;
|
|
import com.baiyue.common.event.DeliveryAck;
|
|
import com.baiyue.common.event.EventEnvelopeValidator;
|
|
import com.baiyue.common.event.EventIdentifiers;
|
|
import com.baiyue.common.event.EventPayloadCanonicalizer;
|
|
import com.baiyue.common.event.EventPayloadHash;
|
|
import com.baiyue.common.event.EventTypes;
|
|
import com.baiyue.common.event.WalletCommandCompletedV1;
|
|
import com.baiyue.common.event.WalletCommandRequestedV1;
|
|
import com.baiyue.finance.dto.FinanceWalletDtos.WalletFreezeRequest;
|
|
import com.baiyue.finance.dto.FinanceWalletDtos.WalletHold;
|
|
import com.baiyue.finance.dto.FinanceWalletDtos.WalletHoldActionRequest;
|
|
import com.fasterxml.jackson.databind.JsonNode;
|
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
|
import org.springframework.dao.DuplicateKeyException;
|
|
import org.springframework.jdbc.core.JdbcTemplate;
|
|
import org.springframework.stereotype.Service;
|
|
import org.springframework.transaction.PlatformTransactionManager;
|
|
import org.springframework.transaction.TransactionDefinition;
|
|
import org.springframework.transaction.support.TransactionTemplate;
|
|
|
|
import java.math.BigDecimal;
|
|
import java.time.Instant;
|
|
import java.util.List;
|
|
import java.util.Locale;
|
|
import java.util.Set;
|
|
|
|
@Service
|
|
public class WalletCommandEventService {
|
|
private static final Set<String> OPERATIONS = Set.of("FREEZE", "SETTLE", "SETTLE_ACTUAL", "RELEASE", "REFUND");
|
|
private static final Set<String> FINAL_REJECTIONS = Set.of(
|
|
"FIN-WALLET-INSUFFICIENT", "FIN-WALLET-HOLD-409", "FIN-IDEMPOTENCY-CONFLICT", "FIN-REQ-400",
|
|
"FIN-WALLET-REFUND-EXCEEDED", "FIN-WALLET-404");
|
|
private final JdbcTemplate jdbc;
|
|
private final ObjectMapper objectMapper;
|
|
private final FinanceWalletService walletService;
|
|
private final TransactionTemplate transactions;
|
|
|
|
public WalletCommandEventService(JdbcTemplate jdbc, ObjectMapper objectMapper,
|
|
FinanceWalletService walletService, PlatformTransactionManager transactionManager) {
|
|
this.jdbc = jdbc;
|
|
this.objectMapper = objectMapper;
|
|
this.walletService = walletService;
|
|
this.transactions = new TransactionTemplate(transactionManager);
|
|
this.transactions.setPropagationBehavior(TransactionDefinition.PROPAGATION_REQUIRES_NEW);
|
|
}
|
|
|
|
public DeliveryAck consume(EventEnvelope envelope) {
|
|
WalletCommandRequestedV1 command = validate(envelope);
|
|
// Keep event identity durable even if processing rolls back.
|
|
transactions.executeWithoutResult(status -> reserve(envelope));
|
|
try {
|
|
return transactions.execute(status -> {
|
|
requireLockedIdentity(envelope);
|
|
if (resultExists(envelope.eventId())) return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
|
|
// Wallet methods join this transaction: money, the Inbox outcome and
|
|
// the result Outbox commit together, or all roll back.
|
|
ExecutionResult execution = execute(command);
|
|
finalizeResult(envelope, completed(envelope.eventId(), command, "SUCCEEDED",
|
|
execution.status(), execution.amount(), null, null, execution.holdId(), execution.ledgerId()));
|
|
return DeliveryAck.of(envelope, DeliveryAck.Outcome.APPLIED);
|
|
});
|
|
} catch (FinanceEventException exception) {
|
|
if (!FINAL_REJECTIONS.contains(exception.code())) throw exception;
|
|
// A joined wallet transaction may have been marked rollback-only. Persist
|
|
// a final business rejection only AFTER that transaction has rolled back.
|
|
return transactions.execute(status -> {
|
|
requireLockedIdentity(envelope);
|
|
if (resultExists(envelope.eventId())) return DeliveryAck.of(envelope, DeliveryAck.Outcome.DUPLICATE);
|
|
finalizeResult(envelope, completed(envelope.eventId(), command, "FAILED",
|
|
null, command.amount(), exception.code(), exception.getMessage(), null, null));
|
|
return DeliveryAck.of(envelope, DeliveryAck.Outcome.APPLIED);
|
|
});
|
|
}
|
|
}
|
|
|
|
private void requireLockedIdentity(EventEnvelope envelope) {
|
|
List<String> hashes = jdbc.queryForList(
|
|
"SELECT payload_hash FROM fin_wallet_command_inbox WHERE event_id=? FOR UPDATE",
|
|
String.class, envelope.eventId());
|
|
if (hashes.size() != 1) throw new IllegalStateException("资金指令身份记录不存在");
|
|
if (!hashes.get(0).equals(envelope.payloadHash())) {
|
|
throw new FinanceEventException("FIN-EVT-002", "同一事件ID对应不同资金指令");
|
|
}
|
|
}
|
|
|
|
private record ExecutionResult(String status, BigDecimal amount, String holdId, String ledgerId) { }
|
|
|
|
private ExecutionResult execute(WalletCommandRequestedV1 command) {
|
|
if ("REFUND".equals(command.operation())) {
|
|
var refund = walletService.refundConsumption(command.customerId(), Long.parseLong(command.originalLedgerId()),
|
|
command.businessNo(), command.idempotencyKey(), command.amount(), 0, "业务系统", command.remark());
|
|
return new ExecutionResult("REFUNDED", refund.amount(), null, String.valueOf(refund.ledgerId()));
|
|
}
|
|
WalletHold hold = switch (command.operation()) {
|
|
case "FREEZE" -> walletService.freeze(command.customerId(), command.idempotencyKey(),
|
|
new WalletFreezeRequest(command.amount(), command.businessType(), command.businessNo(), command.remark()));
|
|
case "SETTLE" -> walletService.settle(command.customerId(), command.businessType(), command.businessNo(),
|
|
command.idempotencyKey(), new WalletHoldActionRequest(command.remark()));
|
|
case "SETTLE_ACTUAL" -> walletService.settleActual(command.customerId(), command.businessType(), command.businessNo(),
|
|
command.idempotencyKey(), command.amount(), command.remark());
|
|
case "RELEASE" -> walletService.release(command.customerId(), command.businessType(), command.businessNo(),
|
|
command.idempotencyKey(), new WalletHoldActionRequest(command.remark()));
|
|
default -> throw new FinanceEventException("FIN-EVT-001", "资金指令操作无效");
|
|
};
|
|
if (hold == null) throw new IllegalStateException("资金操作未返回处理结果");
|
|
if (Set.of("SETTLE", "SETTLE_ACTUAL").contains(command.operation())) {
|
|
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 h.id=? AND f.direction='RECEIVABLE' AND f.status='AVAILABLE' AND f.amount=-l.amount",
|
|
hold.holdId());
|
|
}
|
|
Long ledgerId = "FREEZE".equals(command.operation()) ? hold.freezeLedgerId() : hold.actionLedgerId();
|
|
return new ExecutionResult(hold.status(), "SETTLE_ACTUAL".equals(command.operation()) ? command.amount() : hold.amount(),
|
|
String.valueOf(hold.holdId()), ledgerId == null ? null : String.valueOf(ledgerId));
|
|
}
|
|
|
|
private WalletCommandRequestedV1 validate(EventEnvelope envelope) {
|
|
try { EventEnvelopeValidator.requireValid(envelope); }
|
|
catch (RuntimeException exception) { throw new FinanceEventException("FIN-EVT-001", "资金指令信封无效"); }
|
|
if (!EventTypes.WALLET_COMMAND_REQUESTED.equals(envelope.eventType())
|
|
|| envelope.eventVersion() != WalletCommandRequestedV1.VERSION) {
|
|
throw new FinanceEventException("FIN-EVT-001", "资金指令类型或版本无效");
|
|
}
|
|
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
|
|
throw new FinanceEventException("FIN-EVT-002", "事件信封摘要与实际载荷不一致");
|
|
}
|
|
try {
|
|
WalletCommandRequestedV1 value = objectMapper.treeToValue(envelope.payload(), WalletCommandRequestedV1.class);
|
|
String operation = value.operation() == null ? "" : value.operation().trim().toUpperCase(Locale.ROOT);
|
|
if (!OPERATIONS.contains(operation) || value.customerId() <= 0 || blank(value.businessType())
|
|
|| blank(value.businessNo()) || blank(value.idempotencyKey())
|
|
|| (Set.of("FREEZE", "SETTLE_ACTUAL", "REFUND").contains(operation)
|
|
&& (value.amount() == null || value.amount().signum() <= 0 || value.amount().scale() > 2))) {
|
|
throw new FinanceEventException("FIN-EVT-001", "资金指令字段无效");
|
|
}
|
|
if ("REFUND".equals(operation)) {
|
|
if (blank(value.originalLedgerId()) || Long.parseLong(value.originalLedgerId()) <= 0 || blank(value.remark())) {
|
|
throw new FinanceEventException("FIN-EVT-001", "退款必须关联原扣款并填写原因");
|
|
}
|
|
} else if (value.originalLedgerId() != null) {
|
|
throw new FinanceEventException("FIN-EVT-001", "非退款指令不能指定原扣款");
|
|
}
|
|
if (!String.valueOf(value.customerId()).equals(envelope.tenantId())) {
|
|
throw new FinanceEventException("FIN-EVT-007", "资金指令客户归属不一致");
|
|
}
|
|
return new WalletCommandRequestedV1(operation, value.customerId(), value.businessType(), value.businessNo(),
|
|
value.amount(), value.remark(), value.idempotencyKey(), value.requestedAt(), value.originalLedgerId());
|
|
} catch (FinanceEventException exception) { throw exception; }
|
|
catch (Exception exception) { throw new FinanceEventException("FIN-EVT-001", "资金指令载荷无法解析"); }
|
|
}
|
|
|
|
private void reserve(EventEnvelope envelope) {
|
|
try {
|
|
jdbc.update("INSERT INTO fin_wallet_command_inbox(event_id,event_type,payload_hash,payload_json,status,received_at) VALUES(?,?,?,CAST(? AS JSON),'PROCESSING',NOW(3))",
|
|
envelope.eventId(), envelope.eventType(), envelope.payloadHash(), json(envelope));
|
|
} catch (DuplicateKeyException duplicate) {
|
|
// The durable winner is verified again under SELECT FOR UPDATE before
|
|
// any wallet call, including when both requests initially saw no record.
|
|
}
|
|
}
|
|
|
|
private void finalizeResult(EventEnvelope commandEnvelope, WalletCommandCompletedV1 result) {
|
|
JsonNode payload = objectMapper.valueToTree(result);
|
|
String payloadHash = EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload));
|
|
EventEnvelope completed = new EventEnvelope(EventIdentifiers.newUuid(), EventTypes.WALLET_COMMAND_COMPLETED,
|
|
WalletCommandCompletedV1.VERSION, Instant.now(),
|
|
String.valueOf(result.customerId()), "WALLET_HOLD", result.businessNo(), 1L, null,
|
|
commandEnvelope.traceId(), commandEnvelope.correlationId(), commandEnvelope.eventId(), payload, payloadHash).requireValid();
|
|
jdbc.update("INSERT INTO fin_wallet_event_outbox(event_id,event_key,payload_json,status,next_retry_at,created_at,updated_at) VALUES(?,?,CAST(? AS JSON),'PENDING',NOW(3),NOW(3),NOW(3))",
|
|
completed.eventId(), "WALLET_RESULT:" + commandEnvelope.eventId(), json(completed));
|
|
jdbc.update("UPDATE fin_wallet_command_inbox SET status=?,result_json=CAST(? AS JSON),processed_at=NOW(3) WHERE event_id=?",
|
|
result.result(), json(result), commandEnvelope.eventId());
|
|
}
|
|
|
|
private boolean resultExists(String eventId) {
|
|
Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM fin_wallet_event_outbox WHERE event_key=?", Integer.class,
|
|
"WALLET_RESULT:" + eventId);
|
|
return count != null && count > 0;
|
|
}
|
|
private WalletCommandCompletedV1 completed(String eventId, WalletCommandRequestedV1 command, String result,
|
|
String holdStatus, BigDecimal amount, String errorCode, String errorMessage,
|
|
String holdId, String ledgerId) {
|
|
return new WalletCommandCompletedV1(eventId, command.operation(), command.customerId(), command.businessType(),
|
|
command.businessNo(), result, holdStatus, amount, errorCode, errorMessage, Instant.now(), holdId, ledgerId);
|
|
}
|
|
private String json(Object value) {
|
|
try { return objectMapper.writeValueAsString(value); }
|
|
catch (Exception exception) { throw new FinanceEventException("FIN-EVT-001", "资金事件序列化失败"); }
|
|
}
|
|
private static boolean blank(String value) { return value == null || value.isBlank(); }
|
|
}
|