diff --git a/modules/apps/gb32960-ingest-app/pom.xml b/modules/apps/gb32960-ingest-app/pom.xml
index 49bf5972..c8a70b82 100644
--- a/modules/apps/gb32960-ingest-app/pom.xml
+++ b/modules/apps/gb32960-ingest-app/pom.xml
@@ -33,6 +33,10 @@
com.lingniu.ingest
sink-mq
+
+ com.lingniu.ingest
+ sink-archive
+
org.springframework.boot
spring-boot-starter-web
diff --git a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml b/modules/apps/gb32960-ingest-app/src/main/resources/application.yml
index 92f1591e..0c9bc30d 100644
--- a/modules/apps/gb32960-ingest-app/src/main/resources/application.yml
+++ b/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:
diff --git a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java
index 8640ed51..cc521c67 100644
--- a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppCompositionTest.java
+++ b/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");
});
}
diff --git a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java b/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java
index 765d218a..2796b0f1 100644
--- a/modules/apps/gb32960-ingest-app/src/test/java/com/lingniu/ingest/gb32960app/Gb32960IngestAppDefaultsTest.java
+++ b/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)
diff --git a/modules/apps/jt808-ingest-app/pom.xml b/modules/apps/jt808-ingest-app/pom.xml
index c15e4b92..30cdfbd6 100644
--- a/modules/apps/jt808-ingest-app/pom.xml
+++ b/modules/apps/jt808-ingest-app/pom.xml
@@ -37,6 +37,10 @@
com.lingniu.ingest
sink-mq
+
+ com.lingniu.ingest
+ sink-archive
+
org.springframework.boot
spring-boot-starter-web
diff --git a/modules/apps/jt808-ingest-app/src/main/resources/application.yml b/modules/apps/jt808-ingest-app/src/main/resources/application.yml
index 63660f81..698b5e71 100644
--- a/modules/apps/jt808-ingest-app/src/main/resources/application.yml
+++ b/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:
diff --git a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java
index 8e384b50..b26c87b1 100644
--- a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppCompositionTest.java
+++ b/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");
});
}
diff --git a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java b/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java
index 5679a261..d5d1d973 100644
--- a/modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java
+++ b/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)
diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java
index e148fd3b..5a1e9e8d 100644
--- a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/RawArchiveEventSink.java
+++ b/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) {
- this.store = store;
+ public RawArchiveEventSink(Path root) {
+ 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 publish(VehicleEvent event) {
if (!(event instanceof VehicleEvent.RawArchive raw)) {
return CompletableFuture.completedFuture(null);
}
- try {
- 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);
- }
+ return CompletableFuture.runAsync(() -> write(raw), executor);
}
@Override
- public boolean accepts(VehicleEvent event) {
- return event instanceof VehicleEvent.RawArchive;
+ public void close() {
+ 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);
+ }
}
}
diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java
index f64c7c86..0bf04f4e 100644
--- a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveAutoConfiguration.java
+++ b/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(
- prefix = "lingniu.ingest.sink.archive",
- name = "enabled",
- havingValue = "true",
- matchIfMissing = true)
+@ConditionalOnProperty(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)
- public RawArchiveEventSink rawArchiveEventSink(ArchiveStore store) {
- return new RawArchiveEventSink(store);
+ @ConditionalOnProperty(prefix = "lingniu.ingest.sink.archive", name = "type", havingValue = "local", matchIfMissing = true)
+ public RawArchiveEventSink rawArchiveEventSink(SinkArchiveProperties properties) {
+ return new RawArchiveEventSink(Path.of(properties.getPath()));
}
}
diff --git a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java b/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java
index f64e2acb..32668078 100644
--- a/modules/sinks/sink-archive/src/main/java/com/lingniu/ingest/sink/archive/config/SinkArchiveProperties.java
+++ b/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;
diff --git a/modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/RawArchiveEventSinkTest.java b/modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/RawArchiveEventSinkTest.java
index 7ffe5385..c6534851 100644
--- a/modules/sinks/sink-archive/src/test/java/com/lingniu/ingest/sink/archive/RawArchiveEventSinkTest.java
+++ b/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() {
- RawArchiveEventSink sink = new RawArchiveEventSink(new CapturingArchiveStore());
- VehicleEvent.RawArchive raw = new VehicleEvent.RawArchive(
- "event-1",
- "VIN123",
- ProtocolId.GB32960,
- Instant.parse("2026-06-23T00:00:00Z"),
- Instant.parse("2026-06-23T00:00:01Z"),
- "trace-1",
- Map.of(),
- 2,
- 1,
- null);
+ void writesRawBytesUsingSharedArchiveKey() throws Exception {
+ RawArchiveEventSink sink = new RawArchiveEventSink(tempDir);
+ VehicleEvent.RawArchive raw = rawArchive(Map.of(), new byte[]{0x23, 0x23, 0x01});
+
+ sink.publish(raw).join();
+
+ String key = RawArchiveKeys.key(raw);
+ assertThat(Files.readAllBytes(tempDir.resolve(key))).containsExactly(0x23, 0x23, 0x01);
+ assertThat(RawArchiveKeys.logicalUri(key)).isEqualTo("archive://" + key);
+ }
+
+ @Test
+ 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
- public String put(String key, InputStream data, long length) {
- return "archive://" + key;
- }
+ RawArchiveEventSink sink = new RawArchiveEventSink(root);
+ VehicleEvent.RawArchive raw = rawArchive(
+ Map.of(RawArchiveKeys.META_KEY, "link/nested/raw.bin"),
+ new byte[]{0x01});
- @Override
- public String append(String key, byte[] chunk) {
- return "archive://" + key;
- }
+ assertThatThrownBy(() -> sink.publish(raw).join())
+ .hasRootCauseMessage("raw archive path traverses symlink: " + root.resolve("link"));
+ assertThat(outside.resolve("nested")).doesNotExist();
+ }
- @Override
- public InputStream get(String key) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public boolean exists(String key) {
- return false;
- }
-
- @Override
- public long size(String key) {
- return 0;
- }
+ private static VehicleEvent.RawArchive rawArchive(Map metadata, byte[] rawBytes) {
+ return new VehicleEvent.RawArchive(
+ "raw-1",
+ "VIN001",
+ ProtocolId.GB32960,
+ Instant.parse("2026-06-22T08:00:00Z"),
+ Instant.parse("2026-06-22T08:00:01Z"),
+ "trace-1",
+ metadata,
+ 0x02,
+ 0,
+ rawBytes);
}
}
diff --git a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java b/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java
index 111022f1..05a36ad0 100644
--- a/modules/sinks/sink-mq/src/main/java/com/lingniu/ingest/sink/mq/EnvelopeMapper.java
+++ b/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);
+ }
+ }
}
diff --git a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/EnvelopeMapperTelemetrySnapshotTest.java b/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/EnvelopeMapperTelemetrySnapshotTest.java
index 4ca1f25a..6e5d806e 100644
--- a/modules/sinks/sink-mq/src/test/java/com/lingniu/ingest/sink/mq/EnvelopeMapperTelemetrySnapshotTest.java
+++ b/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:");
}
}