refactor: centralize raw frame identity derivation

2026-07-01 18:12:04 +08:00
parent c8abc721e1
commit f89ac1c23f
4 changed files with 92 additions and 42 deletions
--- a/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
+++ b/modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java
@@ -10,17 +10,13 @@ import com.lingniu.ingest.api.pipeline.RawFrame;
 import com.lingniu.ingest.core.concurrency.AsyncBatchExecutor;
 import com.lingniu.ingest.core.concurrency.DisruptorEventBus;
 import com.lingniu.ingest.core.pipeline.InterceptorChain;
-import com.lingniu.ingest.facts.FactIds;
-import com.lingniu.ingest.facts.VehicleKey;
+import com.lingniu.ingest.facts.RawFrameIdentity;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
 import java.time.Instant;
 import java.util.ArrayList;
-import java.util.HexFormat;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
@@ -187,11 +183,9 @@ public final class Dispatcher {
        Instant ingestTime = frame.receivedAt() != null ? frame.receivedAt() : Instant.now();
        String eventId = nextRawArchiveEventId(ingestTime);
        String key = RawArchiveKeys.key(ingestTime, frame.protocolId(), vin, eventId);
-        String checksum = "sha256:" + sha256(bytes);
-        String vehicleKey = VehicleKey.derive(frame.protocolId(), vin, phone, eventId);
-        String frameId = FactIds.rawFrameId(
-                frame.protocolId(), vehicleKey, frame.command(), frame.infoType(), ingestTime, checksum);
-        Map<String, String> archiveMeta = addRawArchiveMetadata(meta, eventId, key, frameId);
+        RawFrameIdentity identity = RawFrameIdentity.derive(
+                frame.protocolId(), vin, phone, eventId, frame.command(), frame.infoType(), ingestTime, bytes);
+        Map<String, String> archiveMeta = addRawArchiveMetadata(meta, eventId, key, identity.frameId());

        VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
                eventId,
@@ -206,7 +200,7 @@ public final class Dispatcher {
                bytes,
                parsedJson);
        out.add(raw);
-        return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), frameId);
+        return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), identity.frameId());
    }

    private static String parsedJson(Object payload, String explicitParsedJson) {
@@ -231,15 +225,6 @@ public final class Dispatcher {
        return value.replace("\\", "\\\\").replace("\"", "\\\"");
    }

-    private static String sha256(byte[] value) {
-        try {
-            MessageDigest digest = MessageDigest.getInstance("SHA-256");
-            return HexFormat.of().formatHex(digest.digest(value == null ? new byte[0] : value));
-        } catch (NoSuchAlgorithmException e) {
-            throw new IllegalStateException("SHA-256 is unavailable", e);
-        }
-    }
-
    private static String nextRawArchiveEventId(Instant ingestTime) {
        // 文件名需要可排序且单进程内唯一：毫秒时间放大到微秒量级，再用 AtomicLong 递增兜底。
        long base = (ingestTime == null ? Instant.now() : ingestTime).toEpochMilli() * 1000L;
--- a/modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameIdentity.java
+++ b/modules/core/ingest-facts/src/main/java/com/lingniu/ingest/facts/RawFrameIdentity.java
@@ -0,0 +1,43 @@
+package com.lingniu.ingest.facts;
+
+import com.lingniu.ingest.api.ProtocolId;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.time.Instant;
+import java.util.HexFormat;
+
+public record RawFrameIdentity(
+        String checksum,
+        String vehicleKey,
+        String frameId
+) {
+    public RawFrameIdentity {
+        checksum = checksum == null ? "" : checksum;
+        vehicleKey = vehicleKey == null ? "" : vehicleKey;
+        frameId = frameId == null ? "" : frameId;
+    }
+
+    public static RawFrameIdentity derive(ProtocolId protocol,
+                                          String vin,
+                                          String phone,
+                                          String rawIdentitySeed,
+                                          int messageId,
+                                          int subType,
+                                          Instant receivedAt,
+                                          byte[] rawBytes) {
+        String checksum = "sha256:" + sha256(rawBytes);
+        String vehicleKey = VehicleKey.derive(protocol, vin, phone, rawIdentitySeed);
+        String frameId = FactIds.rawFrameId(protocol, vehicleKey, messageId, subType, receivedAt, checksum);
+        return new RawFrameIdentity(checksum, vehicleKey, frameId);
+    }
+
+    private static String sha256(byte[] value) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            return HexFormat.of().formatHex(digest.digest(value == null ? new byte[0] : value));
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException("SHA-256 is unavailable", e);
+        }
+    }
+}
--- a/modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameIdentityTest.java
+++ b/modules/core/ingest-facts/src/test/java/com/lingniu/ingest/facts/RawFrameIdentityTest.java
@@ -0,0 +1,36 @@
+package com.lingniu.ingest.facts;
+
+import com.lingniu.ingest.api.ProtocolId;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class RawFrameIdentityTest {
+
+    @Test
+    void derivesChecksumVehicleKeyAndFrameIdTogether() {
+        RawFrameIdentity identity = RawFrameIdentity.derive(
+                ProtocolId.JT808,
+                "",
+                "13307795425",
+                "1782900000000000",
+                0x0200,
+                0,
+                Instant.parse("2026-07-01T10:00:00Z"),
+                new byte[]{0x7e, 0x02, 0x00, 0x7e});
+
+        assertThat(identity.checksum()).startsWith("sha256:");
+        assertThat(identity.vehicleKey()).isEqualTo("jt808:13307795425");
+        assertThat(identity.frameId())
+                .isEqualTo(FactIds.rawFrameId(
+                        ProtocolId.JT808,
+                        "jt808:13307795425",
+                        0x0200,
+                        0,
+                        Instant.parse("2026-07-01T10:00:00Z"),
+                        identity.checksum()))
+                .startsWith("rf_");
+    }
+}
--- a/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java
+++ b/modules/sinks/sink-kafka/src/main/java/com/lingniu/ingest/sink/kafka/EnvelopeMapper.java
@@ -8,17 +8,12 @@ import com.lingniu.ingest.api.event.TelemetryFieldValue;
 import com.lingniu.ingest.api.event.TelemetrySnapshot;
 import com.lingniu.ingest.api.event.VehicleEvent;
 import com.lingniu.ingest.api.event.VehicleEventTelemetrySnapshotMapper;
-import com.lingniu.ingest.facts.FactIds;
-import com.lingniu.ingest.facts.VehicleKey;
+import com.lingniu.ingest.facts.RawFrameIdentity;
 import com.lingniu.ingest.sink.kafka.proto.ParseStatusProto;
 import com.lingniu.ingest.sink.kafka.proto.TelemetryField;
 import com.lingniu.ingest.sink.kafka.proto.TelemetrySnapshot.Builder;
 import com.lingniu.ingest.sink.kafka.proto.VehicleEnvelope;

-import java.security.MessageDigest;
-import java.security.NoSuchAlgorithmException;
-import java.util.HexFormat;
-
 /**
 * 领域事件 → Protobuf Envelope 的纯函数映射。通过 switch 模式匹配穷尽处理，
 * 新增 {@link VehicleEvent} 子类型时编译期就会提示补齐分支。
@@ -80,29 +75,28 @@ public final class EnvelopeMapper {
                String key = rawArchiveKey(ra);
                String uri = rawArchiveUri(ra, key);
                String phone = metadataValue(ra, "phone");
-                String vehicleKey = VehicleKey.derive(ra.source(), ra.vin(), phone, ra.eventId());
-                String checksum = sha256(rawBytes);
-                String frameId = FactIds.rawFrameId(
-                        ra.source(), vehicleKey, ra.command(), ra.infoType(), ra.ingestTime(), checksum);
+                RawFrameIdentity identity = RawFrameIdentity.derive(
+                        ra.source(), ra.vin(), phone, ra.eventId(), ra.command(), ra.infoType(),
+                        ra.ingestTime(), rawBytes);
                // RAW envelope 只携带 URI/size 引用信息，不把完整原始字节塞进 Kafka，避免 topic 膨胀。
                b.putMetadata(RawArchiveKeys.META_KEY, key);
                b.putMetadata(RawArchiveKeys.META_URI, uri);
                b.putMetadata(RawArchiveKeys.META_EVENT_ID, ra.eventId());
                b.setRawArchive(com.lingniu.ingest.sink.kafka.proto.RawArchiveRef.newBuilder()
                        .setUri(uri)
-                        .setChecksum(checksum)
+                        .setChecksum(identity.checksum())
                        .setSizeBytes(size)
                        .setParsedJson(nullToEmpty(ra.parsedJson()))
                        .build());
                b.setRawFrameFact(com.lingniu.ingest.sink.kafka.proto.RawFrameFactPayload.newBuilder()
-                        .setFrameId(frameId)
-                        .setVehicleKey(vehicleKey)
+                        .setFrameId(identity.frameId())
+                        .setVehicleKey(identity.vehicleKey())
                        .setVin(nullToEmpty(ra.vin()))
                        .setPhone(phone)
                        .setMessageId(ra.command())
                        .setSubType(ra.infoType())
                        .setRawUri(uri)
-                        .setChecksum(checksum)
+                        .setChecksum(identity.checksum())
                        .setRawSizeBytes(size)
                        .setParseStatus(parseStatus(ra))
                        .setParseError(parseError(ra))
@@ -229,12 +223,4 @@ public final class EnvelopeMapper {
        return Boolean.parseBoolean(metadataValue(event, key));
    }

-    private static String sha256(byte[] bytes) {
-        try {
-            MessageDigest digest = MessageDigest.getInstance("SHA-256");
-            return "sha256:" + HexFormat.of().formatHex(digest.digest(bytes == null ? new byte[0] : bytes));
-        } catch (NoSuchAlgorithmException e) {
-            throw new IllegalStateException("SHA-256 is unavailable", e);
-        }
-    }
 }