fix: link derived events to raw frames

modules/core/ingest-core/pom.xml

@@ -20,6 +20,10 @@
             <groupId>com.lingniu.ingest</groupId>
             <artifactId>ingest-codec-common</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lingniu.ingest</groupId>
+            <artifactId>ingest-facts</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter</artifactId>

modules/core/ingest-core/src/main/java/com/lingniu/ingest/core/dispatcher/Dispatcher.java

@@ -10,12 +10,17 @@ 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 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;
@@ -178,10 +183,15 @@ public final class Dispatcher {
                 : new LinkedHashMap<>(frame.sourceMeta());
         String parsedJson = parsedJson(frame.payload(), meta.remove(RawArchiveKeys.META_PARSED_JSON));
         String vin = meta.getOrDefault("vin", "");
+        String phone = meta.getOrDefault("phone", "");
         Instant ingestTime = frame.receivedAt() != null ? frame.receivedAt() : Instant.now();
         String eventId = nextRawArchiveEventId(ingestTime);
         String key = RawArchiveKeys.key(ingestTime, frame.protocolId(), vin, eventId);
-        Map<String, String> archiveMeta = addRawArchiveMetadata(meta, eventId, key);
+        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);
 
         VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
                 eventId,
@@ -196,7 +206,7 @@ public final class Dispatcher {
                 bytes,
                 parsedJson);
         out.add(raw);
-        return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key));
+        return new RawArchiveLookup(eventId, key, RawArchiveKeys.logicalUri(key), frameId);
     }
 
     private static String parsedJson(Object payload, String explicitParsedJson) {
@@ -221,6 +231,15 @@ 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;
@@ -230,11 +249,16 @@ public final class Dispatcher {
 
     private static Map<String, String> addRawArchiveMetadata(Map<String, String> metadata,
                                                              String eventId,
-                                                             String key) {
+                                                             String key,
+                                                             String frameId) {
         Map<String, String> out = new LinkedHashMap<>(metadata == null ? Map.of() : metadata);
         out.putIfAbsent(RawArchiveKeys.META_EVENT_ID, eventId);
         out.putIfAbsent(RawArchiveKeys.META_KEY, key);
         out.putIfAbsent(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(key));
+        if (frameId != null && !frameId.isBlank()) {
+            out.putIfAbsent("rawFrameId", frameId);
+            out.putIfAbsent("frame_id", frameId);
+        }
         return Map.copyOf(out);
     }
 
@@ -243,7 +267,8 @@ public final class Dispatcher {
             return event;
         }
         // 派生事件携带 rawArchiveUri，便于后续服务从 Kafka 事件回查原始报文。
-        Map<String, String> metadata = addRawArchiveMetadata(event.metadata(), rawArchive.eventId(), rawArchive.key());
+        Map<String, String> metadata = addRawArchiveMetadata(
+                event.metadata(), rawArchive.eventId(), rawArchive.key(), rawArchive.frameId());
         return switch (event) {
             case VehicleEvent.Realtime e -> new VehicleEvent.Realtime(
                     e.eventId(), e.vin(), e.source(), e.eventTime(), e.ingestTime(),
@@ -273,9 +298,9 @@ public final class Dispatcher {
         };
     }
 
-    private record RawArchiveLookup(String eventId, String key, String uri) {
+    private record RawArchiveLookup(String eventId, String key, String uri, String frameId) {
         private static RawArchiveLookup empty() {
-            return new RawArchiveLookup("", "", "");
+            return new RawArchiveLookup("", "", "", "");
         }
 
         private boolean isEmpty() {

modules/core/ingest-core/src/test/java/com/lingniu/ingest/core/dispatcher/DispatcherRawArchiveMetadataTest.java

@@ -51,15 +51,20 @@ class DispatcherRawArchiveMetadataTest {
             VehicleEvent.Location location = awaitEvent(sink, VehicleEvent.Location.class);
 
             String rawArchiveKey = raw.metadata().get("rawArchiveKey");
+            String rawFrameId = raw.metadata().get("rawFrameId");
             assertThat(rawArchiveKey).isNotBlank();
+            assertThat(rawFrameId).startsWith("rf_");
             assertThat(raw.parsedJson()).contains("\"name\":\"payload-0200\"");
             assertThat(raw.metadata()).doesNotContainKey("rawArchiveParsedJson");
             assertThat(raw.eventId()).containsOnlyDigits();
             assertThat(rawArchiveKey).endsWith("/" + raw.eventId() + ".bin");
+            assertThat(raw.metadata()).containsEntry("frame_id", rawFrameId);
             assertThat(location.metadata())
                     .containsEntry("rawArchiveEventId", raw.eventId())
                     .containsEntry("rawArchiveKey", rawArchiveKey)
-                    .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey);
+                    .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey)
+                    .containsEntry("rawFrameId", rawFrameId)
+                    .containsEntry("frame_id", rawFrameId);
         } finally {
             batchExecutor.close();
             eventBus.close();
@@ -92,10 +97,14 @@ class DispatcherRawArchiveMetadataTest {
             VehicleEvent.Location location = awaitLocationEvent(sink, "async-location-1");
 
             String rawArchiveKey = raw.metadata().get("rawArchiveKey");
+            String rawFrameId = raw.metadata().get("rawFrameId");
+            assertThat(rawFrameId).startsWith("rf_");
             assertThat(location.metadata())
                     .containsEntry("rawArchiveEventId", raw.eventId())
                     .containsEntry("rawArchiveKey", rawArchiveKey)
-                    .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey);
+                    .containsEntry("rawArchiveUri", "archive://" + rawArchiveKey)
+                    .containsEntry("rawFrameId", rawFrameId)
+                    .containsEntry("frame_id", rawFrameId);
         } finally {
             batchExecutor.close();
             eventBus.close();