feat: add raw archive store contract

modules/sinks/raw-archive-store/pom.xml

@@ -0,0 +1,30 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <groupId>com.lingniu.ingest</groupId>
+        <artifactId>lingniu-vehicle-ingest</artifactId>
+        <version>0.1.0-SNAPSHOT</version>
+        <relativePath>../../../pom.xml</relativePath>
+    </parent>
+    <artifactId>raw-archive-store</artifactId>
+    <name>raw-archive-store</name>
+    <description>Raw frame archive writer and reader contracts.</description>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.lingniu.ingest</groupId>
+            <artifactId>ingest-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.assertj</groupId>
+            <artifactId>assertj-core</artifactId>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStore.java

@@ -0,0 +1,110 @@
+package com.lingniu.ingest.rawarchive;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+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.time.ZoneId;
+import java.util.HexFormat;
+
+public final class LocalRawArchiveStore implements RawArchiveWriter, RawArchiveReader {
+
+    private static final ZoneId PARTITION_ZONE = ZoneId.of("Asia/Shanghai");
+
+    private final Path root;
+    private final String uriPrefix;
+
+    public LocalRawArchiveStore(Path root, String uriPrefix) {
+        if (root == null) {
+            throw new IllegalArgumentException("root must not be null");
+        }
+        this.root = root.toAbsolutePath().normalize();
+        String prefix = uriPrefix == null || uriPrefix.isBlank() ? "archive://raw" : uriPrefix.trim();
+        this.uriPrefix = prefix.endsWith("/") ? prefix.substring(0, prefix.length() - 1) : prefix;
+    }
+
+    @Override
+    public RawArchiveReceipt write(RawArchiveWriteRequest request) throws IOException {
+        String key = key(request);
+        Path target = resolveKey(key);
+        Files.createDirectories(target.getParent());
+        rejectSymlinkPath(target);
+        byte[] bytes = request.rawBytes();
+        Files.write(target, bytes, StandardOpenOption.CREATE_NEW, StandardOpenOption.WRITE);
+        return new RawArchiveReceipt(uriPrefix + "/" + key, checksum(bytes), bytes.length);
+    }
+
+    @Override
+    public byte[] read(String rawUri) throws IOException {
+        String key = keyFromUri(rawUri);
+        Path target = resolveKey(key);
+        rejectSymlinkPath(target);
+        return Files.readAllBytes(target);
+    }
+
+    private String key(RawArchiveWriteRequest request) {
+        var date = request.receivedAt().atZone(PARTITION_ZONE).toLocalDate();
+        return "%04d/%02d/%02d/%s/%s/%s.bin".formatted(
+                date.getYear(),
+                date.getMonthValue(),
+                date.getDayOfMonth(),
+                request.protocol().name(),
+                safeSegment(request.vehicleKey()),
+                safeSegment(request.frameId()));
+    }
+
+    private String keyFromUri(String rawUri) {
+        if (rawUri == null || rawUri.isBlank()) {
+            throw new IllegalArgumentException("rawUri must not be blank");
+        }
+        String normalized = rawUri.trim();
+        String expected = uriPrefix + "/";
+        if (!normalized.startsWith(expected)) {
+            throw new IllegalArgumentException("rawUri does not use expected prefix: " + rawUri);
+        }
+        return normalized.substring(expected.length());
+    }
+
+    private Path resolveKey(String key) {
+        Path target = root.resolve(key).normalize();
+        if (!target.startsWith(root)) {
+            throw new IllegalArgumentException("raw archive key escapes root: " + key);
+        }
+        return target;
+    }
+
+    private void rejectSymlinkPath(Path target) throws IOException {
+        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 static String safeSegment(String value) {
+        String cleaned = value == null ? "" : value.trim();
+        if (cleaned.isBlank()) {
+            return "unknown";
+        }
+        return cleaned.replaceAll("[^A-Za-z0-9._-]", "_");
+    }
+
+    private static String checksum(byte[] bytes) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            return "sha256:" + HexFormat.of().formatHex(digest.digest(bytes));
+        } catch (NoSuchAlgorithmException e) {
+            throw new IllegalStateException("SHA-256 is unavailable", e);
+        }
+    }
+}

modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReader.java

@@ -0,0 +1,7 @@
+package com.lingniu.ingest.rawarchive;
+
+import java.io.IOException;
+
+public interface RawArchiveReader {
+    byte[] read(String rawUri) throws IOException;
+}

modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveReceipt.java

@@ -0,0 +1,19 @@
+package com.lingniu.ingest.rawarchive;
+
+public record RawArchiveReceipt(String rawUri, String checksum, long sizeBytes) {
+    public RawArchiveReceipt {
+        rawUri = required(rawUri, "rawUri");
+        checksum = required(checksum, "checksum");
+        if (sizeBytes < 0) {
+            throw new IllegalArgumentException("sizeBytes must not be negative");
+        }
+    }
+
+    private static String required(String value, String field) {
+        String cleaned = value == null ? "" : value.trim();
+        if (cleaned.isBlank()) {
+            throw new IllegalArgumentException(field + " must not be blank");
+        }
+        return cleaned;
+    }
+}

modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriteRequest.java

@@ -0,0 +1,41 @@
+package com.lingniu.ingest.rawarchive;
+
+import com.lingniu.ingest.api.ProtocolId;
+
+import java.time.Instant;
+
+public record RawArchiveWriteRequest(
+        ProtocolId protocol,
+        String vehicleKey,
+        String frameId,
+        Instant receivedAt,
+        byte[] rawBytes
+) {
+    public RawArchiveWriteRequest {
+        if (protocol == null) {
+            throw new IllegalArgumentException("protocol must not be null");
+        }
+        vehicleKey = required(vehicleKey, "vehicleKey");
+        frameId = required(frameId, "frameId");
+        if (receivedAt == null) {
+            throw new IllegalArgumentException("receivedAt must not be null");
+        }
+        if (rawBytes == null || rawBytes.length == 0) {
+            throw new IllegalArgumentException("rawBytes must not be empty");
+        }
+        rawBytes = rawBytes.clone();
+    }
+
+    @Override
+    public byte[] rawBytes() {
+        return rawBytes.clone();
+    }
+
+    private static String required(String value, String field) {
+        String cleaned = value == null ? "" : value.trim();
+        if (cleaned.isBlank()) {
+            throw new IllegalArgumentException(field + " must not be blank");
+        }
+        return cleaned;
+    }
+}

modules/sinks/raw-archive-store/src/main/java/com/lingniu/ingest/rawarchive/RawArchiveWriter.java

@@ -0,0 +1,7 @@
+package com.lingniu.ingest.rawarchive;
+
+import java.io.IOException;
+
+public interface RawArchiveWriter {
+    RawArchiveReceipt write(RawArchiveWriteRequest request) throws IOException;
+}

modules/sinks/raw-archive-store/src/test/java/com/lingniu/ingest/rawarchive/LocalRawArchiveStoreTest.java

@@ -0,0 +1,50 @@
+package com.lingniu.ingest.rawarchive;
+
+import com.lingniu.ingest.api.ProtocolId;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.io.TempDir;
+
+import java.nio.file.Path;
+import java.time.Instant;
+import java.util.HexFormat;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+
+class LocalRawArchiveStoreTest {
+
+    @TempDir
+    Path tempDir;
+
+    @Test
+    void writesReadableRawObjectWithArchiveUriAndChecksum() throws Exception {
+        LocalRawArchiveStore store = new LocalRawArchiveStore(tempDir, "archive://raw");
+        byte[] bytes = HexFormat.of().parseHex("7e020000007e");
+
+        RawArchiveReceipt receipt = store.write(new RawArchiveWriteRequest(
+                ProtocolId.JT808,
+                "jt808:013912345678",
+                "frame-1",
+                Instant.parse("2026-06-29T00:00:00Z"),
+                bytes));
+
+        assertThat(receipt.rawUri()).isEqualTo("archive://raw/2026/06/29/JT808/jt808_013912345678/frame-1.bin");
+        assertThat(receipt.checksum()).startsWith("sha256:");
+        assertThat(receipt.sizeBytes()).isEqualTo(bytes.length);
+        assertThat(store.read(receipt.rawUri())).containsExactly(bytes);
+    }
+
+    @Test
+    void rejectsBlankFrameId() {
+        LocalRawArchiveStore store = new LocalRawArchiveStore(tempDir, "archive://raw");
+
+        assertThatThrownBy(() -> store.write(new RawArchiveWriteRequest(
+                ProtocolId.GB32960,
+                "VIN001",
+                "",
+                Instant.parse("2026-06-29T00:00:00Z"),
+                new byte[]{1})))
+                .isInstanceOf(IllegalArgumentException.class)
+                .hasMessageContaining("frameId");
+    }
+}

pom.xml

@@ -26,6 +26,7 @@
         <module>modules/core/observability</module>
         <module>modules/sinks/sink-mq</module>
         <module>modules/sinks/sink-archive</module>
+        <module>modules/sinks/raw-archive-store</module>
         <module>modules/sinks/event-file-store</module>
         <module>modules/services/event-history-service</module>
         <module>modules/services/vehicle-state-service</module>
@@ -184,6 +185,11 @@
                 <artifactId>sink-archive</artifactId>
                 <version>${project.version}</version>
             </dependency>
+            <dependency>
+                <groupId>com.lingniu.ingest</groupId>
+                <artifactId>raw-archive-store</artifactId>
+                <version>${project.version}</version>
+            </dependency>
             <dependency>
                 <groupId>com.lingniu.ingest</groupId>
                 <artifactId>event-file-store</artifactId>