refactor: store jt808 mileage metrics directly

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

@@ -13,6 +13,8 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
 
     private static final String DAILY_MILEAGE_KEY = "daily_mileage_km";
     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 final JdbcTemplate jdbcTemplate;
 
@@ -33,15 +35,7 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         Date statDate = Date.valueOf(result.statDate());
         double value = result.dailyMileageKm().getAsDouble();
         String strategy = result.dailyMileageStrategy().name();
-        try {
-            jdbcTemplate.update("""
-                    INSERT INTO vehicle_stat_metric
-                    (vin, stat_date, metric_key, metric_value, metric_unit, calculation_method)
-                    VALUES (?, ?, ?, ?, ?, ?)
-                    """, vin, statDate, DAILY_MILEAGE_KEY, value, DAILY_MILEAGE_UNIT, strategy);
-        } catch (DuplicateKeyException ex) {
-            updateDailyStat(vin, statDate, value, strategy);
-        }
+        upsertMetric(vin, statDate, DAILY_MILEAGE_KEY, value, DAILY_MILEAGE_UNIT, strategy);
     }
 
     @Override
@@ -62,15 +56,72 @@ public final class JdbcVehicleStatMetricRepository implements VehicleStatReposit
         return rows.stream().findFirst();
     }
 
-    private void updateDailyStat(String vin, Date statDate, double value, String strategy) {
-        jdbcTemplate.update("""
-                UPDATE vehicle_stat_metric
-                SET metric_value = ?,
-                    metric_unit = ?,
-                    calculation_method = ?,
-                    updated_at = CURRENT_TIMESTAMP
+    @Override
+    public Optional<VehicleDailyStatResult> recordDailyMileageSample(String vin, LocalDate statDate, double totalMileageKm) {
+        if (!Double.isFinite(totalMileageKm) || totalMileageKm < 0.0) {
+            return Optional.empty();
+        }
+        String normalizedVin = clean(vin);
+        Date date = Date.valueOf(statDate);
+        OptionalDouble existingStart = findMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY);
+        double startTotalMileage = existingStart.orElse(totalMileageKm);
+        if (existingStart.isEmpty()) {
+            upsertMetric(normalizedVin, date, DAILY_MILEAGE_START_TOTAL_KEY, startTotalMileage,
+                    DAILY_MILEAGE_UNIT, DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF.name());
+        }
+
+        OptionalDouble existingLatest = findMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY);
+        if (existingLatest.isPresent() && totalMileageKm < existingLatest.getAsDouble()) {
+            return findDailyStat(normalizedVin, statDate);
+        }
+        if (totalMileageKm < startTotalMileage) {
+            return findDailyStat(normalizedVin, statDate);
+        }
+
+        String strategy = DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF.name();
+        upsertMetric(normalizedVin, date, DAILY_MILEAGE_LATEST_TOTAL_KEY, totalMileageKm,
+                DAILY_MILEAGE_UNIT, strategy);
+        double dailyMileageKm = totalMileageKm - startTotalMileage;
+        VehicleDailyStatResult result = new VehicleDailyStatResult(
+                normalizedVin,
+                statDate,
+                OptionalDouble.of(dailyMileageKm),
+                DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF);
+        saveDailyStat(result);
+        return Optional.of(result);
+    }
+
+    private OptionalDouble findMetric(String vin, Date statDate, String metricKey) {
+        List<Double> rows = jdbcTemplate.query("""
+                SELECT metric_value
+                FROM vehicle_stat_metric
                 WHERE vin = ? AND stat_date = ? AND metric_key = ?
-                """, value, DAILY_MILEAGE_UNIT, strategy, vin, statDate, DAILY_MILEAGE_KEY);
+                """, (rs, rowNum) -> rs.getBigDecimal("metric_value") == null
+                ? Double.NaN
+                : rs.getBigDecimal("metric_value").doubleValue(), vin, statDate, metricKey);
+        if (rows.isEmpty() || !Double.isFinite(rows.getFirst())) {
+            return OptionalDouble.empty();
+        }
+        return OptionalDouble.of(rows.getFirst());
+    }
+
+    private void upsertMetric(String vin, Date statDate, String metricKey, double value, String unit, String strategy) {
+        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);
+        } catch (DuplicateKeyException ex) {
+            jdbcTemplate.update("""
+                    UPDATE vehicle_stat_metric
+                    SET metric_value = ?,
+                        metric_unit = ?,
+                        calculation_method = ?,
+                        updated_at = CURRENT_TIMESTAMP
+                    WHERE vin = ? AND stat_date = ? AND metric_key = ?
+                    """, value, unit, strategy, vin, statDate, metricKey);
+        }
     }
 
     private void ensureSchema() {

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

@@ -7,5 +7,7 @@ public interface VehicleStatRepository {
 
     void saveDailyStat(VehicleDailyStatResult result);
 
+    Optional<VehicleDailyStatResult> recordDailyMileageSample(String vin, LocalDate statDate, double totalMileageKm);
+
     Optional<VehicleDailyStatResult> findDailyStat(String vin, LocalDate statDate);
 }

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

@@ -7,12 +7,8 @@ import com.lingniu.ingest.vehiclestat.VehicleStatController;
 import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor;
 import com.lingniu.ingest.vehiclestat.VehicleStatRepository;
 import com.lingniu.ingest.vehiclestat.jt808.Jt808LocationPointExtractor;
-import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStateStore;
 import com.lingniu.ingest.vehiclestat.jt808.Jt808MileageStreamProcessor;
-import com.lingniu.ingest.vehiclestat.jt808.RedisJt808MileageStateStore;
-import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
-import org.springframework.boot.autoconfigure.data.redis.RedisAutoConfiguration;
 import org.springframework.boot.autoconfigure.jdbc.DataSourceAutoConfiguration;
 import org.springframework.boot.autoconfigure.jdbc.JdbcTemplateAutoConfiguration;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
@@ -20,16 +16,13 @@ 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;
-import org.springframework.data.redis.core.StringRedisTemplate;
 import org.springframework.jdbc.core.JdbcTemplate;
 
-import java.time.Duration;
 import java.time.ZoneId;
 
 @AutoConfiguration(after = {
         DataSourceAutoConfiguration.class,
-        JdbcTemplateAutoConfiguration.class,
-        RedisAutoConfiguration.class
+        JdbcTemplateAutoConfiguration.class
 })
 @EnableConfigurationProperties(VehicleStatProperties.class)
 @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "enabled", havingValue = "true")
@@ -50,27 +43,12 @@ public class VehicleStatAutoConfiguration {
     }
 
     @Bean
-    @ConditionalOnBean({StringRedisTemplate.class, ObjectMapper.class})
-    @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat.jt808", name = "enabled", havingValue = "true")
-    @ConditionalOnMissingBean(Jt808MileageStateStore.class)
-    public Jt808MileageStateStore redisJt808MileageStateStore(StringRedisTemplate redis,
-                                                              ObjectMapper objectMapper,
-                                                              VehicleStatProperties props) {
-        VehicleStatProperties.Jt808 jt808 = props.getJt808();
-        return new RedisJt808MileageStateStore(redis, objectMapper,
-                jt808.getRedisKeyPrefix(), Duration.ofDays(jt808.getStateTtlDays()));
-    }
-
-    @Bean
-    @ConditionalOnBean({Jt808LocationPointExtractor.class, Jt808MileageStateStore.class,
-            VehicleStatRepository.class})
+    @ConditionalOnBean({Jt808LocationPointExtractor.class, VehicleStatRepository.class})
     @ConditionalOnMissingBean
     public Jt808MileageStreamProcessor jt808MileageStreamProcessor(Jt808LocationPointExtractor extractor,
-                                                                    Jt808MileageStateStore stateStore,
                                                                     VehicleStatRepository repository,
                                                                     VehicleStatProperties props) {
-        return new Jt808MileageStreamProcessor(extractor, stateStore, repository,
-                ZoneId.of(props.getZoneId()));
+        return new Jt808MileageStreamProcessor(extractor, repository, ZoneId.of(props.getZoneId()));
     }
 
     @Bean

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

@@ -28,8 +28,6 @@ public class VehicleStatProperties {
 
     public static class Jt808 {
         private boolean enabled;
-        private String redisKeyPrefix = "vehicle:mileage:jt808:daily:";
-        private long stateTtlDays = 3;
 
         public boolean isEnabled() {
             return enabled;
@@ -38,22 +36,5 @@ public class VehicleStatProperties {
         public void setEnabled(boolean enabled) {
             this.enabled = enabled;
         }
-
-        public String getRedisKeyPrefix() {
-            return redisKeyPrefix;
-        }
-
-        public void setRedisKeyPrefix(String redisKeyPrefix) {
-            this.redisKeyPrefix = redisKeyPrefix;
-        }
-
-        public long getStateTtlDays() {
-            return stateTtlDays;
-        }
-
-        public void setStateTtlDays(long stateTtlDays) {
-            this.stateTtlDays = stateTtlDays;
-        }
-
     }
 }

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

@@ -1,181 +0,0 @@
-package com.lingniu.ingest.vehiclestat.jt808;
-
-import com.lingniu.ingest.vehiclestat.DailyMileageStrategy;
-import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult;
-
-import java.time.Instant;
-import java.time.LocalDate;
-import java.util.Optional;
-import java.util.OptionalDouble;
-
-public final class Jt808DailyMileageState {
-
-    private final String vehicleKey;
-    private final String vin;
-    private final String phone;
-    private final LocalDate statDate;
-    private Instant firstEventTime;
-    private Instant lastEventTime;
-    private double firstTotalMileageKm = Double.NaN;
-    private double lastTotalMileageKm = Double.NaN;
-    private int acceptedPoints;
-    private int totalMileageSamples;
-    private int outOfOrderPoints;
-    private int odometerAnomalies;
-
-    private Jt808DailyMileageState(String vehicleKey, String vin, String phone, LocalDate statDate) {
-        if (vehicleKey == null || vehicleKey.isBlank()) {
-            throw new IllegalArgumentException("vehicleKey must not be blank");
-        }
-        if (statDate == null) {
-            throw new IllegalArgumentException("statDate must not be null");
-        }
-        this.vehicleKey = vehicleKey;
-        this.vin = vin == null ? "" : vin;
-        this.phone = phone == null ? "" : phone;
-        this.statDate = statDate;
-    }
-
-    public static Jt808DailyMileageState empty(String vehicleKey, String vin, String phone, LocalDate statDate) {
-        return new Jt808DailyMileageState(vehicleKey, vin, phone, statDate);
-    }
-
-    public static Jt808DailyMileageState restore(
-            String vehicleKey,
-            String vin,
-            String phone,
-            LocalDate statDate,
-            Instant firstEventTime,
-            Instant lastEventTime,
-            double firstTotalMileageKm,
-            double lastTotalMileageKm,
-            int acceptedPoints,
-            int totalMileageSamples,
-            int outOfOrderPoints,
-            int odometerAnomalies) {
-        Jt808DailyMileageState state = new Jt808DailyMileageState(vehicleKey, vin, phone, statDate);
-        state.firstEventTime = firstEventTime;
-        state.lastEventTime = lastEventTime;
-        state.firstTotalMileageKm = firstTotalMileageKm;
-        state.lastTotalMileageKm = lastTotalMileageKm;
-        state.acceptedPoints = acceptedPoints;
-        state.totalMileageSamples = totalMileageSamples;
-        state.outOfOrderPoints = outOfOrderPoints;
-        state.odometerAnomalies = odometerAnomalies;
-        return state;
-    }
-
-    public void apply(Jt808LocationPoint point) {
-        if (point == null) {
-            return;
-        }
-        if (lastEventTime != null && !point.eventTime().isAfter(lastEventTime)) {
-            outOfOrderPoints++;
-            return;
-        }
-        if (firstEventTime == null) {
-            firstEventTime = point.eventTime();
-        }
-        updateOdometer(point);
-        lastEventTime = point.eventTime();
-        acceptedPoints++;
-    }
-
-    public Optional<VehicleDailyStatResult> toVehicleDailyStatResult() {
-        OptionalDouble dailyMileage = totalMileageDifference();
-        if (dailyMileage.isEmpty()) {
-            return Optional.empty();
-        }
-        return Optional.of(new VehicleDailyStatResult(
-                statVehicleId(),
-                statDate,
-                dailyMileage,
-                DailyMileageStrategy.JT808_TOTAL_MILEAGE_DIFF));
-    }
-
-    private void updateOdometer(Jt808LocationPoint point) {
-        if (point.totalMileageKm() == null || !Double.isFinite(point.totalMileageKm())) {
-            return;
-        }
-        double current = point.totalMileageKm();
-        totalMileageSamples++;
-        if (!Double.isFinite(firstTotalMileageKm)) {
-            firstTotalMileageKm = current;
-        }
-        if (!Double.isFinite(lastTotalMileageKm)) {
-            lastTotalMileageKm = current;
-            return;
-        }
-        if (current < lastTotalMileageKm) {
-            odometerAnomalies++;
-            lastTotalMileageKm = current;
-            return;
-        }
-        lastTotalMileageKm = current;
-    }
-
-    private OptionalDouble totalMileageDifference() {
-        if (totalMileageSamples < 2
-                || !Double.isFinite(firstTotalMileageKm)
-                || !Double.isFinite(lastTotalMileageKm)
-                || lastTotalMileageKm < firstTotalMileageKm) {
-            return OptionalDouble.empty();
-        }
-        return OptionalDouble.of(lastTotalMileageKm - firstTotalMileageKm);
-    }
-
-    private String statVehicleId() {
-        if (!vin.isBlank() && !"unknown".equalsIgnoreCase(vin)) {
-            return vin;
-        }
-        return vehicleKey;
-    }
-
-    public String vehicleKey() {
-        return vehicleKey;
-    }
-
-    public String vin() {
-        return vin;
-    }
-
-    public String phone() {
-        return phone;
-    }
-
-    public LocalDate statDate() {
-        return statDate;
-    }
-
-    public Instant firstEventTime() {
-        return firstEventTime;
-    }
-
-    public Instant lastEventTime() {
-        return lastEventTime;
-    }
-
-    public double lastTotalMileageKm() {
-        return lastTotalMileageKm;
-    }
-
-    public double firstTotalMileageKm() {
-        return firstTotalMileageKm;
-    }
-
-    public int acceptedPoints() {
-        return acceptedPoints;
-    }
-
-    public int totalMileageSamples() {
-        return totalMileageSamples;
-    }
-
-    public int outOfOrderPoints() {
-        return outOfOrderPoints;
-    }
-
-    public int odometerAnomalies() {
-        return odometerAnomalies;
-    }
-}

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

@@ -1,11 +0,0 @@
-package com.lingniu.ingest.vehiclestat.jt808;
-
-import java.time.LocalDate;
-import java.util.Optional;
-
-public interface Jt808MileageStateStore {
-
-    Optional<Jt808DailyMileageState> load(String vehicleKey, LocalDate statDate);
-
-    void save(Jt808DailyMileageState state);
-}

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

@@ -9,17 +9,14 @@ import java.time.ZoneId;
 public final class Jt808MileageStreamProcessor {
 
     private final Jt808LocationPointExtractor extractor;
-    private final Jt808MileageStateStore stateStore;
     private final VehicleStatRepository repository;
     private final ZoneId zoneId;
 
     public Jt808MileageStreamProcessor(
             Jt808LocationPointExtractor extractor,
-            Jt808MileageStateStore stateStore,
             VehicleStatRepository repository,
             ZoneId zoneId) {
         this.extractor = extractor;
-        this.stateStore = stateStore;
         this.repository = repository;
         this.zoneId = zoneId;
     }
@@ -29,11 +26,17 @@ public final class Jt808MileageStreamProcessor {
     }
 
     private void processPoint(Jt808LocationPoint point) {
+        if (point.totalMileageKm() == null || !Double.isFinite(point.totalMileageKm())) {
+            return;
+        }
         LocalDate statDate = LocalDate.ofInstant(point.eventTime(), zoneId);
-        Jt808DailyMileageState state = stateStore.load(point.vehicleKey(), statDate)
-                .orElseGet(() -> Jt808DailyMileageState.empty(point.vehicleKey(), point.vin(), point.phone(), statDate));
-        state.apply(point);
-        stateStore.save(state);
-        state.toVehicleDailyStatResult().ifPresent(repository::saveDailyStat);
+        repository.recordDailyMileageSample(statVehicleId(point), statDate, point.totalMileageKm());
+    }
+
+    private static String statVehicleId(Jt808LocationPoint point) {
+        if (!point.vin().isBlank() && !"unknown".equalsIgnoreCase(point.vin())) {
+            return point.vin();
+        }
+        return point.vehicleKey();
     }
 }

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

@@ -1,155 +0,0 @@
-package com.lingniu.ingest.vehiclestat.jt808;
-
-import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.core.JsonProcessingException;
-import com.fasterxml.jackson.databind.ObjectMapper;
-import org.springframework.data.redis.core.StringRedisTemplate;
-
-import java.time.Duration;
-import java.time.Instant;
-import java.time.LocalDate;
-import java.util.Optional;
-
-public final class RedisJt808MileageStateStore implements Jt808MileageStateStore {
-
-    private final StringRedisTemplate redis;
-    private final ObjectMapper objectMapper;
-    private final String keyPrefix;
-    private final Duration ttl;
-
-    public RedisJt808MileageStateStore(
-            StringRedisTemplate redis,
-            ObjectMapper objectMapper,
-            String keyPrefix,
-            Duration ttl) {
-        this.redis = redis;
-        this.objectMapper = objectMapper;
-        this.keyPrefix = keyPrefix;
-        this.ttl = ttl;
-    }
-
-    @Override
-    public Optional<Jt808DailyMileageState> load(String vehicleKey, LocalDate statDate) {
-        String json = redis.opsForValue().get(key(vehicleKey, statDate));
-        if (json == null || json.isBlank()) {
-            return Optional.empty();
-        }
-        try {
-            return Optional.of(objectMapper.readValue(json, StateSnapshot.class).toState());
-        } catch (JsonProcessingException e) {
-            throw new IllegalStateException("Failed to deserialize JT808 mileage state", e);
-        }
-    }
-
-    @Override
-    public void save(Jt808DailyMileageState state) {
-        try {
-            redis.opsForValue().set(key(state.vehicleKey(), state.statDate()),
-                    objectMapper.writeValueAsString(StateSnapshot.from(state)), ttl);
-        } catch (JsonProcessingException e) {
-            throw new IllegalStateException("Failed to serialize JT808 mileage state", e);
-        }
-    }
-
-    private String key(String vehicleKey, LocalDate statDate) {
-        return keyPrefix + statDate + ":" + vehicleKey;
-    }
-
-    @JsonIgnoreProperties(ignoreUnknown = true)
-    record StateSnapshot(
-            String vehicleKey,
-            String vin,
-            String phone,
-            String statDate,
-            String firstEventTime,
-            String lastEventTime,
-            PointSnapshot lastPoint,
-            Double firstTotalMileageKm,
-            Double lastTotalMileageKm,
-            int acceptedPoints,
-            Integer totalMileageSamples,
-            int outOfOrderPoints,
-            int odometerAnomalies) {
-
-        static StateSnapshot from(Jt808DailyMileageState state) {
-            return new StateSnapshot(
-                    state.vehicleKey(),
-                    state.vin(),
-                    state.phone(),
-                    state.statDate().toString(),
-                    format(state.firstEventTime()),
-                    format(state.lastEventTime()),
-                    null,
-                    finiteOrNull(state.firstTotalMileageKm()),
-                    finiteOrNull(state.lastTotalMileageKm()),
-                    state.acceptedPoints(),
-                    state.totalMileageSamples(),
-                    state.outOfOrderPoints(),
-                    state.odometerAnomalies());
-        }
-
-        Jt808DailyMileageState toState() {
-            return Jt808DailyMileageState.restore(
-                    vehicleKey,
-                    vin,
-                    phone,
-                    LocalDate.parse(statDate),
-                    parseInstant(firstEventTime),
-                    restoredLastEventTime(),
-                    nanIfNull(firstTotalMileageKm),
-                    nanIfNull(lastTotalMileageKm),
-                    acceptedPoints,
-                    restoredTotalMileageSamples(),
-                    outOfOrderPoints,
-                    odometerAnomalies);
-        }
-
-        private Instant restoredLastEventTime() {
-            Instant parsedLastEventTime = parseInstant(lastEventTime);
-            if (parsedLastEventTime != null) {
-                return parsedLastEventTime;
-            }
-            return lastPoint == null ? null : parseInstant(lastPoint.eventTime());
-        }
-
-        private int restoredTotalMileageSamples() {
-            if (totalMileageSamples != null) {
-                return totalMileageSamples;
-            }
-            if (firstTotalMileageKm != null && lastTotalMileageKm != null) {
-                return acceptedPoints >= 2 || !lastTotalMileageKm.equals(firstTotalMileageKm) ? 2 : 1;
-            }
-            return firstTotalMileageKm != null || lastTotalMileageKm != null ? 1 : 0;
-        }
-    }
-
-    @JsonIgnoreProperties(ignoreUnknown = true)
-    record PointSnapshot(
-            String vehicleKey,
-            String vin,
-            String phone,
-            String eventTime,
-            double longitude,
-            double latitude,
-            Double speedKmh,
-            Long statusFlag,
-            Double totalMileageKm) {
-
-    }
-
-    private static String format(Instant instant) {
-        return instant == null ? null : instant.toString();
-    }
-
-    private static Instant parseInstant(String value) {
-        return value == null || value.isBlank() ? null : Instant.parse(value);
-    }
-
-    private static Double finiteOrNull(double value) {
-        return Double.isFinite(value) ? value : null;
-    }
-
-    private static double nanIfNull(Double value) {
-        return value == null ? Double.NaN : value;
-    }
-}