fix(stats): preserve realtime baseline from current candidate

.superpowers/sdd/final-review-fix-report.md

@@ -0,0 +1,47 @@
+## final-review-fix (2026-07-08 15:41 CST)
+
+### Changes
+
+- Made realtime duplicate suppression source-aware in `go/vehicle-gateway/internal/stats/daily_metric.go` by keying the in-memory mileage cache with `vin + protocol + source_key + stat_date`, so identical mileage from different sources is retained while same-source repeats are still skipped.
+- Added optional previous-baseline querying and per-source/day caching in `Writer`; realtime candidates now:
+  - become `NO_PREVIOUS_BASELINE` with reason `missing_previous_source` when the same-source previous-day candidate is absent,
+  - use previous-day `latest_total_mileage_km` as `first_total_mileage_km`,
+  - compute `daily_mileage_km` from previous-day baseline when present,
+  - become `INVALID_DELTA` with reason `outside_daily_range` when the delta is negative or above `maxSelectedDailyMileageKM`.
+- Added generic stale-final cleanup to `ProjectDailyMileage` in `go/vehicle-gateway/internal/stats/source_mileage.go` so `vehicle_daily_mileage` is deleted when no candidate remains selected for the exact `(vin, stat_date, protocol)` target.
+- Changed `BACKFILL_METHOD` default from `scan` to `last_diff` in `go/vehicle-gateway/cmd/stats-backfill/main.go`.
+- Expanded regression coverage for source-aware realtime dedupe, realtime baseline/no-baseline behavior, projector cleanup SQL, stale-final cleanup through backfill, and the backfill method default.
+
+### Test Results
+
+- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./internal/stats -run 'TestWriter|TestProjectDailyMileage|TestSource' -count=1`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.435s`
+- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./cmd/stats-backfill ./internal/stats -count=1`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.643s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.279s`
+- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./... -count=1`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/capacity-check	0.419s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/gateway	0.397s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/history-writer	0.728s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/load-sim	0.698s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/nats-fast-writer	0.688s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/nats-kafka-bridge	0.683s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/realtime-api	0.574s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stat-writer	0.534s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.634s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/capacity	0.834s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope	0.738s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus	0.758s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway	0.483s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/health	0.384s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/history	0.502s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity	0.631s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/loadsim	0.640s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics	0.626s`
+  - `?   	lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability	[no test files]`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960	0.622s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808	0.371s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/yutongmqtt	0.384s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime	0.496s`
+  - `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.443s`
+  - `?   	lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics	[no test files]`

.superpowers/sdd/task-1-report.md

@@ -0,0 +1,54 @@
+# Task 1 Report: Source Metadata Schema And Identity Helpers
+
+## Outcome
+- Implemented Task 1 in `go/vehicle-gateway/internal/stats`.
+- Added source metadata schema, identity helpers, and focused tests.
+
+## RED Evidence
+Initial focused test run failed as expected because the new helpers did not exist yet:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1
+```
+
+Result:
+- `undefined: NormalizeSourceIP`
+- `undefined: NewSourceIdentity`
+- `undefined: SourceIdentity`
+- `undefined: UpsertDataSource`
+
+## GREEN Evidence
+After implementation, the focused tests passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.927s`
+
+Full package verification also passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.355s`
+
+## Files Changed
+- `go/vehicle-gateway/internal/stats/schema.go`
+- `go/vehicle-gateway/internal/stats/source.go`
+- `go/vehicle-gateway/internal/stats/source_test.go`
+
+## Notes
+- `DataSourceTableSQL` was added verbatim to `schema.go` for the future schema wiring step.
+- `UpsertDataSource` follows the brief exactly, including the nil exec panic, empty-source short circuit, and SQL shape.
+
+## Self-Review
+- The implementation is tightly scoped to Task 1.
+- The new tests cover IP normalization, identity construction, and the SQL shape for the upsert helper.
+- No additional concerns at this stage.

.superpowers/sdd/task-2-report.md

@@ -0,0 +1,62 @@
+# Task 2 Report: Candidate Mileage Schema And Writer
+
+## Outcome
+- Implemented Task 2 in `go/vehicle-gateway/internal/stats`.
+- Added candidate mileage schema, sample-to-candidate mapping, and the upsert writer.
+
+## RED Evidence
+The focused candidate tests failed before implementation because the new symbols did not exist:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1
+```
+
+Result:
+- `undefined: SourceKey`
+- `undefined: SourceMileageSample`
+- `undefined: QualityOK`
+- `undefined: UpsertSourceMileage`
+
+## GREEN Evidence
+After implementation, the focused candidate tests passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.556s`
+
+Package verification also passed:
+
+```bash
+cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway
+go test ./internal/stats -count=1
+```
+
+Result:
+- `ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.235s`
+
+## Files Changed
+- `go/vehicle-gateway/internal/stats/schema.go`
+- `go/vehicle-gateway/internal/stats/daily_metric.go`
+- `go/vehicle-gateway/internal/stats/source_mileage.go`
+- `go/vehicle-gateway/internal/stats/source_mileage_test.go`
+
+## Self-Review
+- The candidate schema matches the brief's table shape and indexes.
+- `SamplesFromEnvelope` now carries `EventTime` and `DeviceID`, which the new candidate mapping needs.
+- The upsert SQL is focused on the candidate table and reuses the shared `Execer` interface.
+- No unrelated stats files were modified.
+
+## Review Fix Addendum
+- Preserved manual `platform_name` values when the candidate upsert receives blank runtime input by switching to `COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name)`.
+- Bootstrapped `vehicle_data_source`, `vehicle_daily_mileage_source`, and `vehicle_daily_mileage` before running alter statements in `Writer.EnsureSchema`.
+- Added a defensive blank `SourceIP` guard in `UpsertSourceMileage` so malformed candidate rows are skipped instead of written.
+- Removed the SQL comment that existed only to satisfy a string-match test and updated the focused assertions to check the real `daily_mileage_km` and `platform_name` SQL expressions.
+- Verification run:
+  - `go test ./internal/stats -run 'TestUpsertSourceMileage|TestWriterEnsuresSchema' -count=1`
+  - `go test ./internal/stats -count=1`
+  - Both passed.

.superpowers/sdd/task-3-report.md

@@ -0,0 +1,110 @@
+# Task 3 Report: Final Mileage Election Projector
+
+## Outcome
+
+Implemented `ProjectDailyMileage` in `go/vehicle-gateway/internal/stats/source_mileage.go` and added focused projector coverage in `go/vehicle-gateway/internal/stats/source_mileage_test.go`.
+
+## RED
+
+Before implementation, the new projector test failed as expected because the function did not exist yet:
+
+```text
+internal/stats/source_mileage_test.go:87:9: undefined: ProjectDailyMileage
+```
+
+That confirmed the test was exercising the missing task-3 surface rather than passing accidentally.
+
+## GREEN
+
+After implementation:
+
+- `go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1`
+- `go test ./internal/stats -count=1`
+
+Both passed.
+
+## What Changed
+
+- Added `maxSelectedDailyMileageKM = 1000`.
+- Added `ProjectDailyMileage(...)` with three SQL steps:
+  - clear `is_selected` for the VIN/date/protocol slice,
+  - project the elected row from `vehicle_daily_mileage_source` into `vehicle_daily_mileage`,
+  - mark the winning source row as selected.
+- Kept the candidate identity intact: `source_key` still carries protocol + phone/device + source IP, while the data-source join uses protocol + source IP.
+- Added a focused test that asserts the clear/project/mark sequence and the key SQL fragments.
+
+## Files Changed
+
+- `go/vehicle-gateway/internal/stats/source_mileage.go`
+- `go/vehicle-gateway/internal/stats/source_mileage_test.go`
+
+## Self-Review
+
+- The projector stays aligned with the existing stats schema and helper conventions.
+- The SQL ordering and join logic preserve the intended source identity split; there is no IP-only simplification.
+- I did not add an unrelated schema migration because the current repository state already contained the needed `vehicle_daily_mileage_source` and `vehicle_daily_mileage` columns/indexes for this projector.
+- The only notable follow-up risk is that the final selection query relies on MySQL-style `INSERT ... SELECT ... LIMIT 1 ON DUPLICATE KEY UPDATE` behavior, which matches the rest of the stats package but should still be exercised against the target DB in an integration run.
+
+## Commit
+
+`aa12317b5` - `feat(stats): project elected mileage source`
+
+## Fix (reviewer findings)
+
+Applied the reviewer correction for the trusted source endpoint contract:
+
+- In `go/vehicle-gateway/internal/stats/source_mileage.go`, `projectDailyMileageSQL` now writes `trusted_source_endpoint` from `vehicle_data_source.latest_source_endpoint` via `COALESCE(NULLIF(ds.latest_source_endpoint, ''), s.source_endpoint)` so latest connection metadata is authoritative with a simple fallback.
+- Replaced hardcoded quality predicate with the existing constant using `s.quality_status = '` + QualityOK + `'` in `projectDailyMileageSQL`.
+- Updated `go/vehicle-gateway/internal/stats/source_mileage_test.go` to assert both the endpoint expression and constant-based quality predicate.
+
+Validation commands:
+
+```bash
+cd go/vehicle-gateway && go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+
+cd go/vehicle-gateway && go test ./internal/stats -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+```
+
+## Fix (reviewer findings #2)
+
+Corrected the remaining `projectDailyMileageSQL` contract issue so `trusted_source_endpoint` uses only `vehicle_data_source.latest_source_endpoint`:
+
+- In `go/vehicle-gateway/internal/stats/source_mileage.go`, changed the projection from:
+  `COALESCE(NULLIF(ds.latest_source_endpoint, ''), s.source_endpoint)`
+  to:
+  `ds.latest_source_endpoint`.
+- In `go/vehicle-gateway/internal/stats/source_mileage_test.go`, extended `TestProjectDailyMileageSelectsCandidateAndMarksSource` assertions to:
+  - require `ds.latest_source_endpoint` is selected, and
+  - explicitly verify no fallback expression using `s.source_endpoint` remains.
+
+This enforces the strict source-of-truth rule that endpoint data for final projection must come from latest connection metadata only, even when it is null/empty.
+
+## Fix (reviewer findings #3)
+
+Addressed stale candidate selection re-flagging by narrowing source marking to the same eligibility rules used by projection:
+
+- In `go/vehicle-gateway/internal/stats/source_mileage.go`, `markSelectedSourceSQL` now:
+  - No longer joins `vehicle_daily_mileage` to discover the selected source.
+  - Selects the same single winning candidate directly from `vehicle_daily_mileage_source`, constrained by:
+    - VIN/date/protocol
+    - `quality_status = QualityOK`
+    - mileage bounds (`0..maxSelectedDailyMileageKM`)
+    - enabled source metadata
+    - same trust/sample/time ordering
+  - Marks only that candidate as `is_selected = 1`.
+- In `go/vehicle-gateway/internal/stats/source_mileage_test.go`, `TestProjectDailyMileageSelectsCandidateAndMarksSource` now verifies:
+  - the mark query includes candidate filters and ordering, including `s2.quality_status`, `s2.daily_mileage_km`, and `COALESCE(ds.enabled, 1) = 1`.
+  - it no longer references `vehicle_daily_mileage` as a join source.
+  - it passes the selection window arg through to marking (`maxSelectedDailyMileageKM`) and uses seven bind args.
+
+Validation commands:
+
+```bash
+cd go/vehicle-gateway && go test ./internal/stats -run TestProjectDailyMileage -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+
+cd go/vehicle-gateway && go test ./internal/stats -count=1
+# PASS: lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats
+```

.superpowers/sdd/task-4-report.md

@@ -0,0 +1,6 @@
+Status: Completed Task 4.
+Commit: 821ea8e39 (`feat(stats): write realtime mileage candidates`).
+Test Summary: `go test ./internal/stats -run TestWriterAppendWritesSourceCandidateAndProjection -count=1` (PASS) and `go test ./internal/stats -count=1` (PASS).
+Concerns: `Writer.Append` now always runs projector projection per sample event, even when source identity is unavailable; this preserves final row refresh behavior and matches existing schema flow.
+Files changed: `go/vehicle-gateway/internal/stats/daily_metric.go`, `go/vehicle-gateway/internal/stats/daily_metric_test.go`.
+Report path: `.superpowers/sdd/task-4-report.md`.

.superpowers/sdd/task-5-report.md

@@ -0,0 +1,207 @@
+# Task 5 Report: Backfill Candidate Rows And Final Projection
+
+## Summary
+
+Implemented Task 5 in `go/vehicle-gateway/cmd/stats-backfill` so backfill now writes source candidates into `vehicle_daily_mileage_source` and re-projects `vehicle_daily_mileage` through the shared projector instead of writing final rows directly.
+
+## TDD Evidence
+
+1. Added the failing regression in `go/vehicle-gateway/cmd/stats-backfill/main_test.go`:
+   - `TestDailySourceLastBuildsCandidateKeysBySourceIP`
+   - Updated the existing trusted-source test to the new `normalizedSourceKey(protocol, phone, deviceID, endpoint)` signature.
+2. Verified the red state with:
+
+```bash
+go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1
+```
+
+Observed failure:
+
+```text
+too many arguments in call to normalizedSourceKey
+have (string, string, string, string)
+want (string, string)
+```
+
+3. Implemented the production changes.
+4. Verified green with:
+
+```bash
+go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1
+go test ./cmd/stats-backfill ./internal/stats -count=1
+```
+
+Both commands passed.
+
+## Files Changed
+
+- `go/vehicle-gateway/cmd/stats-backfill/main.go`
+- `go/vehicle-gateway/cmd/stats-backfill/main_test.go`
+
+## What Changed
+
+### 1. Source-key normalization now matches runtime semantics
+
+- Changed backfill `normalizedSourceKey` to call:
+
+```go
+stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, stats.NormalizeSourceIP(endpoint))
+```
+
+- This means candidate keys now use:
+  - protocol
+  - phone or device ID
+  - normalized source IP
+- Port changes on the same source endpoint no longer produce different source keys.
+
+### 2. Scan-mode backfill now preserves per-source candidates
+
+- Expanded raw-frame reads to include `phone`, `device_id`, and `source_endpoint`.
+- Populated `envelope.FrameEnvelope` with those fields before calling `stats.SamplesFromEnvelope`.
+- Changed scan-mode aggregation key from:
+
+```text
+vin + stat_date + protocol
+```
+
+to:
+
+```text
+vin + stat_date + protocol + source_key
+```
+
+- Extended `metricAgg` to carry:
+  - `SourceKey`
+  - `Phone`
+  - `DeviceID`
+  - `SourceEndpoint`
+  - `FirstEventTime`
+  - `LatestEventTime`
+  - `QualityStatus`
+  - `QualityReason`
+
+This keeps scan-mode candidates source-separated instead of mixing sources together.
+
+### 3. Backfill writes candidates + re-projects final rows
+
+- Replaced the direct `vehicle_daily_mileage` batch upsert path.
+- `writeAggregates` now, per aggregate:
+  1. Upserts `vehicle_data_source` using protocol + normalized source IP.
+  2. Upserts a `vehicle_daily_mileage_source` candidate row.
+  3. Calls `stats.ProjectDailyMileage` for the VIN/date/protocol.
+
+- Quality handling:
+  - default `OK`
+  - `INVALID_DELTA` when daily delta is `< 0` or `> 1000`
+  - pre-set non-OK statuses from `last_diff` are preserved
+
+### 4. `last_diff` now preserves every source candidate
+
+- `buildLastDiffAggregates` no longer picks a single trusted source.
+- It now creates one aggregate per current-source candidate using:
+
+```text
+vin + stat_date + protocol + source_key
+```
+
+- If a same-source previous-day baseline exists:
+  - `FirstKM = previous.TotalKM`
+  - `LatestKM = current.TotalKM`
+  - `QualityStatus = OK`
+  - `QualityReason = "same_source_previous_day"`
+
+- If there is no same-source previous-day baseline:
+  - `FirstKM = current.TotalKM`
+  - `LatestKM = current.TotalKM`
+  - `QualityStatus = NO_PREVIOUS_BASELINE`
+  - `QualityReason = "missing_previous_source"`
+
+This preserves candidates without projecting them by default, matching the projector’s `quality_status = 'OK'` filter.
+
+### 5. Reset now clears candidate rows too
+
+- Updated `resetStats` to delete both:
+  - `vehicle_daily_mileage_source`
+  - `vehicle_daily_mileage`
+
+within the requested date/protocol scope.
+
+This prevents stale candidates from affecting a subsequent rebuild run.
+
+## Test Results
+
+```bash
+go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.572s
+```
+
+```bash
+go test ./cmd/stats-backfill ./internal/stats -count=1
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.174s
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.557s
+```
+
+## Concerns / Notes
+
+- `chooseTrustedSource` remains in place for its existing unit test, but the `last_diff` aggregation path no longer relies on it.
+- `ProjectDailyMileage` still follows the shared runtime semantics: it selects only `quality_status = 'OK'` candidates and does not synthesize cross-source deltas.
+
+## Review Fix: Stale Final Rows
+
+### Bug Addressed
+
+When all candidates for a backfill target `(vin, stat_date, protocol)` are non-qualifying (`NO_PREVIOUS_BASELINE` / `INVALID_DELTA`), `ProjectDailyMileage` runs but inserts no final row, leaving any prior `vehicle_daily_mileage` row in place. That violated the expectation that a run with no selected candidate should not keep a historical final row for that key.
+
+### Fix Applied
+
+- Added helper in `cmd/stats-backfill/main.go`:
+  - `clearBackfillFinalMileage(ctx, db, vin, statDate, protocol)`
+  - Executes `DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?`
+  - Guarded against empty key values.
+- Updated `writeAggregates` to call the helper once per exact `(vin, stat_date, protocol)` target before calling `ProjectDailyMileage`.
+- Kept existing runtime writer behavior unchanged; only backfill projection sequencing changed.
+
+### Tests Added
+
+- `TestClearBackfillFinalMileageClearsExactKey` in `cmd/stats-backfill/main_test.go`
+  - Verifies the SQL targets exactly the intended triplet.
+- `TestWriteAggregatesClearsFinalBeforeProjection` in `cmd/stats-backfill/main_test.go`
+  - Verifies backfill writes still occur and final-row clear is executed before projection.
+
+### Updated Evidence
+
+```bash
+go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP|Test.*Reset|Test.*Projection|Test.*Final' -count=1
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.447s
+go test ./cmd/stats-backfill ./internal/stats -count=1
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill	0.188s
+ok  	lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats	0.547s
+```
+
+## Follow-up Fix: Non-Reset Backfill Idempotency
+
+### Additional Bug
+
+The prior fix only cleared `vehicle_daily_mileage` before projection. In non-reset backfills, stale `vehicle_daily_mileage_source` rows for the same `(vin, stat_date, protocol)` could still be selected by `ProjectDailyMileage`, causing old OK candidates to win against fresh backfill input.
+
+### Additional Fix
+
+- Added `clearBackfillTargetMileage` in `go/vehicle-gateway/cmd/stats-backfill/main.go` to delete both tables for the exact target key:
+  - `DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?`
+  - `DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?`
+- Updated `writeAggregates` so this clear happens at most once per `(vin, stat_date, protocol)` and before the first candidate upsert for that target.
+- Kept existing backfill runtime flow intact: still upserts sources/candidates and calls `ProjectDailyMileage` for each target aggregate.
+
+### Tests Added/Updated
+
+- `TestClearBackfillTargetMileageClearsExactKey`:
+  - Verifies both table deletes use the exact VIN/date/protocol tuple.
+- `TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert`:
+  - Verifies clear operations happen before candidate writes and projection, and that the stale-candidate flow is tested by using a non-qualifying candidate status.
+
+### Verification
+
+```bash
+go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP|Test.*Reset|Test.*Projection|Test.*Final|Test.*Candidate' -count=1
+go test ./cmd/stats-backfill ./internal/stats -count=1
+```

docs/superpowers/specs/2026-07-08-multi-source-mileage-stats-design.md

@@ -0,0 +1,161 @@
+# Multi-Source Mileage Statistics Design
+
+## Goal
+
+Vehicle mileage statistics must keep every protocol source that reports data, while still exposing one reliable default result for BI and API consumers. The design separates source facts from elected business results so multi-source JT808, GB32960, and Yutong MQTT data can be audited, corrected, and re-elected without losing raw source evidence.
+
+## Current Problem
+
+The current `vehicle_daily_mileage` table stores one row per `vin + stat_date + protocol`. That prevents duplicate counting, but it also forces source election too early. When multiple source platforms report the same VIN, mixing them can create false mileage jumps; selecting only one source can undercount when a reliable alternative exists.
+
+The observed example was `LA9GG64L7PBAF4001` on `2026-07-08` for JT808. Two source IPs reported different total mileage ranges. The correct model is to retain each source as its own candidate and then elect the recommended candidate for default queries.
+
+## Tables
+
+### `vehicle_data_source`
+
+Stores source platform metadata discovered from incoming data and later maintained by operators.
+
+Identity:
+
+- `protocol`
+- `source_ip`
+
+Fields:
+
+- `id`
+- `protocol`
+- `source_ip`
+- `latest_source_endpoint`
+- `platform_name`
+- `trust_priority`
+- `enabled`
+- `first_seen_at`
+- `latest_seen_at`
+- `remark`
+- `created_at`
+- `updated_at`
+
+Rules:
+
+- `protocol + source_ip` is unique.
+- Ports are not part of source identity because sender ports change often.
+- Runtime may update `latest_source_endpoint`, `first_seen_at`, and `latest_seen_at`.
+- Runtime must not overwrite manually maintained `platform_name`, `trust_priority`, `enabled`, or `remark`.
+
+### `vehicle_daily_mileage_source`
+
+Stores one daily candidate per source.
+
+Identity:
+
+- `vin`
+- `stat_date`
+- `protocol`
+- `source_key`
+
+Fields:
+
+- `vin`
+- `stat_date`
+- `protocol`
+- `source_key`
+- `source_ip`
+- `source_endpoint`
+- `phone`
+- `device_id`
+- `platform_name`
+- `first_total_mileage_km`
+- `latest_total_mileage_km`
+- `daily_mileage_km`
+- `sample_count`
+- `first_event_time`
+- `latest_event_time`
+- `quality_status`
+- `quality_reason`
+- `is_selected`
+- `created_at`
+- `updated_at`
+
+Rules:
+
+- Candidate rows are facts by source. They should not be deleted just because they are not selected.
+- `source_key` is normalized as protocol-specific device identity plus source IP. For JT808 that is usually `phone@source_ip`.
+- A candidate can be marked invalid without being removed.
+- Candidate statistics never mix different `source_key` values.
+
+### `vehicle_daily_mileage`
+
+Keeps the existing default query surface for BI and APIs.
+
+Rules:
+
+- One row per `vin + stat_date + protocol`.
+- The row is a projection from one selected `vehicle_daily_mileage_source` row.
+- It stores selected source metadata for fast default queries.
+
+## Election
+
+The default selected source is chosen from candidate rows. Selection priority:
+
+1. Source must be enabled in `vehicle_data_source`.
+2. Prefer lower `trust_priority` when configured.
+3. Prefer candidates with a valid same-source previous-day baseline.
+4. Reject impossible deltas, such as negative mileage or mileage above the configured daily maximum.
+5. Prefer more samples when priority and quality are tied.
+6. Prefer later `latest_event_time` when still tied.
+
+If trusted source A is missing on a day, source B can be selected only if B has its own previous-day baseline. The system must not calculate `B today - A yesterday`.
+
+## Data Flow
+
+1. Gateway/history/realtime parsing continues to emit protocol fields without non-protocol synthetic fields.
+2. Stats writer extracts protocol-specific total mileage from the existing fields event.
+3. Stats writer upserts `vehicle_data_source` by `protocol + source_ip`.
+4. Stats writer upserts `vehicle_daily_mileage_source` by `vin + stat_date + protocol + source_key`.
+5. Election updates `vehicle_daily_mileage` from candidates.
+6. History backfill can rebuild candidates from TDengine raw frames and then rebuild final projections.
+
+## API Behavior
+
+Default daily mileage API continues to read `vehicle_daily_mileage`.
+
+Future query extension:
+
+- `includeCandidates=true` returns all candidates for each selected row.
+- Candidate rows include `platform_name`, `source_ip`, `source_endpoint`, `quality_status`, and `quality_reason`.
+
+## Edge Cases
+
+- Multiple ports from the same platform source are collapsed by source IP.
+- A new source with no previous-day baseline is stored as a candidate with `NO_PREVIOUS_BASELINE` and is not selected by default.
+- A source that reports a total-mileage reset is stored as a candidate with an invalid quality status until a clear policy is applied.
+- If all candidates are invalid, no final row is projected for that VIN/day/protocol.
+- Manual platform names and priority changes should allow re-election without reparsing raw data.
+
+## Performance
+
+- Real-time writes are small upserts keyed by indexed dimensions.
+- Default BI/API reads stay fast because they read the final projection table.
+- Candidate/audit queries are opt-in.
+- Backfill aggregates by source and date instead of writing every raw frame to MySQL.
+
+## Testing
+
+Automated tests should cover:
+
+- Source IP extraction and endpoint normalization.
+- Source table upsert preserving manually maintained fields.
+- Candidate stats never mixing different source keys.
+- Election choosing the continuous source for `LA9GG64L7PBAF4001`-style data.
+- Election skipping `B today - A yesterday`.
+- Backfill rebuilding candidates and final projection.
+
+## Rollout
+
+1. Add schemas and repository helpers.
+2. Write candidate rows in stats writer and backfill.
+3. Add election projection into `vehicle_daily_mileage`.
+4. Rebuild `2026-07-08` JT808 statistics and verify `LA9GG64L7PBAF4001`.
+5. Deploy to ECS and monitor stat-writer lag and write errors.
+

go/vehicle-gateway/cmd/stats-backfill/main.go

@@ -41,6 +41,9 @@ type config struct {
 type rawFrameRow struct {
 	Protocol       envelope.Protocol
 	VIN            string
+	Phone          string
+	DeviceID       string
+	SourceEndpoint string
 	EventID        string
 	MessageID      string
 	EventTimeMS    int64
@@ -57,13 +60,19 @@ type metricAgg struct {
 	Count           int64
 	SourceKey       string
 	Phone           string
+	DeviceID        string
 	SourceEndpoint  string
+	FirstEventTime  time.Time
+	LatestEventTime time.Time
+	QualityStatus   string
+	QualityReason   string
 }
 
 type dailySourceLast struct {
 	VIN            string
 	SourceKey      string
 	Phone          string
+	DeviceID       string
 	SourceEndpoint string
 	TS             time.Time
 	TotalKM        float64
@@ -168,6 +177,9 @@ func main() {
 			Protocol:       row.Protocol,
 			MessageID:      row.MessageID,
 			VIN:            row.VIN,
+			Phone:          row.Phone,
+			DeviceID:       row.DeviceID,
+			SourceEndpoint: row.SourceEndpoint,
 			EventTimeMS:    row.EventTimeMS,
 			ReceivedAtMS:   row.ReceivedAtMS,
 			Fields:         fields,
@@ -205,7 +217,7 @@ func main() {
 
 func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) {
 	for _, sample := range samples {
-		key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol)
+		key := sample.VIN + "|" + sample.StatDate + "|" + string(sample.Protocol) + "|" + sample.SourceKey
 		agg, ok := aggregates[key]
 		if !ok {
 			aggregates[key] = &metricAgg{
@@ -215,6 +227,14 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
 				FirstKM:         sample.TotalMileageKM,
 				LatestKM:        sample.TotalMileageKM,
 				Count:           1,
+				SourceKey:       sample.SourceKey,
+				Phone:           sample.Phone,
+				DeviceID:        sample.DeviceID,
+				SourceEndpoint:  sample.SourceEndpoint,
+				FirstEventTime:  sample.EventTime,
+				LatestEventTime: sample.EventTime,
+				QualityStatus:   stats.QualityNoPreviousBaseline,
+				QualityReason:   "scan_backfill_missing_previous_source",
 			}
 			continue
 		}
@@ -224,6 +244,19 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
 		if sample.TotalMileageKM > agg.LatestKM {
 			agg.LatestKM = sample.TotalMileageKM
 		}
+		if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) {
+			agg.FirstEventTime = sample.EventTime
+		}
+		if sample.EventTime.After(agg.LatestEventTime) {
+			agg.LatestEventTime = sample.EventTime
+			agg.SourceEndpoint = sample.SourceEndpoint
+			if strings.TrimSpace(sample.Phone) != "" {
+				agg.Phone = sample.Phone
+			}
+			if strings.TrimSpace(sample.DeviceID) != "" {
+				agg.DeviceID = sample.DeviceID
+			}
+		}
 		agg.Count++
 	}
 }
@@ -232,63 +265,76 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 	if len(aggregates) == 0 {
 		return 0, nil
 	}
-	if batchSize <= 0 {
-		batchSize = 500
-	}
-	rows := make([]*metricAgg, 0, len(aggregates))
-	for _, agg := range aggregates {
-		rows = append(rows, agg)
-	}
 	var written int64
-	for start := 0; start < len(rows); start += batchSize {
+	clearedTargets := map[string]struct{}{}
-		end := start + batchSize
+	for _, agg := range aggregates {
-		if end > len(rows) {
+		identity := stats.SourceIdentity{
-			end = len(rows)
+			Protocol:       agg.Protocol,
+			SourceIP:       stats.NormalizeSourceIP(agg.SourceEndpoint),
+			SourceEndpoint: agg.SourceEndpoint,
 		}
-		if err := writeAggregateBatch(ctx, db, rows[start:end]); err != nil {
+		if strings.TrimSpace(identity.SourceIP) == "" {
+			continue
+		}
+		target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
+		if _, ok := clearedTargets[target]; !ok {
+			if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
 				return written, err
 			}
-		written += int64(end - start)
+			clearedTargets[target] = struct{}{}
+		}
+		if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil {
+			return written, err
+		}
+		dailyKM := agg.LatestKM - agg.FirstKM
+		candidate := stats.SourceMileageSample{
+			VIN:             agg.VIN,
+			StatDate:        agg.Date,
+			Protocol:        agg.Protocol,
+			SourceKey:       agg.SourceKey,
+			SourceIP:        identity.SourceIP,
+			SourceEndpoint:  agg.SourceEndpoint,
+			Phone:           agg.Phone,
+			DeviceID:        agg.DeviceID,
+			FirstTotalKM:    agg.FirstKM,
+			LatestTotalKM:   agg.LatestKM,
+			DailyKM:         dailyKM,
+			SampleCount:     agg.Count,
+			FirstEventTime:  agg.FirstEventTime,
+			LatestEventTime: agg.LatestEventTime,
+			QualityStatus:   agg.QualityStatus,
+			QualityReason:   agg.QualityReason,
+		}
+		if candidate.QualityStatus == "" {
+			candidate.QualityStatus = stats.QualityOK
+		}
+		if candidate.QualityReason == "" {
+			candidate.QualityReason = "same_source_previous_day"
+		}
+		if candidate.QualityStatus == stats.QualityOK && (candidate.DailyKM < 0 || candidate.DailyKM > maxTrustedDailyMileageKM) {
+			candidate.QualityStatus = stats.QualityInvalidDelta
+			candidate.QualityReason = "outside_daily_range"
+		}
+		if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil {
+			return written, err
+		}
+		if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
+			return written, err
+		}
+		written++
 	}
 	return written, nil
 }
 
-func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) error {
+func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error {
-	placeholders := make([]string, 0, len(rows))
+	if db == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(string(protocol)) == "" {
-	args := make([]any, 0, len(rows)*7)
+		return nil
-	for _, row := range rows {
-		daily := row.LatestKM - row.FirstKM
-		if daily < 0 {
-			daily = 0
 	}
-		placeholders = append(placeholders, "(?,?,?,?,?,?,?,?,?,?)")
+	_, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage_source WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
-		args = append(args,
+	if err != nil {
-			row.VIN,
+		return err
-			row.Date,
-			string(row.Protocol),
-			daily,
-			row.FirstKM,
-			row.LatestKM,
-			row.SourceKey,
-			row.Phone,
-			row.SourceEndpoint,
-			row.Count,
-		)
 	}
-	sqlText := `INSERT INTO vehicle_daily_mileage
+	_, err = db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
-  (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km,
-   trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
-VALUES ` + strings.Join(placeholders, ",") + `
-ON DUPLICATE KEY UPDATE
-  daily_mileage_km = VALUES(daily_mileage_km),
-  first_total_mileage_km = VALUES(first_total_mileage_km),
-  latest_total_mileage_km = VALUES(latest_total_mileage_km),
-  trusted_source_key = VALUES(trusted_source_key),
-  trusted_phone = VALUES(trusted_phone),
-  trusted_source_endpoint = VALUES(trusted_source_endpoint),
-  sample_count = VALUES(sample_count),
-  updated_at = CURRENT_TIMESTAMP`
-	_, err := db.ExecContext(ctx, sqlText, args...)
 	return err
 }
 
@@ -320,21 +366,35 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s
 			current := lastByProtocolDate[protocol][date]
 			previous := lastByProtocolDate[protocol][prevDate]
 			for vin, currentSources := range current {
-				chosen, ok := chooseTrustedSource(currentSources, previous[vin])
+				previousBySource := map[string]dailySourceLast{}
-				if !ok {
+				for _, previousRow := range previous[vin] {
-					continue
+					previousBySource[previousRow.SourceKey] = previousRow
 				}
-				key := vin + "|" + date + "|" + string(protocol)
+				for _, currentRow := range currentSources {
-				aggregates[key] = &metricAgg{
+					key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey
+					agg := &metricAgg{
 						VIN:             vin,
 						Date:            date,
 						Protocol:        protocol,
-					FirstKM:        chosen.previous.TotalKM,
+						FirstKM:         currentRow.TotalKM,
-					LatestKM:       chosen.current.TotalKM,
+						LatestKM:        currentRow.TotalKM,
 						Count:           1,
-					SourceKey:      chosen.current.SourceKey,
+						SourceKey:       currentRow.SourceKey,
-					Phone:          chosen.current.Phone,
+						Phone:           currentRow.Phone,
-					SourceEndpoint: chosen.current.SourceEndpoint,
+						DeviceID:        currentRow.DeviceID,
+						SourceEndpoint:  currentRow.SourceEndpoint,
+						FirstEventTime:  currentRow.TS,
+						LatestEventTime: currentRow.TS,
+						QualityStatus:   stats.QualityNoPreviousBaseline,
+						QualityReason:   "missing_previous_source",
+					}
+					if previousRow, ok := previousBySource[currentRow.SourceKey]; ok {
+						agg.FirstKM = previousRow.TotalKM
+						agg.FirstEventTime = previousRow.TS
+						agg.QualityStatus = stats.QualityOK
+						agg.QualityReason = "same_source_previous_day"
+					}
+					aggregates[key] = agg
 				}
 			}
 		}
@@ -383,10 +443,10 @@ func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, proto
 		fmt.Sprintf("protocol = '%s'", quote(string(protocol))),
 		realtimeMileageFramePredicate(),
 	}
-	sqlText := fmt.Sprintf(`SELECT vin, phone, source_endpoint, LAST(ts), LAST(parsed_json)
+	sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json)
 FROM %s.raw_frames
 WHERE %s
-GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
+GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
 	rows, err := db.QueryContext(ctx, sqlText)
 	if err != nil {
 		return nil, err
@@ -396,10 +456,11 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
 	for rows.Next() {
 		var vin string
 		var phone string
+		var deviceID string
 		var sourceEndpoint string
 		var ts time.Time
 		var parsedJSON string
-		if err := rows.Scan(&vin, &phone, &sourceEndpoint, &ts, &parsedJSON); err != nil {
+		if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON); err != nil {
 			return nil, err
 		}
 		fields := fieldsForStats(protocol, vin, parsedJSON)
@@ -416,7 +477,7 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
 		if err != nil || len(samples) == 0 {
 			continue
 		}
-		sourceKey := normalizedSourceKey(phone, sourceEndpoint)
+		sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint)
 		if sourceKey == "" {
 			continue
 		}
@@ -424,6 +485,7 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
 			VIN:            strings.TrimSpace(vin),
 			SourceKey:      sourceKey,
 			Phone:          strings.TrimSpace(phone),
+			DeviceID:       strings.TrimSpace(deviceID),
 			SourceEndpoint: strings.TrimSpace(sourceEndpoint),
 			TS:             ts,
 			TotalKM:        samples[0].TotalMileageKM,
@@ -443,26 +505,9 @@ GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join
 	return out, nil
 }
 
-func normalizedSourceKey(phone string, endpoint string) string {
+func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string {
-	var parts []string
+	sourceIP := stats.NormalizeSourceIP(endpoint)
-	if phone = strings.TrimSpace(phone); phone != "" {
+	return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP)
-		parts = append(parts, phone)
-	}
-	if host := endpointHost(endpoint); host != "" {
-		parts = append(parts, host)
-	}
-	return strings.Join(parts, "@")
-}
-
-func endpointHost(endpoint string) string {
-	endpoint = strings.TrimSpace(endpoint)
-	if endpoint == "" {
-		return ""
-	}
-	if host, _, ok := strings.Cut(endpoint, ":"); ok {
-		return strings.TrimSpace(host)
-	}
-	return endpoint
 }
 
 func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any {
@@ -537,7 +582,7 @@ func loadConfig() (config, error) {
 		DateFrom:         dateFrom,
 		DateTo:           dateTo,
 		Protocols:        protocols,
-		Method:           env("BACKFILL_METHOD", "scan"),
+		Method:           env("BACKFILL_METHOD", "last_diff"),
 		Limit:            envInt("BACKFILL_LIMIT", 0),
 		DryRun:           envBool("BACKFILL_DRY_RUN", true),
 		Reset:            envBool("BACKFILL_RESET", false),
@@ -563,7 +608,7 @@ func queryRawFrames(ctx context.Context, db *sql.DB, cfg config) (*sql.Rows, err
 		where = append(where, "protocol IN ("+strings.Join(quoted, ",")+")")
 	}
 	where = append(where, realtimeMileageFramePredicate())
-	sqlText := fmt.Sprintf(`SELECT protocol, vin, event_id, message_id, event_time, received_at, parsed_json
+	sqlText := fmt.Sprintf(`SELECT protocol, vin, phone, device_id, source_endpoint, event_id, message_id, event_time, received_at, parsed_json
 FROM %s.raw_frames
 WHERE %s
 ORDER BY ts ASC`, ident(cfg.TDengineDatabase), strings.Join(where, " AND "))
@@ -583,15 +628,18 @@ func realtimeMileageFramePredicate() string {
 }
 
 func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) {
-	var protocol, vin, eventID, parsed string
+	var protocol, vin, phone, deviceID, sourceEndpoint, eventID, parsed string
 	var messageID int64
 	var eventTime, receivedAt time.Time
-	if err := rows.Scan(&protocol, &vin, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil {
+	if err := rows.Scan(&protocol, &vin, &phone, &deviceID, &sourceEndpoint, &eventID, &messageID, &eventTime, &receivedAt, &parsed); err != nil {
 		return rawFrameRow{}, err
 	}
 	return rawFrameRow{
 		Protocol:       envelope.Protocol(strings.TrimSpace(protocol)),
 		VIN:            strings.TrimSpace(vin),
+		Phone:          strings.TrimSpace(phone),
+		DeviceID:       strings.TrimSpace(deviceID),
+		SourceEndpoint: strings.TrimSpace(sourceEndpoint),
 		EventID:        strings.TrimSpace(eventID),
 		MessageID:      fmt.Sprintf("0x%04X", messageID),
 		EventTimeMS:    eventTime.UnixMilli(),
@@ -647,11 +695,19 @@ func resetStats(ctx context.Context, db *sql.DB, cfg config) (int64, error) {
 		}
 		clauses = append(clauses, "protocol IN ("+strings.Join(placeholders, ",")+")")
 	}
-	result, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE "+strings.Join(clauses, " AND "), args...)
+	var deleted int64
+	for _, table := range []string{"vehicle_daily_mileage_source", "vehicle_daily_mileage"} {
+		result, err := db.ExecContext(ctx, "DELETE FROM "+table+" WHERE "+strings.Join(clauses, " AND "), args...)
 		if err != nil {
-		return 0, err
+			return deleted, err
 		}
-	return result.RowsAffected()
+		affected, err := result.RowsAffected()
+		if err != nil {
+			return deleted, err
+		}
+		deleted += affected
+	}
+	return deleted, nil
 }
 
 func parseProtocols(value string) ([]envelope.Protocol, error) {

go/vehicle-gateway/cmd/stats-backfill/main_test.go

@@ -1,11 +1,19 @@
 package main
 
-import "testing"
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
+)
 
 func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) {
 	previous := []dailySourceLast{{
 		VIN:            "LA9GG64L7PBAF4001",
-		SourceKey:      normalizedSourceKey("13307765812", "115.231.168.135:42630"),
+		SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"),
 		Phone:          "13307765812",
 		SourceEndpoint: "115.231.168.135:42630",
 		TotalKM:        4100.8,
@@ -13,14 +21,14 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T)
 	current := []dailySourceLast{
 		{
 			VIN:            "LA9GG64L7PBAF4001",
-			SourceKey:      normalizedSourceKey("13307765812", "115.159.85.149:28316"),
+			SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.159.85.149:28316"),
 			Phone:          "13307765812",
 			SourceEndpoint: "115.159.85.149:28316",
 			TotalKM:        42447.2,
 		},
 		{
 			VIN:            "LA9GG64L7PBAF4001",
-			SourceKey:      normalizedSourceKey("13307765812", "115.231.168.135:20215"),
+			SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"),
 			Phone:          "13307765812",
 			SourceEndpoint: "115.231.168.135:20215",
 			TotalKM:        4123.9,
@@ -31,10 +39,228 @@ func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T)
 	if !ok {
 		t.Fatal("chooseTrustedSource() did not choose a source")
 	}
-	if chosen.current.SourceKey != normalizedSourceKey("13307765812", "115.231.168.135:20215") {
+	if chosen.current.SourceKey != normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215") {
 		t.Fatalf("chosen source = %q", chosen.current.SourceKey)
 	}
 	if delta := chosen.current.TotalKM - chosen.previous.TotalKM; delta < 23 || delta > 24 {
 		t.Fatalf("delta = %v, want about 23.1", delta)
 	}
 }
+
+func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
+	sourceA := dailySourceLast{
+		VIN:            "LA9GG64L7PBAF4001",
+		SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:20215"),
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		TotalKM:        4123.9,
+	}
+	sourceB := dailySourceLast{
+		VIN:            "LA9GG64L7PBAF4001",
+		SourceKey:      normalizedSourceKey("JT808", "13307765812", "", "115.231.168.135:42630"),
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:42630",
+		TotalKM:        4100.8,
+	}
+	if sourceA.SourceKey != sourceB.SourceKey {
+		t.Fatalf("same source IP should produce same source key: %q vs %q", sourceA.SourceKey, sourceB.SourceKey)
+	}
+}
+
+func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage_source WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	if err := clearBackfillTargetMileage(context.Background(), db, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808); err != nil {
+		t.Fatalf("clearBackfillTargetMileage() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+	mock.MatchExpectationsInOrder(true)
+
+	projNow := time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
+	aggregates := map[string]*metricAgg{
+		"LA9GG64L7PBAF4001|2026-07-08|JT808|JT808:13307765812@115.231.168.135": {
+			VIN:             "LA9GG64L7PBAF4001",
+			Date:            "2026-07-08",
+			Protocol:        envelope.ProtocolJT808,
+			FirstKM:         4100,
+			LatestKM:        4101,
+			Count:           1,
+			SourceKey:       "JT808:13307765812@115.231.168.135",
+			Phone:           "13307765812",
+			SourceEndpoint:  "115.231.168.135:20215",
+			FirstEventTime:  projNow,
+			LatestEventTime: projNow,
+			QualityStatus:   stats.QualityInvalidDelta,
+			QualityReason:   "outside_daily_range",
+		},
+	}
+
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage_source WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`INSERT INTO vehicle_data_source`).
+		WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"JT808:13307765812@115.231.168.135",
+			"115.231.168.135",
+			"115.231.168.135:20215",
+			"13307765812",
+			"",
+			"",
+			float64(4100),
+			float64(4101),
+			float64(1),
+			int64(1),
+			projNow,
+			projNow,
+			stats.QualityInvalidDelta,
+			"outside_daily_range",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
+		WillReturnResult(sqlmock.NewResult(1, 0))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			int64(1000),
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	written, err := writeAggregates(context.Background(), db, aggregates, 500)
+	if err != nil {
+		t.Fatalf("writeAggregates() error = %v", err)
+	}
+	if written != 1 {
+		t.Fatalf("written = %d, want 1", written)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriteAggregatesSkipsBlankSourceBeforeClearingTarget(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	written, err := writeAggregates(context.Background(), db, map[string]*metricAgg{
+		"LA9GG64L7PBAF4001|2026-07-08|JT808|JT808:13307765812@": {
+			VIN:             "LA9GG64L7PBAF4001",
+			Date:            "2026-07-08",
+			Protocol:        envelope.ProtocolJT808,
+			FirstKM:         4100,
+			LatestKM:        4101,
+			Count:           1,
+			SourceKey:       "JT808:13307765812@",
+			Phone:           "13307765812",
+			SourceEndpoint:  "",
+			FirstEventTime:  time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+			LatestEventTime: time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+			QualityStatus:   stats.QualityNoPreviousBaseline,
+			QualityReason:   "missing_previous_source",
+		},
+	}, 500)
+	if err != nil {
+		t.Fatalf("writeAggregates() error = %v", err)
+	}
+	if written != 0 {
+		t.Fatalf("written = %d, want 0", written)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	samples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    time.Date(2026, 7, 8, 12, 0, 0, 0, loc).UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}, loc)
+	if err != nil {
+		t.Fatalf("SamplesFromEnvelope() error = %v", err)
+	}
+	aggregates := map[string]*metricAgg{}
+	addSamples(aggregates, samples)
+	if len(aggregates) != 1 {
+		t.Fatalf("aggregate count = %d, want 1", len(aggregates))
+	}
+	for _, agg := range aggregates {
+		if agg.QualityStatus != stats.QualityNoPreviousBaseline {
+			t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityNoPreviousBaseline)
+		}
+		if agg.QualityReason != "scan_backfill_missing_previous_source" {
+			t.Fatalf("quality reason = %q", agg.QualityReason)
+		}
+	}
+}
+
+func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
+	t.Setenv("BACKFILL_METHOD", "")
+	t.Setenv("BACKFILL_DATE_FROM", "2026-07-08")
+	t.Setenv("BACKFILL_DATE_TO", "2026-07-08")
+	t.Setenv("BACKFILL_PROTOCOLS", "JT808")
+	t.Setenv("LOCAL_TZ", "Asia/Shanghai")
+
+	cfg, err := loadConfig()
+	if err != nil {
+		t.Fatalf("loadConfig() error = %v", err)
+	}
+	if cfg.Method != "last_diff" {
+		t.Fatalf("method = %q, want last_diff", cfg.Method)
+	}
+}

go/vehicle-gateway/internal/stats/daily_metric.go

@@ -19,9 +19,11 @@ type Execer interface {
 
 type Writer struct {
 	exec             Execer
+	query            Queryer
 	loc              *time.Location
 	mu               sync.Mutex
 	lastTotalMileage map[string]float64
+	baselineCache    map[string]sourceBaselineCacheEntry
 }
 
 type MetricSample struct {
@@ -29,8 +31,10 @@ type MetricSample struct {
 	Protocol       envelope.Protocol
 	StatDate       string
 	TotalMileageKM float64
+	EventTime      time.Time
 	SourceKey      string
 	Phone          string
+	DeviceID       string
 	SourceEndpoint string
 }
 
@@ -41,13 +45,28 @@ func NewWriter(exec Execer, loc *time.Location) *Writer {
 	if loc == nil {
 		loc = time.FixedZone("Asia/Shanghai", 8*3600)
 	}
-	return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}}
+	writer := &Writer{
+		exec:             exec,
+		loc:              loc,
+		lastTotalMileage: map[string]float64{},
+		baselineCache:    map[string]sourceBaselineCacheEntry{},
+	}
+	if query, ok := exec.(Queryer); ok {
+		writer.query = query
+	}
+	return writer
 }
 
 func (w *Writer) EnsureSchema(ctx context.Context) error {
-	if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil {
+	for _, statement := range []string{
+		DataSourceTableSQL,
+		DailyMileageSourceTableSQL,
+		DailyMileageTableSQL,
+	} {
+		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
 			return err
 		}
+	}
 	for _, statement := range DailyMileageAlterSQL {
 		if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
 			return err
@@ -61,19 +80,25 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
 	if err != nil {
 		return err
 	}
+	identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint)
 	for _, sample := range samples {
 		if w.seenSameMileage(sample) {
 			continue
 		}
-		if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL,
+		if !hasSource {
-			sample.VIN,
+			continue
-			sample.StatDate,
+		}
-			string(sample.Protocol),
+		if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil {
-			sample.TotalMileageKM,
+			return err
-			sample.TotalMileageKM,
+		}
-			sample.SourceKey,
+		candidate := SourceMileageSampleFromMetric(sample, identity)
-			sample.Phone,
+		if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil {
-			sample.SourceEndpoint); err != nil {
+			return err
+		}
+		if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
+			return err
+		}
+		if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
 			return err
 		}
 		w.markMileageWritten(sample)
@@ -82,8 +107,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
 }
 
 func (w *Writer) seenSameMileage(sample MetricSample) bool {
-	prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
+	key := mileageCacheKey(sample)
-	key := prefix + sample.StatDate
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM {
@@ -93,8 +117,8 @@ func (w *Writer) seenSameMileage(sample MetricSample) bool {
 }
 
 func (w *Writer) markMileageWritten(sample MetricSample) {
-	prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol)
+	prefix := mileageCachePrefix(sample)
-	key := prefix + sample.StatDate
+	key := mileageCacheKey(sample)
 	w.mu.Lock()
 	defer w.mu.Unlock()
 	for existing := range w.lastTotalMileage {
@@ -102,9 +126,85 @@ func (w *Writer) markMileageWritten(sample MetricSample) {
 			delete(w.lastTotalMileage, existing)
 		}
 	}
+	for existing := range w.baselineCache {
+		if strings.HasPrefix(existing, prefix) && existing != key {
+			delete(w.baselineCache, existing)
+		}
+	}
 	w.lastTotalMileage[key] = sample.TotalMileageKM
 }
 
+func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMileageSample) error {
+	if candidate == nil {
+		return nil
+	}
+	baseline, found, err := w.previousBaseline(ctx, *candidate)
+	if err != nil {
+		return err
+	}
+	if !found {
+		candidate.QualityStatus = QualityNoPreviousBaseline
+		candidate.QualityReason = "missing_previous_source"
+		return nil
+	}
+	candidate.FirstTotalKM = baseline.LatestTotalKM
+	candidate.FirstEventTime = baseline.LatestEventTime
+	candidate.DailyKM = candidate.LatestTotalKM - baseline.LatestTotalKM
+	candidate.QualityStatus = QualityOK
+	candidate.QualityReason = "same_source_previous_day"
+	if candidate.DailyKM < 0 || candidate.DailyKM > maxSelectedDailyMileageKM {
+		candidate.QualityStatus = QualityInvalidDelta
+		candidate.QualityReason = "outside_daily_range"
+	}
+	return nil
+}
+
+func (w *Writer) previousBaseline(ctx context.Context, candidate SourceMileageSample) (sourceBaseline, bool, error) {
+	cacheKey := mileageCacheKey(MetricSample{
+		VIN:       candidate.VIN,
+		Protocol:  candidate.Protocol,
+		StatDate:  candidate.StatDate,
+		SourceKey: candidate.SourceKey,
+	})
+	w.mu.Lock()
+	if cached, ok := w.baselineCache[cacheKey]; ok {
+		w.mu.Unlock()
+		return cached.baseline, cached.found, nil
+	}
+	w.mu.Unlock()
+
+	baseline, found, err := lookupPreviousSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey)
+	if err != nil {
+		return sourceBaseline{}, false, err
+	}
+	if !found {
+		baseline, found, err = lookupCurrentSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey)
+		if err != nil {
+			return sourceBaseline{}, false, err
+		}
+	}
+
+	if found {
+		w.mu.Lock()
+		w.baselineCache[cacheKey] = sourceBaselineCacheEntry{baseline: baseline, found: true}
+		w.mu.Unlock()
+	}
+	return baseline, found, nil
+}
+
+func mileageCachePrefix(sample MetricSample) string {
+	return fmt.Sprintf("%s|%s|%s|", sample.VIN, sample.Protocol, sample.SourceKey)
+}
+
+func mileageCacheKey(sample MetricSample) string {
+	return mileageCachePrefix(sample) + sample.StatDate
+}
+
+type sourceBaselineCacheEntry struct {
+	baseline sourceBaseline
+	found    bool
+}
+
 func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) {
 	vin := strings.TrimSpace(env.VIN)
 	if vin == "" {
@@ -133,25 +233,16 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr
 		Protocol:       env.Protocol,
 		StatDate:       statDate,
 		TotalMileageKM: totalMileage,
+		EventTime:      time.UnixMilli(eventMS).In(loc),
 		SourceKey:      sourceKey(env),
 		Phone:          strings.TrimSpace(env.Phone),
+		DeviceID:       strings.TrimSpace(env.DeviceID),
 		SourceEndpoint: strings.TrimSpace(env.SourceEndpoint),
 	}}, nil
 }
 
 func sourceKey(env envelope.FrameEnvelope) string {
-	parts := []string{
+	return SourceKey(env.Protocol, env.Phone, env.DeviceID, NormalizeSourceIP(env.SourceEndpoint))
-		strings.TrimSpace(env.Phone),
-		strings.TrimSpace(env.DeviceID),
-		strings.TrimSpace(env.SourceEndpoint),
-	}
-	var kept []string
-	for _, part := range parts {
-		if part != "" {
-			kept = append(kept, part)
-		}
-	}
-	return strings.Join(kept, "@")
 }
 
 func isDuplicateColumnError(err error) bool {
@@ -197,92 +288,6 @@ func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping
 	}
 }
 
-const upsertDailyMileageSQL = `
-INSERT INTO vehicle_daily_mileage
-  (vin, stat_date, protocol, daily_mileage_km,
-   first_total_mileage_km, latest_total_mileage_km,
-   trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
-VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, 1)
-ON DUPLICATE KEY UPDATE
-  first_total_mileage_km = CASE
-    WHEN trusted_source_key IS NOT NULL
-      AND trusted_source_key <> ''
-      AND trusted_source_key <> VALUES(trusted_source_key)
-      AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-      THEN first_total_mileage_km
-    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
-      THEN VALUES(first_total_mileage_km)
-    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
-  END,
-  latest_total_mileage_km = CASE
-    WHEN trusted_source_key IS NOT NULL
-      AND trusted_source_key <> ''
-      AND trusted_source_key <> VALUES(trusted_source_key)
-      AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-      THEN latest_total_mileage_km
-    WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
-      THEN VALUES(latest_total_mileage_km)
-    ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
-  END,
-  daily_mileage_km = GREATEST(
-    CASE
-      WHEN trusted_source_key IS NOT NULL
-        AND trusted_source_key <> ''
-        AND trusted_source_key <> VALUES(trusted_source_key)
-        AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-        THEN latest_total_mileage_km
-      WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
-        THEN VALUES(latest_total_mileage_km)
-      ELSE latest_total_mileage_km
-    END,
-    CASE
-      WHEN trusted_source_key IS NOT NULL
-        AND trusted_source_key <> ''
-        AND trusted_source_key <> VALUES(trusted_source_key)
-        AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-        THEN latest_total_mileage_km
-      ELSE VALUES(latest_total_mileage_km)
-    END
-  ) - CASE
-    WHEN trusted_source_key IS NOT NULL
-      AND trusted_source_key <> ''
-      AND trusted_source_key <> VALUES(trusted_source_key)
-      AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-      THEN first_total_mileage_km
-    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
-      THEN VALUES(first_total_mileage_km)
-    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
-  END,
-  trusted_source_key = CASE
-    WHEN trusted_source_key IS NULL OR trusted_source_key = ''
-      THEN VALUES(trusted_source_key)
-    WHEN trusted_source_key = VALUES(trusted_source_key)
-      THEN trusted_source_key
-    WHEN ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) <= 50
-      THEN VALUES(trusted_source_key)
-    ELSE trusted_source_key
-  END,
-  trusted_phone = CASE
-    WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
-      THEN VALUES(trusted_phone)
-    ELSE trusted_phone
-  END,
-  trusted_source_endpoint = CASE
-    WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key)
-      THEN VALUES(trusted_source_endpoint)
-    ELSE trusted_source_endpoint
-  END,
-  sample_count = CASE
-    WHEN trusted_source_key IS NOT NULL
-      AND trusted_source_key <> ''
-      AND trusted_source_key <> VALUES(trusted_source_key)
-      AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50
-      THEN sample_count
-    ELSE sample_count + 1
-  END,
-  updated_at = CURRENT_TIMESTAMP
-`
-
 func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
 	if env.Fields == nil {
 		return 0, false

go/vehicle-gateway/internal/stats/daily_metric_test.go

@@ -3,11 +3,14 @@ package stats
 import (
 	"context"
 	"database/sql"
+	"database/sql/driver"
 	"errors"
+	"math"
 	"strings"
 	"testing"
 	"time"
 
+	"github.com/DATA-DOG/go-sqlmock"
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
 )
 
@@ -156,6 +159,360 @@ func TestSamplesFromEnvelopeSkipsNonPositiveMileage(t *testing.T) {
 	}
 }
 
+func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) {
+	exec := &recordingExec{}
+	writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
+	event := envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}
+
+	if err := writer.Append(context.Background(), event); err != nil {
+		t.Fatalf("Append() error = %v", err)
+	}
+	if len(exec.calls) != 6 {
+		t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark + cleanup", len(exec.calls))
+	}
+	if !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") {
+		t.Fatalf("first call should upsert source: %s", exec.calls[0].query)
+	}
+	if !strings.Contains(exec.calls[1].query, "INSERT INTO vehicle_daily_mileage_source") {
+		t.Fatalf("second call should upsert candidate: %s", exec.calls[1].query)
+	}
+	if !strings.Contains(exec.calls[3].query, "INSERT INTO vehicle_daily_mileage") {
+		t.Fatalf("fourth call should project final: %s", exec.calls[3].query)
+	}
+}
+
+func TestWriterAppendSkipsMileageWithoutSourceIdentity(t *testing.T) {
+	exec := &recordingExec{}
+	writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
+	event := envelope.FrameEnvelope{
+		Protocol:    envelope.ProtocolJT808,
+		VIN:         "LA9GG64L7PBAF4001",
+		Phone:       "13307765812",
+		EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}
+
+	if err := writer.Append(context.Background(), event); err != nil {
+		t.Fatalf("Append() error = %v", err)
+	}
+	if len(exec.calls) != 0 {
+		t.Fatalf("exec calls = %d, want 0 without source identity", len(exec.calls))
+	}
+}
+
+func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) {
+	exec := &recordingExec{}
+	writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
+	base := envelope.FrameEnvelope{
+		Protocol:    envelope.ProtocolJT808,
+		VIN:         "LA9GG64L7PBAF4001",
+		Phone:       "13307765812",
+		EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}
+
+	sourceA := base
+	sourceA.SourceEndpoint = "115.231.168.135:20215"
+	if err := writer.Append(context.Background(), sourceA); err != nil {
+		t.Fatalf("Append(sourceA) error = %v", err)
+	}
+
+	sourceB := base
+	sourceB.SourceEndpoint = "115.159.85.149:28316"
+	sourceB.ReceivedAtMS = sourceB.EventTimeMS + 1000
+	if err := writer.Append(context.Background(), sourceB); err != nil {
+		t.Fatalf("Append(sourceB) error = %v", err)
+	}
+
+	sourceADuplicate := sourceA
+	sourceADuplicate.ReceivedAtMS = sourceA.EventTimeMS + 2000
+	if err := writer.Append(context.Background(), sourceADuplicate); err != nil {
+		t.Fatalf("Append(sourceADuplicate) error = %v", err)
+	}
+
+	if got := countExecQueries(exec.calls, "INSERT INTO vehicle_daily_mileage_source"); got != 2 {
+		t.Fatalf("candidate upserts = %d, want 2", got)
+	}
+	if got := countExecQueries(exec.calls, "INSERT INTO vehicle_data_source"); got != 2 {
+		t.Fatalf("source upserts = %d, want 2", got)
+	}
+}
+
+func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	writer := NewWriter(db, loc)
+	eventTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc)
+	event := envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    eventTime.UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}
+
+	mock.ExpectExec(`INSERT INTO vehicle_data_source`).
+		WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT first_total_mileage_km, first_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
+		WillReturnRows(sqlmock.NewRows([]string{"first_total_mileage_km", "first_event_time"}))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"JT808:13307765812@115.231.168.135",
+			"115.231.168.135",
+			"115.231.168.135:20215",
+			"13307765812",
+			"",
+			"",
+			float64(4123.9),
+			float64(4123.9),
+			float64(0),
+			int64(1),
+			eventTime,
+			eventTime,
+			QualityNoPreviousBaseline,
+			"missing_previous_source",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			int64(1000),
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	if err := writer.Append(context.Background(), event); err != nil {
+		t.Fatalf("Append() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	writer := NewWriter(db, loc)
+	previousTime := time.Date(2026, 7, 7, 23, 58, 0, 0, loc)
+	currentTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc)
+	event := envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    currentTime.UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}
+
+	mock.ExpectExec(`INSERT INTO vehicle_data_source`).
+		WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}).
+			AddRow(4100.8, previousTime))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"JT808:13307765812@115.231.168.135",
+			"115.231.168.135",
+			"115.231.168.135:20215",
+			"13307765812",
+			"",
+			"",
+			float64(4100.8),
+			float64(4123.9),
+			approxFloat64{want: 23.1, tolerance: 0.000001},
+			int64(1),
+			previousTime,
+			currentTime,
+			QualityOK,
+			"same_source_previous_day",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			int64(1000),
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+
+	if err := writer.Append(context.Background(), event); err != nil {
+		t.Fatalf("Append() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriterAppendUsesCurrentOKCandidateWhenPreviousBaselineMissing(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	writer := NewWriter(db, loc)
+	firstTime := time.Date(2026, 7, 7, 23, 58, 0, 0, loc)
+	currentTime := time.Date(2026, 7, 8, 15, 56, 0, 0, loc)
+	event := envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    currentTime.UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4136.8,
+		},
+	}
+
+	mock.ExpectExec(`INSERT INTO vehicle_data_source`).
+		WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135").
+		WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}))
+	mock.ExpectQuery(`SELECT first_total_mileage_km, first_event_time FROM vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135").
+		WillReturnRows(sqlmock.NewRows([]string{"first_total_mileage_km", "first_event_time"}).
+			AddRow(4100.8, firstTime))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"JT808:13307765812@115.231.168.135",
+			"115.231.168.135",
+			"115.231.168.135:20215",
+			"13307765812",
+			"",
+			"",
+			float64(4100.8),
+			float64(4136.8),
+			approxFloat64{want: 36.0, tolerance: 0.000001},
+			int64(1),
+			firstTime,
+			currentTime,
+			QualityOK,
+			"same_source_previous_day",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			int64(1000),
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 0))
+
+	if err := writer.Append(context.Background(), event); err != nil {
+		t.Fatalf("Append() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
 func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
 	exec := &recordingExec{}
 	writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
@@ -165,6 +522,8 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
 	if err := writer.Append(context.Background(), envelope.FrameEnvelope{
 		Protocol:       envelope.ProtocolGB32960,
 		VIN:            "LNBVIN00000000002",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
 		EventTimeMS:    time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
 		Fields: map[string]any{
 			envelope.FieldTotalMileageKM: "10000.0",
@@ -173,39 +532,55 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
 		t.Fatalf("Append() error = %v", err)
 	}
 
-	if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_daily_mileage") {
+	if len(exec.calls) != 12 {
-		t.Fatalf("unexpected schema sql: %s", exec.calls[0].query)
-	}
-	for _, column := range []string{"vehicle_key", "id BIGINT", "AUTO_INCREMENT", "created_at"} {
-		if strings.Contains(exec.calls[0].query, column) {
-			t.Fatalf("schema should not include %s: %s", column, exec.calls[0].query)
-		}
-	}
-	if !strings.Contains(exec.calls[0].query, "PRIMARY KEY (vin, stat_date, protocol)") {
-		t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[0].query)
-	}
-	if strings.Contains(exec.calls[0].query, "KEY idx_vin (vin)") {
-		t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[0].query)
-	}
-	if len(exec.calls) != 5 {
 		t.Fatalf("exec calls = %d", len(exec.calls))
 	}
-	upsertCall := exec.calls[4]
+	for i, want := range []string{
-	if !strings.Contains(upsertCall.query, "ON DUPLICATE KEY UPDATE") {
+		"CREATE TABLE IF NOT EXISTS vehicle_data_source",
-		t.Fatalf("unexpected upsert sql: %s", upsertCall.query)
+		"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source",
+		"CREATE TABLE IF NOT EXISTS vehicle_daily_mileage",
+	} {
+		if !strings.Contains(exec.calls[i].query, want) {
+			t.Fatalf("schema call %d = %s, want %s", i, exec.calls[i].query, want)
 		}
-	if strings.Contains(upsertCall.query, "metric_key") || strings.Contains(upsertCall.query, "metric_unit") {
-		t.Fatalf("daily mileage upsert should not use generic metric columns: %s", upsertCall.query)
 	}
-	if !strings.Contains(upsertCall.query, "first_total_mileage_km <= 0") {
+	for _, column := range []string{"vehicle_key", "id BIGINT", "AUTO_INCREMENT", "created_at"} {
-		t.Fatalf("upsert should ignore legacy zero first mileage: %s", upsertCall.query)
+		if strings.Contains(exec.calls[2].query, column) {
+			t.Fatalf("schema should not include %s: %s", column, exec.calls[2].query)
 		}
-	if !strings.Contains(upsertCall.query, "trusted_source_key") {
-		t.Fatalf("upsert should track trusted source: %s", upsertCall.query)
 	}
-	if got := upsertCall.args[0]; got != "LNBVIN00000000002" {
+	if !strings.Contains(exec.calls[2].query, "PRIMARY KEY (vin, stat_date, protocol)") {
+		t.Fatalf("daily mileage table should key by vin/stat_date/protocol: %s", exec.calls[2].query)
+	}
+	if strings.Contains(exec.calls[2].query, "KEY idx_vin (vin)") {
+		t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[2].query)
+	}
+	projectCall := exec.calls[9]
+	if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") {
+		t.Fatalf("unexpected project sql: %s", projectCall.query)
+	}
+	if !strings.Contains(projectCall.query, "ON DUPLICATE KEY UPDATE") {
+		t.Fatalf("unexpected upsert sql: %s", projectCall.query)
+	}
+	if strings.Contains(projectCall.query, "metric_key") || strings.Contains(projectCall.query, "metric_unit") {
+		t.Fatalf("daily mileage upsert should not use generic metric columns: %s", projectCall.query)
+	}
+	if !strings.Contains(projectCall.query, "LEFT JOIN vehicle_data_source ds") {
+		t.Fatalf("project sql should join data source: %s", projectCall.query)
+	}
+	if !strings.Contains(projectCall.query, "trusted_source_key") {
+		t.Fatalf("upsert should track trusted source: %s", projectCall.query)
+	}
+	if got := projectCall.args[0]; got != "LNBVIN00000000002" {
 		t.Fatalf("first upsert arg should be vin, got %#v", got)
 	}
+
+	if !strings.Contains(exec.calls[6].query, "INSERT INTO vehicle_data_source") {
+		t.Fatalf("source upsert query missing: %s", exec.calls[6].query)
+	}
+	if !strings.Contains(exec.calls[7].query, "INSERT INTO vehicle_daily_mileage_source") {
+		t.Fatalf("candidate upsert query missing: %s", exec.calls[7].query)
+	}
 }
 
 func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
@@ -214,6 +589,7 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
 	event := envelope.FrameEnvelope{
 		Protocol:       envelope.ProtocolJT808,
 		VIN:            "LNBVIN00000000001",
+		SourceEndpoint: "115.231.168.135:20215",
 		EventTimeMS:    time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
 		Fields: map[string]any{
 			envelope.FieldTotalMileageKM: 10241.2,
@@ -228,8 +604,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
 		t.Fatalf("duplicate Append() error = %v", err)
 	}
 
-	if len(exec.calls) != 1 {
+	if len(exec.calls) != 6 {
-		t.Fatalf("exec calls = %d, want 1", len(exec.calls))
+		t.Fatalf("exec calls = %d, want 6", len(exec.calls))
 	}
 }
 
@@ -239,6 +615,7 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
 	event := envelope.FrameEnvelope{
 		Protocol:       envelope.ProtocolJT808,
 		VIN:            "LNBVIN00000000001",
+		SourceEndpoint: "115.231.168.135:20215",
 		EventTimeMS:    time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(),
 		Fields: map[string]any{
 			envelope.FieldTotalMileageKM: 10241.2,
@@ -252,8 +629,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
 		t.Fatalf("retry Append() error = %v", err)
 	}
 
-	if len(exec.calls) != 2 {
+	if len(exec.calls) != 7 {
-		t.Fatalf("exec calls = %d, want 2", len(exec.calls))
+		t.Fatalf("exec calls = %d, want 7", len(exec.calls))
 	}
 }
 
@@ -263,6 +640,7 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
 	event := envelope.FrameEnvelope{
 		Protocol:       envelope.ProtocolJT808,
 		VIN:            "LNBVIN00000000001",
+		SourceEndpoint: "115.231.168.135:20215",
 		Fields: map[string]any{
 			envelope.FieldTotalMileageKM: 10241.2,
 		},
@@ -277,14 +655,24 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
 		t.Fatalf("next-day Append() error = %v", err)
 	}
 
-	if len(exec.calls) != 2 {
+	if len(exec.calls) != 12 {
-		t.Fatalf("exec calls = %d, want 2", len(exec.calls))
+		t.Fatalf("exec calls = %d, want 12", len(exec.calls))
 	}
 	if len(writer.lastTotalMileage) != 1 {
 		t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))
 	}
 }
 
+func countExecQueries(calls []execCall, fragment string) int {
+	count := 0
+	for _, call := range calls {
+		if strings.Contains(call.query, fragment) {
+			count++
+		}
+	}
+	return count
+}
+
 type execCall struct {
 	query string
 	args  []any
@@ -304,3 +692,16 @@ func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any
 	}
 	return nil, nil
 }
+
+type approxFloat64 struct {
+	want      float64
+	tolerance float64
+}
+
+func (m approxFloat64) Match(value driver.Value) bool {
+	got, ok := value.(float64)
+	if !ok {
+		return false
+	}
+	return math.Abs(got-m.want) <= m.tolerance
+}

go/vehicle-gateway/internal/stats/query.go

@@ -33,6 +33,9 @@ type MetricRow struct {
 	DailyMileageKM        float64  `json:"daily_mileage_km"`
 	FirstTotalMileageKM   *float64 `json:"first_total_mileage_km,omitempty"`
 	LatestTotalMileageKM  *float64 `json:"latest_total_mileage_km,omitempty"`
+	TrustedSourceKey      *string  `json:"trusted_source_key,omitempty"`
+	TrustedPhone          *string  `json:"trusted_phone,omitempty"`
+	TrustedSourceEndpoint *string  `json:"trusted_source_endpoint,omitempty"`
 	SampleCount           int64    `json:"sample_count"`
 	UpdatedAt             string   `json:"updated_at"`
 }
@@ -64,6 +67,9 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr
 		var updatedAt scanDateTime
 		var first sql.NullFloat64
 		var latest sql.NullFloat64
+		var trustedSourceKey sql.NullString
+		var trustedPhone sql.NullString
+		var trustedSourceEndpoint sql.NullString
 		if err := rows.Scan(
 			&row.VIN,
 			&statDate,
@@ -71,6 +77,9 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr
 			&row.DailyMileageKM,
 			&first,
 			&latest,
+			&trustedSourceKey,
+			&trustedPhone,
+			&trustedSourceEndpoint,
 			&row.SampleCount,
 			&updatedAt,
 		); err != nil {
@@ -84,6 +93,15 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr
 		if latest.Valid {
 			row.LatestTotalMileageKM = &latest.Float64
 		}
+		if trustedSourceKey.Valid {
+			row.TrustedSourceKey = &trustedSourceKey.String
+		}
+		if trustedPhone.Valid {
+			row.TrustedPhone = &trustedPhone.String
+		}
+		if trustedSourceEndpoint.Valid {
+			row.TrustedSourceEndpoint = &trustedSourceEndpoint.String
+		}
 		out = append(out, row)
 	}
 	return out, rows.Err()
@@ -120,7 +138,10 @@ func normalizeMetricQuery(query MetricQuery) MetricQuery {
 
 func buildMetricSQL(query MetricQuery) (string, []any) {
 	where, args := buildMetricWhere(query)
-	sqlText := `SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count, updated_at FROM vehicle_daily_mileage`
+	sqlText := `SELECT vin, stat_date, protocol, daily_mileage_km,
+       first_total_mileage_km, latest_total_mileage_km,
+       trusted_source_key, trusted_phone, trusted_source_endpoint,
+       sample_count, updated_at FROM vehicle_daily_mileage`
 	if len(where) > 0 {
 		sqlText += " WHERE " + strings.Join(where, " AND ")
 	}

go/vehicle-gateway/internal/stats/query_test.go

@@ -18,15 +18,19 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
 		t.Fatalf("sqlmock.New() error = %v", err)
 	}
 	defer db.Close()
-	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count, updated_at FROM vehicle_daily_mileage").
+	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage").
 		WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0).
 		WillReturnRows(sqlmock.NewRows([]string{
 			"vin", "stat_date", "protocol", "daily_mileage_km",
-			"first_total_mileage_km", "latest_total_mileage_km", "sample_count",
+			"first_total_mileage_km", "latest_total_mileage_km",
+			"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
+			"sample_count",
 			"updated_at",
 		}).AddRow(
 			"LKLG7C4E3NA774736", time.Date(2026, 7, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), "JT808", 12.3,
-			12345.6, 12357.9, 13, time.Date(2026, 7, 1, 23, 9, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+			12345.6, 12357.9,
+			nil, nil, nil,
+			13, time.Date(2026, 7, 1, 23, 9, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
 		))
 
 	repository := NewMetricRepository(db)
@@ -54,6 +58,39 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) {
 	}
 }
 
+func TestMetricQueryReturnsSelectedSourceFields(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage").
+		WillReturnRows(sqlmock.NewRows([]string{
+			"vin", "stat_date", "protocol", "daily_mileage_km",
+			"first_total_mileage_km", "latest_total_mileage_km",
+			"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
+			"sample_count", "updated_at",
+		}).AddRow(
+			"LA9GG64L7PBAF4001", "2026-07-08", "JT808", 23.1,
+			4100.8, 4123.9,
+			"JT808:13307765812@115.231.168.135", "13307765812", "115.231.168.135:20215",
+			1, "2026-07-08 13:30:57",
+		))
+
+	repo := NewMetricRepository(db)
+	got, err := repo.Query(context.Background(), MetricQuery{
+		VIN:      "LA9GG64L7PBAF4001",
+		Protocol: "JT808",
+		Limit:    1,
+	})
+	if err != nil {
+		t.Fatalf("Query() error = %v", err)
+	}
+	if got[0].TrustedSourceKey == nil || *got[0].TrustedSourceKey != "JT808:13307765812@115.231.168.135" {
+		t.Fatalf("trusted source = %#v", got[0].TrustedSourceKey)
+	}
+}
+
 func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
 	db, mock, err := sqlmock.New()
 	if err != nil {
@@ -63,15 +100,19 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
 	mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_daily_mileage").
 		WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01").
 		WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(42))
-	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count, updated_at FROM vehicle_daily_mileage").
+	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage").
 		WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0).
 		WillReturnRows(sqlmock.NewRows([]string{
 			"vin", "stat_date", "protocol", "daily_mileage_km",
-			"first_total_mileage_km", "latest_total_mileage_km", "sample_count",
+			"first_total_mileage_km", "latest_total_mileage_km",
+			"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
+			"sample_count",
 			"updated_at",
 		}).AddRow(
 			"LB9A32A21R0LS1707", "2020-07-01", "GB32960", 0.0,
-			53490.9, 53490.9, 3, "2026-07-01 22:28:25",
+			53490.9, 53490.9,
+			nil, nil, nil,
+			3, "2026-07-01 22:28:25",
 		))
 
 	handler := NewMetricHandler(NewMetricRepository(db))
@@ -103,11 +144,13 @@ func TestMetricHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) {
 		t.Fatalf("sqlmock.New() error = %v", err)
 	}
 	defer db.Close()
-	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count, updated_at FROM vehicle_daily_mileage").
+	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage").
 		WithArgs("YUTONG_MQTT", 50, 0).
 		WillReturnRows(sqlmock.NewRows([]string{
 			"vin", "stat_date", "protocol", "daily_mileage_km",
-			"first_total_mileage_km", "latest_total_mileage_km", "sample_count",
+			"first_total_mileage_km", "latest_total_mileage_km",
+			"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
+			"sample_count",
 			"updated_at",
 		}))
 
@@ -135,15 +178,19 @@ func TestMetricHandlerSkipsTotalCountByDefault(t *testing.T) {
 		t.Fatalf("sqlmock.New() error = %v", err)
 	}
 	defer db.Close()
-	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count, updated_at FROM vehicle_daily_mileage").
+	mock.ExpectQuery("SELECT vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count, updated_at FROM vehicle_daily_mileage").
 		WithArgs("JT808", 1, 0).
 		WillReturnRows(sqlmock.NewRows([]string{
 			"vin", "stat_date", "protocol", "daily_mileage_km",
-			"first_total_mileage_km", "latest_total_mileage_km", "sample_count",
+			"first_total_mileage_km", "latest_total_mileage_km",
+			"trusted_source_key", "trusted_phone", "trusted_source_endpoint",
+			"sample_count",
 			"updated_at",
 		}).AddRow(
 			"LKLG7C4E3NA774736", "2026-07-02", "JT808", 12.3,
-			8792.8, 8805.1, 30, "2026-07-02 23:59:59",
+			8792.8, 8805.1,
+			nil, nil, nil,
+			30, "2026-07-02 23:59:59",
 		))
 
 	handler := NewMetricHandler(NewMetricRepository(db))

go/vehicle-gateway/internal/stats/schema.go

@@ -22,3 +22,48 @@ var DailyMileageAlterSQL = []string{
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_phone VARCHAR(32) NULL",
 	"ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_endpoint VARCHAR(128) NULL",
 }
+
+const DataSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_data_source (
+  id BIGINT NOT NULL AUTO_INCREMENT,
+  protocol VARCHAR(32) NOT NULL,
+  source_ip VARCHAR(64) NOT NULL,
+  latest_source_endpoint VARCHAR(128) NULL,
+  platform_name VARCHAR(128) NULL,
+  trust_priority INT NOT NULL DEFAULT 100,
+  enabled TINYINT(1) NOT NULL DEFAULT 1,
+  first_seen_at DATETIME NULL,
+  latest_seen_at DATETIME NULL,
+  remark VARCHAR(512) NULL,
+  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (id),
+  UNIQUE KEY uk_protocol_source_ip (protocol, source_ip),
+  KEY idx_protocol_enabled_priority (protocol, enabled, trust_priority)
+)`
+
+const DailyMileageSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage_source (
+  vin VARCHAR(32) NOT NULL,
+  stat_date DATE NOT NULL,
+  protocol VARCHAR(32) NOT NULL,
+  source_key VARCHAR(256) NOT NULL,
+  source_ip VARCHAR(64) NOT NULL,
+  source_endpoint VARCHAR(128) NULL,
+  phone VARCHAR(32) NULL,
+  device_id VARCHAR(64) NULL,
+  platform_name VARCHAR(128) NULL,
+  first_total_mileage_km DECIMAL(18,3) NULL,
+  latest_total_mileage_km DECIMAL(18,3) NULL,
+  daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0,
+  sample_count BIGINT NOT NULL DEFAULT 0,
+  first_event_time DATETIME NULL,
+  latest_event_time DATETIME NULL,
+  quality_status VARCHAR(32) NOT NULL DEFAULT 'OK',
+  quality_reason VARCHAR(255) NULL,
+  is_selected TINYINT(1) NOT NULL DEFAULT 0,
+  created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+  PRIMARY KEY (vin, stat_date, protocol, source_key),
+  KEY idx_protocol_date (protocol, stat_date),
+  KEY idx_source_ip (protocol, source_ip),
+  KEY idx_selected (stat_date, protocol, is_selected)
+)`

go/vehicle-gateway/internal/stats/source.go

@@ -0,0 +1,68 @@
+package stats
+
+import (
+	"context"
+	"strings"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+type SourceIdentity struct {
+	Protocol       envelope.Protocol
+	SourceIP       string
+	SourceEndpoint string
+}
+
+func NewSourceIdentity(protocol envelope.Protocol, endpoint string) (SourceIdentity, bool) {
+	sourceIP := NormalizeSourceIP(endpoint)
+	if sourceIP == "" {
+		return SourceIdentity{}, false
+	}
+	return SourceIdentity{
+		Protocol:       protocol,
+		SourceIP:       sourceIP,
+		SourceEndpoint: strings.TrimSpace(endpoint),
+	}, true
+}
+
+func NormalizeSourceIP(endpoint string) string {
+	endpoint = strings.TrimSpace(endpoint)
+	if endpoint == "" {
+		return ""
+	}
+	if host, _, ok := strings.Cut(endpoint, ":"); ok {
+		return strings.TrimSpace(host)
+	}
+	return endpoint
+}
+
+func UpsertDataSource(ctx context.Context, exec Execer, identity SourceIdentity, now time.Time) error {
+	if exec == nil {
+		panic("stats execer must not be nil")
+	}
+	if identity.SourceIP == "" {
+		return nil
+	}
+	if now.IsZero() {
+		now = time.Now()
+	}
+	_, err := exec.ExecContext(ctx, upsertDataSourceSQL,
+		string(identity.Protocol),
+		identity.SourceIP,
+		identity.SourceEndpoint,
+		now,
+		now,
+	)
+	return err
+}
+
+const upsertDataSourceSQL = `
+INSERT INTO vehicle_data_source
+  (protocol, source_ip, latest_source_endpoint, first_seen_at, latest_seen_at)
+VALUES (?, ?, ?, ?, ?)
+ON DUPLICATE KEY UPDATE
+  latest_source_endpoint = VALUES(latest_source_endpoint),
+  latest_seen_at = VALUES(latest_seen_at),
+  updated_at = CURRENT_TIMESTAMP
+`

go/vehicle-gateway/internal/stats/source_mileage.go

@@ -0,0 +1,379 @@
+package stats
+
+import (
+	"context"
+	"database/sql"
+	"strings"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+const (
+	QualityOK                 = "OK"
+	QualityNoPreviousBaseline = "NO_PREVIOUS_BASELINE"
+	QualityInvalidDelta       = "INVALID_DELTA"
+	maxSelectedDailyMileageKM = 1000
+)
+
+type SourceMileageSample struct {
+	VIN             string
+	StatDate        string
+	Protocol        envelope.Protocol
+	SourceKey       string
+	SourceIP        string
+	SourceEndpoint  string
+	Phone           string
+	DeviceID        string
+	PlatformName    string
+	FirstTotalKM    float64
+	LatestTotalKM   float64
+	DailyKM         float64
+	SampleCount     int64
+	FirstEventTime  time.Time
+	LatestEventTime time.Time
+	QualityStatus   string
+	QualityReason   string
+}
+
+func SourceKey(protocol envelope.Protocol, phone string, deviceID string, sourceIP string) string {
+	identity := strings.TrimSpace(phone)
+	if identity == "" {
+		identity = strings.TrimSpace(deviceID)
+	}
+	if identity == "" {
+		identity = "unknown"
+	}
+	return string(protocol) + ":" + identity + "@" + strings.TrimSpace(sourceIP)
+}
+
+func SourceMileageSampleFromMetric(sample MetricSample, identity SourceIdentity) SourceMileageSample {
+	eventTime := sample.EventTime
+	return SourceMileageSample{
+		VIN:             sample.VIN,
+		StatDate:        sample.StatDate,
+		Protocol:        sample.Protocol,
+		SourceKey:       SourceKey(sample.Protocol, sample.Phone, sample.DeviceID, identity.SourceIP),
+		SourceIP:        identity.SourceIP,
+		SourceEndpoint:  identity.SourceEndpoint,
+		Phone:           sample.Phone,
+		DeviceID:        sample.DeviceID,
+		FirstTotalKM:    sample.TotalMileageKM,
+		LatestTotalKM:   sample.TotalMileageKM,
+		DailyKM:         0,
+		SampleCount:     1,
+		FirstEventTime:  eventTime,
+		LatestEventTime: eventTime,
+		QualityStatus:   QualityOK,
+		QualityReason:   "realtime_sample",
+	}
+}
+
+func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageSample) error {
+	if exec == nil {
+		panic("stats execer must not be nil")
+	}
+	if sample.VIN == "" || sample.SourceKey == "" || strings.TrimSpace(sample.SourceIP) == "" {
+		return nil
+	}
+	if sample.QualityStatus == "" {
+		sample.QualityStatus = QualityOK
+	}
+	_, err := exec.ExecContext(ctx, upsertSourceMileageSQL,
+		sample.VIN,
+		sample.StatDate,
+		string(sample.Protocol),
+		sample.SourceKey,
+		sample.SourceIP,
+		sample.SourceEndpoint,
+		sample.Phone,
+		sample.DeviceID,
+		sample.PlatformName,
+		sample.FirstTotalKM,
+		sample.LatestTotalKM,
+		sample.DailyKM,
+		sample.SampleCount,
+		sample.FirstEventTime,
+		sample.LatestEventTime,
+		sample.QualityStatus,
+		sample.QualityReason,
+	)
+	return err
+}
+
+func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error {
+	if exec == nil {
+		panic("stats execer must not be nil")
+	}
+	if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" {
+		return nil
+	}
+	if _, err := exec.ExecContext(ctx, clearSelectedSourceSQL, vin, statDate, string(protocol)); err != nil {
+		return err
+	}
+	if _, err := exec.ExecContext(ctx, projectDailyMileageSQL,
+		vin,
+		statDate,
+		string(protocol),
+		maxSelectedDailyMileageKM,
+	); err != nil {
+		return err
+	}
+	_, err := exec.ExecContext(ctx, markSelectedSourceSQL,
+		vin,
+		statDate,
+		string(protocol),
+		maxSelectedDailyMileageKM,
+		vin,
+		statDate,
+		string(protocol),
+	)
+	if err != nil {
+		return err
+	}
+	_, err = exec.ExecContext(ctx, cleanupProjectedDailyMileageSQL,
+		vin,
+		statDate,
+		string(protocol),
+		vin,
+		statDate,
+		string(protocol),
+	)
+	return err
+}
+
+const upsertSourceMileageSQL = `
+INSERT INTO vehicle_daily_mileage_source
+  (vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name,
+   first_total_mileage_km, latest_total_mileage_km, daily_mileage_km, sample_count,
+   first_event_time, latest_event_time, quality_status, quality_reason)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+	ON DUPLICATE KEY UPDATE
+	  source_ip = VALUES(source_ip),
+	  source_endpoint = VALUES(source_endpoint),
+	  phone = VALUES(phone),
+	  device_id = VALUES(device_id),
+	  platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name),
+	  first_total_mileage_km = CASE
+	    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
+	      THEN VALUES(first_total_mileage_km)
+	    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
+	  END,
+	  latest_total_mileage_km = CASE
+	    WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
+	      THEN VALUES(latest_total_mileage_km)
+	    ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km))
+	  END,
+	  daily_mileage_km = GREATEST(
+	    CASE
+	      WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0
+	        THEN VALUES(latest_total_mileage_km)
+      ELSE latest_total_mileage_km
+    END,
+    VALUES(latest_total_mileage_km)
+  ) - CASE
+    WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0
+      THEN VALUES(first_total_mileage_km)
+    ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km))
+  END,
+  sample_count = sample_count + VALUES(sample_count),
+  first_event_time = CASE
+    WHEN first_event_time IS NULL OR VALUES(first_event_time) < first_event_time
+      THEN VALUES(first_event_time)
+    ELSE first_event_time
+  END,
+  latest_event_time = CASE
+    WHEN latest_event_time IS NULL OR VALUES(latest_event_time) > latest_event_time
+      THEN VALUES(latest_event_time)
+    ELSE latest_event_time
+  END,
+  quality_status = VALUES(quality_status),
+  quality_reason = VALUES(quality_reason),
+  updated_at = CURRENT_TIMESTAMP
+`
+
+const clearSelectedSourceSQL = `
+UPDATE vehicle_daily_mileage_source
+SET is_selected = 0
+WHERE vin = ? AND stat_date = ? AND protocol = ?
+`
+
+const projectDailyMileageSQL = `
+INSERT INTO vehicle_daily_mileage
+  (vin, stat_date, protocol, daily_mileage_km,
+   first_total_mileage_km, latest_total_mileage_km,
+   trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count)
+SELECT
+  s.vin,
+  s.stat_date,
+  s.protocol,
+  s.daily_mileage_km,
+  s.first_total_mileage_km,
+  s.latest_total_mileage_km,
+  s.source_key,
+  s.phone,
+  ds.latest_source_endpoint,
+  s.sample_count
+FROM vehicle_daily_mileage_source s
+LEFT JOIN vehicle_data_source ds
+  ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip
+WHERE s.vin = ?
+  AND s.stat_date = ?
+  AND s.protocol = ?
+  AND s.quality_status = '` + QualityOK + `'
+  AND s.daily_mileage_km BETWEEN 0 AND ?
+  AND COALESCE(ds.enabled, 1) = 1
+ORDER BY COALESCE(ds.trust_priority, 100),
+         s.sample_count DESC,
+         s.latest_event_time DESC,
+         s.source_key ASC
+LIMIT 1
+ON DUPLICATE KEY UPDATE
+  daily_mileage_km = VALUES(daily_mileage_km),
+  first_total_mileage_km = VALUES(first_total_mileage_km),
+  latest_total_mileage_km = VALUES(latest_total_mileage_km),
+  trusted_source_key = VALUES(trusted_source_key),
+  trusted_phone = VALUES(trusted_phone),
+  trusted_source_endpoint = VALUES(trusted_source_endpoint),
+  sample_count = VALUES(sample_count),
+  updated_at = CURRENT_TIMESTAMP
+`
+
+const markSelectedSourceSQL = `
+UPDATE vehicle_daily_mileage_source s
+JOIN (
+  SELECT
+    s2.source_key,
+    s2.vin,
+    s2.stat_date,
+    s2.protocol
+  FROM vehicle_daily_mileage_source s2
+  LEFT JOIN vehicle_data_source ds
+    ON ds.protocol = s2.protocol AND ds.source_ip = s2.source_ip
+  WHERE s2.vin = ?
+    AND s2.stat_date = ?
+    AND s2.protocol = ?
+    AND s2.quality_status = '` + QualityOK + `'
+    AND s2.daily_mileage_km BETWEEN 0 AND ?
+    AND COALESCE(ds.enabled, 1) = 1
+  ORDER BY COALESCE(ds.trust_priority, 100),
+           s2.sample_count DESC,
+           s2.latest_event_time DESC,
+           s2.source_key ASC
+  LIMIT 1
+) selected_source
+  ON selected_source.source_key = s.source_key
+ AND selected_source.vin = s.vin
+ AND selected_source.stat_date = s.stat_date
+ AND selected_source.protocol = s.protocol
+SET s.is_selected = 1
+WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ?
+`
+
+const cleanupProjectedDailyMileageSQL = `
+DELETE FROM vehicle_daily_mileage
+WHERE vin = ? AND stat_date = ? AND protocol = ?
+  AND NOT EXISTS (
+    SELECT 1
+    FROM vehicle_daily_mileage_source
+    WHERE vin = ? AND stat_date = ? AND protocol = ? AND is_selected = 1
+  )
+`
+
+type sourceBaseline struct {
+	LatestTotalKM   float64
+	LatestEventTime time.Time
+}
+
+func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (sourceBaseline, bool, error) {
+	if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(sourceKey) == "" {
+		return sourceBaseline{}, false, nil
+	}
+	previousDate, ok := previousStatDate(statDate)
+	if !ok {
+		return sourceBaseline{}, false, nil
+	}
+	rows, err := query.QueryContext(ctx, previousSourceBaselineSQL, vin, previousDate, string(protocol), sourceKey)
+	if err != nil {
+		return sourceBaseline{}, false, err
+	}
+	defer rows.Close()
+	if !rows.Next() {
+		if err := rows.Err(); err != nil {
+			return sourceBaseline{}, false, err
+		}
+		return sourceBaseline{}, false, nil
+	}
+	var latestTotal sql.NullFloat64
+	var latestEvent sql.NullTime
+	if err := rows.Scan(&latestTotal, &latestEvent); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	if err := rows.Err(); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	return sourceBaseline{
+		LatestTotalKM:   latestTotal.Float64,
+		LatestEventTime: latestEvent.Time,
+	}, latestTotal.Valid, nil
+}
+
+func lookupCurrentSourceBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (sourceBaseline, bool, error) {
+	if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(sourceKey) == "" {
+		return sourceBaseline{}, false, nil
+	}
+	rows, err := query.QueryContext(ctx, currentSourceBaselineSQL, vin, statDate, string(protocol), sourceKey)
+	if err != nil {
+		return sourceBaseline{}, false, err
+	}
+	defer rows.Close()
+	if !rows.Next() {
+		if err := rows.Err(); err != nil {
+			return sourceBaseline{}, false, err
+		}
+		return sourceBaseline{}, false, nil
+	}
+	var firstTotal sql.NullFloat64
+	var firstEvent sql.NullTime
+	if err := rows.Scan(&firstTotal, &firstEvent); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	if err := rows.Err(); err != nil {
+		return sourceBaseline{}, false, err
+	}
+	return sourceBaseline{
+		LatestTotalKM:   firstTotal.Float64,
+		LatestEventTime: firstEvent.Time,
+	}, firstTotal.Valid, nil
+}
+
+func previousStatDate(statDate string) (string, bool) {
+	day, err := time.Parse("2006-01-02", strings.TrimSpace(statDate))
+	if err != nil {
+		return "", false
+	}
+	return day.AddDate(0, 0, -1).Format("2006-01-02"), true
+}
+
+const previousSourceBaselineSQL = `
+SELECT latest_total_mileage_km, latest_event_time
+FROM vehicle_daily_mileage_source
+WHERE vin = ?
+  AND stat_date = ?
+  AND protocol = ?
+  AND source_key = ?
+ORDER BY latest_event_time DESC
+LIMIT 1
+`
+
+const currentSourceBaselineSQL = `
+SELECT first_total_mileage_km, first_event_time
+FROM vehicle_daily_mileage_source
+WHERE vin = ?
+  AND stat_date = ?
+  AND protocol = ?
+  AND source_key = ?
+  AND quality_status = '` + QualityOK + `'
+ORDER BY latest_event_time DESC
+LIMIT 1
+`

go/vehicle-gateway/internal/stats/source_mileage_test.go

@@ -0,0 +1,159 @@
+package stats
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+func TestSourceKeyUsesProtocolDeviceAndSourceIP(t *testing.T) {
+	got := SourceKey(envelope.ProtocolJT808, "13307765812", "", "115.231.168.135")
+	if got != "JT808:13307765812@115.231.168.135" {
+		t.Fatalf("source key = %q", got)
+	}
+
+	got = SourceKey(envelope.ProtocolGB32960, "", "iccid-1", "202.98.117.132")
+	if got != "GB32960:iccid-1@202.98.117.132" {
+		t.Fatalf("source key = %q", got)
+	}
+}
+
+func TestUpsertSourceMileageWritesCandidateRow(t *testing.T) {
+	exec := &recordingExec{}
+	sample := SourceMileageSample{
+		VIN:             "LA9GG64L7PBAF4001",
+		StatDate:        "2026-07-08",
+		Protocol:        envelope.ProtocolJT808,
+		SourceKey:       "JT808:13307765812@115.231.168.135",
+		SourceIP:        "115.231.168.135",
+		SourceEndpoint:  "115.231.168.135:20215",
+		Phone:           "13307765812",
+		FirstTotalKM:    4100.8,
+		LatestTotalKM:   4123.9,
+		DailyKM:         23.1,
+		SampleCount:     2,
+		FirstEventTime:  time.Date(2026, 7, 8, 0, 1, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+		LatestEventTime: time.Date(2026, 7, 8, 23, 59, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+		QualityStatus:   QualityOK,
+		QualityReason:   "same_source_continuous",
+	}
+	if err := UpsertSourceMileage(context.Background(), exec, sample); err != nil {
+		t.Fatalf("UpsertSourceMileage() error = %v", err)
+	}
+	if len(exec.calls) != 1 {
+		t.Fatalf("exec calls = %d", len(exec.calls))
+	}
+	sql := exec.calls[0].query
+	for _, want := range []string{
+		"INSERT INTO vehicle_daily_mileage_source",
+		"ON DUPLICATE KEY UPDATE",
+		"daily_mileage_km = GREATEST(",
+		"quality_status = VALUES(quality_status)",
+		"platform_name = COALESCE(NULLIF(TRIM(VALUES(platform_name)), ''), platform_name)",
+	} {
+		if !strings.Contains(sql, want) {
+			t.Fatalf("candidate upsert missing %q: %s", want, sql)
+		}
+	}
+	if got := exec.calls[0].args[0]; got != "LA9GG64L7PBAF4001" {
+		t.Fatalf("first arg = %#v", got)
+	}
+}
+
+func TestUpsertSourceMileageSkipsBlankSourceIP(t *testing.T) {
+	exec := &recordingExec{}
+	sample := SourceMileageSample{
+		VIN:           "LA9GG64L7PBAF4001",
+		StatDate:      "2026-07-08",
+		Protocol:      envelope.ProtocolJT808,
+		SourceKey:     "JT808:13307765812@",
+		SourceIP:      "   ",
+		Phone:         "13307765812",
+		QualityStatus: QualityOK,
+	}
+	if err := UpsertSourceMileage(context.Background(), exec, sample); err != nil {
+		t.Fatalf("UpsertSourceMileage() error = %v", err)
+	}
+	if len(exec.calls) != 0 {
+		t.Fatalf("exec calls = %d, want 0", len(exec.calls))
+	}
+}
+
+func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) {
+	exec := &recordingExec{}
+	err := ProjectDailyMileage(context.Background(), exec, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808)
+	if err != nil {
+		t.Fatalf("ProjectDailyMileage() error = %v", err)
+	}
+	if len(exec.calls) != 4 {
+		t.Fatalf("exec calls = %d, want 4", len(exec.calls))
+	}
+	updateSources := exec.calls[0].query
+	projectFinal := exec.calls[1].query
+	markSelected := exec.calls[2].query
+	cleanupFinal := exec.calls[3].query
+	if !strings.Contains(updateSources, "UPDATE vehicle_daily_mileage_source") || !strings.Contains(updateSources, "is_selected = 0") {
+		t.Fatalf("first query should clear selected candidates: %s", updateSources)
+	}
+	for _, want := range []string{
+		"INSERT INTO vehicle_daily_mileage",
+		"FROM vehicle_daily_mileage_source s",
+		"LEFT JOIN vehicle_data_source ds",
+		"ORDER BY COALESCE(ds.trust_priority, 100)",
+		"s.quality_status = '" + QualityOK + "'",
+		"s.daily_mileage_km BETWEEN 0 AND",
+		"ds.latest_source_endpoint",
+	} {
+		if !strings.Contains(projectFinal, want) {
+			t.Fatalf("project query missing %q: %s", want, projectFinal)
+		}
+	}
+	if strings.Contains(projectFinal, "COALESCE(NULLIF(ds.latest_source_endpoint, ''), s.source_endpoint)") {
+		t.Fatalf("project query should not fallback to source endpoint: %s", projectFinal)
+	}
+	if strings.Contains(projectFinal, ", s.source_endpoint") {
+		t.Fatalf("project query should only project latest_source_endpoint: %s", projectFinal)
+	}
+	if !strings.Contains(markSelected, "UPDATE vehicle_daily_mileage_source s") || !strings.Contains(markSelected, "SET s.is_selected = 1") {
+		t.Fatalf("mark query should flag the elected source: %s", markSelected)
+	}
+	if strings.Contains(markSelected, "JOIN vehicle_daily_mileage m") {
+		t.Fatalf("mark query should not rejoin final mileage for candidate selection: %s", markSelected)
+	}
+	for _, want := range []string{
+		"JOIN (",
+		"SELECT",
+		"s2.source_key",
+		"s2.quality_status = '" + QualityOK + "'",
+		"s2.daily_mileage_km BETWEEN 0 AND",
+		"COALESCE(ds.enabled, 1) = 1",
+		"ORDER BY COALESCE(ds.trust_priority, 100)",
+		"LIMIT 1",
+	} {
+		if !strings.Contains(markSelected, want) {
+			t.Fatalf("mark query missing %q: %s", want, markSelected)
+		}
+	}
+	if len(exec.calls[2].args) != 7 {
+		t.Fatalf("mark query args = %d, want 7", len(exec.calls[2].args))
+	}
+	if got := exec.calls[2].args[3]; got != maxSelectedDailyMileageKM {
+		t.Fatalf("mark query max mileage arg = %#v, want %d", got, maxSelectedDailyMileageKM)
+	}
+	for _, want := range []string{
+		"DELETE FROM vehicle_daily_mileage",
+		"NOT EXISTS (",
+		"FROM vehicle_daily_mileage_source",
+		"is_selected = 1",
+	} {
+		if !strings.Contains(cleanupFinal, want) {
+			t.Fatalf("cleanup query missing %q: %s", want, cleanupFinal)
+		}
+	}
+	if len(exec.calls[3].args) != 6 {
+		t.Fatalf("cleanup query args = %d, want 6", len(exec.calls[3].args))
+	}
+}

go/vehicle-gateway/internal/stats/source_test.go

@@ -0,0 +1,79 @@
+package stats
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+func TestNormalizeSourceIPDropsPort(t *testing.T) {
+	tests := map[string]string{
+		"115.231.168.135:20215":  "115.231.168.135",
+		"115.231.168.135":        "115.231.168.135",
+		" 115.159.85.149:28316 ": "115.159.85.149",
+		"":                       "",
+	}
+	for input, want := range tests {
+		if got := NormalizeSourceIP(input); got != want {
+			t.Fatalf("NormalizeSourceIP(%q) = %q, want %q", input, got, want)
+		}
+	}
+}
+
+func TestNewSourceIdentityRequiresSourceIP(t *testing.T) {
+	identity, ok := NewSourceIdentity(envelope.ProtocolJT808, "115.231.168.135:20215")
+	if !ok {
+		t.Fatal("NewSourceIdentity() ok = false")
+	}
+	if identity.Protocol != envelope.ProtocolJT808 {
+		t.Fatalf("protocol = %q", identity.Protocol)
+	}
+	if identity.SourceIP != "115.231.168.135" {
+		t.Fatalf("source ip = %q", identity.SourceIP)
+	}
+	if identity.SourceEndpoint != "115.231.168.135:20215" {
+		t.Fatalf("endpoint = %q", identity.SourceEndpoint)
+	}
+
+	if _, ok := NewSourceIdentity(envelope.ProtocolJT808, ""); ok {
+		t.Fatal("empty endpoint should not produce identity")
+	}
+}
+
+func TestUpsertDataSourcePreservesManualFields(t *testing.T) {
+	exec := &recordingExec{}
+	identity := SourceIdentity{
+		Protocol:       envelope.ProtocolJT808,
+		SourceIP:       "115.231.168.135",
+		SourceEndpoint: "115.231.168.135:20215",
+	}
+	if err := UpsertDataSource(context.Background(), exec, identity, time.Date(2026, 7, 8, 13, 0, 0, 0, time.UTC)); err != nil {
+		t.Fatalf("UpsertDataSource() error = %v", err)
+	}
+	if len(exec.calls) != 1 {
+		t.Fatalf("exec calls = %d", len(exec.calls))
+	}
+	sql := exec.calls[0].query
+	for _, want := range []string{
+		"INSERT INTO vehicle_data_source",
+		"latest_source_endpoint = VALUES(latest_source_endpoint)",
+		"latest_seen_at = VALUES(latest_seen_at)",
+	} {
+		if !strings.Contains(sql, want) {
+			t.Fatalf("source upsert missing %q: %s", want, sql)
+		}
+	}
+	for _, forbidden := range []string{
+		"platform_name = VALUES(platform_name)",
+		"trust_priority = VALUES(trust_priority)",
+		"enabled = VALUES(enabled)",
+		"remark = VALUES(remark)",
+	} {
+		if strings.Contains(sql, forbidden) {
+			t.Fatalf("source upsert should preserve manual field %q: %s", forbidden, sql)
+		}
+	}
+}