shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(stats): write realtime mileage candidates

Browse Source
This commit is contained in:
lingniu
2026-07-08 14:48:51 +08:00
parent 67aa521dd9
commit 762c7265d7
3 changed files with 80 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
.superpowers/sdd/task-4-report.md Normal file
View File

@@ -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`.

106
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -69,19 +69,21 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error {
if err != nil { if err != nil {
return err return err
} }
identity, hasSource := NewSourceIdentity(env.Protocol, env.SourceEndpoint)
for _, sample := range samples { for _, sample := range samples {
if w.seenSameMileage(sample) { if w.seenSameMileage(sample) {
continue continue
} }
if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL, if hasSource {
sample.VIN, if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil {
sample.StatDate, return err
string(sample.Protocol), }
sample.TotalMileageKM, candidate := SourceMileageSampleFromMetric(sample, identity)
sample.TotalMileageKM, if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil {
sample.SourceKey, return err
sample.Phone, }
sample.SourceEndpoint); err != nil { }
if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil {
return err return err
} }
w.markMileageWritten(sample) w.markMileageWritten(sample)
@@ -196,92 +198,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) { func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
if env.Fields == nil { if env.Fields == nil {
return 0, false return 0, false

83
go/vehicle-gateway/internal/stats/daily_metric_test.go
View File

@@ -156,6 +156,37 @@ 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) != 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)
}
}
func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
exec := &recordingExec{} exec := &recordingExec{}
writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600))
@@ -163,9 +194,11 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("EnsureSchema() error = %v", err) t.Fatalf("EnsureSchema() error = %v", err)
} }
if err := writer.Append(context.Background(), envelope.FrameEnvelope{ if err := writer.Append(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolGB32960, Protocol: envelope.ProtocolGB32960,
VIN: "LNBVIN00000000002", VIN: "LNBVIN00000000002",
EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), 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{ Fields: map[string]any{
envelope.FieldTotalMileageKM: "10000.0", envelope.FieldTotalMileageKM: "10000.0",
}, },
@@ -173,7 +206,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
t.Fatalf("Append() error = %v", err) t.Fatalf("Append() error = %v", err)
} }
if len(exec.calls) != 7 { if len(exec.calls) != 11 {
t.Fatalf("exec calls = %d", len(exec.calls)) t.Fatalf("exec calls = %d", len(exec.calls))
} }
for i, want := range []string{ for i, want := range []string{
@@ -196,22 +229,32 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) {
if strings.Contains(exec.calls[2].query, "KEY idx_vin (vin)") { 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) t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[2].query)
} }
upsertCall := exec.calls[6] projectCall := exec.calls[9]
if !strings.Contains(upsertCall.query, "ON DUPLICATE KEY UPDATE") { if !strings.Contains(projectCall.query, "INSERT INTO vehicle_daily_mileage") {
t.Fatalf("unexpected upsert sql: %s", upsertCall.query) t.Fatalf("unexpected project sql: %s", projectCall.query)
} }
if strings.Contains(upsertCall.query, "metric_key") || strings.Contains(upsertCall.query, "metric_unit") { if !strings.Contains(projectCall.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("daily mileage upsert should not use generic metric columns: %s", upsertCall.query) t.Fatalf("unexpected upsert sql: %s", projectCall.query)
} }
if !strings.Contains(upsertCall.query, "first_total_mileage_km <= 0") { if strings.Contains(projectCall.query, "metric_key") || strings.Contains(projectCall.query, "metric_unit") {
t.Fatalf("upsert should ignore legacy zero first mileage: %s", upsertCall.query) t.Fatalf("daily mileage upsert should not use generic metric columns: %s", projectCall.query)
} }
if !strings.Contains(upsertCall.query, "trusted_source_key") { if !strings.Contains(projectCall.query, "LEFT JOIN vehicle_data_source ds") {
t.Fatalf("upsert should track trusted source: %s", upsertCall.query) t.Fatalf("project sql should join data source: %s", projectCall.query)
} }
if got := upsertCall.args[0]; got != "LNBVIN00000000002" { 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) 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) { func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
@@ -234,8 +277,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) {
t.Fatalf("duplicate Append() error = %v", err) t.Fatalf("duplicate Append() error = %v", err)
} }
if len(exec.calls) != 1 { if len(exec.calls) != 3 {
t.Fatalf("exec calls = %d, want 1", len(exec.calls)) t.Fatalf("exec calls = %d, want 3", len(exec.calls))
} }
} }
@@ -258,8 +301,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) {
t.Fatalf("retry Append() error = %v", err) t.Fatalf("retry Append() error = %v", err)
} }
if len(exec.calls) != 2 { if len(exec.calls) != 4 {
t.Fatalf("exec calls = %d, want 2", len(exec.calls)) t.Fatalf("exec calls = %d, want 4", len(exec.calls))
} }
} }
@@ -283,8 +326,8 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) {
t.Fatalf("next-day Append() error = %v", err) t.Fatalf("next-day Append() error = %v", err)
} }
if len(exec.calls) != 2 { if len(exec.calls) != 6 {
t.Fatalf("exec calls = %d, want 2", len(exec.calls)) t.Fatalf("exec calls = %d, want 6", len(exec.calls))
} }
if len(writer.lastTotalMileage) != 1 { if len(writer.lastTotalMileage) != 1 {
t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage)) t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))