fix: restore sink archive implementation

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

@@ -0,0 +1,17 @@
+package com.lingniu.ingest.sink.archive;
+
+import java.io.IOException;
+import java.io.InputStream;
+
+public interface ArchiveStore {
+
+    String put(String key, InputStream data, long length) throws IOException;
+
+    String append(String key, byte[] chunk) throws IOException;
+
+    InputStream get(String key) throws IOException;
+
+    boolean exists(String key);
+
+    long size(String key) throws IOException;
+}

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

@@ -0,0 +1,88 @@
+package com.lingniu.ingest.sink.archive;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.net.URI;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+
+public final class LocalArchiveStore implements ArchiveStore {
+
+    private final Path root;
+
+    public LocalArchiveStore(String root) {
+        this(rootPath(root));
+    }
+
+    public LocalArchiveStore(Path root) {
+        this.root = root.toAbsolutePath().normalize();
+    }
+
+    @Override
+    public String put(String key, InputStream data, long length) throws IOException {
+        if (data == null) {
+            throw new IllegalArgumentException("data must not be null");
+        }
+        Path target = resolve(key);
+        Files.createDirectories(target.getParent());
+        try (OutputStream out = Files.newOutputStream(
+                target,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.TRUNCATE_EXISTING,
+                StandardOpenOption.WRITE)) {
+            data.transferTo(out);
+        }
+        return target.toUri().toString();
+    }
+
+    @Override
+    public String append(String key, byte[] chunk) throws IOException {
+        Path target = resolve(key);
+        Files.createDirectories(target.getParent());
+        Files.write(
+                target,
+                chunk == null ? new byte[0] : chunk,
+                StandardOpenOption.CREATE,
+                StandardOpenOption.APPEND,
+                StandardOpenOption.WRITE);
+        return target.toUri().toString();
+    }
+
+    @Override
+    public InputStream get(String key) throws IOException {
+        return Files.newInputStream(resolve(key), StandardOpenOption.READ);
+    }
+
+    @Override
+    public boolean exists(String key) {
+        return Files.exists(resolve(key));
+    }
+
+    @Override
+    public long size(String key) throws IOException {
+        return Files.size(resolve(key));
+    }
+
+    private Path resolve(String key) {
+        if (key == null || key.isBlank()) {
+            throw new IllegalArgumentException("archive key must not be blank");
+        }
+        Path target = root.resolve(key).normalize();
+        if (!target.startsWith(root)) {
+            throw new IllegalArgumentException("archive key escapes root: " + key);
+        }
+        return target;
+    }
+
+    private static Path rootPath(String value) {
+        if (value == null || value.isBlank()) {
+            return Path.of(System.getProperty("java.io.tmpdir"), "lingniu-archive");
+        }
+        if (value.startsWith("file://")) {
+            return Path.of(URI.create(value));
+        }
+        return Path.of(value);
+    }
+}

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

@@ -0,0 +1,41 @@
+package com.lingniu.ingest.sink.archive;
+
+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.util.concurrent.CompletableFuture;
+
+public final class RawArchiveEventSink implements EventSink {
+
+    private final ArchiveStore store;
+
+    public RawArchiveEventSink(ArchiveStore store) {
+        this.store = store;
+    }
+
+    @Override
+    public String name() {
+        return "raw-archive";
+    }
+
+    @Override
+    public CompletableFuture<Void> publish(VehicleEvent event) {
+        if (!(event instanceof VehicleEvent.RawArchive raw)) {
+            return CompletableFuture.completedFuture(null);
+        }
+        byte[] bytes = raw.rawBytes() == null ? new byte[0] : raw.rawBytes();
+        try {
+            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) {
+        return event instanceof VehicleEvent.RawArchive;
+    }
+}

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

@@ -0,0 +1,33 @@
+package com.lingniu.ingest.sink.archive.config;
+
+import com.lingniu.ingest.sink.archive.ArchiveStore;
+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.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+
+@AutoConfiguration
+@EnableConfigurationProperties(SinkArchiveProperties.class)
+@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());
+    }
+
+    @Bean
+    @ConditionalOnMissingBean
+    public RawArchiveEventSink rawArchiveEventSink(ArchiveStore store) {
+        return new RawArchiveEventSink(store);
+    }
+}

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

@@ -0,0 +1,35 @@
+package com.lingniu.ingest.sink.archive.config;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@ConfigurationProperties(prefix = "lingniu.ingest.sink.archive")
+public class SinkArchiveProperties {
+
+    private boolean enabled = true;
+    private String type = "local";
+    private String path = System.getProperty("java.io.tmpdir") + "/lingniu-archive";
+
+    public boolean isEnabled() {
+        return enabled;
+    }
+
+    public void setEnabled(boolean enabled) {
+        this.enabled = enabled;
+    }
+
+    public String getType() {
+        return type;
+    }
+
+    public void setType(String type) {
+        this.type = type;
+    }
+
+    public String getPath() {
+        return path;
+    }
+
+    public void setPath(String path) {
+        this.path = path;
+    }
+}