fix(stats): clear stale backfill daily mileage rows

go/vehicle-gateway/cmd/stats-backfill/main.go

@@ -266,6 +266,7 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 		return 0, nil
 	}
 	var written int64
+	targets := map[string]struct{}{}
 	for _, agg := range aggregates {
 		identity := stats.SourceIdentity{
 			Protocol:       agg.Protocol,
@@ -307,6 +308,13 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 		if err := stats.UpsertSourceMileage(ctx, db, candidate); err != nil {
 			return written, err
 		}
+		target := agg.VIN + "|" + agg.Date + "|" + string(agg.Protocol)
+		if _, ok := targets[target]; !ok {
+			if err := clearBackfillFinalMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
+				return written, err
+			}
+			targets[target] = struct{}{}
+		}
 		if err := stats.ProjectDailyMileage(ctx, db, agg.VIN, agg.Date, agg.Protocol); err != nil {
 			return written, err
 		}
@@ -315,6 +323,14 @@ func writeAggregates(ctx context.Context, db *sql.DB, aggregates map[string]*met
 	return written, nil
 }
 
+func clearBackfillFinalMileage(ctx context.Context, db *sql.DB, vin string, statDate string, protocol envelope.Protocol) error {
+	if db == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(string(protocol)) == "" {
+		return nil
+	}
+	_, err := db.ExecContext(ctx, "DELETE FROM vehicle_daily_mileage WHERE vin = ? AND stat_date = ? AND protocol = ?", vin, statDate, string(protocol))
+	return err
+}
+
 func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) {
 	dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo)
 	if err != nil {

go/vehicle-gateway/cmd/stats-backfill/main_test.go

@@ -1,6 +1,14 @@
 package main
 
-import "testing"
+import (
+	"context"
+	"testing"
+	"time"
+
+	"github.com/DATA-DOG/go-sqlmock"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats"
+)
 
 func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) {
 	previous := []dailySourceLast{{
@@ -58,3 +66,104 @@ func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) {
 		t.Fatalf("same source IP should produce same source key: %q vs %q", sourceA.SourceKey, sourceB.SourceKey)
 	}
 }
+
+func TestClearBackfillFinalMileageClearsExactKey(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	if err := clearBackfillFinalMileage(context.Background(), db, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808); err != nil {
+		t.Fatalf("clearBackfillFinalMileage() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
+func TestWriteAggregatesClearsFinalBeforeProjection(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatalf("sqlmock.New() error = %v", err)
+	}
+	defer db.Close()
+	mock.MatchExpectationsInOrder(true)
+
+	projNow := time.Date(2026, 7, 8, 12, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600))
+	aggregates := map[string]*metricAgg{
+		"LA9GG64L7PBAF4001|2026-07-08|JT808|JT808:13307765812@115.231.168.135": {
+			VIN:             "LA9GG64L7PBAF4001",
+			Date:            "2026-07-08",
+			Protocol:        envelope.ProtocolJT808,
+			FirstKM:         4100,
+			LatestKM:        4101,
+			Count:           1,
+			SourceKey:       "JT808:13307765812@115.231.168.135",
+			Phone:           "13307765812",
+			SourceEndpoint:  "115.231.168.135:20215",
+			FirstEventTime:  projNow,
+			LatestEventTime: projNow,
+			QualityStatus:   stats.QualityInvalidDelta,
+			QualityReason:   "outside_daily_range",
+		},
+	}
+
+	mock.ExpectExec(`INSERT INTO vehicle_data_source`).
+		WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			"JT808:13307765812@115.231.168.135",
+			"115.231.168.135",
+			"115.231.168.135:20215",
+			"13307765812",
+			"",
+			"",
+			float64(4100),
+			float64(4101),
+			float64(1),
+			int64(1),
+			projNow,
+			projNow,
+			stats.QualityInvalidDelta,
+			"outside_daily_range",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`DELETE FROM vehicle_daily_mileage WHERE vin = \? AND stat_date = \? AND protocol = \?`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 1))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`).
+		WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)).
+		WillReturnResult(sqlmock.NewResult(1, 0))
+	mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`).
+		WithArgs(
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+			int64(1000),
+			"LA9GG64L7PBAF4001",
+			"2026-07-08",
+			"JT808",
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	written, err := writeAggregates(context.Background(), db, aggregates, 500)
+	if err != nil {
+		t.Fatalf("writeAggregates() error = %v", err)
+	}
+	if written != 1 {
+		t.Fatalf("written = %d, want 1", written)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}