diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 1bcb785a..1d11a3d7 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -49,12 +49,24 @@ type rawFrameRow struct { } type metricAgg struct { - VIN string - Date string - Protocol envelope.Protocol - FirstKM float64 - LatestKM float64 - Count int64 + VIN string + Date string + Protocol envelope.Protocol + FirstKM float64 + LatestKM float64 + Count int64 + SourceKey string + Phone string + SourceEndpoint string +} + +type dailySourceLast struct { + VIN string + SourceKey string + Phone string + SourceEndpoint string + TS time.Time + TotalKM float64 } func main() { @@ -249,7 +261,7 @@ func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) err if daily < 0 { daily = 0 } - placeholders = append(placeholders, "(?,?,?,?,?,?,?)") + placeholders = append(placeholders, "(?,?,?,?,?,?,?,?,?,?)") args = append(args, row.VIN, row.Date, @@ -257,16 +269,23 @@ func writeAggregateBatch(ctx context.Context, db *sql.DB, rows []*metricAgg) err daily, row.FirstKM, row.LatestKM, + row.SourceKey, + row.Phone, + row.SourceEndpoint, row.Count, ) } sqlText := `INSERT INTO vehicle_daily_mileage - (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, sample_count) + (vin, stat_date, protocol, daily_mileage_km, first_total_mileage_km, latest_total_mileage_km, + trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count) VALUES ` + strings.Join(placeholders, ",") + ` ON DUPLICATE KEY UPDATE daily_mileage_km = VALUES(daily_mileage_km), first_total_mileage_km = VALUES(first_total_mileage_km), latest_total_mileage_km = VALUES(latest_total_mileage_km), + trusted_source_key = VALUES(trusted_source_key), + trusted_phone = VALUES(trusted_phone), + trusted_source_endpoint = VALUES(trusted_source_endpoint), sample_count = VALUES(sample_count), updated_at = CURRENT_TIMESTAMP` _, err := db.ExecContext(ctx, sqlText, args...) @@ -278,11 +297,11 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s if err != nil { return nil, err } - lastByProtocolDate := map[envelope.Protocol]map[string]map[string]float64{} + lastByProtocolDate := map[envelope.Protocol]map[string]map[string][]dailySourceLast{} for _, protocol := range cfg.Protocols { - lastByProtocolDate[protocol] = map[string]map[string]float64{} + lastByProtocolDate[protocol] = map[string]map[string][]dailySourceLast{} for _, date := range dates { - rows, err := queryDailyLastRows(ctx, db, cfg, protocol, date) + rows, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date) if err != nil { return nil, err } @@ -300,19 +319,22 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s prevDate := previousDate(date) current := lastByProtocolDate[protocol][date] previous := lastByProtocolDate[protocol][prevDate] - for vin, latest := range current { - first, ok := previous[vin] - if !ok || latest < first { + for vin, currentSources := range current { + chosen, ok := chooseTrustedSource(currentSources, previous[vin]) + if !ok { continue } key := vin + "|" + date + "|" + string(protocol) aggregates[key] = &metricAgg{ - VIN: vin, - Date: date, - Protocol: protocol, - FirstKM: first, - LatestKM: latest, - Count: 1, + VIN: vin, + Date: date, + Protocol: protocol, + FirstKM: chosen.previous.TotalKM, + LatestKM: chosen.current.TotalKM, + Count: 1, + SourceKey: chosen.current.SourceKey, + Phone: chosen.current.Phone, + SourceEndpoint: chosen.current.SourceEndpoint, } } } @@ -320,7 +342,38 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s return aggregates, nil } -func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string]float64, error) { +type trustedChoice struct { + current dailySourceLast + previous dailySourceLast +} + +const maxTrustedDailyMileageKM = 1000 + +func chooseTrustedSource(current []dailySourceLast, previous []dailySourceLast) (trustedChoice, bool) { + previousBySource := map[string]dailySourceLast{} + for _, row := range previous { + previousBySource[row.SourceKey] = row + } + var chosen trustedChoice + var chosenDelta float64 + for _, currentRow := range current { + previousRow, ok := previousBySource[currentRow.SourceKey] + if !ok { + continue + } + delta := currentRow.TotalKM - previousRow.TotalKM + if delta < 0 || delta > maxTrustedDailyMileageKM { + continue + } + if chosen.current.SourceKey == "" || delta < chosenDelta { + chosen = trustedChoice{current: currentRow, previous: previousRow} + chosenDelta = delta + } + } + return chosen, chosen.current.SourceKey != "" +} + +func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, date string) (map[string][]dailySourceLast, error) { where := []string{ fmt.Sprintf("ts >= '%s 00:00:00'", quote(date)), fmt.Sprintf("ts < '%s 00:00:00'", quote(nextDate(date))), @@ -330,20 +383,23 @@ func queryDailyLastRows(ctx context.Context, db *sql.DB, cfg config, protocol en fmt.Sprintf("protocol = '%s'", quote(string(protocol))), realtimeMileageFramePredicate(), } - sqlText := fmt.Sprintf(`SELECT vin, LAST(parsed_json) + sqlText := fmt.Sprintf(`SELECT vin, phone, source_endpoint, LAST(ts), LAST(parsed_json) FROM %s.raw_frames WHERE %s -GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) +GROUP BY vin, phone, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) rows, err := db.QueryContext(ctx, sqlText) if err != nil { return nil, err } defer rows.Close() - out := map[string]float64{} + latestBySource := map[string]dailySourceLast{} for rows.Next() { var vin string + var phone string + var sourceEndpoint string + var ts time.Time var parsedJSON string - if err := rows.Scan(&vin, &parsedJSON); err != nil { + if err := rows.Scan(&vin, &phone, &sourceEndpoint, &ts, &parsedJSON); err != nil { return nil, err } fields := fieldsForStats(protocol, vin, parsedJSON) @@ -360,9 +416,53 @@ GROUP BY vin`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) if err != nil || len(samples) == 0 { continue } - out[strings.TrimSpace(vin)] = samples[0].TotalMileageKM + sourceKey := normalizedSourceKey(phone, sourceEndpoint) + if sourceKey == "" { + continue + } + row := dailySourceLast{ + VIN: strings.TrimSpace(vin), + SourceKey: sourceKey, + Phone: strings.TrimSpace(phone), + SourceEndpoint: strings.TrimSpace(sourceEndpoint), + TS: ts, + TotalKM: samples[0].TotalMileageKM, + } + key := row.VIN + "|" + row.SourceKey + if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) { + latestBySource[key] = row + } } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + out := map[string][]dailySourceLast{} + for _, row := range latestBySource { + out[row.VIN] = append(out[row.VIN], row) + } + return out, nil +} + +func normalizedSourceKey(phone string, endpoint string) string { + var parts []string + if phone = strings.TrimSpace(phone); phone != "" { + parts = append(parts, phone) + } + if host := endpointHost(endpoint); host != "" { + parts = append(parts, host) + } + return strings.Join(parts, "@") +} + +func endpointHost(endpoint string) string { + endpoint = strings.TrimSpace(endpoint) + if endpoint == "" { + return "" + } + if host, _, ok := strings.Cut(endpoint, ":"); ok { + return strings.TrimSpace(host) + } + return endpoint } func fieldsForStats(protocol envelope.Protocol, vin string, text string) map[string]any { diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go new file mode 100644 index 00000000..076aa15c --- /dev/null +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -0,0 +1,40 @@ +package main + +import "testing" + +func TestChooseTrustedSourceKeepsContinuingSourceAndRejectsNewJump(t *testing.T) { + previous := []dailySourceLast{{ + VIN: "LA9GG64L7PBAF4001", + SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:42630"), + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:42630", + TotalKM: 4100.8, + }} + current := []dailySourceLast{ + { + VIN: "LA9GG64L7PBAF4001", + SourceKey: normalizedSourceKey("13307765812", "115.159.85.149:28316"), + Phone: "13307765812", + SourceEndpoint: "115.159.85.149:28316", + TotalKM: 42447.2, + }, + { + VIN: "LA9GG64L7PBAF4001", + SourceKey: normalizedSourceKey("13307765812", "115.231.168.135:20215"), + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + TotalKM: 4123.9, + }, + } + + chosen, ok := chooseTrustedSource(current, previous) + if !ok { + t.Fatal("chooseTrustedSource() did not choose a source") + } + if chosen.current.SourceKey != normalizedSourceKey("13307765812", "115.231.168.135:20215") { + t.Fatalf("chosen source = %q", chosen.current.SourceKey) + } + if delta := chosen.current.TotalKM - chosen.previous.TotalKM; delta < 23 || delta > 24 { + t.Fatalf("delta = %v, want about 23.1", delta) + } +} diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index b4e03912..665eedb3 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -29,6 +29,9 @@ type MetricSample struct { Protocol envelope.Protocol StatDate string TotalMileageKM float64 + SourceKey string + Phone string + SourceEndpoint string } func NewWriter(exec Execer, loc *time.Location) *Writer { @@ -45,6 +48,11 @@ func (w *Writer) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, DailyMileageTableSQL); err != nil { return err } + for _, statement := range DailyMileageAlterSQL { + if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) { + return err + } + } return nil } @@ -62,7 +70,10 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { sample.StatDate, string(sample.Protocol), sample.TotalMileageKM, - sample.TotalMileageKM); err != nil { + sample.TotalMileageKM, + sample.SourceKey, + sample.Phone, + sample.SourceEndpoint); err != nil { return err } w.markMileageWritten(sample) @@ -122,9 +133,35 @@ func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]Metr Protocol: env.Protocol, StatDate: statDate, TotalMileageKM: totalMileage, + SourceKey: sourceKey(env), + Phone: strings.TrimSpace(env.Phone), + SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), }}, nil } +func sourceKey(env envelope.FrameEnvelope) string { + parts := []string{ + strings.TrimSpace(env.Phone), + strings.TrimSpace(env.DeviceID), + strings.TrimSpace(env.SourceEndpoint), + } + var kept []string + for _, part := range parts { + if part != "" { + kept = append(kept, part) + } + } + return strings.Join(kept, "@") +} + +func isDuplicateColumnError(err error) bool { + if err == nil { + return false + } + text := strings.ToLower(err.Error()) + return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060") +} + type mileageFieldMapping struct { key string scale float64 @@ -163,32 +200,86 @@ func mileageMappingsByProtocol(protocol envelope.Protocol) []mileageFieldMapping const upsertDailyMileageSQL = ` INSERT INTO vehicle_daily_mileage (vin, stat_date, protocol, daily_mileage_km, - first_total_mileage_km, latest_total_mileage_km, sample_count) -VALUES (?, ?, ?, 0, ?, ?, 1) + first_total_mileage_km, latest_total_mileage_km, + trusted_source_key, trusted_phone, trusted_source_endpoint, sample_count) +VALUES (?, ?, ?, 0, ?, ?, ?, ?, ?, 1) ON DUPLICATE KEY UPDATE first_total_mileage_km = CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN first_total_mileage_km WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0 THEN VALUES(first_total_mileage_km) ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)) END, latest_total_mileage_km = CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN latest_total_mileage_km WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0 THEN VALUES(latest_total_mileage_km) ELSE GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)) END, daily_mileage_km = GREATEST( CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN latest_total_mileage_km WHEN latest_total_mileage_km IS NULL OR latest_total_mileage_km <= 0 THEN VALUES(latest_total_mileage_km) ELSE latest_total_mileage_km END, - VALUES(latest_total_mileage_km) + CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN latest_total_mileage_km + ELSE VALUES(latest_total_mileage_km) + END ) - CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN first_total_mileage_km WHEN first_total_mileage_km IS NULL OR first_total_mileage_km <= 0 THEN VALUES(first_total_mileage_km) ELSE LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)) END, - sample_count = sample_count + 1, + trusted_source_key = CASE + WHEN trusted_source_key IS NULL OR trusted_source_key = '' + THEN VALUES(trusted_source_key) + WHEN trusted_source_key = VALUES(trusted_source_key) + THEN trusted_source_key + WHEN ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) <= 50 + THEN VALUES(trusted_source_key) + ELSE trusted_source_key + END, + trusted_phone = CASE + WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key) + THEN VALUES(trusted_phone) + ELSE trusted_phone + END, + trusted_source_endpoint = CASE + WHEN trusted_source_key IS NULL OR trusted_source_key = '' OR trusted_source_key = VALUES(trusted_source_key) + THEN VALUES(trusted_source_endpoint) + ELSE trusted_source_endpoint + END, + sample_count = CASE + WHEN trusted_source_key IS NOT NULL + AND trusted_source_key <> '' + AND trusted_source_key <> VALUES(trusted_source_key) + AND ABS(COALESCE(latest_total_mileage_km, VALUES(latest_total_mileage_km)) - VALUES(latest_total_mileage_km)) > 50 + THEN sample_count + ELSE sample_count + 1 + END, updated_at = CURRENT_TIMESTAMP ` diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index f2c13e2a..41e616f1 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -187,19 +187,23 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { if strings.Contains(exec.calls[0].query, "KEY idx_vin (vin)") { t.Fatalf("daily mileage table should not keep redundant vin index covered by the primary key: %s", exec.calls[0].query) } - if len(exec.calls) != 2 { + if len(exec.calls) != 5 { t.Fatalf("exec calls = %d", len(exec.calls)) } - if !strings.Contains(exec.calls[1].query, "ON DUPLICATE KEY UPDATE") { - t.Fatalf("unexpected upsert sql: %s", exec.calls[1].query) + upsertCall := exec.calls[4] + if !strings.Contains(upsertCall.query, "ON DUPLICATE KEY UPDATE") { + t.Fatalf("unexpected upsert sql: %s", upsertCall.query) } - if strings.Contains(exec.calls[1].query, "metric_key") || strings.Contains(exec.calls[1].query, "metric_unit") { - t.Fatalf("daily mileage upsert should not use generic metric columns: %s", exec.calls[1].query) + if strings.Contains(upsertCall.query, "metric_key") || strings.Contains(upsertCall.query, "metric_unit") { + t.Fatalf("daily mileage upsert should not use generic metric columns: %s", upsertCall.query) } - 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 !strings.Contains(upsertCall.query, "first_total_mileage_km <= 0") { + t.Fatalf("upsert should ignore legacy zero first mileage: %s", upsertCall.query) } - if got := exec.calls[1].args[0]; got != "LNBVIN00000000002" { + if !strings.Contains(upsertCall.query, "trusted_source_key") { + t.Fatalf("upsert should track trusted source: %s", upsertCall.query) + } + if got := upsertCall.args[0]; got != "LNBVIN00000000002" { t.Fatalf("first upsert arg should be vin, got %#v", got) } } diff --git a/go/vehicle-gateway/internal/stats/schema.go b/go/vehicle-gateway/internal/stats/schema.go index eb846a66..502d94d4 100644 --- a/go/vehicle-gateway/internal/stats/schema.go +++ b/go/vehicle-gateway/internal/stats/schema.go @@ -7,9 +7,18 @@ const DailyMileageTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_daily_mileage ( daily_mileage_km DECIMAL(18,3) NOT NULL DEFAULT 0, first_total_mileage_km DECIMAL(18,3) NULL, latest_total_mileage_km DECIMAL(18,3) NULL, + trusted_source_key VARCHAR(256) NULL, + trusted_phone VARCHAR(32) NULL, + trusted_source_endpoint VARCHAR(128) NULL, sample_count BIGINT NOT NULL DEFAULT 0, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, PRIMARY KEY (vin, stat_date, protocol), KEY idx_stat_date (stat_date), KEY idx_protocol_date (protocol, stat_date) )` + +var DailyMileageAlterSQL = []string{ + "ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_key VARCHAR(256) NULL", + "ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_phone VARCHAR(32) NULL", + "ALTER TABLE vehicle_daily_mileage ADD COLUMN trusted_source_endpoint VARCHAR(128) NULL", +}