shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
843690cc7185cc1f218e3b21d89b644cd326b132
lingniu-vehicle-ingest/docs/superpowers/plans/2026-07-08-multi-source-mileage-stats-implementation.md

1313 lines
40 KiB
Markdown
Raw Blame History

# 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`.
Reference in New Issue View Git Blame Copy Permalink