From 17937a1ad831e533d53171e58f36b72bd6f68291 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 15:44:31 +0800 Subject: [PATCH] fix(stats): skip realtime mileage without source identity --- .../internal/stats/daily_metric.go | 23 ++++---- .../internal/stats/daily_metric_test.go | 52 ++++++++++++++----- 2 files changed, 50 insertions(+), 25 deletions(-) diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 41be538c..c1657cdc 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -85,17 +85,18 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { 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 := w.applyRealtimeBaseline(ctx, &candidate); err != nil { - return err - } - if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil { - return err - } + if !hasSource { + continue + } + if err := UpsertDataSource(ctx, w.exec, identity, sample.EventTime); err != nil { + return err + } + candidate := SourceMileageSampleFromMetric(sample, identity) + if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { + return err + } + if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil { + return err } if err := ProjectDailyMileage(ctx, w.exec, sample.VIN, sample.StatDate, sample.Protocol); err != nil { return err diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 6840b559..50d33733 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -190,6 +190,27 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) { } } +func TestWriterAppendSkipsMileageWithoutSourceIdentity(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4123.9, + }, + } + + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("Append() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0 without source identity", len(exec.calls)) + } +} + func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) @@ -474,9 +495,10 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) event := envelope.FrameEnvelope{ - Protocol: envelope.ProtocolJT808, - VIN: "LNBVIN00000000001", - EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), Fields: map[string]any{ envelope.FieldTotalMileageKM: 10241.2, }, @@ -490,8 +512,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { t.Fatalf("duplicate Append() error = %v", err) } - if len(exec.calls) != 4 { - t.Fatalf("exec calls = %d, want 4", len(exec.calls)) + if len(exec.calls) != 6 { + t.Fatalf("exec calls = %d, want 6", len(exec.calls)) } } @@ -499,9 +521,10 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) { exec := &recordingExec{errs: []error{errors.New("mysql unavailable"), nil}} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) event := envelope.FrameEnvelope{ - Protocol: envelope.ProtocolJT808, - VIN: "LNBVIN00000000001", - EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: time.Date(2026, 7, 1, 9, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), Fields: map[string]any{ envelope.FieldTotalMileageKM: 10241.2, }, @@ -514,8 +537,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) { t.Fatalf("retry Append() error = %v", err) } - if len(exec.calls) != 5 { - t.Fatalf("exec calls = %d, want 5", len(exec.calls)) + if len(exec.calls) != 7 { + t.Fatalf("exec calls = %d, want 7", len(exec.calls)) } } @@ -523,8 +546,9 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) event := envelope.FrameEnvelope{ - Protocol: envelope.ProtocolJT808, - VIN: "LNBVIN00000000001", + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + SourceEndpoint: "115.231.168.135:20215", Fields: map[string]any{ envelope.FieldTotalMileageKM: 10241.2, }, @@ -539,8 +563,8 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { t.Fatalf("next-day Append() error = %v", err) } - if len(exec.calls) != 8 { - t.Fatalf("exec calls = %d, want 8", len(exec.calls)) + if len(exec.calls) != 12 { + t.Fatalf("exec calls = %d, want 12", len(exec.calls)) } if len(writer.lastTotalMileage) != 1 { t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage))