From bd6703492777fe1bf7555fd26b4e624cb71d0dd7 Mon Sep 17 00:00:00 2001
From: lingniu <lingniu@lingnius-Mac-mini.local>
Date: Wed, 8 Jul 2026 15:50:38 +0800
Subject: [PATCH] fix(stats): guard backfill source-less mileage

---
 go/vehicle-gateway/cmd/stats-backfill/main.go | 17 +++--
 .../cmd/stats-backfill/main_test.go           | 65 +++++++++++++++++++
 2 files changed, 75 insertions(+), 7 deletions(-)

diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go
index 910643b8..672220fb 100644
--- a/go/vehicle-gateway/cmd/stats-backfill/main.go
+++ b/go/vehicle-gateway/cmd/stats-backfill/main.go
@@ -233,8 +233,8 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample)
 				SourceEndpoint:  sample.SourceEndpoint,
 				FirstEventTime:  sample.EventTime,
 				LatestEventTime: sample.EventTime,
-				QualityStatus:   stats.QualityOK,
-				QualityReason:   "scan_backfill",
+				QualityStatus:   stats.QualityNoPreviousBaseline,
+				QualityReason:   "scan_backfill_missing_previous_source",
 			}
 			continue
 		}
@@ -268,6 +268,14 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 	var written int64
 	clearedTargets := map[string]struct{}{}
 	for _, agg := range aggregates {
+		identity := stats.SourceIdentity{
+			Protocol:       agg.Protocol,
+			SourceIP:       stats.NormalizeSourceIP(agg.SourceEndpoint),
+			SourceEndpoint: agg.SourceEndpoint,
+		}
+		if strings.TrimSpace(identity.SourceIP) == "" {
+			continue
+		}
 		target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
 		if _, ok := clearedTargets[target]; !ok {
 			if err := clearBackfillTargetMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
@@ -275,11 +283,6 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 			}
 			clearedTargets[target] = struct{}{}
 		}
-		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
 		}
diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go
index 80a27e41..3d42e7f1 100644
--- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go
+++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go
@@ -184,6 +184,71 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T)
 	}
 }
 
+func TestWriteAggregatesSkipsBlankSourceBeforeClearingTarget(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+
+	written, err := writeAggregates(context.Background(), db, map[string]*metricAgg{
+		"LA9GG64L7PBAF4001|2026-07-08|JT808|JT808:13307765812@": {
+			VIN:             "LA9GG64L7PBAF4001",
+			Date:            "2026-07-08",
+			Protocol:        envelope.ProtocolJT808,
+			FirstKM:         4100,
+			LatestKM:        4101,
+			Count:           1,
+			SourceKey:       "JT808:13307765812@",
+			Phone:           "13307765812",
+			SourceEndpoint:  "",
+			FirstEventTime:  time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+			LatestEventTime: time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
+			QualityStatus:   stats.QualityNoPreviousBaseline,
+			QualityReason:   "missing_previous_source",
+		},
+	}, 500)
+	if err != nil {
+		t.Fatalf("writeAggregates() error = %v", err)
+	}
+	if written != 0 {
+		t.Fatalf("written = %d, want 0", written)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) {
+	loc := time.FixedZone("Asia/Shanghai", 8*3600)
+	samples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		VIN:            "LA9GG64L7PBAF4001",
+		Phone:          "13307765812",
+		SourceEndpoint: "115.231.168.135:20215",
+		EventTimeMS:    time.Date(2026, 7, 8, 12, 0, 0, 0, loc).UnixMilli(),
+		Fields: map[string]any{
+			"jt808.location.total_mileage_km": 4123.9,
+		},
+	}, loc)
+	if err != nil {
+		t.Fatalf("SamplesFromEnvelope() error = %v", err)
+	}
+	aggregates := map[string]*metricAgg{}
+	addSamples(aggregates, samples)
+	if len(aggregates) != 1 {
+		t.Fatalf("aggregate count = %d, want 1", len(aggregates))
+	}
+	for _, agg := range aggregates {
+		if agg.QualityStatus != stats.QualityNoPreviousBaseline {
+			t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityNoPreviousBaseline)
+		}
+		if agg.QualityReason != "scan_backfill_missing_previous_source" {
+			t.Fatalf("quality reason = %q", agg.QualityReason)
+		}
+	}
+}
+
 func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) {
 	t.Setenv("BACKFILL_METHOD", "")
 	t.Setenv("BACKFILL_DATE_FROM", "2026-07-08")