refactor: store jt808 daily mileage as single metric

docs/operations/gb32960-service-split-runbook.md

@@ -115,7 +115,12 @@ Vehicle history:
 ```bash
 KAFKA_BROKERS=127.0.0.1:9092 \
 HTTP_PORT=20200 \
-EVENT_FILE_STORE_PATH=./target/split-event-store \
+TDENGINE_HISTORY_ENABLED=true \
+TDENGINE_HISTORY_DATABASE=vehicle_ts \
+TDENGINE_JDBC_URL='jdbc:TAOS-WS://<tdengine-host>:6041/vehicle_ts' \
+TDENGINE_USERNAME=root \
+TDENGINE_PASSWORD='<tdengine-password>' \
+TDENGINE_TELEMETRY_FIELDS_ENABLED=false \
 java --sun-misc-unsafe-memory-access=allow \
   -jar modules/apps/vehicle-history-app/target/vehicle-history-app.jar
 ```
@@ -170,14 +175,23 @@ Verify only what was actually run in your environment:
 ```bash
 kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.raw.gb32960.v1 --from-beginning --max-messages 1 --timeout-ms 10000
 kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic vehicle.event.gb32960.v1 --from-beginning --max-messages 1 --timeout-ms 10000
-find target/split-event-store -type f | head
-find target/split-vehicle-stat -type f | head
+```
+
+Use TDengine CLI or a JDBC client to verify history writes:
+
+```sql
+USE vehicle_ts;
+SELECT COUNT(*) FROM raw_frames WHERE protocol = 'GB32960';
+SELECT COUNT(*) FROM vehicle_locations WHERE protocol = 'GB32960';
+SELECT COUNT(*) FROM raw_frames WHERE protocol = 'JT808';
+SELECT COUNT(*) FROM jt808_locations WHERE protocol = 'JT808';
 ```
 
 Useful HTTP query checks after events are consumed. For `realtime_001.hex`, the fixture VIN is `LTEST000000000001`; replace `<date-from>`, `<date-to>`, and `<stat-date>` with values that cover the consumed record's event time:
 
 ```bash
-curl -sS 'http://127.0.0.1:20200/api/event-history/records?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
+curl -sS 'http://127.0.0.1:20200/api/event-history/locations?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
+curl -sS 'http://127.0.0.1:20200/api/event-history/raw-frames?protocol=GB32960&dateFrom=<date-from>&dateTo=<date-to>&vin=LTEST000000000001&limit=10'
 curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/dictionary'
 curl -sS 'http://127.0.0.1:20300/api/vehicle-stat/LTEST000000000001/daily?date=<stat-date>'
 ```
@@ -185,9 +199,10 @@ curl -sS 'http://127.0.0.1:20300/api/vehicle-stat/LTEST000000000001/daily?date=<
 Expected E2E result when Kafka and all services are running:
 
 - A GB32960 client receives a binary success ACK only after required Kafka production succeeds, and the captured ACK bytes are inspected.
-- `vehicle.raw.gb32960.v1` receives a raw archive reference envelope.
+- `vehicle.raw.gb32960.v1` receives a raw frame envelope.
 - `vehicle.event.gb32960.v1` receives one or more normalized records.
-- `target/split-event-store` receives event-store files after the history service flushes consumed Kafka records, including raw archive reference records.
+- TDengine `raw_frames` receives GB32960/JT808 RAW rows with parsed JSON and metadata.
+- TDengine location tables receive compact GB32960/JT808 location rows.
 - JT808 analytics writes daily mileage rows to MySQL `vehicle_stat_metric` after consuming applicable `vehicle.event.jt808.v1` records.
 
 Do not expect `vehicle-history-app` to create raw `.bin` archive files from Kafka raw records in the current implementation. `RawArchiveEventSink` can write archive files only when it receives `VehicleEvent.RawArchive.rawBytes()` inside the same JVM; the Kafka envelope carries only `RawArchiveRef` metadata.
@@ -226,7 +241,7 @@ Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`:
 - Repository-local Kafka setup inspection found no Kafka script, no Docker Compose file, and no compose YAML within the searched repository paths.
 - `nc -z -w 2 127.0.0.1 9092` exited `1`, so no local Kafka broker was reachable at `127.0.0.1:9092`.
 - `kafka-topics`, `kafka-topics.sh`, `docker`, and `docker-compose` were not found on PATH.
-- Kafka topic creation, service startup, health checks, Kafka record checks, archive checks, event-store checks, stat output checks, and ACK observation were not run because the local Kafka prerequisite was absent.
+- Kafka topic creation, service startup, health checks, Kafka record checks, archive checks, TDengine checks, stat output checks, and ACK observation were not run because the local Kafka prerequisite was absent.
 
 ## Latest Local Verification
 
@@ -238,7 +253,7 @@ Observed on 2026-06-23 in worktree `.worktrees/gb32960-service-split`:
 - Split app startup: not run in this verification pass because local Kafka was still absent.
 - Kafka raw records: not verified; no local Kafka broker or Kafka CLI tools were available.
 - Kafka event records: not verified; no local Kafka broker or Kafka CLI tools were available.
-- Event-store files: not verified; downstream services were not started without Kafka.
+- TDengine history rows: not verified; downstream services were not started without Kafka.
 - Analytics output: not verified; downstream services were not started without Kafka.
 - ACK behavior: not verified against a live broker in this pass. Unit tests cover the GB32960 ACK boundary and ordering, but an operator should still run the ACK capture command above in a Kafka-backed local or staging environment.
 

docs/operations/jt808-daily-mileage-runbook.md

@@ -6,7 +6,7 @@ The analytics app can calculate daily mileage from JT808 telemetry only. It cons
 
 ## Storage
 
-There is no separate JT808 daily-mileage table and no Redis mileage state. The current day start total, latest total, and derived daily mileage are all stored in the common JDBC/MySQL metric table `vehicle_stat_metric`.
+There is no separate JT808 daily-mileage table and no Redis mileage state. The derived value is stored as one `daily_mileage_km` metric row in the common JDBC/MySQL table `vehicle_stat_metric`; the first and latest GPS total mileage values are kept on that same row as calculation source columns.
 
 The JT808 daily-mileage value is calculated from the GPS total mileage reported in location additional information:
 
@@ -15,7 +15,7 @@ daily_mileage_km = last_total_mileage_km - first_total_mileage_km
 calculation_method = JT808_TOTAL_MILEAGE_DIFF
 ```
 
-The first valid JT808 location point for a vehicle and local day stores the day start total and writes `daily_mileage_km=0.0`. Later ordered points update the latest total and the derived `daily_mileage_km`. If a later total mileage is lower than the stored start or latest total, the update is ignored and the previous metric remains.
+The first valid JT808 location point for a vehicle and local day stores the day start total on the metric row and writes `daily_mileage_km=0.0`. Later ordered points update the latest total on the same row and rewrite the derived `daily_mileage_km`. If a later total mileage is lower than the stored latest total, the previous latest total remains.
 
 ## Runtime Settings
 
@@ -29,4 +29,4 @@ MYSQL_USERNAME=<user>
 MYSQL_PASSWORD=<password>
 ```
 
-Algorithm defaults use the Asia/Shanghai daily boundary. Restart recovery reads the same `vehicle_stat_metric` rows, so no separate mileage state store is required.
+Algorithm defaults use the Asia/Shanghai daily boundary. Restart recovery reads the same `daily_mileage_km` metric row, so no separate mileage state store is required.

docs/target-architecture.md

@@ -125,12 +125,12 @@ calculation_method = JT808_TOTAL_MILEAGE_DIFF
 ```
 
 - Store the metric in `vehicle_stat_metric` with
-  `metric_key = daily_mileage_km`.
+  `metric_key = daily_mileage_km`; the first and latest GPS total mileage used
+  for the subtraction stay on the same metric row as calculation source columns.
 
 There is no JT808-specific daily mileage table and no in-memory production
-state-store mode. Daily start total, latest total, and derived mileage are all
-stored in `vehicle_stat_metric`, so restart recovery does not require a separate
-Redis mileage state.
+state-store mode. Restart recovery reads the same metric row, so it does not
+require a separate Redis mileage state.
 
 ### Latest State
 

modules/apps/vehicle-history-app/src/test/java/com/lingniu/ingest/historyapp/PortainerComposeResourceLimitsTest.java

@@ -107,6 +107,21 @@ class PortainerComposeResourceLimitsTest {
                 .doesNotContain("Vehicle state can also be enabled from this runtime");
     }
 
+    @Test
+    void splitRunbookUsesTdengineHistoryVerificationInsteadOfEventFileStore() throws IOException {
+        String splitRunbook = Files.readString(repositoryRoot()
+                .resolve("docs/operations/gb32960-service-split-runbook.md"));
+
+        assertThat(splitRunbook)
+                .contains("TDENGINE_HISTORY_ENABLED=true")
+                .contains("SELECT COUNT(*) FROM raw_frames")
+                .contains("SELECT COUNT(*) FROM vehicle_locations")
+                .doesNotContain("EVENT_FILE_STORE_PATH")
+                .doesNotContain("target/split-event-store")
+                .doesNotContain("event-store files")
+                .doesNotContain("/api/event-history/records?");
+    }
+
     @Test
     void vehicleAnalyticsComposeOnlyExposesStatConsumerEnvironment() throws IOException {
         String compose = Files.readString(repositoryRoot().resolve("deploy/portainer/docker-compose.yml"));

modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java

@@ -1,9 +1,12 @@
 package com.lingniu.ingest.vehiclestat;
 
 import org.springframework.dao.DuplicateKeyException;
+import org.springframework.jdbc.core.ConnectionCallback;
 import org.springframework.jdbc.core.JdbcTemplate;
 
 import java.sql.Date;
+import java.sql.ResultSet;
+import java.sql.SQLException;
 import java.time.LocalDate;
 import java.util.List;
 import java.util.Optional;
@@ -15,6 +18,8 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
     private static final String DAILY_MILEAGE_UNIT = "km";
     private static final String DAILY_MILEAGE_START_TOTAL_KEY = "daily_mileage_start_total_km";
     private static final String DAILY_MILEAGE_LATEST_TOTAL_KEY = "daily_mileage_latest_total_km";
+    private static final String FIRST_TOTAL_COLUMN = "first_total_mileage_km";
+    private static final String LATEST_TOTAL_COLUMN = "latest_total_mileage_km";
 
     private final JdbcTemplate jdbcTemplate;
 
@@ -26,17 +31,6 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         ensureSchema();
     }
 
-    private void saveDailyMileageMetric(VehicleDailyStatResult result) {
-        if (result == null || result.dailyMileageKm().isEmpty()) {
-            return;
-        }
-        String vin = clean(result.vin());
-        Date statDate = Date.valueOf(result.statDate());
-        double value = result.dailyMileageKm().getAsDouble();
-        String strategy = result.dailyMileageStrategy().name();
-        upsertMetric(vin, statDate, DAILY_MILEAGE_KEY, value, DAILY_MILEAGE_UNIT, strategy);
-    }
-
     @Override
     public Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate) {
         String normalizedVin = clean(vin);
@@ -63,34 +57,48 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         String normalizedVin = clean(vin);
         Date date = Date.valueOf(statDate);
         String strategy = DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF.name();
-        OptionalDouble existingMin = findMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY);
-        double minTotalMileage = existingMin.isPresent()
-                ? Math.min(existingMin.getAsDouble(), totalMileageKm)
-                : totalMileageKm;
-        if (existingMin.isEmpty() || Double.compare(minTotalMileage, existingMin.getAsDouble()) != 0) {
-            upsertMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY, minTotalMileage,
-                    DAILY_MILEAGE_UNIT, strategy);
-        }
-
-        OptionalDouble existingMax = findMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY);
-        double maxTotalMileage = existingMax.isPresent()
-                ? Math.max(existingMax.getAsDouble(), totalMileageKm)
-                : totalMileageKm;
-        if (existingMax.isEmpty() || Double.compare(maxTotalMileage, existingMax.getAsDouble()) != 0) {
-            upsertMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY, maxTotalMileage,
-                    DAILY_MILEAGE_UNIT, strategy);
-        }
-
-        double dailyMileageKm = maxTotalMileage - minTotalMileage;
+        DailyMileageState existingState = findDailyMileageState(normalizedVin, date)
+                .orElseGet(() -> legacyDailyMileageState(normalizedVin, date).orElse(null));
+        double firstTotalMileage = existingState == null || !Double.isFinite(existingState.firstTotalMileageKm())
+                ? totalMileageKm
+                : existingState.firstTotalMileageKm();
+        double latestTotalMileage = existingState == null || !Double.isFinite(existingState.latestTotalMileageKm())
+                ? totalMileageKm
+                : Math.max(existingState.latestTotalMileageKm(), totalMileageKm);
+        double dailyMileageKm = latestTotalMileage - firstTotalMileage;
+        upsertDailyMileageMetric(normalizedVin, date, dailyMileageKm, DAILY_MILEAGE_UNIT, strategy,
+                firstTotalMileage, latestTotalMileage);
+        deleteLegacyMileageStateMetrics(normalizedVin, date);
         VehicleDailyStatResult result = new VehicleDailyStatResult(
                 normalizedVin,
                 statDate,
                 OptionalDouble.of(dailyMileageKm),
                 DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF);
-        saveDailyMileageMetric(result);
         return Optional.of(result);
     }
 
+    private Optional<DailyMileageState> findDailyMileageState(String vin, Date statDate) {
+        List<DailyMileageState> rows = jdbcTemplate.query("""
+                SELECT first_total_mileage_km, latest_total_mileage_km
+                FROM vehicle_stat_metric
+                WHERE vin = ? AND stat_date = ? AND metric_key = ?
+                """, (rs, rowNum) -> new DailyMileageState(
+                nullableDouble(rs, FIRST_TOTAL_COLUMN),
+                nullableDouble(rs, LATEST_TOTAL_COLUMN)), vin, statDate, DAILY_MILEAGE_KEY);
+        return rows.stream()
+                .filter(DailyMileageState::hasCompleteState)
+                .findFirst();
+    }
+
+    private Optional<DailyMileageState> legacyDailyMileageState(String vin, Date statDate) {
+        OptionalDouble start = findMetric(vin, statDate, DAILY_MILEAGE_START_TOTAL_KEY);
+        OptionalDouble latest = findMetric(vin, statDate, DAILY_MILEAGE_LATEST_TOTAL_KEY);
+        if (start.isEmpty() || latest.isEmpty()) {
+            return Optional.empty();
+        }
+        return Optional.of(new DailyMileageState(start.getAsDouble(), latest.getAsDouble()));
+    }
+
     private OptionalDouble findMetric(String vin, Date statDate, String metricKey) {
         List<Double> rows = jdbcTemplate.query("""
                 SELECT metric_value
@@ -105,25 +113,38 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         return OptionalDouble.of(rows.getFirst());
     }
 
-    private void upsertMetric(String vin, Date statDate, String metricKey, double value, String unit, String strategy) {
+    private void upsertDailyMileageMetric(String vin, Date statDate, double value, String unit, String strategy,
+                                          double firstTotalMileageKm, double latestTotalMileageKm) {
         try {
             jdbcTemplate.update("""
                     INSERT INTO vehicle_stat_metric
-                    (vin, stat_date, metric_key, metric_value, metric_unit, calculation_method)
-                    VALUES (?, ?, ?, ?, ?, ?)
-                    """, vin, statDate, metricKey, value, unit, strategy);
+                    (vin, stat_date, metric_key, metric_value, metric_unit, calculation_method,
+                     first_total_mileage_km, latest_total_mileage_km)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+                    """, vin, statDate, DAILY_MILEAGE_KEY, value, unit, strategy,
+                    firstTotalMileageKm, latestTotalMileageKm);
         } catch (DuplicateKeyException ex) {
             jdbcTemplate.update("""
                     UPDATE vehicle_stat_metric
                     SET metric_value = ?,
                         metric_unit = ?,
                         calculation_method = ?,
+                        first_total_mileage_km = ?,
+                        latest_total_mileage_km = ?,
                         updated_at = CURRENT_TIMESTAMP
                     WHERE vin = ? AND stat_date = ? AND metric_key = ?
-                    """, value, unit, strategy, vin, statDate, metricKey);
+                    """, value, unit, strategy, firstTotalMileageKm, latestTotalMileageKm,
+                    vin, statDate, DAILY_MILEAGE_KEY);
         }
     }
 
+    private void deleteLegacyMileageStateMetrics(String vin, Date statDate) {
+        jdbcTemplate.update("""
+                DELETE FROM vehicle_stat_metric
+                WHERE vin = ? AND stat_date = ? AND metric_key IN (?, ?)
+                """, vin, statDate, DAILY_MILEAGE_START_TOTAL_KEY, DAILY_MILEAGE_LATEST_TOTAL_KEY);
+    }
+
     private void ensureSchema() {
         jdbcTemplate.execute("""
                 CREATE TABLE IF NOT EXISTS vehicle_stat_metric (
@@ -133,11 +154,43 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
                   metric_value DECIMAL(18,6) NULL,
                   metric_unit VARCHAR(16) NOT NULL,
                   calculation_method VARCHAR(64) NOT NULL,
+                  first_total_mileage_km DECIMAL(18,6) NULL,
+                  latest_total_mileage_km DECIMAL(18,6) NULL,
                   created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                   updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
                   PRIMARY KEY (vin, stat_date, metric_key)
                 )
                 """);
+        ensureColumn(FIRST_TOTAL_COLUMN);
+        ensureColumn(LATEST_TOTAL_COLUMN);
+    }
+
+    private void ensureColumn(String columnName) {
+        if (columnExists(columnName)) {
+            return;
+        }
+        jdbcTemplate.execute("ALTER TABLE vehicle_stat_metric ADD COLUMN " + columnName + " DECIMAL(18,6) NULL");
+    }
+
+    private boolean columnExists(String columnName) {
+        ConnectionCallback<Boolean> callback = connection -> {
+            try (ResultSet columns = connection.getMetaData().getColumns(
+                    connection.getCatalog(), null, "vehicle_stat_metric", columnName)) {
+                if (columns.next()) {
+                    return true;
+                }
+            }
+            try (ResultSet columns = connection.getMetaData().getColumns(
+                    connection.getCatalog(), null, "VEHICLE_STAT_METRIC", columnName.toUpperCase())) {
+                return columns.next();
+            }
+        };
+        return Boolean.TRUE.equals(jdbcTemplate.execute(callback));
+    }
+
+    private static double nullableDouble(ResultSet rs, String columnName) throws SQLException {
+        double value = rs.getDouble(columnName);
+        return rs.wasNull() ? Double.NaN : value;
     }
 
     private static String clean(String value) {
@@ -146,4 +199,10 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         }
         return value.trim();
     }
+
+    private record DailyMileageState(double firstTotalMileageKm, double latestTotalMileageKm) {
+        boolean hasCompleteState() {
+            return Double.isFinite(firstTotalMileageKm) && Double.isFinite(latestTotalMileageKm);
+        }
+    }
 }

modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java

@@ -27,13 +27,23 @@ class JdbcVehicleStatMetricRepositoryTest {
                 .extracting(result -> result.dailyMileageKm().orElseThrow())
                 .isEqualTo(15.5);
         assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM vehicle_stat_metric", Integer.class))
-                .isEqualTo(3);
+                .isOne();
         assertThat(jdbcTemplate.queryForObject(
                 "SELECT metric_key FROM vehicle_stat_metric WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'",
                 String.class)).isEqualTo("daily_mileage_km");
         assertThat(jdbcTemplate.queryForObject(
                 "SELECT calculation_method FROM vehicle_stat_metric WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'",
                 String.class)).isEqualTo("JT808_TOTAL_MILEAGE_DIFF");
+        assertThat(jdbcTemplate.queryForObject("""
+                SELECT first_total_mileage_km
+                FROM vehicle_stat_metric
+                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
+                """, Double.class)).isEqualTo(100.0);
+        assertThat(jdbcTemplate.queryForObject("""
+                SELECT latest_total_mileage_km
+                FROM vehicle_stat_metric
+                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
+                """, Double.class)).isEqualTo(115.5);
         assertThat(jdbcTemplate.queryForObject("""
                 SELECT COUNT(*)
                 FROM information_schema.tables
@@ -58,7 +68,7 @@ class JdbcVehicleStatMetricRepositoryTest {
         assertThat(second.dailyMileageKm()).hasValue(8.25);
         assertThat(jdbcTemplate.queryForObject(
                 "SELECT COUNT(*) FROM vehicle_stat_metric WHERE vin = 'VIN001' AND stat_date = DATE '2026-07-01'",
-                Integer.class)).isEqualTo(3);
+                Integer.class)).isOne();
         assertThat(jdbcTemplate.queryForObject("""
                 SELECT metric_value
                 FROM vehicle_stat_metric
@@ -70,30 +80,33 @@ class JdbcVehicleStatMetricRepositoryTest {
     }
 
     @Test
-    void recalculatesJt808MileageFromDailyMinAndMaxTotalsWhenSamplesArriveOutOfOrder() {
+    void recalculatesJt808MileageFromSingleMetricRowAfterRestart() {
         JdbcTemplate jdbcTemplate = new JdbcTemplate(new DriverManagerDataSource(
                 "jdbc:h2:mem:vehicle_stat_metric_jt808_out_of_order;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1",
                 "sa",
                 ""));
         JdbcVehicleStatMetricRepository repository = new JdbcVehicleStatMetricRepository(jdbcTemplate);
 
-        repository.recordDailyMileageSample("VIN001", LocalDate.of(2026, 7, 1), 1008.75).orElseThrow();
-        VehicleDailyStatResult lowerSample = repository.recordDailyMileageSample(
-                "VIN001", LocalDate.of(2026, 7, 1), 1000.5).orElseThrow();
-        VehicleDailyStatResult higherSample = repository.recordDailyMileageSample(
+        repository.recordDailyMileageSample("VIN001", LocalDate.of(2026, 7, 1), 1000.5).orElseThrow();
+        JdbcVehicleStatMetricRepository restartedRepository = new JdbcVehicleStatMetricRepository(jdbcTemplate);
+        VehicleDailyStatResult higherSample = restartedRepository.recordDailyMileageSample(
                 "VIN001", LocalDate.of(2026, 7, 1), 1010.0).orElseThrow();
 
-        assertThat(lowerSample.dailyMileageKm()).hasValue(8.25);
         assertThat(higherSample.dailyMileageKm()).hasValue(9.5);
         assertThat(jdbcTemplate.queryForObject("""
-                SELECT metric_value
+                SELECT COUNT(*)
                 FROM vehicle_stat_metric
-                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_start_total_km'
+                WHERE vin = 'VIN001' AND stat_date = DATE '2026-07-01'
+                """, Integer.class)).isOne();
+        assertThat(jdbcTemplate.queryForObject("""
+                SELECT first_total_mileage_km
+                FROM vehicle_stat_metric
+                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
                 """, Double.class)).isEqualTo(1000.5);
         assertThat(jdbcTemplate.queryForObject("""
-                SELECT metric_value
+                SELECT latest_total_mileage_km
                 FROM vehicle_stat_metric
-                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_latest_total_km'
+                WHERE vin = 'VIN001' AND metric_key = 'daily_mileage_km'
                 """, Double.class)).isEqualTo(1010.0);
         assertThat(jdbcTemplate.queryForObject("""
                 SELECT metric_value