From fe770132c79fcc693f4fe032bd32da8a85d12ff7 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 14:12:40 +0800 Subject: [PATCH] docs(stats): plan multi-source mileage implementation --- ...lti-source-mileage-stats-implementation.md | 1312 +++++++++++++++++ 1 file changed, 1312 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md diff --git a/docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md b/docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md new file mode 100644 index 00000000..aee1d4b0 --- /dev/null +++ b/docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md @@ -0,0 +1,1312 @@ +# Multi-Source Mileage Statistics Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Store every mileage source candidate, maintain source platform metadata, and project one elected daily mileage result for default BI/API queries. + +**Architecture:** Add a source metadata table and a per-source daily mileage candidate table in MySQL. Real-time stat writer and historical backfill write candidates first, then run an election projector that updates the existing `vehicle_daily_mileage` final table. Default reads stay on `vehicle_daily_mileage`; source audit reads can be added after the write path is stable. + +**Tech Stack:** Go 1.26, MySQL via `github.com/go-sql-driver/mysql`, TDengine via `github.com/taosdata/driver-go/v3/taosWS`, Kafka via `github.com/segmentio/kafka-go`, existing `internal/stats` package. + +## Global Constraints + +- Runtime flow must not add non-protocol synthetic fields to `vehicle.fields.go.*`. +- Source identity uses `protocol + source_ip`; source endpoint port is latest connection metadata only. +- Platform name, trust priority, enabled flag, and remark are operator-maintained and must not be overwritten by runtime upserts. +- Candidate statistics use `vin + stat_date + protocol + source_key` and never mix sources. +- Final statistics use `vin + stat_date + protocol` and are projected from one selected candidate. +- The projector must not calculate `B today - A yesterday`. +- A source with no same-source previous-day baseline remains a candidate and is not selected by default. +- Default API queries continue to read `vehicle_daily_mileage`. + +--- + +## File Structure + +- Modify `go/vehicle-gateway/internal/stats/schema.go` + - Owns MySQL schema strings for source metadata, source candidates, final table, and compatibility alters. +- Create `go/vehicle-gateway/internal/stats/source.go` + - Owns source endpoint normalization, `SourceIdentity`, and source metadata upsert. +- Create `go/vehicle-gateway/internal/stats/source_test.go` + - Tests source IP extraction and runtime upsert preserving manual fields. +- Create `go/vehicle-gateway/internal/stats/source_mileage.go` + - Owns `SourceMileageSample`, candidate upsert, final election projection, and shared constants. +- Create `go/vehicle-gateway/internal/stats/source_mileage_test.go` + - Tests candidate writes and election rules. +- Modify `go/vehicle-gateway/internal/stats/daily_metric.go` + - Keep protocol mileage extraction, route `Writer.Append` through source metadata, candidate upsert, and election projection. +- Modify `go/vehicle-gateway/internal/stats/daily_metric_test.go` + - Update expectations for source candidate writes and final projection. +- Modify `go/vehicle-gateway/internal/stats/query.go` + - Keep default response compatible; optionally include selected source columns in existing rows if already present. +- Modify `go/vehicle-gateway/cmd/stats-backfill/main.go` + - Rebuild source candidates by day/source, then project final rows. +- Modify `go/vehicle-gateway/cmd/stats-backfill/main_test.go` + - Update trusted-source tests to assert source candidates and final selection. +- Deploy binaries: + - `/opt/lingniu-go-native/current/stat-writer` + - `/opt/lingniu-go-native/current/stats-backfill` + +--- + +### Task 1: Source Metadata Schema And Identity Helpers + +**Files:** +- Modify: `go/vehicle-gateway/internal/stats/schema.go` +- Create: `go/vehicle-gateway/internal/stats/source.go` +- Create: `go/vehicle-gateway/internal/stats/source_test.go` + +**Interfaces:** +- Produces: `type SourceIdentity struct { Protocol envelope.Protocol; SourceIP string; SourceEndpoint string }` +- Produces: `func NewSourceIdentity(protocol envelope.Protocol, endpoint string) (SourceIdentity, bool)` +- Produces: `func NormalizeSourceIP(endpoint string) string` +- Produces: `func UpsertDataSource(ctx context.Context, exec Execer, identity SourceIdentity, now time.Time) error` +- Consumes: `stats.Execer` + +- [ ] **Step 1: Write failing source identity tests** + +Add to `go/vehicle-gateway/internal/stats/source_test.go`: + +```go +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) + } + } +} +``` + +- [ ] **Step 2: Run source tests and verify failure** + +Run: + +```bash +cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway +go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1 +``` + +Expected: FAIL because `NormalizeSourceIP`, `NewSourceIdentity`, and `UpsertDataSource` are undefined. + +- [ ] **Step 3: Add source schema** + +Add to `go/vehicle-gateway/internal/stats/schema.go`: + +```go +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) +)` +``` + +Update `Writer.EnsureSchema` in Task 3 to execute this statement. + +- [ ] **Step 4: Implement source helpers** + +Create `go/vehicle-gateway/internal/stats/source.go`: + +```go +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 +` +``` + +- [ ] **Step 5: Run source tests and verify pass** + +Run: + +```bash +go test ./internal/stats -run 'TestNormalizeSourceIP|TestNewSourceIdentity|TestUpsertDataSource' -count=1 +``` + +Expected: PASS. + +- [ ] **Step 6: Commit source metadata foundation** + +Run: + +```bash +git add go/vehicle-gateway/internal/stats/schema.go go/vehicle-gateway/internal/stats/source.go go/vehicle-gateway/internal/stats/source_test.go +git commit -m "feat(stats): add vehicle data source metadata" +``` + +--- + +### Task 2: Candidate Mileage Schema And Writer + +**Files:** +- Modify: `go/vehicle-gateway/internal/stats/schema.go` +- Create: `go/vehicle-gateway/internal/stats/source_mileage.go` +- Create: `go/vehicle-gateway/internal/stats/source_mileage_test.go` + +**Interfaces:** +- Consumes: `MetricSample` +- Produces: `type SourceMileageSample struct` +- Produces: `func SourceMileageSampleFromMetric(sample MetricSample, identity SourceIdentity) SourceMileageSample` +- Produces: `func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageSample) error` +- Produces: `func SourceKey(protocol envelope.Protocol, phone string, deviceID string, sourceIP string) string` + +- [ ] **Step 1: Write failing candidate tests** + +Add to `go/vehicle-gateway/internal/stats/source_mileage_test.go`: + +```go +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 = VALUES(daily_mileage_km)", + "quality_status = VALUES(quality_status)", + } { + 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) + } +} +``` + +- [ ] **Step 2: Run candidate tests and verify failure** + +Run: + +```bash +go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1 +``` + +Expected: FAIL because candidate interfaces are undefined. + +- [ ] **Step 3: Add candidate schema** + +Add to `go/vehicle-gateway/internal/stats/schema.go`: + +```go +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) +)` +``` + +- [ ] **Step 4: Implement candidate writer** + +Create `go/vehicle-gateway/internal/stats/source_mileage.go`: + +```go +package stats + +import ( + "context" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +const ( + QualityOK = "OK" + QualityNoPreviousBaseline = "NO_PREVIOUS_BASELINE" + QualityInvalidDelta = "INVALID_DELTA" +) + +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 sample.VIN == "" || sample.SourceKey == "" { + 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 +} + +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(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 +` +``` + +- [ ] **Step 5: Extend MetricSample with event and device fields** + +Modify `MetricSample` in `go/vehicle-gateway/internal/stats/daily_metric.go`: + +```go +type MetricSample struct { + VIN string + Protocol envelope.Protocol + StatDate string + TotalMileageKM float64 + EventTime time.Time + SourceKey string + Phone string + DeviceID string + SourceEndpoint string +} +``` + +Populate `EventTime` and `DeviceID` in `SamplesFromEnvelope`. + +- [ ] **Step 6: Run candidate tests and verify pass** + +Run: + +```bash +go test ./internal/stats -run 'TestSourceKey|TestUpsertSourceMileage' -count=1 +``` + +Expected: PASS. + +- [ ] **Step 7: Commit candidate writer** + +Run: + +```bash +git add go/vehicle-gateway/internal/stats/schema.go go/vehicle-gateway/internal/stats/source_mileage.go go/vehicle-gateway/internal/stats/source_mileage_test.go go/vehicle-gateway/internal/stats/daily_metric.go +git commit -m "feat(stats): store source mileage candidates" +``` + +--- + +### Task 3: Final Mileage Election Projector + +**Files:** +- Modify: `go/vehicle-gateway/internal/stats/source_mileage.go` +- Modify: `go/vehicle-gateway/internal/stats/source_mileage_test.go` +- Modify: `go/vehicle-gateway/internal/stats/schema.go` + +**Interfaces:** +- Produces: `func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error` +- Produces: `const maxSelectedDailyMileageKM = 1000` +- Consumes: `vehicle_daily_mileage_source` +- Produces: selected row in `vehicle_daily_mileage` + +- [ ] **Step 1: Write failing projector SQL tests** + +Add to `go/vehicle-gateway/internal/stats/source_mileage_test.go`: + +```go +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) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + } + updateSources := exec.calls[0].query + projectFinal := exec.calls[1].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 = 'OK'", + "s.daily_mileage_km BETWEEN 0 AND", + } { + if !strings.Contains(projectFinal, want) { + t.Fatalf("project query missing %q: %s", want, projectFinal) + } + } +} +``` + +- [ ] **Step 2: Run projector test and verify failure** + +Run: + +```bash +go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1 +``` + +Expected: FAIL because `ProjectDailyMileage` is undefined. + +- [ ] **Step 3: Implement projector** + +Add to `go/vehicle-gateway/internal/stats/source_mileage.go`: + +```go +const maxSelectedDailyMileageKM = 1000 + +func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error { + if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" { + return nil + } + if _, err := exec.ExecContext(ctx, clearSelectedSourceSQL, vin, statDate, string(protocol)); err != nil { + return err + } + _, err := exec.ExecContext(ctx, projectDailyMileageSQL, + vin, + statDate, + string(protocol), + maxSelectedDailyMileageKM, + vin, + statDate, + string(protocol), + ) + return err +} + +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, + s.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 = 'OK' + 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 +` +``` + +Then add a third exec in `ProjectDailyMileage` to mark the selected candidate: + +```go +const markSelectedSourceSQL = ` +UPDATE vehicle_daily_mileage_source s +JOIN vehicle_daily_mileage m + ON m.vin = s.vin + AND m.stat_date = s.stat_date + AND m.protocol = s.protocol + AND m.trusted_source_key = s.source_key +SET s.is_selected = 1 +WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ? +` +``` + +Update the test to expect three calls and assert `markSelectedSourceSQL` contains `SET s.is_selected = 1`. + +- [ ] **Step 4: Run projector tests and verify pass** + +Run: + +```bash +go test ./internal/stats -run TestProjectDailyMileageSelectsCandidateAndMarksSource -count=1 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit projector** + +Run: + +```bash +git add go/vehicle-gateway/internal/stats/source_mileage.go go/vehicle-gateway/internal/stats/source_mileage_test.go +git commit -m "feat(stats): project elected mileage source" +``` + +--- + +### Task 4: Rewire Real-Time Stat Writer To Candidates + +**Files:** +- Modify: `go/vehicle-gateway/internal/stats/daily_metric.go` +- Modify: `go/vehicle-gateway/internal/stats/daily_metric_test.go` + +**Interfaces:** +- Consumes: `UpsertDataSource` +- Consumes: `UpsertSourceMileage` +- Consumes: `ProjectDailyMileage` +- Produces: `Writer.Append` writes source metadata, candidate row, and final projection. + +- [ ] **Step 1: Write failing Writer.Append flow test** + +Add to `go/vehicle-gateway/internal/stats/daily_metric_test.go`: + +```go +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) != 5 { + t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark", 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) + } +} +``` + +- [ ] **Step 2: Run flow test and verify failure** + +Run: + +```bash +go test ./internal/stats -run TestWriterAppendWritesSourceCandidateAndProjection -count=1 +``` + +Expected: FAIL because `Writer.Append` still calls the old direct final upsert. + +- [ ] **Step 3: Rewrite Writer.Append** + +Modify `Writer.Append` in `go/vehicle-gateway/internal/stats/daily_metric.go`: + +```go +func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { + samples, err := SamplesFromEnvelope(env, w.loc) + if err != nil { + return err + } + identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint) + for _, sample := range samples { + if w.seenSameMileage(sample) { + continue + } + if hasSource { + if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil { + return err + } + candidate := SourceMileageSampleFromMetric(sample, identity) + 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) + } + return nil +} +``` + +Keep `upsertDailyMileageSQL` only if older tests still reference it; remove it after tests prove no code path uses it. + +- [ ] **Step 4: Update schema bootstrap** + +Modify `EnsureSchema` in `go/vehicle-gateway/internal/stats/daily_metric.go`: + +```go +func (w *Writer) EnsureSchema(ctx context.Context) error { + 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 + } + } + return nil +} +``` + +- [ ] **Step 5: Run stats tests** + +Run: + +```bash +go test ./internal/stats -count=1 +``` + +Expected: PASS. + +- [ ] **Step 6: Commit realtime writer rewire** + +Run: + +```bash +git add go/vehicle-gateway/internal/stats/daily_metric.go go/vehicle-gateway/internal/stats/daily_metric_test.go +git commit -m "feat(stats): write realtime mileage candidates" +``` + +--- + +### Task 5: Backfill Candidate Rows And Final Projection + +**Files:** +- Modify: `go/vehicle-gateway/cmd/stats-backfill/main.go` +- Modify: `go/vehicle-gateway/cmd/stats-backfill/main_test.go` + +**Interfaces:** +- Consumes: `UpsertDataSource` +- Consumes: `UpsertSourceMileage` +- Consumes: `ProjectDailyMileage` +- Produces: backfill can rebuild source candidates and final result for a date/protocol range. + +- [ ] **Step 1: Write failing backfill election test** + +Modify `go/vehicle-gateway/cmd/stats-backfill/main_test.go` to keep the existing `TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump` and add: + +```go +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) + } +} +``` + +- [ ] **Step 2: Run backfill tests and verify failure** + +Run: + +```bash +go test ./cmd/stats-backfill -run 'TestChooseTrustedSource|TestDailySourceLastBuildsCandidateKeysBySourceIP' -count=1 +``` + +Expected: FAIL until `normalizedSourceKey` accepts protocol and device fields. + +- [ ] **Step 3: Update source key normalization in backfill** + +Change `normalizedSourceKey` in `go/vehicle-gateway/cmd/stats-backfill/main.go`: + +```go +func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string { + sourceIP := stats.NormalizeSourceIP(endpoint) + return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP) +} +``` + +Update all calls: + +```go +sourceKey := normalizedSourceKey(string(protocol), phone, "", sourceEndpoint) +``` + +- [ ] **Step 4: Change backfill write path to candidates** + +Replace final-only `writeAggregates` with candidate write and projection: + +```go +func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*metricAgg, batchSize int) (int64, error) { + var written int64 + for _, agg := range aggregates { + identity := stats.SourceIdentity{ + Protocol: agg.Protocol, + SourceIP: stats.NormalizeSourceIP(agg.SourceEndpoint), + SourceEndpoint: agg.SourceEndpoint, + } + if err := stats.UpsertDataSource(ctx, db, identity, agg.LatestEventTime); err != nil { + return written, err + } + 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: agg.LatestKM - agg.FirstKM, + SampleCount: agg.Count, + FirstEventTime: agg.FirstEventTime, + LatestEventTime: agg.LatestEventTime, + QualityStatus: stats.QualityOK, + QualityReason: "same_source_previous_day", + } + if candidate.DailyKM < 0 || candidate.DailyKM > 1000 { + 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 +} +``` + +Extend `metricAgg` with `DeviceID`, `FirstEventTime`, and `LatestEventTime`. + +- [ ] **Step 5: Preserve every source candidate** + +Update `buildLastDiffAggregates` so it writes one aggregate per source candidate when same-source previous baseline exists, not just one chosen source. + +Use this key: + +```go +key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey +``` + +When a current source has no previous source, create a candidate with: + +```go +QualityStatus: stats.QualityNoPreviousBaseline +QualityReason: "missing_previous_source" +FirstKM: currentRow.TotalKM +LatestKM: currentRow.TotalKM +DailyKM: 0 +``` + +Do not project invalid candidates; projection SQL already filters `quality_status = 'OK'`. + +- [ ] **Step 6: Run backfill tests** + +Run: + +```bash +go test ./cmd/stats-backfill ./internal/stats -count=1 +``` + +Expected: PASS. + +- [ ] **Step 7: Commit backfill candidate support** + +Run: + +```bash +git add go/vehicle-gateway/cmd/stats-backfill/main.go go/vehicle-gateway/cmd/stats-backfill/main_test.go +git commit -m "feat(stats): backfill mileage source candidates" +``` + +--- + +### Task 6: API Query Compatibility And Candidate Audit Hook + +**Files:** +- Modify: `go/vehicle-gateway/internal/stats/query.go` +- Modify: `go/vehicle-gateway/internal/stats/query_test.go` + +**Interfaces:** +- Produces: existing `/api/stats/daily-metrics` response remains backward-compatible. +- Produces: selected source metadata is visible in final rows when present. + +- [ ] **Step 1: Write failing query test for selected source fields** + +Add to `go/vehicle-gateway/internal/stats/query_test.go`: + +```go +func TestMetricQueryReturnsSelectedSourceFields(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + rows := 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", + ) + mock.ExpectQuery("SELECT vin, stat_date, protocol").WillReturnRows(rows) + + 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) + } +} +``` + +- [ ] **Step 2: Run query test and verify failure** + +Run: + +```bash +go test ./internal/stats -run TestMetricQueryReturnsSelectedSourceFields -count=1 +``` + +Expected: FAIL because `MetricRow` does not expose trusted source fields. + +- [ ] **Step 3: Extend MetricRow and SQL** + +Modify `MetricRow` in `go/vehicle-gateway/internal/stats/query.go`: + +```go +TrustedSourceKey *string `json:"trusted_source_key,omitempty"` +TrustedPhone *string `json:"trusted_phone,omitempty"` +TrustedSourceEndpoint *string `json:"trusted_source_endpoint,omitempty"` +``` + +Modify selected columns: + +```go +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 +``` + +Scan into `sql.NullString` values and set pointers when valid. + +- [ ] **Step 4: Run query tests** + +Run: + +```bash +go test ./internal/stats -run 'TestMetricQuery|TestMetricHandler' -count=1 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit query compatibility** + +Run: + +```bash +git add go/vehicle-gateway/internal/stats/query.go go/vehicle-gateway/internal/stats/query_test.go +git commit -m "feat(stats): expose selected mileage source" +``` + +--- + +### Task 7: Deploy, Rebuild, And Verify Production Data + +**Files:** +- Uses built binaries from `go/vehicle-gateway/cmd/stat-writer` +- Uses built binaries from `go/vehicle-gateway/cmd/stats-backfill` + +**Interfaces:** +- Consumes: ECS env files at `/opt/lingniu-go-native/env/stat-writer.env` and `/opt/lingniu-go-native/env/history-writer.env` +- Produces: source tables and final table populated on ECS. + +- [ ] **Step 1: Run full test suite locally** + +Run: + +```bash +cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway +go test ./... -count=1 +``` + +Expected: PASS for all packages. + +- [ ] **Step 2: Build Linux binaries** + +Run: + +```bash +mkdir -p ../../outputs/multi_source_stats_20260708/bin +GOOS=linux GOARCH=amd64 go build -trimpath -ldflags='-s -w' -o ../../outputs/multi_source_stats_20260708/bin/stat-writer ./cmd/stat-writer +GOOS=linux GOARCH=amd64 go build -trimpath -ldflags='-s -w' -o ../../outputs/multi_source_stats_20260708/bin/stats-backfill ./cmd/stats-backfill +``` + +Expected: both binaries exist in `../../outputs/multi_source_stats_20260708/bin`. + +- [ ] **Step 3: Upload and restart stat writer** + +Run: + +```bash +scp -q ../../outputs/multi_source_stats_20260708/bin/stat-writer root@115.29.187.205:/tmp/stat-writer.new +scp -q ../../outputs/multi_source_stats_20260708/bin/stats-backfill root@115.29.187.205:/tmp/stats-backfill.new +ssh root@115.29.187.205 'set -e +stamp=$(date +%Y%m%d%H%M%S) +cp /opt/lingniu-go-native/current/stat-writer /opt/lingniu-go-native/current/stat-writer.bak.$stamp +cp /opt/lingniu-go-native/current/stats-backfill /opt/lingniu-go-native/current/stats-backfill.bak.$stamp || true +install -m 755 /tmp/stat-writer.new /opt/lingniu-go-native/current/stat-writer +install -m 755 /tmp/stats-backfill.new /opt/lingniu-go-native/current/stats-backfill +rm -f /tmp/stat-writer.new /tmp/stats-backfill.new +systemctl restart lingniu-go-stat-writer.service +systemctl is-active lingniu-go-stat-writer.service +journalctl -u lingniu-go-stat-writer.service -n 20 --no-pager' +``` + +Expected: service status is `active`. + +- [ ] **Step 4: Rebuild 2026-07-08 JT808** + +Run: + +```bash +ssh root@115.29.187.205 'BACKFILL_ENV_FILES=/opt/lingniu-go-native/env/stat-writer.env,/opt/lingniu-go-native/env/history-writer.env BACKFILL_METHOD=last_diff BACKFILL_DATE_FROM=2026-07-08 BACKFILL_DATE_TO=2026-07-08 BACKFILL_PROTOCOLS=JT808 BACKFILL_DRY_RUN=false BACKFILL_RESET=true /opt/lingniu-go-native/current/stats-backfill' +``` + +Expected output includes `stats backfill complete` and non-zero candidate/final counts. + +- [ ] **Step 5: Verify anomaly VIN** + +Run: + +```bash +curl -fsS 'http://115.29.187.205:20200/api/stats/daily-metrics?vin=LA9GG64L7PBAF4001&protocol=JT808&dateFrom=2026-07-08&dateTo=2026-07-08&limit=10&includeTotal=true' +``` + +Expected JSON contains: + +```json +{ + "vin": "LA9GG64L7PBAF4001", + "daily_mileage_km": 23.1, + "first_total_mileage_km": 4100.8, + "latest_total_mileage_km": 4123.9 +} +``` + +- [ ] **Step 6: Verify candidates exist** + +Run a temporary MySQL query binary or existing inspection helper to execute: + +```sql +SELECT vin, stat_date, protocol, source_key, source_ip, daily_mileage_km, quality_status, is_selected +FROM vehicle_daily_mileage_source +WHERE vin = 'LA9GG64L7PBAF4001' + AND stat_date = '2026-07-08' + AND protocol = 'JT808' +ORDER BY source_key; +``` + +Expected: + +- At least one selected source row with `is_selected = 1`. +- At least one non-selected row when alternate 424xx source data exists. +- Non-selected invalid rows remain available for audit. + +- [ ] **Step 7: Verify stat writer health** + +Run: + +```bash +ssh root@115.29.187.205 "curl -fsS http://127.0.0.1:20213/metrics | grep -E 'vehicle_stat_(kafka_lag|writes)' | tail -40" +``` + +Expected: + +- `vehicle_stat_kafka_lag` is `0` for GB32960, JT808, and Yutong MQTT fields topics. +- `vehicle_stat_writes_total{status="ok"}` is increasing. + +- [ ] **Step 8: Commit deployment-ready changes** + +Run: + +```bash +git status --short +git add go/vehicle-gateway/internal/stats go/vehicle-gateway/cmd/stats-backfill +git commit -m "feat(stats): persist multi-source mileage candidates" +git push origin go +``` + +Expected: branch `go` is pushed successfully. + +--- + +## Self-Review + +- Spec coverage: source metadata, source candidates, final projection, source election, API compatibility, backfill, and production verification are each covered by at least one task. +- Placeholder scan: this plan contains no placeholder markers or open-ended implementation instructions. +- Type consistency: source identity flows from `SourceIdentity` to `SourceMileageSample`; final projection always consumes candidate rows and updates `vehicle_daily_mileage`.