diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index f5462179..807212e7 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -4,8 +4,10 @@ import ( "context" "database/sql" "errors" + "fmt" "strconv" "strings" + "sync" "time" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" @@ -16,8 +18,10 @@ type Execer interface { } type Writer struct { - exec Execer - loc *time.Location + exec Execer + loc *time.Location + mu sync.Mutex + lastTotalMileage map[string]float64 } type MetricSample struct { @@ -34,7 +38,7 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { if loc == nil { loc = time.FixedZone("Asia/Shanghai", 8*3600) } - return &Writer{exec: exec, loc: loc} + return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}} } func (w *Writer) EnsureSchema(ctx context.Context) error { @@ -50,6 +54,9 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { return err } for _, sample := range samples { + if w.seenSameMileage(sample) { + continue + } if _, err := w.exec.ExecContext(ctx, upsertDailyMileageSQL, sample.VIN, sample.StatDate, @@ -62,6 +69,23 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { return nil } +func (w *Writer) seenSameMileage(sample MetricSample) bool { + prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol) + key := prefix + sample.StatDate + w.mu.Lock() + defer w.mu.Unlock() + if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM { + return true + } + for existing := range w.lastTotalMileage { + if strings.HasPrefix(existing, prefix) && existing != key { + delete(w.lastTotalMileage, existing) + } + } + w.lastTotalMileage[key] = sample.TotalMileageKM + return false +} + func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) { vin := strings.TrimSpace(env.VIN) if vin == "" { diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 325411cf..4be39fbe 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -149,6 +149,59 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { } } +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(), + Fields: map[string]any{ + envelope.FieldTotalMileageKM: 10241.2, + }, + } + + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("first Append() error = %v", err) + } + event.ReceivedAtMS += 1000 + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("duplicate Append() error = %v", err) + } + + if len(exec.calls) != 1 { + t.Fatalf("exec calls = %d, want 1", len(exec.calls)) + } +} + +func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LNBVIN00000000001", + Fields: map[string]any{ + envelope.FieldTotalMileageKM: 10241.2, + }, + } + + event.EventTimeMS = time.Date(2026, 7, 1, 23, 59, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli() + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("first Append() error = %v", err) + } + event.EventTimeMS = time.Date(2026, 7, 2, 0, 1, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli() + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("next-day Append() error = %v", err) + } + + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + } + if len(writer.lastTotalMileage) != 1 { + t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage)) + } +} + type execCall struct { query string args []any