docs: plan gb32960 duckdb hot history

2026-06-23 19:39:06 +08:00
parent 9d7af780cd
commit 620821b624
1 changed files with 810 additions and 0 deletions
--- a/docs/superpowers/plans/2026-06-23-gb32960-history-duckdb-hot-store.md
+++ b/docs/superpowers/plans/2026-06-23-gb32960-history-duckdb-hot-store.md
@@ -0,0 +1,810 @@
+# GB32960 History DuckDB Hot Store Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Replace the current high-write-risk Parquet rewrite history path with a DuckDB hot history store that supports fast `vin + time` queries and full RAW frame replay.
+
+**Architecture:** Add a new `DuckDbHotEventFileStore` implementation behind the existing `EventFileStore` interface, backed by append/upsert DuckDB tables instead of rewriting per-vehicle Parquet files. Keep the current HTTP history API and RAW archive reader intact, then switch `vehicle-history-app` to the hot DuckDB store by configuration.
+
+**Tech Stack:** Java 26, Spring Boot, DuckDB JDBC, JUnit 5, AssertJ, existing Kafka `VehicleEnvelope`, existing `ArchiveStore`.
+
+---
+
+## Scope
+
+This plan implements Phase 1 from `docs/superpowers/specs/2026-06-23-gb32960-production-readiness-design.md`.
+
+Included:
+
+- DuckDB hot event table for `EventFileRecord`.
+- Idempotent writes by `event_id`.
+- Query by protocol/date/VIN/eventType/eventTime/order/limit.
+- Lookup by `rawArchiveUri`.
+- Spring configuration to select the hot store.
+- Tests that prove history endpoints still replay RAW frames.
+- Local verification using the existing `vehicle-history-app`.
+
+Not included in this plan:
+
+- MySQL daily statistics.
+- VIN-to-platform local mapping.
+- Full telemetry point columnar table.
+- Alarm timeline MySQL tables.
+- Cold Parquet export.
+
+Those will be separate plans after this foundation lands.
+
+## Files
+
+- Create: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
+- Create: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
+- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java`
+- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java`
+- Modify: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java`
+- Modify: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
+- Modify: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
+- Modify: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java`
+
+## Task 1: Add Hot Store Configuration
+
+**Files:**
+- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreProperties.java`
+- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfiguration.java`
+- Test: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/config/EventFileStoreAutoConfigurationTest.java`
+
+- [ ] **Step 1: Write the failing auto-configuration test**
+
+Add a test that sets `lingniu.ingest.event-file-store.storage=duckdb-hot` and expects the bean class to be `DuckDbHotEventFileStore`.
+
+```java
+@Test
+void createsDuckDbHotStoreWhenStorageIsDuckDbHot() {
+    contextRunner
+            .withPropertyValues(
+                    "lingniu.ingest.event-file-store.enabled=true",
+                    "lingniu.ingest.event-file-store.storage=duckdb-hot",
+                    "lingniu.ingest.event-file-store.path=" + tempDir.resolve("history"))
+            .run(context -> assertThat(context)
+                    .hasSingleBean(EventFileStore.class)
+                    .getBean(EventFileStore.class)
+                    .isInstanceOf(DuckDbHotEventFileStore.class));
+}
+```
+
+- [ ] **Step 2: Run the new test and verify it fails**
+
+Run:
+
+```bash
+mvn -pl :event-file-store -Dtest=EventFileStoreAutoConfigurationTest#createsDuckDbHotStoreWhenStorageIsDuckDbHot test
+```
+
+Expected: compilation failure because `DuckDbHotEventFileStore` and `storage` property do not exist.
+
+- [ ] **Step 3: Add the `storage` property**
+
+Add to `EventFileStoreProperties`:
+
+```java
+/**
+ * Storage backend. `duckdb-hot` is the production backend; `parquet-sidecar`
+ * keeps the previous implementation available for compatibility tests.
+ */
+private String storage = "duckdb-hot";
+
+public String getStorage() {
+    return storage;
+}
+
+public void setStorage(String storage) {
+    this.storage = storage;
+}
+```
+
+- [ ] **Step 4: Switch auto-configuration by storage mode**
+
+Change `eventFileStore(...)` to:
+
+```java
+@Bean
+@ConditionalOnMissingBean
+public EventFileStore eventFileStore(EventFileStoreProperties properties,
+                                     ObjectProvider<ObjectMapper> objectMapper) {
+    ObjectMapper mapper = mapper(objectMapper);
+    Path root = Path.of(properties.getPath());
+    ZoneId zoneId = ZoneId.of(properties.getZoneId());
+    String storage = properties.getStorage() == null ? "" : properties.getStorage().trim();
+    return switch (storage) {
+        case "", "duckdb-hot" -> new DuckDbHotEventFileStore(root, zoneId, mapper);
+        case "parquet-sidecar" -> new DuckDbParquetEventFileStore(root, zoneId, mapper);
+        default -> throw new IllegalStateException(
+                "unsupported event-file-store storage: " + properties.getStorage());
+    };
+}
+```
+
+Import `DuckDbHotEventFileStore`.
+
+- [ ] **Step 5: Run configuration tests**
+
+Run:
+
+```bash
+mvn -pl :event-file-store -Dtest=EventFileStoreAutoConfigurationTest test
+```
+
+Expected: tests pass after the store class exists in Task 2.
+
+## Task 2: Implement DuckDB Hot Store Schema and Writes
+
+**Files:**
+- Create: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
+- Create: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
+
+- [ ] **Step 1: Write failing append/query/idempotency tests**
+
+Create `DuckDbHotEventFileStoreTest`:
+
+```java
+class DuckDbHotEventFileStoreTest {
+
+    @TempDir
+    Path tempDir;
+
+    @Test
+    void appendsRecordsAndQueriesByVinTypeAndTime() throws Exception {
+        EventFileStore store = store();
+        store.append(rawRecord("raw-1", "VIN001", "2026-06-23T01:00:00Z"));
+        store.append(rawRecord("raw-2", "VIN002", "2026-06-23T01:00:01Z"));
+        store.append(rawRecord("raw-3", "VIN001", "2026-06-23T01:00:02Z"));
+
+        EventFileQuery query = new EventFileQuery(
+                ProtocolId.GB32960,
+                LocalDate.parse("2026-06-23"),
+                LocalDate.parse("2026-06-23"),
+                Instant.parse("2026-06-23T01:00:00Z"),
+                Instant.parse("2026-06-23T01:00:03Z"),
+                EventFileQuery.Order.DESC,
+                2,
+                "VIN001",
+                "RAW_ARCHIVE");
+
+        assertThat(store.query(query))
+                .extracting(EventFileRecord::eventId)
+                .containsExactly("raw-3", "raw-1");
+    }
+
+    @Test
+    void appendAllIsIdempotentByEventId() throws Exception {
+        EventFileStore store = store();
+        EventFileRecord original = rawRecord("same-id", "VIN001", "2026-06-23T01:00:00Z");
+        EventFileRecord replacement = new EventFileRecord(
+                "same-id",
+                ProtocolId.GB32960,
+                "RAW_ARCHIVE",
+                "VIN001",
+                Instant.parse("2026-06-23T01:00:05Z"),
+                Instant.parse("2026-06-23T01:00:06Z"),
+                "archive://replacement.bin",
+                Map.of("source", "replacement"),
+                "{\"replacement\":true}");
+
+        store.appendAll(List.of(original, replacement));
+
+        assertThat(store.query(new EventFileQuery(
+                ProtocolId.GB32960,
+                LocalDate.parse("2026-06-23"),
+                LocalDate.parse("2026-06-23"),
+                EventFileQuery.Order.ASC,
+                10,
+                "VIN001",
+                "RAW_ARCHIVE")))
+                .singleElement()
+                .satisfies(record -> {
+                    assertThat(record.eventId()).isEqualTo("same-id");
+                    assertThat(record.rawArchiveUri()).isEqualTo("archive://replacement.bin");
+                });
+    }
+
+    @Test
+    void findsRecordByRawArchiveUri() throws Exception {
+        EventFileStore store = store();
+        EventFileRecord record = rawRecord("raw-uri", "VIN001", "2026-06-23T01:00:00Z");
+        store.append(record);
+
+        EventFileRecord found = store.findByRawArchiveUri(record.rawArchiveUri());
+
+        assertThat(found).isNotNull();
+        assertThat(found.eventId()).isEqualTo("raw-uri");
+    }
+
+    private EventFileStore store() {
+        return new DuckDbHotEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), new ObjectMapper());
+    }
+
+    private static EventFileRecord rawRecord(String id, String vin, String eventTime) {
+        return new EventFileRecord(
+                id,
+                ProtocolId.GB32960,
+                "RAW_ARCHIVE",
+                vin,
+                Instant.parse(eventTime),
+                Instant.parse(eventTime).plusMillis(100),
+                "archive://" + id + ".bin",
+                Map.of("platformAccount", "Hyundai", "command", "REALTIME_REPORT"),
+                "{\"eventId\":\"" + id + "\"}");
+    }
+}
+```
+
+- [ ] **Step 2: Run the test and verify it fails**
+
+Run:
+
+```bash
+mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
+```
+
+Expected: compilation failure because `DuckDbHotEventFileStore` does not exist.
+
+- [ ] **Step 3: Create the hot store class**
+
+Create `DuckDbHotEventFileStore` with this structure:
+
+```java
+public final class DuckDbHotEventFileStore implements EventFileStore {
+
+    private static final TypeReference<Map<String, String>> STRING_MAP = new TypeReference<>() {};
+
+    private final Path root;
+    private final Path dbPath;
+    private final ZoneId partitionZone;
+    private final ObjectMapper objectMapper;
+    private volatile boolean initialized;
+
+    public DuckDbHotEventFileStore(Path root, ZoneId partitionZone) {
+        this(root, partitionZone, new ObjectMapper());
+    }
+
+    public DuckDbHotEventFileStore(Path root, ZoneId partitionZone, ObjectMapper objectMapper) {
+        if (root == null) {
+            throw new IllegalArgumentException("root must not be null");
+        }
+        this.root = root.toAbsolutePath();
+        this.dbPath = this.root.resolve("events.duckdb");
+        this.partitionZone = partitionZone == null ? ZoneId.of("Asia/Shanghai") : partitionZone;
+        this.objectMapper = objectMapper == null ? new ObjectMapper() : objectMapper;
+    }
+
+    @Override
+    public synchronized void appendAll(List<EventFileRecord> records) throws IOException {
+        if (records == null || records.isEmpty()) {
+            return;
+        }
+        ensureInitialized();
+        try (Connection connection = DriverManager.getConnection(jdbcUrl())) {
+            connection.setAutoCommit(false);
+            try {
+                upsertRecords(connection, records);
+                connection.commit();
+            } catch (SQLException | IOException e) {
+                connection.rollback();
+                throw e;
+            } finally {
+                connection.setAutoCommit(true);
+            }
+        } catch (SQLException e) {
+            throw new IOException("write duckdb hot event store failed", e);
+        }
+    }
+
+    @Override
+    public List<EventFileRecord> query(EventFileQuery query) throws IOException {
+        ensureInitialized();
+        // Implement in Task 3.
+        return List.of();
+    }
+
+    @Override
+    public EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
+        ensureInitialized();
+        // Implement in Task 3.
+        return null;
+    }
+}
+```
+
+- [ ] **Step 4: Add schema initialization**
+
+Add:
+
+```java
+private void ensureInitialized() throws IOException {
+    if (initialized) {
+        return;
+    }
+    synchronized (this) {
+        if (initialized) {
+            return;
+        }
+        Files.createDirectories(root);
+        try (Connection connection = DriverManager.getConnection(jdbcUrl());
+             Statement statement = connection.createStatement()) {
+            statement.execute("""
+                    CREATE TABLE IF NOT EXISTS event_records (
+                        event_id VARCHAR PRIMARY KEY,
+                        protocol VARCHAR NOT NULL,
+                        event_type VARCHAR NOT NULL,
+                        vin VARCHAR NOT NULL,
+                        event_time_ms BIGINT NOT NULL,
+                        ingest_time_ms BIGINT NOT NULL,
+                        partition_date DATE NOT NULL,
+                        raw_archive_uri VARCHAR NOT NULL,
+                        metadata_json VARCHAR NOT NULL,
+                        payload_json VARCHAR NOT NULL
+                    )
+                    """);
+            statement.execute("""
+                    CREATE INDEX IF NOT EXISTS event_records_protocol_date_time_idx
+                    ON event_records(protocol, partition_date, event_time_ms)
+                    """);
+            statement.execute("""
+                    CREATE INDEX IF NOT EXISTS event_records_vin_date_time_idx
+                    ON event_records(protocol, vin, partition_date, event_time_ms)
+                    """);
+            statement.execute("""
+                    CREATE INDEX IF NOT EXISTS event_records_vin_type_date_time_idx
+                    ON event_records(protocol, vin, event_type, partition_date, event_time_ms)
+                    """);
+            statement.execute("""
+                    CREATE INDEX IF NOT EXISTS event_records_raw_archive_uri_idx
+                    ON event_records(raw_archive_uri)
+                    """);
+            initialized = true;
+        } catch (SQLException e) {
+            throw new IOException("initialize duckdb hot event store failed", e);
+        }
+    }
+}
+```
+
+- [ ] **Step 5: Add idempotent batch upsert**
+
+Use DuckDB `INSERT OR REPLACE` inside one transaction:
+
+```java
+private void upsertRecords(Connection connection, List<EventFileRecord> records)
+        throws SQLException, IOException {
+    try (PreparedStatement ps = connection.prepareStatement("""
+            INSERT OR REPLACE INTO event_records VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """)) {
+        for (EventFileRecord record : records) {
+            ps.setString(1, record.eventId());
+            ps.setString(2, record.protocol().name());
+            ps.setString(3, record.eventType());
+            ps.setString(4, record.vin());
+            ps.setLong(5, record.eventTime().toEpochMilli());
+            ps.setLong(6, record.ingestTime().toEpochMilli());
+            ps.setString(7, LocalDate.ofInstant(record.eventTime(), partitionZone).toString());
+            ps.setString(8, record.rawArchiveUri());
+            ps.setString(9, objectMapper.writeValueAsString(record.metadata()));
+            ps.setString(10, record.payloadJson());
+            ps.addBatch();
+        }
+        ps.executeBatch();
+    }
+}
+```
+
+- [ ] **Step 6: Run the hot store tests**
+
+Run:
+
+```bash
+mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
+```
+
+Expected: query tests still fail until Task 3 implements reads; append initialization should compile.
+
+## Task 3: Implement Hot Store Queries and Raw URI Lookup
+
+**Files:**
+- Modify: `modules/sinks/event-file-store/src/main/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStore.java`
+- Test: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbHotEventFileStoreTest.java`
+
+- [ ] **Step 1: Implement `query(EventFileQuery)`**
+
+Use prepared statements for every external value:
+
+```java
+@Override
+public List<EventFileRecord> query(EventFileQuery query) throws IOException {
+    ensureInitialized();
+    String order = query.order() == EventFileQuery.Order.DESC ? "DESC" : "ASC";
+    StringBuilder where = new StringBuilder("""
+            WHERE protocol = ?
+              AND partition_date BETWEEN CAST(? AS DATE) AND CAST(? AS DATE)
+            """);
+    if (query.vin() != null) {
+        where.append("  AND vin = ?\n");
+    }
+    if (query.eventType() != null) {
+        where.append("  AND event_type = ?\n");
+    }
+    if (query.eventTimeFrom() != null) {
+        where.append("  AND event_time_ms >= ?\n");
+    }
+    if (query.eventTimeTo() != null) {
+        where.append("  AND event_time_ms <= ?\n");
+    }
+    String sql = """
+            SELECT event_id, protocol, event_type, vin, event_time_ms, ingest_time_ms,
+                   raw_archive_uri, metadata_json, payload_json
+            FROM event_records
+            %s
+            ORDER BY event_time_ms %s, ingest_time_ms %s, event_id %s
+            LIMIT ?
+            """.formatted(where, order, order, order);
+    try (Connection connection = DriverManager.getConnection(jdbcUrl());
+         PreparedStatement ps = connection.prepareStatement(sql)) {
+        bindQuery(ps, query);
+        try (ResultSet rs = ps.executeQuery()) {
+            List<EventFileRecord> out = new ArrayList<>();
+            while (rs.next()) {
+                out.add(record(rs));
+            }
+            return out;
+        }
+    } catch (SQLException e) {
+        throw new IOException("query duckdb hot event store failed", e);
+    }
+}
+```
+
+- [ ] **Step 2: Add query binding helper**
+
+```java
+private static void bindQuery(PreparedStatement ps, EventFileQuery query) throws SQLException {
+    int index = 1;
+    ps.setString(index++, query.protocol().name());
+    ps.setString(index++, query.dateFrom().toString());
+    ps.setString(index++, query.dateTo().toString());
+    if (query.vin() != null) {
+        ps.setString(index++, query.vin());
+    }
+    if (query.eventType() != null) {
+        ps.setString(index++, query.eventType());
+    }
+    if (query.eventTimeFrom() != null) {
+        ps.setLong(index++, query.eventTimeFrom().toEpochMilli());
+    }
+    if (query.eventTimeTo() != null) {
+        ps.setLong(index++, query.eventTimeTo().toEpochMilli());
+    }
+    ps.setInt(index, query.limit());
+}
+```
+
+- [ ] **Step 3: Implement `findByRawArchiveUri`**
+
+```java
+@Override
+public EventFileRecord findByRawArchiveUri(String rawArchiveUri) throws IOException {
+    if (rawArchiveUri == null || rawArchiveUri.isBlank()) {
+        return null;
+    }
+    ensureInitialized();
+    try (Connection connection = DriverManager.getConnection(jdbcUrl());
+         PreparedStatement ps = connection.prepareStatement("""
+                 SELECT event_id, protocol, event_type, vin, event_time_ms, ingest_time_ms,
+                        raw_archive_uri, metadata_json, payload_json
+                 FROM event_records
+                 WHERE raw_archive_uri = ?
+                 ORDER BY event_type = 'RAW_ARCHIVE' DESC, ingest_time_ms DESC, event_id DESC
+                 LIMIT 1
+                 """)) {
+        ps.setString(1, rawArchiveUri);
+        try (ResultSet rs = ps.executeQuery()) {
+            return rs.next() ? record(rs) : null;
+        }
+    } catch (SQLException e) {
+        throw new IOException("query duckdb hot store by raw archive uri failed", e);
+    }
+}
+```
+
+- [ ] **Step 4: Add record mapper and helpers**
+
+```java
+private EventFileRecord record(ResultSet rs) throws SQLException, IOException {
+    return new EventFileRecord(
+            rs.getString("event_id"),
+            protocol(rs.getString("protocol")),
+            rs.getString("event_type"),
+            rs.getString("vin"),
+            Instant.ofEpochMilli(rs.getLong("event_time_ms")),
+            Instant.ofEpochMilli(rs.getLong("ingest_time_ms")),
+            rs.getString("raw_archive_uri"),
+            readMetadata(rs.getString("metadata_json")),
+            rs.getString("payload_json"));
+}
+
+private Map<String, String> readMetadata(String json) throws IOException {
+    if (json == null || json.isBlank()) {
+        return Map.of();
+    }
+    return objectMapper.readValue(json, STRING_MAP);
+}
+
+private static ProtocolId protocol(String value) {
+    if (value == null || value.isBlank()) {
+        return ProtocolId.UNKNOWN;
+    }
+    try {
+        return ProtocolId.valueOf(value);
+    } catch (IllegalArgumentException ex) {
+        return ProtocolId.UNKNOWN;
+    }
+}
+
+private String jdbcUrl() {
+    return "jdbc:duckdb:" + dbPath;
+}
+```
+
+- [ ] **Step 5: Run hot store tests**
+
+Run:
+
+```bash
+mvn -pl :event-file-store -Dtest=DuckDbHotEventFileStoreTest test
+```
+
+Expected: all `DuckDbHotEventFileStoreTest` tests pass.
+
+## Task 4: Preserve Legacy Store Tests and Update Defaults
+
+**Files:**
+- Modify: `modules/sinks/event-file-store/src/test/java/com/lingniu/ingest/eventfilestore/DuckDbParquetEventFileStoreTest.java`
+- Modify: `modules/apps/vehicle-history-app/src/main/resources/application.yml`
+- Modify: `modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/VehicleHistoryAppDefaultsTest.java`
+
+- [ ] **Step 1: Keep Parquet tests explicitly legacy**
+
+No behavior change is needed in `DuckDbParquetEventFileStoreTest`; leave it instantiating `DuckDbParquetEventFileStore` directly. Add a class comment:
+
+```java
+/**
+ * Compatibility coverage for the legacy Parquet sidecar backend.
+ * Production history uses DuckDbHotEventFileStore through auto-configuration.
+ */
+class DuckDbParquetEventFileStoreTest {
+```
+
+- [ ] **Step 2: Set vehicle-history-app storage default**
+
+Add to `modules/apps/vehicle-history-app/src/main/resources/application.yml`:
+
+```yaml
+    event-file-store:
+      enabled: ${EVENT_FILE_STORE_ENABLED:true}
+      storage: ${EVENT_FILE_STORE_STORAGE:duckdb-hot}
+      path: ${EVENT_FILE_STORE_PATH:./target/event-store/}
+      zone-id: ${EVENT_FILE_STORE_ZONE_ID:Asia/Shanghai}
+      batch-size: ${EVENT_FILE_STORE_BATCH_SIZE:1000}
+      flush-interval-millis: ${EVENT_FILE_STORE_FLUSH_INTERVAL_MILLIS:1000}
+```
+
+Keep existing indentation and only add `storage`; if `batch-size` already exists, update it to `1000`.
+
+- [ ] **Step 3: Update app default test**
+
+In `VehicleHistoryAppDefaultsTest`, assert:
+
+```java
+assertThat(context.getEnvironment()
+        .getProperty("lingniu.ingest.event-file-store.storage"))
+        .isEqualTo("duckdb-hot");
+```
+
+- [ ] **Step 4: Run app default tests**
+
+Run:
+
+```bash
+mvn -pl :vehicle-history-app -Dtest=VehicleHistoryAppDefaultsTest test
+```
+
+Expected: default configuration test passes.
+
+## Task 5: Verify History Ingest Still Archives RAW Bytes
+
+**Files:**
+- Modify: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/EventHistoryEnvelopeIngestorTest.java`
+- Test: `modules/services/event-history-service/src/test/java/com/lingniu/ingest/eventhistory/Gb32960DecodedFrameServiceTest.java`
+
+- [ ] **Step 1: Add an integration-style test using the hot store**
+
+Add a test that writes a RAW envelope through `EventHistoryEnvelopeIngestor` into `DuckDbHotEventFileStore`, then finds it by URI.
+
+```java
+@Test
+void rawArchiveEnvelopeCanBeFoundFromDuckDbHotStoreByUri(@TempDir Path tempDir) throws Exception {
+    EventFileStore store = new DuckDbHotEventFileStore(tempDir, ZoneId.of("Asia/Shanghai"), OBJECT_MAPPER);
+    CapturingArchiveStore archive = new CapturingArchiveStore();
+    EventHistoryEnvelopeIngestor ingestor =
+            new EventHistoryEnvelopeIngestor(store, new TelemetryEnvelopeRecordMapper(), archive);
+    String rawArchiveKey = "2026/06/23/GB32960/VINRAW001/raw-event-hot.bin";
+    String rawArchiveUri = "archive://" + rawArchiveKey;
+    byte[] rawBytes = new byte[]{0x23, 0x23, 0x02, 0x01};
+
+    VehicleEnvelope envelope = VehicleEnvelope.newBuilder()
+            .setSchemaVersion("1.0")
+            .setEventId("raw-event-hot")
+            .setVin("VINRAW001")
+            .setSource("GB32960")
+            .setProtocolVersion("V2016")
+            .setEventTimeMs(1_782_112_400_000L)
+            .setIngestTimeMs(1_782_112_401_000L)
+            .putMetadata(RawArchiveKeys.META_KEY, rawArchiveKey)
+            .putMetadata(RawArchiveKeys.META_URI, rawArchiveUri)
+            .setRawArchive(RawArchiveRef.newBuilder()
+                    .setUri(rawArchiveUri)
+                    .setSizeBytes(rawBytes.length)
+                    .setData(ByteString.copyFrom(rawBytes))
+                    .build())
+            .build();
+
+    EnvelopeIngestResult result = ingestor.tryIngest(envelope.toByteArray());
+
+    assertThat(result.status()).isEqualTo(EnvelopeIngestResult.Status.STORED);
+    assertThat(store.findByRawArchiveUri(rawArchiveUri))
+            .isNotNull()
+            .extracting(EventFileRecord::eventId)
+            .isEqualTo("raw-event-hot");
+    assertThat(archive.bytesByKey).containsEntry(rawArchiveKey, rawBytes);
+}
+```
+
+- [ ] **Step 2: Run the event history ingestor test**
+
+Run:
+
+```bash
+mvn -pl :event-history-service -Dtest=EventHistoryEnvelopeIngestorTest test
+```
+
+Expected: test passes.
+
+- [ ] **Step 3: Run decoded frame service tests**
+
+Run:
+
+```bash
+mvn -pl :event-history-service -Dtest=Gb32960DecodedFrameServiceTest test
+```
+
+Expected: existing replay/snapshot tests pass. If they use a fake store, no change is needed.
+
+## Task 6: Full Module Verification
+
+**Files:**
+- No source changes unless failures expose missing imports or config assertions.
+
+- [ ] **Step 1: Run sink module tests**
+
+Run:
+
+```bash
+mvn -pl :event-file-store test
+```
+
+Expected: all event-file-store tests pass, including both hot and legacy stores.
+
+- [ ] **Step 2: Run event-history-service tests**
+
+Run:
+
+```bash
+mvn -pl :event-history-service test
+```
+
+Expected: all event-history-service tests pass.
+
+- [ ] **Step 3: Package vehicle-history-app**
+
+Run:
+
+```bash
+mvn -pl :vehicle-history-app -am package -DskipTests
+```
+
+Expected: package succeeds.
+
+## Task 7: Local Runtime Verification
+
+**Files:**
+- No committed source changes.
+
+- [ ] **Step 1: Stop any old history service on port 20200**
+
+Run:
+
+```bash
+lsof -tiTCP:20200 -sTCP:LISTEN | xargs -r kill
+```
+
+Expected: no command output, or the old process exits.
+
+- [ ] **Step 2: Start vehicle-history-app with hot DuckDB store**
+
+Run:
+
+```bash
+EVENT_FILE_STORE_STORAGE=duckdb-hot \
+EVENT_FILE_STORE_PATH=./target/live-history-event-store \
+SINK_ARCHIVE_PATH=./target/live-history-archive \
+KAFKA_BROKERS=114.55.58.251:9092 \
+KAFKA_CONSUMER_ENABLED=true \
+KAFKA_GROUP_HISTORY=vehicle-history-hot-$(date +%Y%m%d%H%M%S) \
+java --sun-misc-unsafe-memory-access=allow \
+  -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
+```
+
+Expected:
+
+- app starts on `http://127.0.0.1:20200`;
+- logs show Kafka consumer subscribed;
+- `target/live-history-event-store/events.duckdb` is created.
+
+- [ ] **Step 3: Verify health**
+
+Run:
+
+```bash
+curl -sS http://127.0.0.1:20200/actuator/health
+```
+
+Expected:
+
+```json
+{"status":"UP"}
+```
+
+- [ ] **Step 4: Verify a recent VIN query**
+
+Run with a VIN observed in live archive:
+
+```bash
+curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/telemetry-snapshots?vin=LNXNEGRR9SR318194&platformAccount=Hyundai&dateFrom=2026-06-23&dateTo=2026-06-24&order=DESC&limit=3'
+```
+
+Expected:
+
+- response is a JSON array;
+- if live traffic for that VIN exists after service start, at least one snapshot appears;
+- `rawArchiveUris` point to existing files under `target/live-history-archive`.
+
+- [ ] **Step 5: Verify raw frame replay**
+
+Use one `rawArchiveUri` from Step 4:
+
+```bash
+curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/frame?rawArchiveUri=archive://REPLACE_ME&platformAccount=Hyundai'
+```
+
+Expected:
+
+- response contains `vin`, `command`, `eventTime`, and parsed `blocks`;
+- no `raw archive is missing` warning for the selected URI.
+
+## Self-Review Checklist
+
+- [ ] The new hot store does not rewrite Parquet files on every append.
+- [ ] `appendAll` writes one batch in one transaction.
+- [ ] Duplicate `event_id` replay is idempotent.
+- [ ] Existing history HTTP APIs still use `EventFileStore`, so controller contracts are unchanged.
+- [ ] RAW bytes still land in `ArchiveStore`.
+- [ ] `findByRawArchiveUri` is backed by DuckDB index.
+- [ ] No MySQL credential, RDS host, username, or password appears in the plan or source files.
+- [ ] This plan does not claim the full production goal is complete; it only lands the history foundation.