shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(stats): preserve realtime baseline from current candidate

Browse Source
This commit is contained in:
lingniu
2026-07-08 16:19:50 +08:00
parent bd67034927
commit 44e119331d
3 changed files with 144 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
go/vehicle-gateway/internal/stats/daily_metric.go
View File

@@ -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
}

92
go/vehicle-gateway/internal/stats/daily_metric_test.go
View File

@@ -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))

41
go/vehicle-gateway/internal/stats/source_mileage.go
View File

@@ -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
`