diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 672220fb..7a20380a 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -74,8 +74,11 @@ type dailySourceLast struct { Phone string DeviceID string SourceEndpoint string + FirstTS time.Time TS time.Time + FirstTotalKM float64 TotalKM float64 + RawSampleCount int64 } func main() { @@ -233,21 +236,17 @@ func addSamples(aggregates map[string]*metricAgg, samples []stats.MetricSample) SourceEndpoint: sample.SourceEndpoint, FirstEventTime: sample.EventTime, LatestEventTime: sample.EventTime, - QualityStatus: stats.QualityNoPreviousBaseline, - QualityReason: "scan_backfill_missing_previous_source", + QualityStatus: stats.QualityOK, + QualityReason: "current_day_first_sample", } continue } - if sample.TotalMileageKM < agg.FirstKM { - agg.FirstKM = sample.TotalMileageKM - } - if sample.TotalMileageKM > agg.LatestKM { - agg.LatestKM = sample.TotalMileageKM - } if agg.FirstEventTime.IsZero() || sample.EventTime.Before(agg.FirstEventTime) { + agg.FirstKM = sample.TotalMileageKM agg.FirstEventTime = sample.EventTime } if sample.EventTime.After(agg.LatestEventTime) { + agg.LatestKM = sample.TotalMileageKM agg.LatestEventTime = sample.EventTime agg.SourceEndpoint = sample.SourceEndpoint if strings.TrimSpace(sample.Phone) != "" { @@ -339,22 +338,6 @@ func clearBackfillTargetMileage(ctx context.Context, db *sql.DB, vin string, sta } func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[string]*metricAgg, error) { - dates, err := dateRangeWithPrevious(cfg.DateFrom, cfg.DateTo) - if err != nil { - return nil, err - } - lastByProtocolDate := map[envelope.Protocol]map[string]map[string][]dailySourceLast{} - for _, protocol := range cfg.Protocols { - lastByProtocolDate[protocol] = map[string]map[string][]dailySourceLast{} - for _, date := range dates { - rows, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date) - if err != nil { - return nil, err - } - lastByProtocolDate[protocol][date] = rows - slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(rows)) - } - } targetDates, err := dateRange(cfg.DateFrom, cfg.DateTo) if err != nil { return nil, err @@ -362,9 +345,15 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s aggregates := map[string]*metricAgg{} for _, protocol := range cfg.Protocols { for _, date := range targetDates { - prevDate := previousDate(date) - current := lastByProtocolDate[protocol][date] - previous := lastByProtocolDate[protocol][prevDate] + current, err := queryDailyLastSourceRows(ctx, db, cfg, protocol, date) + if err != nil { + return nil, err + } + previous, err := queryPreviousLastSourceRows(ctx, db, cfg, protocol, date) + if err != nil { + return nil, err + } + slog.Info("daily last loaded", "protocol", protocol, "date", date, "vehicles", len(current), "previousVehicles", len(previous)) for vin, currentSources := range current { previousBySource := map[string]dailySourceLast{} for _, previousRow := range previous[vin] { @@ -372,29 +361,8 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s } for _, currentRow := range currentSources { key := vin + "|" + date + "|" + string(protocol) + "|" + currentRow.SourceKey - agg := &metricAgg{ - VIN: vin, - Date: date, - Protocol: protocol, - FirstKM: currentRow.TotalKM, - LatestKM: currentRow.TotalKM, - Count: 1, - SourceKey: currentRow.SourceKey, - Phone: currentRow.Phone, - DeviceID: currentRow.DeviceID, - SourceEndpoint: currentRow.SourceEndpoint, - FirstEventTime: currentRow.TS, - LatestEventTime: currentRow.TS, - QualityStatus: stats.QualityNoPreviousBaseline, - QualityReason: "missing_previous_source", - } - if previousRow, ok := previousBySource[currentRow.SourceKey]; ok { - agg.FirstKM = previousRow.TotalKM - agg.FirstEventTime = previousRow.TS - agg.QualityStatus = stats.QualityOK - agg.QualityReason = "same_source_previous_day" - } - aggregates[key] = agg + previousRow, hasPrevious := previousBySource[currentRow.SourceKey] + aggregates[key] = aggregateFromDailySource(date, protocol, currentRow, previousRow, hasPrevious) } } } @@ -402,6 +370,40 @@ func buildLastDiffAggregates(ctx context.Context, db *sql.DB, cfg config) (map[s return aggregates, nil } +func aggregateFromDailySource(date string, protocol envelope.Protocol, current dailySourceLast, previous dailySourceLast, hasPrevious bool) *metricAgg { + firstKM := current.FirstTotalKM + firstEventTime := current.FirstTS + qualityReason := "current_day_first_sample" + if hasPrevious { + firstKM = previous.TotalKM + firstEventTime = previous.TS + qualityReason = "historical_source_baseline" + } + if firstEventTime.IsZero() { + firstEventTime = current.TS + } + count := current.RawSampleCount + if count <= 0 { + count = 1 + } + return &metricAgg{ + VIN: current.VIN, + Date: date, + Protocol: protocol, + FirstKM: firstKM, + LatestKM: current.TotalKM, + Count: count, + SourceKey: current.SourceKey, + Phone: current.Phone, + DeviceID: current.DeviceID, + SourceEndpoint: current.SourceEndpoint, + FirstEventTime: firstEventTime, + LatestEventTime: current.TS, + QualityStatus: stats.QualityOK, + QualityReason: qualityReason, + } +} + type trustedChoice struct { current dailySourceLast previous dailySourceLast @@ -443,10 +445,30 @@ func queryDailyLastSourceRows(ctx context.Context, db *sql.DB, cfg config, proto fmt.Sprintf("protocol = '%s'", quote(string(protocol))), realtimeMileageFramePredicate(), } - sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json) + 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 GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) + return querySourceRows(ctx, db, cfg, protocol, sqlText, true) +} + +func queryPreviousLastSourceRows(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)), + "parse_status = 'OK'", + "vin IS NOT NULL", + "vin <> ''", + fmt.Sprintf("protocol = '%s'", quote(string(protocol))), + realtimeMileageFramePredicate(), + } + sqlText := fmt.Sprintf(`SELECT vin, phone, device_id, source_endpoint, LAST(ts), LAST(parsed_json), COUNT(*) +FROM %s.raw_frames +WHERE %s +GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), strings.Join(where, " AND ")) + return querySourceRows(ctx, db, cfg, protocol, sqlText, false) +} + +func querySourceRows(ctx context.Context, db *sql.DB, cfg config, protocol envelope.Protocol, sqlText string, includeFirst bool) (map[string][]dailySourceLast, error) { rows, err := db.QueryContext(ctx, sqlText) if err != nil { return nil, err @@ -458,25 +480,32 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s var phone string var deviceID string var sourceEndpoint string + var firstTS time.Time + var firstParsedJSON string var ts time.Time var parsedJSON string - if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON); err != nil { - return nil, err + var rawSampleCount int64 + if includeFirst { + if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &firstTS, &firstParsedJSON, &ts, &parsedJSON, &rawSampleCount); err != nil { + return nil, err + } + } else { + if err := rows.Scan(&vin, &phone, &deviceID, &sourceEndpoint, &ts, &parsedJSON, &rawSampleCount); err != nil { + return nil, err + } + firstTS = ts + firstParsedJSON = parsedJSON } - fields := fieldsForStats(protocol, vin, parsedJSON) - env := envelope.FrameEnvelope{ - Protocol: protocol, - VIN: strings.TrimSpace(vin), - EventTimeMS: time.Now().UnixMilli(), - ReceivedAtMS: time.Now().UnixMilli(), - Fields: fields, - ParsedFields: fields, - ParseStatus: envelope.ParseOK, - } - samples, err := stats.SamplesFromEnvelope(env, cfg.Location) - if err != nil || len(samples) == 0 { + latestTotalKM, ok := mileageFromParsed(protocol, vin, parsedJSON, ts, cfg.Location) + if !ok { continue } + firstTotalKM := latestTotalKM + if includeFirst { + if parsedFirst, ok := mileageFromParsed(protocol, vin, firstParsedJSON, firstTS, cfg.Location); ok { + firstTotalKM = parsedFirst + } + } sourceKey := normalizedSourceKey(string(protocol), phone, deviceID, sourceEndpoint) if sourceKey == "" { continue @@ -487,8 +516,11 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s Phone: strings.TrimSpace(phone), DeviceID: strings.TrimSpace(deviceID), SourceEndpoint: strings.TrimSpace(sourceEndpoint), + FirstTS: firstTS.In(cfg.Location), TS: ts, - TotalKM: samples[0].TotalMileageKM, + FirstTotalKM: firstTotalKM, + TotalKM: latestTotalKM, + RawSampleCount: rawSampleCount, } key := row.VIN + "|" + row.SourceKey if existing, ok := latestBySource[key]; !ok || row.TS.After(existing.TS) { @@ -505,6 +537,24 @@ GROUP BY vin, phone, device_id, source_endpoint`, ident(cfg.TDengineDatabase), s return out, nil } +func mileageFromParsed(protocol envelope.Protocol, vin string, parsedJSON string, eventTime time.Time, loc *time.Location) (float64, bool) { + fields := fieldsForStats(protocol, vin, parsedJSON) + env := envelope.FrameEnvelope{ + Protocol: protocol, + VIN: strings.TrimSpace(vin), + EventTimeMS: eventTime.UnixMilli(), + ReceivedAtMS: eventTime.UnixMilli(), + Fields: fields, + ParsedFields: fields, + ParseStatus: envelope.ParseOK, + } + samples, err := stats.SamplesFromEnvelope(env, loc) + if err != nil || len(samples) == 0 { + return 0, false + } + return samples[0].TotalMileageKM, true +} + func normalizedSourceKey(protocol string, phone string, deviceID string, endpoint string) string { sourceIP := stats.NormalizeSourceIP(endpoint) return stats.SourceKey(envelope.Protocol(protocol), phone, deviceID, sourceIP) diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index 3d42e7f1..d118432d 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -67,6 +67,69 @@ func TestDailySourceLastBuildsCandidateKeysBySourceIP(t *testing.T) { } } +func TestAggregateFromDailySourceUsesOlderHistoricalBaseline(t *testing.T) { + current := dailySourceLast{ + VIN: "LMRKH9AC2R1004087", + SourceKey: normalizedSourceKey("YUTONG_MQTT", "", "LMRKH9AC2R1004087", "mqtt://yutong/ytforward/shln/3"), + DeviceID: "LMRKH9AC2R1004087", + SourceEndpoint: "mqtt://yutong/ytforward/shln/3", + FirstTS: time.Date(2026, 7, 8, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + TS: time.Date(2026, 7, 8, 17, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + FirstTotalKM: 120778, + TotalKM: 120788, + RawSampleCount: 15, + } + previous := dailySourceLast{ + VIN: "LMRKH9AC2R1004087", + SourceKey: current.SourceKey, + DeviceID: "LMRKH9AC2R1004087", + SourceEndpoint: "mqtt://yutong/ytforward/shln/3", + TS: time.Date(2026, 7, 4, 23, 58, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + TotalKM: 120672, + } + + agg := aggregateFromDailySource("2026-07-08", envelope.ProtocolYutongMQTT, current, previous, true) + + if agg.FirstKM != 120672 || agg.LatestKM != 120788 { + t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM) + } + if agg.FirstEventTime != previous.TS || agg.LatestEventTime != current.TS { + t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime) + } + if agg.QualityStatus != stats.QualityOK || agg.QualityReason != "historical_source_baseline" { + t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason) + } + if agg.Count != 15 { + t.Fatalf("sample count = %d, want 15", agg.Count) + } +} + +func TestAggregateFromDailySourceUsesCurrentFirstSampleWithoutHistory(t *testing.T) { + current := dailySourceLast{ + VIN: "LMRKH9AC2R1004087", + SourceKey: normalizedSourceKey("YUTONG_MQTT", "", "LMRKH9AC2R1004087", "mqtt://yutong/ytforward/shln/3"), + DeviceID: "LMRKH9AC2R1004087", + SourceEndpoint: "mqtt://yutong/ytforward/shln/3", + FirstTS: time.Date(2026, 7, 8, 8, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + TS: time.Date(2026, 7, 8, 17, 0, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)), + FirstTotalKM: 120778, + TotalKM: 120788, + RawSampleCount: 15, + } + + agg := aggregateFromDailySource("2026-07-08", envelope.ProtocolYutongMQTT, current, dailySourceLast{}, false) + + if agg.FirstKM != 120778 || agg.LatestKM != 120788 { + t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM) + } + if agg.FirstEventTime != current.FirstTS || agg.LatestEventTime != current.TS { + t.Fatalf("event range = %v -> %v", agg.FirstEventTime, agg.LatestEventTime) + } + if agg.QualityStatus != stats.QualityOK || agg.QualityReason != "current_day_first_sample" { + t.Fatalf("quality = %s/%s", agg.QualityStatus, agg.QualityReason) + } +} + func TestClearBackfillTargetMileageClearsExactKey(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { @@ -219,9 +282,9 @@ func TestWriteAggregatesSkipsBlankSourceBeforeClearingTarget(t *testing.T) { } } -func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) { +func TestAddSamplesUsesCurrentDayFirstSampleBaseline(t *testing.T) { loc := time.FixedZone("Asia/Shanghai", 8*3600) - samples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{ + firstSamples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, VIN: "LA9GG64L7PBAF4001", Phone: "13307765812", @@ -234,18 +297,38 @@ func TestAddSamplesKeepsScanCandidatesNoPreviousBaseline(t *testing.T) { if err != nil { t.Fatalf("SamplesFromEnvelope() error = %v", err) } + secondSamples, err := stats.SamplesFromEnvelope(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: time.Date(2026, 7, 8, 13, 0, 0, 0, loc).UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4129.9, + }, + }, loc) + if err != nil { + t.Fatalf("SamplesFromEnvelope() error = %v", err) + } aggregates := map[string]*metricAgg{} - addSamples(aggregates, samples) + addSamples(aggregates, firstSamples) + addSamples(aggregates, secondSamples) if len(aggregates) != 1 { t.Fatalf("aggregate count = %d, want 1", len(aggregates)) } for _, agg := range aggregates { - if agg.QualityStatus != stats.QualityNoPreviousBaseline { - t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityNoPreviousBaseline) + if agg.QualityStatus != stats.QualityOK { + t.Fatalf("quality = %q, want %q", agg.QualityStatus, stats.QualityOK) } - if agg.QualityReason != "scan_backfill_missing_previous_source" { + if agg.QualityReason != "current_day_first_sample" { t.Fatalf("quality reason = %q", agg.QualityReason) } + if agg.FirstKM != 4123.9 || agg.LatestKM != 4129.9 { + t.Fatalf("km range = %v -> %v", agg.FirstKM, agg.LatestKM) + } + if agg.FirstEventTime != time.Date(2026, 7, 8, 12, 0, 0, 0, loc) { + t.Fatalf("first event time = %v", agg.FirstEventTime) + } } } diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index ae4dd5ac..f80338b9 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -182,15 +182,18 @@ func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMil return err } if !found { - candidate.QualityStatus = QualityNoPreviousBaseline - candidate.QualityReason = "missing_previous_source" + candidate.QualityStatus = QualityOK + candidate.QualityReason = "current_day_first_sample" return nil } candidate.FirstTotalKM = baseline.LatestTotalKM candidate.FirstEventTime = baseline.LatestEventTime candidate.DailyKM = candidate.LatestTotalKM - baseline.LatestTotalKM candidate.QualityStatus = QualityOK - candidate.QualityReason = "same_source_previous_day" + candidate.QualityReason = baseline.QualityReason + if candidate.QualityReason == "" { + candidate.QualityReason = "historical_source_baseline" + } if candidate.DailyKM < 0 || candidate.DailyKM > maxSelectedDailyMileageKM { candidate.QualityStatus = QualityInvalidDelta candidate.QualityReason = "outside_daily_range" diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 08a12179..ccaaae7d 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -278,7 +278,7 @@ func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) { } } -func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *testing.T) { +func TestWriterAppendUsesCurrentSampleWhenNoHistoricalBaseline(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) @@ -303,7 +303,7 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`). - WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135"). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135"). WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"})) mock.ExpectQuery(`SELECT first_total_mileage_km, first_event_time FROM vehicle_daily_mileage_source`). WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135"). @@ -325,8 +325,8 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes int64(1), eventTime, eventTime, - QualityNoPreviousBaseline, - "missing_previous_source", + QualityOK, + "current_day_first_sample", ). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`). @@ -334,7 +334,7 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`). WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", int64(1000)). - WillReturnResult(sqlmock.NewResult(0, 0)) + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`). WithArgs( "LA9GG64L7PBAF4001", @@ -345,7 +345,7 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes "2026-07-08", "JT808", ). - WillReturnResult(sqlmock.NewResult(0, 0)) + WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`). WithArgs( "LA9GG64L7PBAF4001", @@ -355,7 +355,7 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes "2026-07-08", "JT808", ). - WillReturnResult(sqlmock.NewResult(0, 1)) + WillReturnResult(sqlmock.NewResult(0, 0)) if err := writer.Append(context.Background(), event); err != nil { t.Fatalf("Append() error = %v", err) @@ -391,7 +391,7 @@ func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`). - WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135"). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135"). WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}). AddRow(4100.8, previousTime)) mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`). @@ -412,7 +412,7 @@ func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T previousTime, currentTime, QualityOK, - "same_source_previous_day", + "historical_source_baseline", ). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`). @@ -451,6 +451,92 @@ func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T } } +func TestWriterAppendUsesOlderHistoricalSourceBaseline(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) + writer := NewWriter(db, loc) + historicalTime := time.Date(2026, 7, 4, 23, 58, 0, 0, loc) + currentTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + VIN: "LMRKH9AC2R1004087", + DeviceID: "LMRKH9AC2R1004087", + SourceEndpoint: "mqtt://yutong/ytforward/shln/3", + EventTimeMS: currentTime.UnixMilli(), + Fields: map[string]any{ + "yutong_mqtt.data.total_mileage": 120788000, + }, + } + + mock.ExpectExec(`INSERT INTO vehicle_data_source`). + WithArgs("YUTONG_MQTT", "mqtt", "mqtt://yutong/ytforward/shln/3", sqlmock.AnyArg(), sqlmock.AnyArg()). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`). + WithArgs("LMRKH9AC2R1004087", "2026-07-08", "YUTONG_MQTT", "YUTONG_MQTT:LMRKH9AC2R1004087@mqtt"). + WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"}). + AddRow(120672.0, historicalTime)) + mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`). + WithArgs( + "LMRKH9AC2R1004087", + "2026-07-08", + "YUTONG_MQTT", + "YUTONG_MQTT:LMRKH9AC2R1004087@mqtt", + "mqtt", + "mqtt://yutong/ytforward/shln/3", + "", + "LMRKH9AC2R1004087", + "", + float64(120672.0), + float64(120788.0), + approxFloat64{want: 116.0, tolerance: 0.000001}, + int64(1), + historicalTime, + currentTime, + QualityOK, + "historical_source_baseline", + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`). + WithArgs("LMRKH9AC2R1004087", "2026-07-08", "YUTONG_MQTT"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec(`INSERT INTO vehicle_daily_mileage`). + WithArgs("LMRKH9AC2R1004087", "2026-07-08", "YUTONG_MQTT", int64(1000)). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(`UPDATE vehicle_daily_mileage_source s`). + WithArgs( + "LMRKH9AC2R1004087", + "2026-07-08", + "YUTONG_MQTT", + int64(1000), + "LMRKH9AC2R1004087", + "2026-07-08", + "YUTONG_MQTT", + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`). + WithArgs( + "LMRKH9AC2R1004087", + "2026-07-08", + "YUTONG_MQTT", + "LMRKH9AC2R1004087", + "2026-07-08", + "YUTONG_MQTT", + ). + WillReturnResult(sqlmock.NewResult(0, 0)) + + if err := writer.Append(context.Background(), event); err != nil { + t.Fatalf("Append() error = %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestWriterAppendUsesCurrentOKCandidateWhenPreviousBaselineMissing(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { @@ -477,7 +563,7 @@ func TestWriterAppendUsesCurrentOKCandidateWhenPreviousBaselineMissing(t *testin WithArgs("JT808", "115.231.168.135", "115.231.168.135:20215", sqlmock.AnyArg(), sqlmock.AnyArg()). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectQuery(`SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source`). - WithArgs("LA9GG64L7PBAF4001", "2026-07-07", "JT808", "JT808:13307765812@115.231.168.135"). + WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135"). WillReturnRows(sqlmock.NewRows([]string{"latest_total_mileage_km", "latest_event_time"})) mock.ExpectQuery(`SELECT first_total_mileage_km, first_event_time FROM vehicle_daily_mileage_source`). WithArgs("LA9GG64L7PBAF4001", "2026-07-08", "JT808", "JT808:13307765812@115.231.168.135"). @@ -501,7 +587,7 @@ func TestWriterAppendUsesCurrentOKCandidateWhenPreviousBaselineMissing(t *testin firstTime, currentTime, QualityOK, - "same_source_previous_day", + "current_day_first_sample", ). WillReturnResult(sqlmock.NewResult(0, 1)) mock.ExpectExec(`UPDATE vehicle_daily_mileage_source`). diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index f5c4c857..4d433d3d 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -273,17 +273,14 @@ WHERE vin = ? AND stat_date = ? AND protocol = ? type sourceBaseline struct { LatestTotalKM float64 LatestEventTime time.Time + QualityReason string } func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string, statDate string, protocol envelope.Protocol, sourceKey string) (sourceBaseline, bool, error) { if query == nil || strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" || strings.TrimSpace(sourceKey) == "" { return sourceBaseline{}, false, nil } - previousDate, ok := previousStatDate(statDate) - if !ok { - return sourceBaseline{}, false, nil - } - rows, err := query.QueryContext(ctx, previousSourceBaselineSQL, vin, previousDate, string(protocol), sourceKey) + rows, err := query.QueryContext(ctx, previousSourceBaselineSQL, vin, statDate, string(protocol), sourceKey) if err != nil { return sourceBaseline{}, false, err } @@ -305,6 +302,7 @@ func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string return sourceBaseline{ LatestTotalKM: latestTotal.Float64, LatestEventTime: latestEvent.Time, + QualityReason: "historical_source_baseline", }, latestTotal.Valid, nil } @@ -334,25 +332,19 @@ func lookupCurrentSourceBaseline(ctx context.Context, query Queryer, vin string, return sourceBaseline{ LatestTotalKM: firstTotal.Float64, LatestEventTime: firstEvent.Time, + QualityReason: "current_day_first_sample", }, firstTotal.Valid, nil } -func previousStatDate(statDate string) (string, bool) { - day, err := time.Parse("2006-01-02", strings.TrimSpace(statDate)) - if err != nil { - return "", false - } - return day.AddDate(0, 0, -1).Format("2006-01-02"), true -} - const previousSourceBaselineSQL = ` SELECT latest_total_mileage_km, latest_event_time FROM vehicle_daily_mileage_source WHERE vin = ? - AND stat_date = ? + AND stat_date < ? AND protocol = ? AND source_key = ? -ORDER BY latest_event_time DESC + AND quality_status = '` + QualityOK + `' +ORDER BY stat_date DESC, latest_event_time DESC LIMIT 1 `