From 620821b6249bc8570e32dc7d1a82589b0477ae72 Mon Sep 17 00:00:00 2001 From: lingniu Date: Tue, 23 Jun 2026 19:39:06 +0800 Subject: [PATCH] docs: plan gb32960 duckdb hot history --- ...-06-23-gb32960-history-duckdb-hot-store.md | 810 ++++++++++++++++++ 1 file changed, 810 insertions(+) create mode 100644 docs/superpowers/plans/2026-06-23-gb32960-history-duckdb-hot-store.md diff --git 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 new file mode 100644 index 00000000..ecc6aa8a --- /dev/null +++ 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 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> 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 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 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 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 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 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 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.