From b452be3b94ba759d70d178f152f9b3697fb9d797 Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 8 Jul 2026 18:39:20 +0800 Subject: [PATCH] fix(stats): backfill mqtt mileage from mileage frames --- go/vehicle-gateway/cmd/stats-backfill/main.go | 18 ++++ .../cmd/stats-backfill/main_test.go | 85 +++++++++++++++++++ 2 files changed, 103 insertions(+) diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 7a20380a..1578b219 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -445,6 +445,9 @@ func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, proto fmt.Sprintf("protocol = '%s'", quote(string(protocol))), realtimeMileageFramePredicate(), } + if predicate := mileageBearingFramePredicate(protocol); predicate != "" { + where = append(where, predicate) + } sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, FIRST(ts), FIRST(parsed_json), LAST(ts), LAST(parsed_json), COUNT(*) FROM %s.raw_frames WHERE %s @@ -461,6 +464,9 @@ func queryPreviousLastSourceRows(ctx context.Context, db *sql.DB, cfg config, pr fmt.Sprintf("protocol = '%s'", quote(string(protocol))), realtimeMileageFramePredicate(), } + if predicate := mileageBearingFramePredicate(protocol); predicate != "" { + where = append(where, predicate) + } sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json), COUNT(*) FROM %s.raw_frames WHERE %s @@ -677,6 +683,18 @@ func realtimeMileageFramePredicate() string { )` } +func mileageBearingFramePredicate(protocol envelope.Protocol) string { + keys := mileageKeys(protocol) + if len(keys) == 0 { + return "" + } + conditions := make([]string, 0, len(keys)) + for _, key := range keys { + conditions = append(conditions, fmt.Sprintf("parsed_json LIKE '%%%s%%'", quote(key))) + } + return "(" + strings.Join(conditions, " OR ") + ")" +} + func scanRawFrame(rows *sql.Rows) (rawFrameRow, error) { var protocol, vin, phone, deviceID, sourceEndpoint, eventID, parsed string var messageID int64 diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index d118432d..eb64566e 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -130,6 +130,91 @@ func TestAggregateFromDailySourceUsesCurrentFirstSampleWithoutHistory(t *testing } } +func TestQueryDailyLastSourceRowsFiltersYutongToMileageFrames(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + loc := time.FixedZone("Asia/Shanghai", 8*3600) + firstTS := time.Date(2026, 7, 8, 8, 0, 0, 0, loc) + lastTS := time.Date(2026, 7, 8, 18, 0, 0, 0, loc) + mock.ExpectQuery("(?s)FROM lingniu_vehicle_ts\\.raw_frames.*protocol = 'YUTONG_MQTT'.*parsed_json LIKE '%yutong_mqtt\\.data\\.total_mileage%'"). + WillReturnRows(sqlmock.NewRows([]string{ + "vin", "phone", "device_id", "source_endpoint", "FIRST(ts)", "FIRST(parsed_json)", "LAST(ts)", "LAST(parsed_json)", "COUNT(*)", + }).AddRow( + "LMRKH9AC6R1004108", + "", + "LMRKH9AC6R1004108", + "mqtt://yutong/ytforward/shln/3", + firstTS, + `{"yutong_mqtt.data.total_mileage":"65422000"}`, + lastTS, + `{"yutong_mqtt.data.total_mileage":"65423000"}`, + int64(42), + )) + + rows, err := queryDailyLastSourceRows(context.Background(), db, config{ + TDengineDatabase: "lingniu_vehicle_ts", + Location: loc, + }, envelope.ProtocolYutongMQTT, "2026-07-08") + if err != nil { + t.Fatalf("queryDailyLastSourceRows() error = %v", err) + } + sourceRows := rows["LMRKH9AC6R1004108"] + if len(sourceRows) != 1 { + t.Fatalf("rows = %d, want 1", len(sourceRows)) + } + if sourceRows[0].FirstTotalKM != 65422 || sourceRows[0].TotalKM != 65423 { + t.Fatalf("km range = %v -> %v", sourceRows[0].FirstTotalKM, sourceRows[0].TotalKM) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + +func TestQueryPreviousLastSourceRowsFiltersYutongToMileageFrames(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + loc := time.FixedZone("Asia/Shanghai", 8*3600) + lastTS := time.Date(2026, 7, 7, 23, 58, 0, 0, loc) + mock.ExpectQuery("(?s)FROM lingniu_vehicle_ts\\.raw_frames.*protocol = 'YUTONG_MQTT'.*parsed_json LIKE '%yutong_mqtt\\.data\\.total_mileage%'"). + WillReturnRows(sqlmock.NewRows([]string{ + "vin", "phone", "device_id", "source_endpoint", "LAST(ts)", "LAST(parsed_json)", "COUNT(*)", + }).AddRow( + "LMRKH9AC6R1004108", + "", + "LMRKH9AC6R1004108", + "mqtt://yutong/ytforward/shln/3", + lastTS, + `{"yutong_mqtt.data.total_mileage":"65377000"}`, + int64(31), + )) + + rows, err := queryPreviousLastSourceRows(context.Background(), db, config{ + TDengineDatabase: "lingniu_vehicle_ts", + Location: loc, + }, envelope.ProtocolYutongMQTT, "2026-07-08") + if err != nil { + t.Fatalf("queryPreviousLastSourceRows() error = %v", err) + } + sourceRows := rows["LMRKH9AC6R1004108"] + if len(sourceRows) != 1 { + t.Fatalf("rows = %d, want 1", len(sourceRows)) + } + if sourceRows[0].TotalKM != 65377 { + t.Fatalf("previous total = %v, want 65377", sourceRows[0].TotalKM) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) { db, mock, err := sqlmock.New() if err != nil {