From c6a647ce36a29579f63db5b87d051e9c68560575 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 01:39:13 +0800 Subject: [PATCH] feat: store jt808 daily mileage as metric --- .../operations/jt808-daily-mileage-runbook.md | 7 +- .../2026-06-30-kafka-streaming-mileage.md | 8 +- .../src/main/resources/application.yml | 1 + .../VehicleAnalyticsAppDefaultsTest.java | 1 + .../JdbcVehicleStatMetricRepository.java | 108 ++++++++++++++++++ .../config/VehicleStatAutoConfiguration.java | 13 ++- .../config/VehicleStatProperties.java | 11 ++ .../JdbcVehicleStatMetricRepositoryTest.java | 44 +++++++ .../VehicleStatAutoConfigurationTest.java | 22 ++++ 9 files changed, 210 insertions(+), 5 deletions(-) create mode 100644 modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java create mode 100644 modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java diff --git a/docs/operations/jt808-daily-mileage-runbook.md b/docs/operations/jt808-daily-mileage-runbook.md index 48ebe602..52d5b2ad 100644 --- a/docs/operations/jt808-daily-mileage-runbook.md +++ b/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. Runtime state is stored under the configured JT808 state store so the stream can continue after restart. The derived metric is written through `VehicleStatRepository.saveDailyStat(...)`; the default repository persists it under `VEHICLE_STAT_FILE_PATH` as `daily-stats.tsv`. +There is no separate JT808 daily-mileage table. Runtime state is stored under the configured JT808 state store so the stream can continue after restart. The derived metric is written through `VehicleStatRepository.saveDailyStat(...)`; production uses the common JDBC/MySQL metric table `vehicle_stat_metric`. Local development can force the file fallback with `VEHICLE_STAT_REPOSITORY_TYPE=file`, which writes `daily-stats.tsv` under `VEHICLE_STAT_FILE_PATH`. The JT808 daily-mileage value is calculated from the GPS total mileage reported in location additional information: @@ -23,8 +23,11 @@ Set these in Portainer or Nacos, without committing secrets: ```text KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1 VEHICLE_STAT_JT808_MILEAGE_ENABLED=true +VEHICLE_STAT_REPOSITORY_TYPE=jdbc VEHICLE_STAT_JT808_STATE_STORE=redis -VEHICLE_STAT_FILE_PATH=/data/vehicle-stat +MYSQL_JDBC_URL= +MYSQL_USERNAME= +MYSQL_PASSWORD= REDIS_HOST= REDIS_PORT=6379 REDIS_DATABASE=50 diff --git a/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md b/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md index a544d38b..3b37052c 100644 --- a/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md +++ b/docs/superpowers/plans/2026-06-30-kafka-streaming-mileage.md @@ -10,7 +10,8 @@ Consume JT808 Kafka location events, keep rolling daily state, and write the der - Runtime app: `vehicle-analytics-app` - State: Redis or in-memory `Jt808MileageStateStore` - Metric output: `VehicleStatRepository.saveDailyStat(...)` -- Default metric storage: `VEHICLE_STAT_FILE_PATH/daily-stats.tsv` +- Production metric storage: JDBC/MySQL `vehicle_stat_metric` +- Local fallback storage: `VEHICLE_STAT_FILE_PATH/daily-stats.tsv` - Date boundary: `Asia/Shanghai` JT808 daily mileage is calculated only from the GPS total mileage reported by JT808 location additional information: @@ -27,8 +28,11 @@ The stream saves a metric only when at least two ordered location points for the KAFKA_TOPIC_JT808_EVENT=vehicle.event.jt808.v1 VEHICLE_STAT_ENABLED=true VEHICLE_STAT_JT808_MILEAGE_ENABLED=true +VEHICLE_STAT_REPOSITORY_TYPE=jdbc VEHICLE_STAT_JT808_STATE_STORE=redis -VEHICLE_STAT_FILE_PATH=/data/vehicle-stat +MYSQL_JDBC_URL= +MYSQL_USERNAME= +MYSQL_PASSWORD= REDIS_HOST= REDIS_PORT=6379 REDIS_DATABASE=50 diff --git a/modules/apps/vehicle-analytics-app/src/main/resources/application.yml b/modules/apps/vehicle-analytics-app/src/main/resources/application.yml index 94df661f..55e7af8a 100644 --- a/modules/apps/vehicle-analytics-app/src/main/resources/application.yml +++ b/modules/apps/vehicle-analytics-app/src/main/resources/application.yml @@ -78,6 +78,7 @@ lingniu: enabled: ${VEHICLE_STATE_ENABLED:false} vehicle-stat: enabled: ${VEHICLE_STAT_ENABLED:true} + repository-type: ${VEHICLE_STAT_REPOSITORY_TYPE:jdbc} file-path: ${VEHICLE_STAT_FILE_PATH:./target/vehicle-stat/} zone-id: ${VEHICLE_STAT_ZONE_ID:Asia/Shanghai} jt808: diff --git a/modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java b/modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java index 5319e64a..66bad7b6 100644 --- a/modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java +++ b/modules/apps/vehicle-analytics-app/src/test/java/com/lingniu/ingest/analyticsapp/VehicleAnalyticsAppDefaultsTest.java @@ -25,6 +25,7 @@ class VehicleAnalyticsAppDefaultsTest { .containsEntry("lingniu.ingest.event-file-store.enabled", false) .containsEntry("lingniu.ingest.event-history.enabled", false) .containsEntry("lingniu.ingest.vehicle-stat.enabled", "${VEHICLE_STAT_ENABLED:true}") + .containsEntry("lingniu.ingest.vehicle-stat.repository-type", "${VEHICLE_STAT_REPOSITORY_TYPE:jdbc}") .containsEntry("lingniu.ingest.vehicle-state.enabled", "${VEHICLE_STATE_ENABLED:false}") .containsEntry("lingniu.ingest.sink.mq.consumer.enabled", "${KAFKA_CONSUMER_ENABLED:true}") .containsEntry( diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java new file mode 100644 index 00000000..55f1164d --- /dev/null +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepository.java @@ -0,0 +1,108 @@ +package com.lingniu.ingest.vehiclestat; + +import org.springframework.dao.DuplicateKeyException; +import org.springframework.jdbc.core.JdbcTemplate; + +import java.sql.Date; +import java.time.LocalDate; +import java.util.List; +import java.util.Optional; +import java.util.OptionalDouble; + +public final class JdbcVehicleStatMetricRepository implements VehicleStatRepository { + + private static final String DAILY_MILEAGE_KEY = "daily_mileage_km"; + private static final String DAILY_MILEAGE_UNIT = "km"; + + private final JdbcTemplate jdbcTemplate; + + public JdbcVehicleStatMetricRepository(JdbcTemplate jdbcTemplate) { + if (jdbcTemplate == null) { + throw new IllegalArgumentException("jdbcTemplate must not be null"); + } + this.jdbcTemplate = jdbcTemplate; + ensureSchema(); + } + + @Override + public void appendMileagePoint(String vin, MileagePoint point) { + // Production metrics are written as final metric rows; raw source points stay in RAW/locations history. + } + + @Override + public List mileagePoints(String vin, LocalDate statDate) { + return List.of(); + } + + @Override + public void saveDailyStat(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(); + 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); + } + } + + @Override + public Optional findDailyStat(String vin, LocalDate statDate) { + String normalizedVin = clean(vin); + List rows = jdbcTemplate.query(""" + SELECT metric_value, calculation_method + FROM vehicle_stat_metric + WHERE vin = ? AND stat_date = ? AND metric_key = ? + """, (rs, rowNum) -> new VehicleDailyStatResult( + normalizedVin, + statDate, + rs.getBigDecimal("metric_value") == null + ? OptionalDouble.empty() + : OptionalDouble.of(rs.getBigDecimal("metric_value").doubleValue()), + DailyMileageStrategy.valueOf(rs.getString("calculation_method"))), + normalizedVin, Date.valueOf(statDate), DAILY_MILEAGE_KEY); + 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 + WHERE vin = ? AND stat_date = ? AND metric_key = ? + """, value, DAILY_MILEAGE_UNIT, strategy, vin, statDate, DAILY_MILEAGE_KEY); + } + + private void ensureSchema() { + jdbcTemplate.execute(""" + CREATE TABLE IF NOT EXISTS vehicle_stat_metric ( + vin VARCHAR(64) NOT NULL, + stat_date DATE NOT NULL, + metric_key VARCHAR(64) NOT NULL, + metric_value DECIMAL(18,6) NULL, + metric_unit VARCHAR(16) NOT NULL, + calculation_method VARCHAR(64) NOT NULL, + created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, + PRIMARY KEY (vin, stat_date, metric_key) + ) + """); + } + + private static String clean(String value) { + if (value == null || value.isBlank()) { + throw new IllegalArgumentException("vin must not be blank"); + } + return value.trim(); + } +} diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java index 9a63659b..cfdfd54b 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java @@ -6,6 +6,7 @@ import com.lingniu.ingest.vehiclestat.DailyMileageCalculator; import com.lingniu.ingest.vehiclestat.DailyMileageStrategy; import com.lingniu.ingest.vehiclestat.DailyVehicleStatService; import com.lingniu.ingest.vehiclestat.FileVehicleStatRepository; +import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository; import com.lingniu.ingest.vehiclestat.VehicleStatController; import com.lingniu.ingest.vehiclestat.VehicleStatEnvelopeIngestor; import com.lingniu.ingest.vehiclestat.VehicleStatEventSink; @@ -30,6 +31,7 @@ import org.springframework.boot.context.properties.EnableConfigurationProperties import org.springframework.beans.factory.ObjectProvider; import org.springframework.context.annotation.Bean; import org.springframework.data.redis.core.StringRedisTemplate; +import org.springframework.jdbc.core.JdbcTemplate; import java.nio.file.Path; import java.time.Duration; @@ -44,10 +46,19 @@ import java.time.ZoneId; @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "enabled", havingValue = "true") public class VehicleStatAutoConfiguration { + @Bean + @ConditionalOnBean(JdbcTemplate.class) + @ConditionalOnProperty(prefix = "lingniu.ingest.vehicle-stat", name = "repository-type", + havingValue = "jdbc", matchIfMissing = true) + @ConditionalOnMissingBean + public VehicleStatRepository jdbcVehicleStatMetricRepository(JdbcTemplate jdbcTemplate) { + return new JdbcVehicleStatMetricRepository(jdbcTemplate); + } + @Bean @ConditionalOnMissingBean public VehicleStatRepository vehicleStatRepository(VehicleStatProperties props) { - // 派生统计默认落本地文件，便于独立开关；主历史库仍由 event-file-store 管理。 + // 无 JDBC 指标库时才回落本地文件，主要用于开发和单元验证。 return new FileVehicleStatRepository(Path.of(props.getFilePath())); } diff --git a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java index 62f9a32b..a5fdaadd 100644 --- a/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java +++ b/modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatProperties.java @@ -8,6 +8,9 @@ public class VehicleStatProperties { /** 派生统计本地文件根目录；不是 32960 RAW archive 或 DuckDB 历史库目录。 */ private String filePath = "./target/vehicle-stat/"; + /** 指标仓储类型：有 JDBC 数据源时生产使用 jdbc；本地无数据库时回落 file。 */ + private String repositoryType = "jdbc"; + /** 统计自然日口径，默认按国内业务使用东八区。 */ private String zoneId = "Asia/Shanghai"; @@ -21,6 +24,14 @@ public class VehicleStatProperties { this.filePath = filePath; } + public String getRepositoryType() { + return repositoryType; + } + + public void setRepositoryType(String repositoryType) { + this.repositoryType = repositoryType; + } + public String getZoneId() { return zoneId; } diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java new file mode 100644 index 00000000..24183d98 --- /dev/null +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/JdbcVehicleStatMetricRepositoryTest.java @@ -0,0 +1,44 @@ +package com.lingniu.ingest.vehiclestat; + +import org.junit.jupiter.api.Test; +import org.springframework.jdbc.core.JdbcTemplate; +import org.springframework.jdbc.datasource.DriverManagerDataSource; + +import java.time.LocalDate; +import java.util.OptionalDouble; + +import static org.assertj.core.api.Assertions.assertThat; + +class JdbcVehicleStatMetricRepositoryTest { + + @Test + void upsertsDailyMileageIntoCommonMetricTable() { + JdbcTemplate jdbcTemplate = new JdbcTemplate(new DriverManagerDataSource( + "jdbc:h2:mem:vehicle_stat_metric;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1", + "sa", + "")); + JdbcVehicleStatMetricRepository repository = new JdbcVehicleStatMetricRepository(jdbcTemplate); + + repository.saveDailyStat(new VehicleDailyStatResult( + "VIN001", + LocalDate.of(2026, 6, 30), + OptionalDouble.of(12.345), + DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST)); + repository.saveDailyStat(new VehicleDailyStatResult( + "VIN001", + LocalDate.of(2026, 6, 30), + OptionalDouble.of(15.5), + DailyMileageStrategy.CURRENT_LAST_MINUS_PREVIOUS_LAST)); + + assertThat(repository.findDailyStat("VIN001", LocalDate.of(2026, 6, 30))) + .isPresent() + .get() + .extracting(result -> result.dailyMileageKm().orElseThrow()) + .isEqualTo(15.5); + assertThat(jdbcTemplate.queryForObject("SELECT COUNT(*) FROM vehicle_stat_metric", Integer.class)) + .isEqualTo(1); + assertThat(jdbcTemplate.queryForObject( + "SELECT metric_key FROM vehicle_stat_metric WHERE vin = 'VIN001'", + String.class)).isEqualTo("daily_mileage_km"); + } +} diff --git a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java index bdf6a9c8..9ccaf3be 100644 --- a/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java +++ b/modules/services/vehicle-stat-service/src/test/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfigurationTest.java @@ -6,6 +6,7 @@ import com.lingniu.ingest.vehiclestat.DailyMileageCalculator; import com.lingniu.ingest.vehiclestat.DailyMileageStrategy; import com.lingniu.ingest.vehiclestat.DailyVehicleStatService; import com.lingniu.ingest.vehiclestat.FileVehicleStatRepository; +import com.lingniu.ingest.vehiclestat.JdbcVehicleStatMetricRepository; import com.lingniu.ingest.vehiclestat.MileagePoint; import com.lingniu.ingest.vehiclestat.VehicleStatController; import com.lingniu.ingest.vehiclestat.VehicleDailyStatResult; @@ -134,11 +135,32 @@ class VehicleStatAutoConfigurationTest { "lingniu.ingest.vehicle-stat.jt808.state-store=memory") .run(context -> { assertThat(context).hasSingleBean(JdbcTemplate.class); + assertThat(context).hasSingleBean(JdbcVehicleStatMetricRepository.class); assertThat(context).hasSingleBean(Jt808MileageStreamProcessor.class); assertThat(context).hasSingleBean(VehicleStatEnvelopeIngestor.class); }); } + @Test + void canForceFileRepositoryEvenWhenJdbcTemplateExists() { + new ApplicationContextRunner() + .withConfiguration(AutoConfigurations.of( + JdbcTemplateAutoConfiguration.class, + VehicleStatAutoConfiguration.class)) + .withBean(DataSource.class, () -> new DriverManagerDataSource( + "jdbc:h2:mem:vehicle_stat_file_mode;MODE=MySQL;DATABASE_TO_LOWER=TRUE;DB_CLOSE_DELAY=-1", + "sa", + "")) + .withPropertyValues( + "lingniu.ingest.vehicle-stat.enabled=true", + "lingniu.ingest.vehicle-stat.repository-type=file") + .run(context -> { + assertThat(context).hasSingleBean(VehicleStatRepository.class); + assertThat(context).hasSingleBean(FileVehicleStatRepository.class); + assertThat(context).doesNotHaveBean(JdbcVehicleStatMetricRepository.class); + }); + } + private static final class InMemoryVehicleStatRepository implements VehicleStatRepository { @Override public void appendMileagePoint(String vin, MileagePoint point) {} @Override public List mileagePoints(String vin, LocalDate statDate) { return List.of(); }