From bcf7cdae2eb84690bb4cd0f059cad20ca9e12d37 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 00:18:35 +0800 Subject: [PATCH] fix: key daily metrics by vehicle identity --- .../internal/stats/daily_metric.go | 97 ++++++++++++++++++- .../internal/stats/daily_metric_test.go | 37 ++++++- go/vehicle-gateway/internal/stats/query.go | 40 +++++--- .../internal/stats/query_test.go | 52 ++++++++-- go/vehicle-gateway/internal/stats/schema.go | 6 +- 5 files changed, 201 insertions(+), 31 deletions(-) diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index 89c1224f..8759dc96 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -26,6 +26,7 @@ type Writer struct { } type MetricSample struct { + VehicleKey string VIN string Protocol envelope.Protocol StatDate string @@ -45,8 +46,13 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { } func (w *Writer) EnsureSchema(ctx context.Context) error { - _, err := w.exec.ExecContext(ctx, DailyMetricTableSQL) - return err + if _, err := w.exec.ExecContext(ctx, DailyMetricTableSQL); err != nil { + return err + } + if db, ok := w.exec.(metadataDB); ok { + return migrateDailyMetricTable(ctx, db) + } + return nil } func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { @@ -56,6 +62,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { } for _, sample := range samples { if _, err := w.exec.ExecContext(ctx, upsertDailyMetricSQL, + sample.VehicleKey, sample.VIN, sample.StatDate, string(sample.Protocol), @@ -71,7 +78,8 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) { vin := strings.TrimSpace(env.VIN) - if vin == "" { + vehicleKey := strings.TrimSpace(env.VehicleKey()) + if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { return nil, nil } totalMileage, ok := floatField(env, envelope.FieldTotalMileageKM) @@ -94,6 +102,7 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr statDate := time.UnixMilli(eventMS).In(loc).Format("2006-01-02") return []MetricSample{ { + VehicleKey: vehicleKey, VIN: vin, Protocol: env.Protocol, StatDate: statDate, @@ -102,6 +111,7 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr TotalMileageKM: totalMileage, }, { + VehicleKey: vehicleKey, VIN: vin, Protocol: env.Protocol, StatDate: statDate, @@ -114,10 +124,11 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr const upsertDailyMetricSQL = ` INSERT INTO vehicle_daily_metric - (vin, stat_date, protocol, metric_key, metric_value, metric_unit, + (vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method) -VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF') +VALUES (?, ?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF') ON DUPLICATE KEY UPDATE + vin = IF(VALUES(vin) <> '', VALUES(vin), vin), first_total_mileage_km = CASE WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0 THEN VALUES(first_total_mileage_km) @@ -158,6 +169,82 @@ ON DUPLICATE KEY UPDATE updated_at = CURRENT_TIMESTAMP ` +type metadataDB interface { + Execer + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +func migrateDailyMetricTable(ctx context.Context, db metadataDB) error { + columnExists, err := informationSchemaExists(ctx, db, ` +SELECT COUNT(*) +FROM information_schema.columns +WHERE table_schema = DATABASE() + AND table_name = 'vehicle_daily_metric' + AND column_name = 'vehicle_key'`) + if err != nil { + return err + } + if !columnExists { + if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD COLUMN vehicle_key VARCHAR(96) NOT NULL DEFAULT '' AFTER id`); err != nil { + return err + } + } + if _, err := db.ExecContext(ctx, `UPDATE vehicle_daily_metric SET vehicle_key = vin WHERE vehicle_key = ''`); err != nil { + return err + } + vehicleIndexExists, err := informationSchemaExists(ctx, db, ` +SELECT COUNT(*) +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND table_name = 'vehicle_daily_metric' + AND index_name = 'uk_daily_metric_vehicle'`) + if err != nil { + return err + } + if !vehicleIndexExists { + if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD UNIQUE KEY uk_daily_metric_vehicle (vehicle_key, stat_date, protocol, metric_key)`); err != nil { + return err + } + } + oldIndexExists, err := informationSchemaExists(ctx, db, ` +SELECT COUNT(*) +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND table_name = 'vehicle_daily_metric' + AND index_name = 'uk_daily_metric'`) + if err != nil { + return err + } + if oldIndexExists { + if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric DROP INDEX uk_daily_metric`); err != nil { + return err + } + } + vinIndexExists, err := informationSchemaExists(ctx, db, ` +SELECT COUNT(*) +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND table_name = 'vehicle_daily_metric' + AND index_name = 'idx_vin'`) + if err != nil { + return err + } + if !vinIndexExists { + if _, err := db.ExecContext(ctx, `ALTER TABLE vehicle_daily_metric ADD KEY idx_vin (vin)`); err != nil { + return err + } + } + return nil +} + +func informationSchemaExists(ctx context.Context, db metadataDB, query string) (bool, error) { + var count int + if err := db.QueryRowContext(ctx, query).Scan(&count); err != nil { + return false, err + } + return count > 0, nil +} + func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { if env.Fields == nil { return 0, false diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 2d250be8..8a4f2046 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -31,6 +31,9 @@ func TestSamplesFromEnvelopeDerivesDailyMileageAndTotal(t *testing.T) { if samples[0].MetricKey != MetricDailyMileageKM || samples[0].MetricValue != 0 { t.Fatalf("unexpected daily mileage sample: %#v", samples[0]) } + if samples[0].VehicleKey != "LNBVIN00000000001" { + t.Fatalf("vehicle key = %q", samples[0].VehicleKey) + } if samples[1].MetricKey != MetricDailyTotalMileageKM || samples[1].MetricValue != 10241.2 { t.Fatalf("unexpected daily total sample: %#v", samples[1]) } @@ -39,7 +42,31 @@ func TestSamplesFromEnvelopeDerivesDailyMileageAndTotal(t *testing.T) { } } -func TestSamplesFromEnvelopeSkipsMissingVINOrMileage(t *testing.T) { +func TestSamplesFromEnvelopeUsesVehicleKeyWhenVINIsMissing(t *testing.T) { + loc := time.FixedZone("Asia/Shanghai", 8*3600) + samples, err := SamplesFromEnvelope(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + Phone: "013307811254", + EventTimeMS: time.Date(2026, 7, 2, 0, 3, 0, 0, loc).UnixMilli(), + Fields: map[string]any{ + envelope.FieldTotalMileageKM: 10985.7, + }, + }, loc) + if err != nil { + t.Fatalf("SamplesFromEnvelope() error = %v", err) + } + if len(samples) != 2 { + t.Fatalf("sample count = %d", len(samples)) + } + if samples[0].VehicleKey != "JT808:013307811254" { + t.Fatalf("vehicle key = %q", samples[0].VehicleKey) + } + if samples[0].VIN != "" { + t.Fatalf("vin should stay empty for unresolved JT808 identity, got %q", samples[0].VIN) + } +} + +func TestSamplesFromEnvelopeSkipsMissingVehicleKeyOrMileage(t *testing.T) { samples, err := SamplesFromEnvelope(envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, Fields: map[string]any{envelope.FieldTotalMileageKM: 1.2}, @@ -48,7 +75,7 @@ func TestSamplesFromEnvelopeSkipsMissingVINOrMileage(t *testing.T) { t.Fatalf("SamplesFromEnvelope() error = %v", err) } if len(samples) != 0 { - t.Fatalf("expected no samples without vin, got %#v", samples) + t.Fatalf("expected no samples without vehicle key, got %#v", samples) } samples, err = SamplesFromEnvelope(envelope.FrameEnvelope{ @@ -103,6 +130,9 @@ func TestWriterEnsuresSchemaAndUpsertsTwoMetrics(t *testing.T) { if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_daily_metric") { t.Fatalf("unexpected schema sql: %s", exec.calls[0].query) } + if !strings.Contains(exec.calls[0].query, "vehicle_key") { + t.Fatalf("schema should include vehicle_key: %s", exec.calls[0].query) + } if len(exec.calls) != 3 { t.Fatalf("exec calls = %d", len(exec.calls)) } @@ -112,6 +142,9 @@ func TestWriterEnsuresSchemaAndUpsertsTwoMetrics(t *testing.T) { if !strings.Contains(exec.calls[1].query, "first_total_mileage_km <= 0") { t.Fatalf("upsert should ignore legacy zero first mileage: %s", exec.calls[1].query) } + if got := exec.calls[1].args[0]; got != "LNBVIN00000000002" { + t.Fatalf("first upsert arg should be vehicle_key, got %#v", got) + } } type execCall struct { diff --git a/go/vehicle-gateway/internal/stats/query.go b/go/vehicle-gateway/internal/stats/query.go index 1cb3402a..ecb8b74a 100644 --- a/go/vehicle-gateway/internal/stats/query.go +++ b/go/vehicle-gateway/internal/stats/query.go @@ -17,16 +17,18 @@ type Queryer interface { } type MetricQuery struct { - VIN string - Protocol string - MetricKey string - DateFrom string - DateTo string - Limit int - Offset int + VehicleKey string + VIN string + Protocol string + MetricKey string + DateFrom string + DateTo string + Limit int + Offset int } type MetricRow struct { + VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` StatDate string `json:"stat_date"` Protocol string `json:"protocol"` @@ -70,6 +72,7 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr var first sql.NullFloat64 var latest sql.NullFloat64 if err := rows.Scan( + &row.VehicleKey, &row.VIN, &statDate, &row.Protocol, @@ -100,6 +103,7 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr } func normalizeMetricQuery(query MetricQuery) MetricQuery { + query.VehicleKey = strings.TrimSpace(query.VehicleKey) query.VIN = strings.TrimSpace(query.VIN) query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) query.MetricKey = strings.TrimSpace(query.MetricKey) @@ -121,6 +125,9 @@ func buildMetricSQL(query MetricQuery) (string, []any) { if query.VIN != "" { add("vin = ?", query.VIN) } + if query.VehicleKey != "" { + add("vehicle_key = ?", query.VehicleKey) + } if query.Protocol != "" { add("protocol = ?", query.Protocol) } @@ -133,11 +140,11 @@ func buildMetricSQL(query MetricQuery) (string, []any) { if query.DateTo != "" { add("stat_date <= ?", query.DateTo) } - sqlText := `SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric` + sqlText := `SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric` if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") } - sqlText += " ORDER BY stat_date DESC, vin ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?" + sqlText += " ORDER BY stat_date DESC, vehicle_key ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?" args = append(args, query.Limit, query.Offset) return sqlText, args } @@ -192,13 +199,14 @@ func parseMetricQuery(r *http.Request) (MetricQuery, error) { return MetricQuery{}, err } query := MetricQuery{ - VIN: values.Get("vin"), - Protocol: values.Get("protocol"), - MetricKey: values.Get("metricKey"), - DateFrom: values.Get("dateFrom"), - DateTo: values.Get("dateTo"), - Limit: limit, - Offset: offset, + VehicleKey: values.Get("vehicleKey"), + VIN: values.Get("vin"), + Protocol: values.Get("protocol"), + MetricKey: values.Get("metricKey"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, } if !validDate(query.DateFrom) || !validDate(query.DateTo) { return MetricQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD") diff --git a/go/vehicle-gateway/internal/stats/query_test.go b/go/vehicle-gateway/internal/stats/query_test.go index 75ce6f22..4600a854 100644 --- a/go/vehicle-gateway/internal/stats/query_test.go +++ b/go/vehicle-gateway/internal/stats/query_test.go @@ -18,14 +18,14 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). + mock.ExpectQuery("SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). WithArgs("LKLG7C4E3NA774736", "JT808", "2026-07-01", "2026-07-01", 20, 0). WillReturnRows(sqlmock.NewRows([]string{ - "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", + "vehicle_key", "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", "first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method", "created_at", "updated_at", }).AddRow( - "LKLG7C4E3NA774736", time.Date(2026, 7, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), "JT808", MetricDailyTotalMileageKM, 12345.6, "km", + "LKLG7C4E3NA774736", "LKLG7C4E3NA774736", time.Date(2026, 7, 1, 0, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), "JT808", MetricDailyTotalMileageKM, 12345.6, "km", 12345.6, 12345.6, 13, "TOTAL_MILEAGE_DIFF", time.Date(2026, 7, 1, 22, 49, 11, 0, time.FixedZone("Asia/Shanghai", 8*3600)), time.Date(2026, 7, 1, 23, 9, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)), )) @@ -46,6 +46,9 @@ func TestMetricRepositoryQueriesDailyMetricsWithFilters(t *testing.T) { if rows[0].MetricKey != MetricDailyTotalMileageKM || rows[0].MetricValue != 12345.6 { t.Fatalf("unexpected row: %#v", rows[0]) } + if rows[0].VehicleKey != "LKLG7C4E3NA774736" { + t.Fatalf("vehicle key = %q", rows[0].VehicleKey) + } if rows[0].StatDate != "2026-07-01" || rows[0].UpdatedAt != "2026-07-01 23:09:36" { t.Fatalf("unexpected time formatting: %#v", rows[0]) } @@ -60,14 +63,14 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) { t.Fatalf("sqlmock.New() error = %v", err) } defer db.Close() - mock.ExpectQuery("SELECT vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). + mock.ExpectQuery("SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0). WillReturnRows(sqlmock.NewRows([]string{ - "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", + "vehicle_key", "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", "first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method", "created_at", "updated_at", }).AddRow( - "LB9A32A21R0LS1707", "2020-07-01", "GB32960", MetricDailyMileageKM, 0.0, "km", + "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "2020-07-01", "GB32960", MetricDailyMileageKM, 0.0, "km", 53490.9, 53490.9, 3, "TOTAL_MILEAGE_DIFF", "2026-07-01 22:07:58", "2026-07-01 22:28:25", )) @@ -91,6 +94,43 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) { } } +func TestMetricHandlerFiltersByVehicleKey(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + mock.ExpectQuery("SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric"). + WithArgs("JT808:013307811254", "JT808", 50, 0). + WillReturnRows(sqlmock.NewRows([]string{ + "vehicle_key", "vin", "stat_date", "protocol", "metric_key", "metric_value", "metric_unit", + "first_total_mileage_km", "latest_total_mileage_km", "sample_count", "calculation_method", + "created_at", "updated_at", + }).AddRow( + "JT808:013307811254", "", "2026-07-02", "JT808", MetricDailyTotalMileageKM, 10985.7, "km", + 10985.7, 10985.7, 1, "TOTAL_MILEAGE_DIFF", "2026-07-02 00:03:00", "2026-07-02 00:03:00", + )) + + handler := NewMetricHandler(NewMetricRepository(db)) + request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?vehicleKey=JT808:013307811254&protocol=JT808", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"vehicle_key":"JT808:013307811254"`, `"vin":""`, `"metric_value":10985.7`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestMetricHandlerRejectsInvalidPagination(t *testing.T) { handler := NewMetricHandler(NewMetricRepository(&sql.DB{})) request := httptest.NewRequest(http.MethodGet, "/api/stats/daily-metrics?limit=2001", nil) diff --git a/go/vehicle-gateway/internal/stats/schema.go b/go/vehicle-gateway/internal/stats/schema.go index 7bf860e8..2094dc7b 100644 --- a/go/vehicle-gateway/internal/stats/schema.go +++ b/go/vehicle-gateway/internal/stats/schema.go @@ -2,7 +2,8 @@ package stats const DailyMetricTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_metric ( id BIGINT PRIMARY KEY AUTO_INCREMENT, - vin VARCHAR(32) NOT NULL, + vehicle_key VARCHAR(96) NOT NULL, + vin VARCHAR(32) NOT NULL DEFAULT '', stat_date DATE NOT NULL, protocol VARCHAR(32) NOT NULL, metric_key VARCHAR(64) NOT NULL, @@ -14,7 +15,8 @@ const DailyMetricTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_metric ( calculation_method VARCHAR(64) NOT NULL, created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_daily_metric (vin, stat_date, protocol, metric_key), + UNIQUE KEY uk_daily_metric_vehicle (vehicle_key, stat_date, protocol, metric_key), + KEY idx_vin (vin), KEY idx_stat_date (stat_date), KEY idx_protocol_metric (protocol, metric_key) )`