diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 743b2cae..360c4495 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/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 { diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index 4a20c67d..9b71433c 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/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) + } +}