diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index c1657cdc..5e37e461 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -177,10 +177,18 @@ func (w *Writer) previousBaseline(ctx context.Context, candidate SourceMileageSa if err != nil { return sourceBaseline{}, false, err } + if !found { + baseline, found, err = lookupCurrentSourceBaseline(ctx, w.query, candidate.VIN, candidate.StatDate, candidate.Protocol, candidate.SourceKey) + if err != nil { + return sourceBaseline{}, false, err + } + } - w.mu.Lock() - w.baselineCache[cacheKey] = sourceBaselineCacheEntry{baseline: baseline, found: found} - w.mu.Unlock() + if found { + w.mu.Lock() + w.baselineCache[cacheKey] = sourceBaselineCacheEntry{baseline: baseline, found: true} + w.mu.Unlock() + } return baseline, found, nil } diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 50d33733..b9725f39 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -278,6 +278,9 @@ func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(t *tes 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"). 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"). + WillReturnRows(sqlmock.NewRows([]string{"first_total_mileage_km", "first_event_time"})) mock.ExpectExec(`INSERT INTO vehicle_daily_mileage_source`). WithArgs( "LA9GG64L7PBAF4001", @@ -421,6 +424,95 @@ func TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(t *testing.T } } +func TestWriterAppendUsesCurrentOKCandidateWhenPreviousBaselineMissing(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) + firstTime := time.Date(2026, 7, 7, 23, 58, 0, 0, loc) + currentTime := time.Date(2026, 7, 8, 15, 56, 0, 0, loc) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: currentTime.UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4136.8, + }, + } + + 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.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"). + 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"). + WillReturnRows(sqlmock.NewRows([]string{"first_total_mileage_km", "first_event_time"}). + AddRow(4100.8, firstTime)) + 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.8), + float64(4136.8), + approxFloat64{want: 36.0, tolerance: 0.000001}, + int64(1), + firstTime, + currentTime, + QualityOK, + "same_source_previous_day", + ). + 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(0, 1)) + 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)) + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`). + WithArgs( + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + ). + 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 TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index 01976899..1abbeb82 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -318,6 +318,35 @@ func lookupPreviousSourceBaseline(ctx context.Context, query Queryer, vin string }, latestTotal.Valid, nil } +func lookupCurrentSourceBaseline(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 + } + rows, err := query.QueryContext(ctx, currentSourceBaselineSQL, vin, statDate, string(protocol), sourceKey) + if err != nil { + return sourceBaseline{}, false, err + } + defer rows.Close() + if !rows.Next() { + if err := rows.Err(); err != nil { + return sourceBaseline{}, false, err + } + return sourceBaseline{}, false, nil + } + var firstTotal sql.NullFloat64 + var firstEvent sql.NullTime + if err := rows.Scan(&firstTotal, &firstEvent); err != nil { + return sourceBaseline{}, false, err + } + if err := rows.Err(); err != nil { + return sourceBaseline{}, false, err + } + return sourceBaseline{ + LatestTotalKM: firstTotal.Float64, + LatestEventTime: firstEvent.Time, + }, firstTotal.Valid, nil +} + func previousStatDate(statDate string) (string, bool) { day, err := time.Parse("2006-01-02", strings.TrimSpace(statDate)) if err != nil { @@ -336,3 +365,15 @@ WHERE vin = ? ORDER BY latest_event_time DESC LIMIT 1 ` + +const currentSourceBaselineSQL = ` +SELECT first_total_mileage_km, first_event_time +FROM vehicle_daily_mileage_source +WHERE vin = ? + AND stat_date = ? + AND protocol = ? + AND source_key = ? + AND quality_status = '` + QualityOK + `' +ORDER BY latest_event_time DESC +LIMIT 1 +`