# JT808 Kafka Streaming Mileage 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:** Consume only JT808 Kafka location events, calculate daily mileage in real time, keep rolling state in Redis or memory, and upsert final/current daily results into MySQL.

**Architecture:** Reuse `vehicle-analytics-app` and `vehicle-stat-service` instead of adding a new runtime. The Kafka consumer subscribes to `vehicle.event.jt808.v1`, extracts JT808 location points from `VehicleEnvelope.telemetry_snapshot`, updates one rolling state per `(vehicleKey, statDate)`, stores that state in Redis DB 50 when available, and writes the current daily aggregate to MySQL table `vehicle_daily_mileage_jt808`. No GB32960 comparison and no historical backfill are required for v1.

**Tech Stack:** Java 21, Spring Boot auto-configuration, existing `sink-mq` Kafka consumer, protobuf `VehicleEnvelope`, Spring Data Redis `StringRedisTemplate`, Spring JDBC `JdbcTemplate`, MySQL `INSERT ... ON DUPLICATE KEY UPDATE`.

---

## Confirmed Runtime Inputs

- Kafka topic: `vehicle.event.jt808.v1`
- Statistics source: JT808 only
- MySQL database: reuse the configured production vehicle database
- MySQL user: use the configured production vehicle database user
- Redis endpoint: configure through `REDIS_HOST` and `REDIS_PORT`
- Redis DB: use `50` by default; `50-60` are available
- Date zone: `Asia/Shanghai`
- Segment rules:
  - max segment gap: `PT5M`
  - max implied speed: `200 km/h`
  - short-gap jump filter: `PT10S` and `300 m`
- Historical backfill: not needed for v1
- Vehicle key:
  - if VIN is present and not `unknown`, use VIN
  - otherwise use `jt808:{phone}`

## Configuration To Add

Add these deployment properties to `vehicle-analytics-app`:

```yaml
spring:
  data:
    redis:
      host: ${REDIS_HOST:}
      port: ${REDIS_PORT:6379}
      password: ${REDIS_PASSWORD:}
      database: ${REDIS_DATABASE:50}

lingniu:
  ingest:
    sink:
      mq:
        consumer:
          bindings:
            vehicleStatEnvelopeConsumerProcessor:
              topics:
                - ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
    vehicle-stat:
      jt808-mileage:
        enabled: ${JT808_MILEAGE_ENABLED:true}
        mysql:
          jdbc-url: ${JT808_MILEAGE_MYSQL_JDBC_URL:}
          username: ${JT808_MILEAGE_MYSQL_USERNAME:}
          password: ${JT808_MILEAGE_MYSQL_PASSWORD:}
          table-name: ${JT808_MILEAGE_MYSQL_TABLE:vehicle_daily_mileage_jt808}
          initialize-schema: ${JT808_MILEAGE_MYSQL_INITIALIZE_SCHEMA:true}
        state:
          store: ${JT808_MILEAGE_STATE_STORE:redis}
          redis-key-prefix: ${JT808_MILEAGE_REDIS_KEY_PREFIX:vehicle:mileage:jt808:daily:}
          ttl: ${JT808_MILEAGE_REDIS_TTL:P3D}
        max-segment-gap: ${JT808_MILEAGE_MAX_SEGMENT_GAP:PT5M}
        max-implied-speed-kmh: ${JT808_MILEAGE_MAX_IMPLIED_SPEED_KMH:200}
        short-gap: ${JT808_MILEAGE_SHORT_GAP:PT10S}
        short-gap-jump-meters: ${JT808_MILEAGE_SHORT_GAP_JUMP_METERS:300}
```

Do not commit real passwords to `application.yml`; pass them through Nacos or environment variables.

## MySQL Table

Create automatically when `initialize-schema=true`:

```sql
CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_jt808 (
  id BIGINT AUTO_INCREMENT PRIMARY KEY,
  stat_date DATE NOT NULL,
  vehicle_key VARCHAR(64) NOT NULL,
  vin VARCHAR(32) NOT NULL DEFAULT '',
  phone VARCHAR(32) NOT NULL DEFAULT '',
  first_event_time DATETIME(3) NULL,
  last_event_time DATETIME(3) NULL,
  gps_mileage_km DECIMAL(12,3) NOT NULL DEFAULT 0,
  speed_integral_km DECIMAL(12,3) NOT NULL DEFAULT 0,
  odometer_mileage_km DECIMAL(12,3) NULL,
  accepted_points INT NOT NULL DEFAULT 0,
  bad_jump_segments INT NOT NULL DEFAULT 0,
  long_gap_segments INT NOT NULL DEFAULT 0,
  out_of_order_points INT NOT NULL DEFAULT 0,
  odometer_anomalies INT NOT NULL DEFAULT 0,
  data_quality VARCHAR(16) NOT NULL DEFAULT 'PARTIAL',
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_jt808_daily_mileage (stat_date, vehicle_key),
  KEY idx_jt808_daily_mileage_vin_date (vin, stat_date),
  KEY idx_jt808_daily_mileage_phone_date (phone, stat_date)
);
```

## File Structure

- Modify `modules/services/vehicle-stat-service/pom.xml`
  - Add `spring-boot-starter-data-redis`, `spring-jdbc`, and `mysql-connector-j` if not already present through module dependencies.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageProperties.java`
  - Configuration properties for MySQL, Redis state, and GPS filters.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPoint.java`
  - Normalized JT808 point.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808LocationPointExtractor.java`
  - Extracts only JT808 location snapshots from Kafka envelopes.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808GpsMileageCalculator.java`
  - Haversine segment calculator with filtering.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageState.java`
  - Rolling mutable daily state.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808DailyMileageResult.java`
  - Persisted result model.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStateStore.java`
  - State load/save abstraction.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/RedisJt808MileageStateStore.java`
  - Redis JSON state store using `StringRedisTemplate`.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/InMemoryJt808MileageStateStore.java`
  - Local fallback state store.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/JdbcJt808DailyMileageRepository.java`
  - MySQL schema initialization and upsert.
- Create `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/jt808/Jt808MileageStreamProcessor.java`
  - One-envelope processing flow.
- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/VehicleStatEnvelopeIngestor.java`
  - Fan out to existing odometer processor and optional JT808 mileage stream processor.
- Modify `modules/services/vehicle-stat-service/src/main/java/com/lingniu/ingest/vehiclestat/config/VehicleStatAutoConfiguration.java`
  - Wire the new JT808 mileage beans.
- Modify `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
  - Change stat binding topic to `vehicle.event.jt808.v1` when JT808 mileage is enabled.

## Task 1: JT808 Point Extraction

**Files:**
- Create: `.../jt808/Jt808LocationPoint.java`
- Create: `.../jt808/Jt808LocationPointExtractor.java`
- Test: `.../Jt808LocationPointExtractorTest.java`

- [ ] **Step 1: Write failing tests**

Test cases:
- Extracts a `source=JT808`, `eventType=LOCATION` envelope with `longitude`, `latitude`, `speed_kmh`, `location_status_raw`, and `total_mileage_km`.
- Returns empty for `source=GB32960`.
- Returns empty when longitude or latitude is missing.
- Uses VIN as `vehicleKey` when VIN is not blank and not `unknown`.
- Uses `jt808:{phone}` when VIN is `unknown`.

- [ ] **Step 2: Implement extractor**

The normalized record should contain:

```java
public record Jt808LocationPoint(
        String vehicleKey,
        String vin,
        String phone,
        Instant eventTime,
        double longitude,
        double latitude,
        Double speedKmh,
        Long statusFlag,
        Double totalMileageKm
) {}
```

- [ ] **Step 3: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service -Dtest=Jt808LocationPointExtractorTest test
```

Expected: tests pass.

## Task 2: GPS Segment Calculator

**Files:**
- Create: `.../jt808/Jt808GpsMileageCalculator.java`
- Test: `.../Jt808GpsMileageCalculatorTest.java`

- [ ] **Step 1: Write failing tests**

Cover:
- Valid adjacent points accumulate Haversine distance.
- `dt <= 0` increments out-of-order/non-positive counter.
- `dt > PT5M` increments long-gap counter and does not accumulate.
- implied speed `> 200 km/h` increments bad-jump counter and does not accumulate.
- speed integral uses adjacent average speed.

- [ ] **Step 2: Implement calculator**

Return a segment result with:
- `gpsKm`
- `speedIntegralKm`
- `used`
- `badJump`
- `longGap`
- `outOfOrder`

- [ ] **Step 3: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service -Dtest=Jt808GpsMileageCalculatorTest test
```

Expected: tests pass.

## Task 3: Rolling State In Redis/Memory

**Files:**
- Create: `.../jt808/Jt808DailyMileageState.java`
- Create: `.../jt808/Jt808MileageStateStore.java`
- Create: `.../jt808/RedisJt808MileageStateStore.java`
- Create: `.../jt808/InMemoryJt808MileageStateStore.java`
- Test: `.../Jt808DailyMileageStateTest.java`
- Test: `.../RedisJt808MileageStateStoreTest.java`

- [ ] **Step 1: Write failing state tests**

Cover:
- First point initializes state without distance.
- Ordered point updates GPS and speed mileage.
- Duplicate/older point is ignored.
- Monotonic `totalMileageKm` updates `odometerMileageKm`.
- Odometer rollback increments anomaly counter.

- [ ] **Step 2: Implement state key**

Use Redis key:

```text
vehicle:mileage:jt808:daily:{statDate}:{vehicleKey}
```

Example:

```text
vehicle:mileage:jt808:daily:2026-06-30:LKLG7C4E7NA774755
vehicle:mileage:jt808:daily:2026-06-30:jt808:40692934289
```

- [ ] **Step 3: Implement Redis store**

Use JSON value and TTL `P3D`. Use DB selection through Spring Redis database config, not by issuing `SELECT` manually.

- [ ] **Step 4: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service -Dtest=Jt808DailyMileageStateTest,RedisJt808MileageStateStoreTest test
```

Expected: tests pass.

## Task 4: MySQL Result Repository

**Files:**
- Create: `.../jt808/Jt808DailyMileageResult.java`
- Create: `.../jt808/JdbcJt808DailyMileageRepository.java`
- Test: `.../JdbcJt808DailyMileageRepositoryTest.java`

- [ ] **Step 1: Write failing repository tests**

Use H2 MySQL mode. Cover:
- Schema initialization creates `vehicle_daily_mileage_jt808`.
- Upsert inserts first result.
- Upsert updates same `(stat_date, vehicle_key)`.
- Query by date and vehicle key returns latest result.

- [ ] **Step 2: Implement repository**

Use `JdbcTemplate` and:

```sql
INSERT INTO vehicle_daily_mileage_jt808 (...)
VALUES (...)
ON DUPLICATE KEY UPDATE
  vin = VALUES(vin),
  phone = VALUES(phone),
  first_event_time = VALUES(first_event_time),
  last_event_time = VALUES(last_event_time),
  gps_mileage_km = VALUES(gps_mileage_km),
  speed_integral_km = VALUES(speed_integral_km),
  odometer_mileage_km = VALUES(odometer_mileage_km),
  accepted_points = VALUES(accepted_points),
  bad_jump_segments = VALUES(bad_jump_segments),
  long_gap_segments = VALUES(long_gap_segments),
  out_of_order_points = VALUES(out_of_order_points),
  odometer_anomalies = VALUES(odometer_anomalies),
  data_quality = VALUES(data_quality)
```

- [ ] **Step 3: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service -Dtest=JdbcJt808DailyMileageRepositoryTest test
```

Expected: tests pass.

## Task 5: Kafka Processor Fan-Out

**Files:**
- Create: `.../jt808/Jt808MileageStreamProcessor.java`
- Modify: `.../VehicleStatEnvelopeIngestor.java`
- Test: `.../Jt808MileageStreamProcessorTest.java`
- Test: `.../VehicleStatEnvelopeIngestorTest.java`

- [ ] **Step 1: Write failing processor tests**

Cover:
- Two JT808 location envelopes update state and upsert MySQL result.
- GB32960 envelope is ignored.
- VIN unknown uses phone key.
- Different natural days create different states.

- [ ] **Step 2: Implement processor flow**

Flow:

```text
VehicleEnvelope bytes
-> Jt808LocationPointExtractor
-> statDate from eventTime at Asia/Shanghai
-> stateStore.load(key) or new state
-> state.apply(point, calculator)
-> stateStore.save(state)
-> mysqlRepository.upsert(state.toResult())
```

- [ ] **Step 3: Modify ingestor**

Keep the old `VehicleStatEventProcessor` path for compatibility, but add optional `Jt808MileageStreamProcessor`. The new processor ignores non-JT808 envelopes internally.

- [ ] **Step 4: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service -Dtest=Jt808MileageStreamProcessorTest,VehicleStatEnvelopeIngestorTest test
```

Expected: tests pass.

## Task 6: Auto-Configuration And App Properties

**Files:**
- Modify: `.../config/VehicleStatAutoConfiguration.java`
- Modify: `modules/apps/vehicle-analytics-app/src/main/resources/application.yml`
- Test: `.../config/VehicleStatAutoConfigurationTest.java`

- [ ] **Step 1: Write failing auto-config tests**

Cover:
- JT808 mileage beans are present when `lingniu.ingest.vehicle-stat.jt808-mileage.enabled=true`.
- Redis state store is used when `state.store=redis` and `StringRedisTemplate` exists.
- Memory state store is used when `state.store=memory`.
- MySQL repository initializes when config is present.

- [ ] **Step 2: Wire beans**

Create beans for:
- `Jt808LocationPointExtractor`
- `Jt808GpsMileageCalculator`
- `Jt808MileageStateStore`
- `JdbcJt808DailyMileageRepository`
- `Jt808MileageStreamProcessor`

- [ ] **Step 3: Update analytics app Kafka binding**

For `vehicleStatEnvelopeConsumerProcessor`, topic must be:

```yaml
- ${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}
```

- [ ] **Step 4: Verify**

Run:

```bash
mvn -pl :vehicle-stat-service,:vehicle-analytics-app -am test
```

Expected: tests pass.

## Task 7: Deployment Smoke Test

**Files:**
- Modify: `deploy/portainer/docker-compose.yml`
- Modify: `docs/operations/gb32960-service-split-runbook.md`

- [ ] **Step 1: Add environment variables**

Add to `vehicle-analytics-app`:

```yaml
REDIS_HOST: ${REDIS_HOST:-}
REDIS_PORT: ${REDIS_PORT:-6379}
REDIS_DATABASE: ${REDIS_DATABASE:-50}
REDIS_PASSWORD: ${REDIS_PASSWORD:-}
KAFKA_TOPIC_JT808_EVENT: ${KAFKA_TOPIC_JT808_EVENT:-vehicle.event.jt808.v1}
JT808_MILEAGE_ENABLED: ${JT808_MILEAGE_ENABLED:-true}
JT808_MILEAGE_STATE_STORE: ${JT808_MILEAGE_STATE_STORE:-redis}
JT808_MILEAGE_MYSQL_JDBC_URL: ${JT808_MILEAGE_MYSQL_JDBC_URL:-}
JT808_MILEAGE_MYSQL_USERNAME: ${JT808_MILEAGE_MYSQL_USERNAME:-}
JT808_MILEAGE_MYSQL_PASSWORD: ${JT808_MILEAGE_MYSQL_PASSWORD:-}
```

- [ ] **Step 2: Run live smoke**

Start analytics app and verify:
- Redis keys appear under `vehicle:mileage:jt808:daily:*`.
- MySQL table `vehicle_daily_mileage_jt808` receives/upserts rows.
- Kafka consumer group `vehicle-stat` advances on `vehicle.event.jt808.v1`.

- [ ] **Step 3: Check one vehicle**

For a known active 808 vehicle, query:

```sql
SELECT *
FROM vehicle_daily_mileage_jt808
WHERE stat_date = CURRENT_DATE
ORDER BY gps_mileage_km DESC
LIMIT 20;
```

Expected: rows show increasing `accepted_points`, `last_event_time`, and mileage during the day.

## Remaining Needed Support

- MySQL password should be supplied through Nacos/env as `JT808_MILEAGE_MYSQL_PASSWORD`; do not commit it.
- Redis password value should be supplied through Nacos/env as `REDIS_PASSWORD`; do not commit it.

## Risks And Decisions

- Redis is rolling state, not the source of truth. MySQL is the durable daily result.
- Memory state is acceptable only for local/dev or emergency Redis outage; restart loses segment continuity until the next point.
- No historical backfill means the first deployment day only calculates from the consumer start offset unless Kafka group is reset.
- JT808 GPS-only mileage is an estimate. Rows with high `bad_jump_segments`, high `long_gap_segments`, or no VIN should be marked `PARTIAL`.

## Self-Review

- Spec coverage: JT808-only Kafka streaming, Redis/memory state, MySQL result persistence, confirmed topic, confirmed daily algorithm defaults, and no backfill are covered.
- Placeholder scan: The only open item is deliberately listed under Remaining Needed Support: MySQL connection identity if not already inherited from deployment.
- Type consistency: The plan uses the `Jt808*` class family throughout and removes cross-protocol verifier scope.