feat: close overseas wallet and pallet cost flow

2026-09-05 22:26:25 +08:00
parent f62283fb7f
commit 5fbd2bbd04
69 changed files with 1598 additions and 53 deletions
@@ -0,0 +1,73 @@
+package com.baiyue.lastleg.wallet;
+
+import com.baiyue.common.event.EventEnvelope;
+import com.baiyue.common.event.EventEnvelopeValidator;
+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.baiyue.common.event.WalletCommandCompletedV1;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
+import org.springframework.web.server.ResponseStatusException;
+
+import java.util.List;
+
+import static org.springframework.http.HttpStatus.BAD_REQUEST;
+
+@Service
+@RequiredArgsConstructor
+public class WalletCommandCompletedConsumer {
+    private final JdbcTemplate jdbc;
+    private final ObjectMapper objectMapper;
+    private final List<WalletResultHandler> handlers;
+
+    @Transactional
+    public int consume(EventEnvelope envelope) {
+        WalletCommandCompletedV1 result = validate(envelope);
+        List<String> existing = jdbc.queryForList("SELECT payload_hash FROM ll_wallet_event_inbox WHERE event_id=?", String.class, envelope.eventId());
+        if (!existing.isEmpty()) {
+            if (!existing.get(0).equals(envelope.payloadHash())) throw bad("同一资金结果事件ID对应不同载荷");
+            return 0;
+        }
+        try {
+            jdbc.update("INSERT INTO ll_wallet_event_inbox(event_id,command_event_id,payload_hash,payload_json,status,received_at) VALUES(?,?,?,CAST(? AS JSON),'RECEIVED',NOW(3))",
+                envelope.eventId(), result.commandEventId(), envelope.payloadHash(), json(envelope));
+        } catch (DuplicateKeyException exception) { return 0; }
+        WalletResultHandler handler = handlers.stream().filter(value -> value.supports(result.businessType())).findFirst()
+            .orElseThrow(() -> bad("没有可处理该资金结果的业务模块"));
+        handler.handle(result);
+        jdbc.update("UPDATE ll_wallet_event_inbox SET status='PROCESSED',processed_at=NOW(3) WHERE event_id=?", envelope.eventId());
+        return 1;
+    }
+
+    private WalletCommandCompletedV1 validate(EventEnvelope envelope) {
+        try { EventEnvelopeValidator.requireValid(envelope); }
+        catch (RuntimeException exception) { throw bad("资金结果事件信封无效"); }
+        if (!EventTypes.WALLET_COMMAND_COMPLETED.equals(envelope.eventType())
+            || envelope.eventVersion() != WalletCommandCompletedV1.VERSION
+            || envelope.sourceRegion() != RegionCode.CN || envelope.targetRegion() != RegionCode.OVERSEAS) {
+            throw bad("资金结果事件类型、版本或区域无效");
+        }
+        if (!EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(envelope.payload())).equals(envelope.payloadHash())) {
+            throw bad("资金结果事件载荷哈希无效");
+        }
+        try {
+            WalletCommandCompletedV1 result = objectMapper.treeToValue(envelope.payload(), WalletCommandCompletedV1.class);
+            if (result.customerId() <= 0 || result.commandEventId() == null || result.commandEventId().isBlank()
+                || result.businessType() == null || result.businessType().isBlank()
+                || result.businessNo() == null || result.businessNo().isBlank()) throw bad("资金结果事件字段无效");
+            return result;
+        } catch (ResponseStatusException exception) { throw exception; }
+        catch (Exception exception) { throw bad("资金结果事件载荷无法解析"); }
+    }
+    private String json(Object value) {
+        try { return objectMapper.writeValueAsString(value); }
+        catch (Exception exception) { throw bad("资金结果事件序列化失败"); }
+    }
+    private static ResponseStatusException bad(String message) { return new ResponseStatusException(BAD_REQUEST, message); }
+}
@@ -0,0 +1,61 @@
+package com.baiyue.lastleg.wallet;
+
+import com.baomidou.mybatisplus.core.toolkit.IdWorker;
+import com.baiyue.common.event.EventEnvelope;
+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.RegionCode;
+import com.baiyue.common.event.WalletCommandRequestedV1;
+import com.baiyue.lastleg.entity.LastlegEventOutbox;
+import com.baiyue.lastleg.mapper.LastlegEventOutboxMapper;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Service;
+
+import java.math.BigDecimal;
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZoneOffset;
+
+@Service
+@RequiredArgsConstructor
+public class WalletCommandOutboxService {
+    private final LastlegEventOutboxMapper mapper;
+    private final ObjectMapper objectMapper;
+
+    public String publish(String operation, long customerId, String businessType, String businessNo,
+                          BigDecimal amount, String remark, long aggregateVersion) {
+        String eventKey = "WALLET_COMMAND:" + businessType + ':' + businessNo + ':' + operation + ':' + aggregateVersion;
+        LastlegEventOutbox existing = mapper.findByEventKey(eventKey);
+        if (existing != null) return eventId(existing.getPayloadJson());
+        String idempotencyKey = businessType + ':' + businessNo + ':' + operation + ':' + aggregateVersion;
+        WalletCommandRequestedV1 value = new WalletCommandRequestedV1(operation, customerId, businessType,
+            businessNo, amount, remark, idempotencyKey, Instant.now());
+        JsonNode payload = objectMapper.valueToTree(value);
+        EventEnvelope envelope = new EventEnvelope(EventIdentifiers.newUuid(), EventTypes.WALLET_COMMAND_REQUESTED,
+            WalletCommandRequestedV1.VERSION, Instant.now(), RegionCode.OVERSEAS, RegionCode.CN,
+            String.valueOf(customerId), businessType, businessNo, aggregateVersion, null,
+            EventIdentifiers.newUuid(), eventKey, null, payload,
+            EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(payload))).requireValid();
+        LastlegEventOutbox row = new LastlegEventOutbox();
+        row.setId(IdWorker.getId()); row.setCustomerId(String.valueOf(customerId));
+        row.setAggregateType(businessType); row.setAggregateId(businessNo);
+        row.setEventType(envelope.eventType()); row.setEventVersion(envelope.eventVersion());
+        row.setEventKey(eventKey); row.setPayloadJson(json(envelope)); row.setStatus("PENDING");
+        row.setRetryCount(0); row.setCreatedAt(LocalDateTime.now(ZoneOffset.UTC));
+        mapper.insert(row);
+        return envelope.eventId();
+    }
+
+    private String eventId(String json) {
+        try { return objectMapper.readValue(json, EventEnvelope.class).eventId(); }
+        catch (Exception exception) { throw new IllegalStateException("资金指令事件快照损坏", exception); }
+    }
+    private String json(Object value) {
+        try { return objectMapper.writeValueAsString(value); }
+        catch (Exception exception) { throw new IllegalStateException("资金指令事件序列化失败", exception); }
+    }
+}
@@ -0,0 +1,8 @@
+package com.baiyue.lastleg.wallet;
+
+import com.baiyue.common.event.WalletCommandCompletedV1;
+
+public interface WalletResultHandler {
+    boolean supports(String businessType);
+    void handle(WalletCommandCompletedV1 result);
+}