feat: persist raw frames and publish raw facts

modules/apps/gb32960-ingest-app/pom.xml

@@ -33,6 +33,10 @@
             <groupId>com.lingniu.ingest</groupId>
             <artifactId>sink-mq</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lingniu.ingest</groupId>
+            <artifactId>sink-archive</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>

modules/apps/gb32960-ingest-app/src/main/resources/application.yml

@@ -102,6 +102,10 @@ lingniu:
           dlq: ${KAFKA_TOPIC_GB32960_DLQ:vehicle.dlq.gb32960.v1}
         consumer:
           enabled: false
+      archive:
+        enabled: ${SINK_ARCHIVE_ENABLED:true}
+        type: local
+        path: ${SINK_ARCHIVE_PATH:./archive/}
     event-file-store:
       enabled: false
     event-history:

modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java

@@ -5,6 +5,9 @@ import com.lingniu.ingest.protocol.gb32960.codec.Gb32960MessageDecoder;
 import com.lingniu.ingest.protocol.gb32960.config.Gb32960AutoConfiguration;
 import com.lingniu.ingest.protocol.gb32960.inbound.Gb32960NettyServer;
 import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration;
+import com.lingniu.ingest.sink.archive.ArchiveStore;
+import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
+import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
 import com.lingniu.ingest.sink.mq.KafkaEventSink;
 import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
 import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration;
@@ -25,6 +28,7 @@ class Gb32960IngestAppCompositionTest {
                     IngestCoreAutoConfiguration.class,
                     SessionCoreAutoConfiguration.class,
                     SinkMqAutoConfiguration.class,
+                    SinkArchiveAutoConfiguration.class,
                     Gb32960AutoConfiguration.class))
             .withAllowBeanDefinitionOverriding(true)
             .withBean("kafkaProducer", KafkaProducer.class, Gb32960IngestAppCompositionTest::kafkaProducer)
@@ -37,6 +41,8 @@ class Gb32960IngestAppCompositionTest {
                     "lingniu.ingest.sink.mq.type=kafka",
                     "lingniu.ingest.sink.mq.bootstrap-servers=localhost:9092",
                     "lingniu.ingest.sink.mq.consumer.enabled=false",
+                    "lingniu.ingest.sink.archive.enabled=true",
+                    "lingniu.ingest.sink.archive.path=target/test-archive-gb32960",
                     "lingniu.ingest.event-file-store.enabled=false",
                     "lingniu.ingest.event-history.enabled=false",
                     "lingniu.ingest.vehicle-state.enabled=false",
@@ -49,8 +55,9 @@ class Gb32960IngestAppCompositionTest {
             assertThat(context).hasSingleBean(Gb32960NettyServer.class);
             assertThat(context).hasSingleBean(KafkaEventSink.class);
             assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
+            assertThat(context).hasSingleBean(ArchiveStore.class);
+            assertThat(context).hasSingleBean(RawArchiveEventSink.class);
 
-            assertTypeNotPresent(context, "com.lingniu.ingest.sink.archive.ArchiveStore");
             assertTypeNotPresent(context, "com.lingniu.ingest.eventfilestore.EventFileStore");
         });
     }

modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java

@@ -25,6 +25,8 @@ class Gb32960IngestAppDefaultsTest {
                 .containsEntry("lingniu.ingest.gb32960.port", "${GB32960_PORT:32960}")
                 .containsEntry("lingniu.ingest.sink.mq.enabled", "${KAFKA_ENABLED:true}")
                 .containsEntry("lingniu.ingest.sink.mq.consumer.enabled", false)
+                .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
+                .containsEntry("lingniu.ingest.sink.archive.path", "${SINK_ARCHIVE_PATH:./archive/}")
                 .containsEntry("lingniu.ingest.event-file-store.enabled", false)
                 .containsEntry("lingniu.ingest.event-history.enabled", false)
                 .containsEntry("lingniu.ingest.vehicle-state.enabled", false)

modules/apps/jt808-ingest-app/pom.xml

@@ -37,6 +37,10 @@
             <groupId>com.lingniu.ingest</groupId>
             <artifactId>sink-mq</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.lingniu.ingest</groupId>
+            <artifactId>sink-archive</artifactId>
+        </dependency>
         <dependency>
             <groupId>org.springframework.boot</groupId>
             <artifactId>spring-boot-starter-web</artifactId>

modules/apps/jt808-ingest-app/src/main/resources/application.yml

@@ -72,6 +72,10 @@ lingniu:
           dlq: ${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1}
         consumer:
           enabled: false
+      archive:
+        enabled: ${SINK_ARCHIVE_ENABLED:true}
+        type: local
+        path: ${SINK_ARCHIVE_PATH:./archive/}
     event-file-store:
       enabled: false
     event-history:

modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java

@@ -6,6 +6,9 @@ import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder;
 import com.lingniu.ingest.protocol.jt808.config.Jt808AutoConfiguration;
 import com.lingniu.ingest.protocol.jt808.inbound.Jt808NettyServer;
 import com.lingniu.ingest.session.config.SessionCoreAutoConfiguration;
+import com.lingniu.ingest.sink.archive.ArchiveStore;
+import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
+import com.lingniu.ingest.sink.archive.config.SinkArchiveAutoConfiguration;
 import com.lingniu.ingest.sink.mq.KafkaEnvelopeDeadLetterSink;
 import com.lingniu.ingest.sink.mq.KafkaEventSink;
 import com.lingniu.ingest.sink.mq.SinkMqAutoConfiguration;
@@ -27,6 +30,7 @@ class Jt808IngestAppCompositionTest {
                     SessionCoreAutoConfiguration.class,
                     VehicleIdentityAutoConfiguration.class,
                     SinkMqAutoConfiguration.class,
+                    SinkArchiveAutoConfiguration.class,
                     Jt808AutoConfiguration.class))
             .withAllowBeanDefinitionOverriding(true)
             .withBean("kafkaProducer", KafkaProducer.class, Jt808IngestAppCompositionTest::kafkaProducer)
@@ -38,6 +42,8 @@ class Jt808IngestAppCompositionTest {
                     "lingniu.ingest.sink.mq.type=kafka",
                     "lingniu.ingest.sink.mq.bootstrap-servers=localhost:9092",
                     "lingniu.ingest.sink.mq.consumer.enabled=false",
+                    "lingniu.ingest.sink.archive.enabled=true",
+                    "lingniu.ingest.sink.archive.path=target/test-archive-jt808",
                     "lingniu.ingest.event-file-store.enabled=false",
                     "lingniu.ingest.event-history.enabled=false",
                     "lingniu.ingest.vehicle-state.enabled=false",
@@ -50,8 +56,9 @@ class Jt808IngestAppCompositionTest {
             assertThat(context).hasSingleBean(Jt808NettyServer.class);
             assertThat(context).hasSingleBean(KafkaEventSink.class);
             assertThat(context).hasSingleBean(KafkaEnvelopeDeadLetterSink.class);
+            assertThat(context).hasSingleBean(ArchiveStore.class);
+            assertThat(context).hasSingleBean(RawArchiveEventSink.class);
 
-            assertTypeNotPresent(context, "com.lingniu.ingest.sink.archive.ArchiveStore");
             assertTypeNotPresent(context, "com.lingniu.ingest.eventfilestore.EventFileStore");
         });
     }

modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java

@@ -28,6 +28,8 @@ class Jt808IngestAppDefaultsTest {
                 .containsEntry("lingniu.ingest.sink.mq.topics.realtime", "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}")
                 .containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}")
                 .containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1}")
+                .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
+                .containsEntry("lingniu.ingest.sink.archive.path", "${SINK_ARCHIVE_PATH:./archive/}")
                 .containsEntry("lingniu.ingest.event-file-store.enabled", false)
                 .containsEntry("lingniu.ingest.event-history.enabled", false)
                 .containsEntry("lingniu.ingest.vehicle-state.enabled", false)

modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java

@@ -4,15 +4,31 @@ import com.lingniu.ingest.api.event.RawArchiveKeys;
 import com.lingniu.ingest.api.event.VehicleEvent;
 import com.lingniu.ingest.api.sink.EventSink;
 
-import java.io.ByteArrayInputStream;
+import java.io.IOException;
+import java.nio.file.FileAlreadyExistsException;
+import java.nio.file.Files;
+import java.nio.file.LinkOption;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+import java.util.HexFormat;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 
-public final class RawArchiveEventSink implements EventSink {
+public final class RawArchiveEventSink implements EventSink, AutoCloseable {
 
-    private final ArchiveStore store;
+    private final Path root;
+    private final ExecutorService executor;
 
-    public RawArchiveEventSink(ArchiveStore store) {
+    public RawArchiveEventSink(Path root) {
-        this.store = store;
+        if (root == null) {
+            throw new IllegalArgumentException("root must not be null");
+        }
+        this.root = root.toAbsolutePath().normalize();
+        this.executor = Executors.newVirtualThreadPerTaskExecutor();
     }
 
     @Override
@@ -20,25 +36,115 @@ public final class RawArchiveEventSink implements EventSink {
         return "raw-archive";
     }
 
+    @Override
+    public boolean accepts(VehicleEvent event) {
+        return event instanceof VehicleEvent.RawArchive;
+    }
+
     @Override
     public CompletableFuture<Void> publish(VehicleEvent event) {
         if (!(event instanceof VehicleEvent.RawArchive raw)) {
             return CompletableFuture.completedFuture(null);
         }
-        try {
+        return CompletableFuture.runAsync(() -> write(raw), executor);
-            byte[] bytes = raw.rawBytes();
-            if (bytes == null) {
-                throw new IllegalArgumentException("raw archive bytes must not be null");
-            }
-            store.put(RawArchiveKeys.key(raw), new ByteArrayInputStream(bytes), bytes.length);
-            return CompletableFuture.completedFuture(null);
-        } catch (Exception e) {
-            return CompletableFuture.failedFuture(e);
-        }
     }
 
     @Override
-    public boolean accepts(VehicleEvent event) {
+    public void close() {
-        return event instanceof VehicleEvent.RawArchive;
+        executor.shutdown();
+    }
+
+    public Path root() {
+        return root;
+    }
+
+    private void write(VehicleEvent.RawArchive raw) {
+        byte[] bytes = raw.rawBytes() == null ? new byte[0] : raw.rawBytes();
+        if (bytes.length == 0) {
+            return;
+        }
+        String key = RawArchiveKeys.key(raw);
+        Path target = resolveKey(key);
+        try {
+            createDirectoriesInsideRoot(target.getParent());
+            rejectSymlinkPath(target);
+            try {
+                Files.write(target, bytes, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
+            } catch (FileAlreadyExistsException duplicate) {
+                byte[] existing = Files.readAllBytes(target);
+                if (!Arrays.equals(sha256(existing), sha256(bytes))) {
+                    throw new IOException("raw archive duplicate key has different content: " + key, duplicate);
+                }
+            }
+        } catch (IOException e) {
+            throw new RawArchiveWriteException("failed to write raw archive " + RawArchiveKeys.logicalUri(key), e);
+        }
+    }
+
+    private Path resolveKey(String key) {
+        Path target = root.resolve(key == null ? "" : key).normalize();
+        if (!target.startsWith(root)) {
+            throw new IllegalArgumentException("raw archive key escapes root: " + key);
+        }
+        return target;
+    }
+
+    private void createDirectoriesInsideRoot(Path directory) throws IOException {
+        rejectEscapedPath(directory);
+        if (Files.notExists(root, LinkOption.NOFOLLOW_LINKS)) {
+            Files.createDirectories(root);
+        }
+        Path current = root;
+        Path relative = root.relativize(directory);
+        for (Path segment : relative) {
+            current = current.resolve(segment);
+            if (Files.isSymbolicLink(current)) {
+                throw new IOException("raw archive path traverses symlink: " + current);
+            }
+            if (Files.notExists(current, LinkOption.NOFOLLOW_LINKS)) {
+                Files.createDirectory(current);
+            } else if (!Files.isDirectory(current, LinkOption.NOFOLLOW_LINKS)) {
+                throw new IOException("raw archive path segment is not a directory: " + current);
+            }
+        }
+    }
+
+    private void rejectSymlinkPath(Path target) throws IOException {
+        rejectEscapedPath(target);
+        Path current = root;
+        Path relative = root.relativize(target);
+        for (Path segment : relative) {
+            current = current.resolve(segment);
+            if (Files.isSymbolicLink(current)) {
+                throw new IOException("raw archive path traverses symlink: " + current);
+            }
+            if (Files.notExists(current, LinkOption.NOFOLLOW_LINKS)) {
+                return;
+            }
+        }
+    }
+
+    private void rejectEscapedPath(Path path) {
+        if (!path.normalize().startsWith(root)) {
+            throw new IllegalArgumentException("raw archive path escapes root: " + path);
+        }
+    }
+
+    private static byte[] sha256(byte[] bytes) {
+        try {
+            return MessageDigest.getInstance("SHA-256").digest(bytes);
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException("SHA-256 is unavailable", e);
+        }
+    }
+
+    static String checksum(byte[] bytes) {
+        return "sha256:" + HexFormat.of().formatHex(sha256(bytes == null ? new byte[0] : bytes));
+    }
+
+    public static final class RawArchiveWriteException extends RuntimeException {
+        public RawArchiveWriteException(String message, Throwable cause) {
+            super(message, cause);
+        }
     }
 }

modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java

@@ -5,31 +5,28 @@ import com.lingniu.ingest.sink.archive.LocalArchiveStore;
 import com.lingniu.ingest.sink.archive.RawArchiveEventSink;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
-import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 
+import java.nio.file.Path;
+
 @AutoConfiguration
 @EnableConfigurationProperties(SinkArchiveProperties.class)
-@ConditionalOnProperty(
+@ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "enabled", havingValue = "true", matchIfMissing = true)
-        prefix = "lingniu.ingest.sink.archive",
-        name = "enabled",
-        havingValue = "true",
-        matchIfMissing = true)
 public class SinkArchiveAutoConfiguration {
 
     @Bean
     @ConditionalOnMissingBean
     @ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "type", havingValue = "local", matchIfMissing = true)
     public ArchiveStore archiveStore(SinkArchiveProperties properties) {
-        return new LocalArchiveStore(properties.getPath());
+        return new LocalArchiveStore(Path.of(properties.getPath()));
     }
 
     @Bean
     @ConditionalOnMissingBean
-    @ConditionalOnBean(ArchiveStore.class)
+    @ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "type", havingValue = "local", matchIfMissing = true)
-    public RawArchiveEventSink rawArchiveEventSink(ArchiveStore store) {
+    public RawArchiveEventSink rawArchiveEventSink(SinkArchiveProperties properties) {
-        return new RawArchiveEventSink(store);
+        return new RawArchiveEventSink(Path.of(properties.getPath()));
     }
 }

modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java

@@ -7,7 +7,7 @@ public class SinkArchiveProperties {
 
     private boolean enabled = true;
     private String type = "local";
-    private String path = System.getProperty("java.io.tmpdir") + "/lingniu-archive";
+    private String path = "./archive/";
 
     public boolean isEnabled() {
         return enabled;

modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/RawArchiveEventSinkTest.java

@@ -1,63 +1,77 @@
 package com.lingniu.ingest.sink.archive;
 
 import com.lingniu.ingest.api.ProtocolId;
+import com.lingniu.ingest.api.event.RawArchiveKeys;
 import com.lingniu.ingest.api.event.VehicleEvent;
 import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
 
-import java.io.InputStream;
+import java.nio.file.Files;
+import java.nio.file.Path;
 import java.time.Instant;
 import java.util.Map;
-import java.util.concurrent.CompletionException;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.assertj.core.api.Assertions.assertThatThrownBy;
 
 class RawArchiveEventSinkTest {
 
+    @TempDir
+    Path tempDir;
+
     @Test
-    void failsFastWhenRawBytesAreNull() {
+    void writesRawBytesUsingSharedArchiveKey() throws Exception {
-        RawArchiveEventSink sink = new RawArchiveEventSink(new CapturingArchiveStore());
+        RawArchiveEventSink sink = new RawArchiveEventSink(tempDir);
-        VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
+        VehicleEvent.RawArchive raw = rawArchive(Map.of(), new byte[]{0x23, 0x23, 0x01});
-                "event-1",
+
-                "VIN123",
+        sink.publish(raw).join();
-                ProtocolId.GB32960,
+
-                Instant.parse("2026-06-23T00:00:00Z"),
+        String key = RawArchiveKeys.key(raw);
-                Instant.parse("2026-06-23T00:00:01Z"),
+        assertThat(Files.readAllBytes(tempDir.resolve(key))).containsExactly(0x23, 0x23, 0x01);
-                "trace-1",
+        assertThat(RawArchiveKeys.logicalUri(key)).isEqualTo("archive://" + key);
-                Map.of(),
+    }
-                2,
+
-                1,
+    @Test
-                null);
+    void rejectsArchiveKeysThatEscapeRoot() {
+        RawArchiveEventSink sink = new RawArchiveEventSink(tempDir);
+        VehicleEvent.RawArchive raw = rawArchive(
+                Map.of(RawArchiveKeys.META_KEY, "../escape.bin"),
+                new byte[]{0x01});
 
         assertThatThrownBy(() -> sink.publish(raw).join())
-                .isInstanceOf(CompletionException.class)
+                .hasCauseInstanceOf(IllegalArgumentException.class)
-                .hasCauseInstanceOf(IllegalArgumentException.class);
+                .hasMessageContaining("raw archive key escapes root");
     }
 
-    private static final class CapturingArchiveStore implements ArchiveStore {
+    @Test
+    void rejectsSymlinkTraversalBeforeCreatingDirectories() throws Exception {
+        Path root = tempDir.resolve("archive");
+        Path outside = tempDir.resolve("outside");
+        Files.createDirectories(root);
+        Files.createDirectories(outside);
+        Files.createSymbolicLink(root.resolve("link"), outside);
 
-        @Override
+        RawArchiveEventSink sink = new RawArchiveEventSink(root);
-        public String put(String key, InputStream data, long length) {
+        VehicleEvent.RawArchive raw = rawArchive(
-            return "archive://" + key;
+                Map.of(RawArchiveKeys.META_KEY, "link/nested/raw.bin"),
-        }
+                new byte[]{0x01});
 
-        @Override
+        assertThatThrownBy(() -> sink.publish(raw).join())
-        public String append(String key, byte[] chunk) {
+                .hasRootCauseMessage("raw archive path traverses symlink: " + root.resolve("link"));
-            return "archive://" + key;
+        assertThat(outside.resolve("nested")).doesNotExist();
-        }
+    }
 
-        @Override
+    private static VehicleEvent.RawArchive rawArchive(Map<String, String> metadata, byte[] rawBytes) {
-        public InputStream get(String key) {
+        return new VehicleEvent.RawArchive(
-            throw new UnsupportedOperationException();
+                "raw-1",
-        }
+                "VIN001",
-
+                ProtocolId.GB32960,
-        @Override
+                Instant.parse("2026-06-22T08:00:00Z"),
-        public boolean exists(String key) {
+                Instant.parse("2026-06-22T08:00:01Z"),
-            return false;
+                "trace-1",
-        }
+                metadata,
-
+                0x02,
-        @Override
+                0,
-        public long size(String key) {
+                rawBytes);
-            return 0;
-        }
     }
 }

modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java

@@ -8,10 +8,17 @@ 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.sink.mq.proto.ParseStatusProto;
 import com.lingniu.ingest.sink.mq.proto.TelemetryField;
 import com.lingniu.ingest.sink.mq.proto.TelemetrySnapshot.Builder;
 import com.lingniu.ingest.sink.mq.proto.VehicleEnvelope;
 
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+
 /**
  * 领域事件 → Protobuf Envelope 的纯函数映射。通过 switch 模式匹配穷尽处理，
  * 新增 {@link VehicleEvent} 子类型时编译期就会提示补齐分支。
@@ -68,17 +75,39 @@ public final class EnvelopeMapper {
                                     p.data() == null ? new byte[0] : p.data()))
                             .build());
             case VehicleEvent.RawArchive ra -> {
-                int size = ra.rawBytes() == null ? 0 : ra.rawBytes().length;
+                byte[] rawBytes = ra.rawBytes() == null ? new byte[0] : ra.rawBytes();
+                int size = rawBytes.length;
                 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);
                 // 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.mq.proto.RawArchiveRef.newBuilder()
                         .setUri(uri)
+                        .setChecksum(checksum)
                         .setSizeBytes(size)
                         .build());
+                b.setRawFrameFact(com.lingniu.ingest.sink.mq.proto.RawFrameFactPayload.newBuilder()
+                        .setFrameId(frameId)
+                        .setVehicleKey(vehicleKey)
+                        .setVin(nullToEmpty(ra.vin()))
+                        .setPhone(phone)
+                        .setMessageId(ra.command())
+                        .setSubType(ra.infoType())
+                        .setRawUri(uri)
+                        .setChecksum(checksum)
+                        .setRawSizeBytes(size)
+                        .setParseStatus(parseStatus(ra))
+                        .setParseError(parseError(ra))
+                        .setPeer(metadataValue(ra, "peer"))
+                        .putAllMetadata(ra.metadata() == null ? java.util.Map.of() : ra.metadata())
+                        .build());
             }
         }
         return b.build();
@@ -166,4 +195,45 @@ public final class EnvelopeMapper {
         String uri = raw.metadata() == null ? "" : raw.metadata().getOrDefault(RawArchiveKeys.META_URI, "");
         return uri == null || uri.isBlank() ? RawArchiveKeys.logicalUri(key) : uri;
     }
+
+    private static String metadataValue(VehicleEvent event, String key) {
+        if (event.metadata() == null) {
+            return "";
+        }
+        return nullToEmpty(event.metadata().get(key));
+    }
+
+    private static ParseStatusProto parseStatus(VehicleEvent.RawArchive raw) {
+        if (hasTruthyMetadata(raw, "frameError")
+                || hasTruthyMetadata(raw, "parseError")
+                || hasTruthyMetadata(raw, "processingError")) {
+            return ParseStatusProto.PARSE_STATUS_FAILED;
+        }
+        return ParseStatusProto.PARSE_STATUS_NOT_PARSED;
+    }
+
+    private static String parseError(VehicleEvent.RawArchive raw) {
+        String frameError = metadataValue(raw, "frameErrorMessage");
+        if (!frameError.isBlank()) {
+            return frameError;
+        }
+        String parseError = metadataValue(raw, "parseErrorMessage");
+        if (!parseError.isBlank()) {
+            return parseError;
+        }
+        return metadataValue(raw, "processingErrorMessage");
+    }
+
+    private static boolean hasTruthyMetadata(VehicleEvent event, String key) {
+        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);
+        }
+    }
 }

modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/EnvelopeMapperTelemetrySnapshotTest.java

@@ -90,5 +90,13 @@ class EnvelopeMapperTelemetrySnapshotTest {
         assertThat(envelope.getRawArchive().getUri()).isEqualTo(RawArchiveKeys.logicalUri(key));
         assertThat(envelope.getRawArchive().getSizeBytes()).isEqualTo(4);
         assertThat(envelope.getMetadataMap()).containsEntry(RawArchiveKeys.META_URI, RawArchiveKeys.logicalUri(key));
+        assertThat(envelope.hasRawFrameFact()).isTrue();
+        assertThat(envelope.getRawFrameFact().getFrameId()).startsWith("rf_");
+        assertThat(envelope.getRawFrameFact().getVehicleKey()).isEqualTo("VIN-JT808-001");
+        assertThat(envelope.getRawFrameFact().getVin()).isEqualTo("VIN-JT808-001");
+        assertThat(envelope.getRawFrameFact().getMessageId()).isEqualTo(0x0200);
+        assertThat(envelope.getRawFrameFact().getRawUri()).isEqualTo(RawArchiveKeys.logicalUri(key));
+        assertThat(envelope.getRawFrameFact().getRawSizeBytes()).isEqualTo(4);
+        assertThat(envelope.getRawFrameFact().getChecksum()).startsWith("sha256:");
     }
 }