暂存提交
This commit is contained in:
+33
-1
@@ -22,8 +22,20 @@ public final class EventPayloadCanonicalizer {
|
|||||||
if (payload == null || payload.isNull()) {
|
if (payload == null || payload.isNull()) {
|
||||||
throw new IllegalArgumentException("事件负载不能为空");
|
throw new IllegalArgumentException("事件负载不能为空");
|
||||||
}
|
}
|
||||||
|
return write(sort(payload));
|
||||||
|
}
|
||||||
|
|
||||||
|
/** 仅用于接收规范化数字规则上线前已经生成的历史摘要。 */
|
||||||
|
public static byte[] canonicalizeLegacy(JsonNode payload) {
|
||||||
|
if (payload == null || payload.isNull()) {
|
||||||
|
throw new IllegalArgumentException("事件负载不能为空");
|
||||||
|
}
|
||||||
|
return write(sortLegacy(payload));
|
||||||
|
}
|
||||||
|
|
||||||
|
private static byte[] write(JsonNode payload) {
|
||||||
try {
|
try {
|
||||||
return OBJECT_MAPPER.writeValueAsBytes(sort(payload));
|
return OBJECT_MAPPER.writeValueAsBytes(payload);
|
||||||
} catch (JsonProcessingException exception) {
|
} catch (JsonProcessingException exception) {
|
||||||
throw new IllegalArgumentException("事件负载无法规范化", exception);
|
throw new IllegalArgumentException("事件负载无法规范化", exception);
|
||||||
}
|
}
|
||||||
@@ -43,6 +55,26 @@ public final class EventPayloadCanonicalizer {
|
|||||||
node.forEach(child -> result.add(sort(child)));
|
node.forEach(child -> result.add(sort(child)));
|
||||||
return result;
|
return result;
|
||||||
}
|
}
|
||||||
|
if (node.isNumber()) {
|
||||||
|
return com.fasterxml.jackson.databind.node.DecimalNode.valueOf(node.decimalValue().stripTrailingZeros());
|
||||||
|
}
|
||||||
|
return node;
|
||||||
|
}
|
||||||
|
|
||||||
|
private static JsonNode sortLegacy(JsonNode node) {
|
||||||
|
if (node.isObject()) {
|
||||||
|
var names = new java.util.ArrayList<String>();
|
||||||
|
node.fieldNames().forEachRemaining(names::add);
|
||||||
|
names.sort(String::compareTo);
|
||||||
|
var result = OBJECT_MAPPER.createObjectNode();
|
||||||
|
names.forEach(name -> result.set(name, sortLegacy(node.get(name))));
|
||||||
|
return result;
|
||||||
|
}
|
||||||
|
if (node.isArray()) {
|
||||||
|
var result = OBJECT_MAPPER.createArrayNode();
|
||||||
|
node.forEach(child -> result.add(sortLegacy(child)));
|
||||||
|
return result;
|
||||||
|
}
|
||||||
return node;
|
return node;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
+29
@@ -0,0 +1,29 @@
|
|||||||
|
package com.baiyue.common.event;
|
||||||
|
|
||||||
|
import com.fasterxml.jackson.databind.ObjectMapper;
|
||||||
|
import org.junit.jupiter.api.Test;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertArrayEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertEquals;
|
||||||
|
import static org.junit.jupiter.api.Assertions.assertNotEquals;
|
||||||
|
|
||||||
|
class EventPayloadCanonicalizerTest {
|
||||||
|
private final ObjectMapper json = new ObjectMapper();
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void equivalentNumberLexemesHaveTheSameCanonicalFormAndHash() throws Exception {
|
||||||
|
var scaled = json.readTree("{\"weight\":30.0,\"volume\":0.3150,\"quantity\":3}");
|
||||||
|
var normalized = json.readTree("{\"quantity\":3.0,\"volume\":0.315,\"weight\":3E+1}");
|
||||||
|
assertArrayEquals(EventPayloadCanonicalizer.canonicalize(scaled), EventPayloadCanonicalizer.canonicalize(normalized));
|
||||||
|
assertEquals(EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(scaled)),
|
||||||
|
EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(normalized)));
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
void negativeZeroIsZeroAndLegacyHashRemainsAvailable() throws Exception {
|
||||||
|
var negativeZero = json.readTree("{\"value\":-0.000}");
|
||||||
|
var zero = json.readTree("{\"value\":0}");
|
||||||
|
assertArrayEquals(EventPayloadCanonicalizer.canonicalize(negativeZero), EventPayloadCanonicalizer.canonicalize(zero));
|
||||||
|
assertNotEquals(EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalize(negativeZero)),
|
||||||
|
EventPayloadHash.sha256(EventPayloadCanonicalizer.canonicalizeLegacy(negativeZero)));
|
||||||
|
}
|
||||||
|
}
|
||||||
@@ -14959,6 +14959,8 @@ USE baiyue_firstleg;
|
|||||||
-- 标准跨区事件摘要包含 sha256: 前缀,共 71 字符;兼容既有 Finance 64 位摘要。
|
-- 标准跨区事件摘要包含 sha256: 前缀,共 71 字符;兼容既有 Finance 64 位摘要。
|
||||||
ALTER TABLE fl_inbound_outbox
|
ALTER TABLE fl_inbound_outbox
|
||||||
MODIFY COLUMN payload_hash VARCHAR(71) NULL COMMENT '事件规范化负载SHA-256;标准跨区事件含sha256:前缀';
|
MODIFY COLUMN payload_hash VARCHAR(71) NULL COMMENT '事件规范化负载SHA-256;标准跨区事件含sha256:前缀';
|
||||||
|
ALTER TABLE fl_inbound_outbox
|
||||||
|
MODIFY COLUMN payload_json LONGTEXT NOT NULL COMMENT '事件负载快照;文本类型保留跨区事件摘要所依赖的数值词法';
|
||||||
CALL baiyue_firstleg.assert_unified_final_schema();
|
CALL baiyue_firstleg.assert_unified_final_schema();
|
||||||
DROP PROCEDURE IF EXISTS baiyue_firstleg.assert_unified_final_schema;
|
DROP PROCEDURE IF EXISTS baiyue_firstleg.assert_unified_final_schema;
|
||||||
SELECT 'PASS' AS migration_status,
|
SELECT 'PASS' AS migration_status,
|
||||||
|
|||||||
+3
-1
@@ -35,7 +35,9 @@ public class EventInboxService {
|
|||||||
if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
|
if (!envelope.targetRegion().name().equalsIgnoreCase(localRegion)) {
|
||||||
throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
|
throw new IllegalArgumentException("事件目标区域与当前事件桥区域不一致");
|
||||||
}
|
}
|
||||||
if (!EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash())) {
|
boolean currentHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalize(envelope.payload()), envelope.payloadHash());
|
||||||
|
boolean legacyHash = EventPayloadHash.matches(EventPayloadCanonicalizer.canonicalizeLegacy(envelope.payload()), envelope.payloadHash());
|
||||||
|
if (!currentHash && !legacyHash) {
|
||||||
throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
|
throw new IllegalArgumentException("payloadHash 与规范化事件负载不一致");
|
||||||
}
|
}
|
||||||
BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());
|
BridgeInbox existing = inboxStore.findByEventId(envelope.eventId());
|
||||||
|
|||||||
@@ -8,7 +8,7 @@ import org.springframework.web.bind.annotation.RequestBody;
|
|||||||
import org.springframework.web.bind.annotation.RequestHeader;
|
import org.springframework.web.bind.annotation.RequestHeader;
|
||||||
|
|
||||||
/** 将本地业务 Outbox 投递到跨区事件桥。 */
|
/** 将本地业务 Outbox 投递到跨区事件桥。 */
|
||||||
@FeignClient(name = "event-bridge-service", contextId = "loadingCompletedEventBridgeClient")
|
@FeignClient(name = "event-bridge-service", contextId = "loadingCompletedEventBridgeClient", url = "${baiyue.lastleg.event-bridge-url:}")
|
||||||
public interface EventBridgeClient {
|
public interface EventBridgeClient {
|
||||||
@PostMapping("/internal/event-bridge/events")
|
@PostMapping("/internal/event-bridge/events")
|
||||||
ApiResponse<Acknowledgement> receive(
|
ApiResponse<Acknowledgement> receive(
|
||||||
|
|||||||
@@ -28,6 +28,8 @@ baiyue:
|
|||||||
security:
|
security:
|
||||||
jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start}
|
jwt-secret: ${JWT_SECRET:change-me-to-at-least-32-bytes-long-before-start}
|
||||||
internal-service-token: ${INTERNAL_SERVICE_TOKEN:change-me-to-an-internal-token-at-least-32-bytes-long}
|
internal-service-token: ${INTERNAL_SERVICE_TOKEN:change-me-to-an-internal-token-at-least-32-bytes-long}
|
||||||
|
lastleg:
|
||||||
|
event-bridge-url: ${LASTLEG_EVENT_BRIDGE_URL:http://127.0.0.1:9312}
|
||||||
clients:
|
clients:
|
||||||
order:
|
order:
|
||||||
url: ${ORDER_SERVICE_URL:}
|
url: ${ORDER_SERVICE_URL:}
|
||||||
|
|||||||
@@ -733,7 +733,7 @@ CREATE TABLE IF NOT EXISTS fl_inbound_outbox (
|
|||||||
aggregate_type VARCHAR(32) NOT NULL DEFAULT 'INBOUND' COMMENT '聚合类型',
|
aggregate_type VARCHAR(32) NOT NULL DEFAULT 'INBOUND' COMMENT '聚合类型',
|
||||||
aggregate_id BIGINT UNSIGNED NOT NULL COMMENT '入库单ID',
|
aggregate_id BIGINT UNSIGNED NOT NULL COMMENT '入库单ID',
|
||||||
event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
|
event_type VARCHAR(64) NOT NULL COMMENT '事件类型',
|
||||||
payload_json JSON NOT NULL COMMENT '事件负载快照',
|
payload_json LONGTEXT NOT NULL COMMENT '事件负载快照;文本类型保留跨区事件摘要所依赖的数值词法',
|
||||||
publish_status VARCHAR(16) NOT NULL DEFAULT 'PENDING' COMMENT '事件发布状态:PENDING待发布、PUBLISHED已发布、FAILED发布失败',
|
publish_status VARCHAR(16) NOT NULL DEFAULT 'PENDING' COMMENT '事件发布状态:PENDING待发布、PUBLISHED已发布、FAILED发布失败',
|
||||||
retry_count INT NOT NULL DEFAULT 0 COMMENT '重试次数',
|
retry_count INT NOT NULL DEFAULT 0 COMMENT '重试次数',
|
||||||
next_retry_at DATETIME(3) NULL COMMENT '下次重试时间',
|
next_retry_at DATETIME(3) NULL COMMENT '下次重试时间',
|
||||||
|
|||||||
@@ -1,2 +1,4 @@
|
|||||||
-- 装柜完成跨区事件使用标准 sha256: 摘要,共 71 字符。
|
-- 装柜完成跨区事件使用标准 sha256: 摘要,共 71 字符。
|
||||||
ALTER TABLE fl_inbound_outbox MODIFY COLUMN payload_hash VARCHAR(71) NULL COMMENT '事件规范化负载SHA-256;标准跨区事件含sha256:前缀';
|
ALTER TABLE fl_inbound_outbox MODIFY COLUMN payload_hash VARCHAR(71) NULL COMMENT '事件规范化负载SHA-256;标准跨区事件含sha256:前缀';
|
||||||
|
-- MySQL JSON 会重写 0.3150 等数值词法,导致入库前计算的 payloadHash 失效。
|
||||||
|
ALTER TABLE fl_inbound_outbox MODIFY COLUMN payload_json LONGTEXT NOT NULL COMMENT '事件负载快照;文本类型保留跨区事件摘要所依赖的数值词法';
|
||||||
|
|||||||
Reference in New Issue
Block a user