diff --git a/.superpowers/sdd/final-review-fix-report.md b/.superpowers/sdd/final-review-fix-report.md new file mode 100644 index 00000000..376c5884 --- /dev/null +++ b/.superpowers/sdd/final-review-fix-report.md @@ -0,0 +1,47 @@ +## final-review-fix (2026-07-08 15:41 CST) + +### Changes + +- Made realtime duplicate suppression source-aware in `go/vehicle-gateway/internal/stats/daily_metric.go` by keying the in-memory mileage cache with `vin + protocol + source_key + stat_date`, so identical mileage from different sources is retained while same-source repeats are still skipped. +- Added optional previous-baseline querying and per-source/day caching in `Writer`; realtime candidates now: + - become `NO_PREVIOUS_BASELINE` with reason `missing_previous_source` when the same-source previous-day candidate is absent, + - use previous-day `latest_total_mileage_km` as `first_total_mileage_km`, + - compute `daily_mileage_km` from previous-day baseline when present, + - become `INVALID_DELTA` with reason `outside_daily_range` when the delta is negative or above `maxSelectedDailyMileageKM`. +- Added generic stale-final cleanup to `ProjectDailyMileage` in `go/vehicle-gateway/internal/stats/source_mileage.go` so `vehicle_daily_mileage` is deleted when no candidate remains selected for the exact `(vin, stat_date, protocol)` target. +- Changed `BACKFILL_METHOD` default from `scan` to `last_diff` in `go/vehicle-gateway/cmd/stats-backfill/main.go`. +- Expanded regression coverage for source-aware realtime dedupe, realtime baseline/no-baseline behavior, projector cleanup SQL, stale-final cleanup through backfill, and the backfill method default. + +### Test Results + +- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./internal/stats -run 'TestWriter|TestProjectDailyMileage|TestSource' -count=1` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats 0.435s` +- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./cmd/stats-backfill ./internal/stats -count=1` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.643s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats 0.279s` +- `cd /Users/lingniu/project/ai-coding/lingniu-vehicle-ingest/go/vehicle-gateway && go test ./... -count=1` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/capacity-check 0.419s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/gateway 0.397s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/history-writer 0.728s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/load-sim 0.698s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/nats-fast-writer 0.688s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/nats-kafka-bridge 0.683s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/realtime-api 0.574s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stat-writer 0.534s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/cmd/stats-backfill 0.634s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/capacity 0.834s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope 0.738s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/eventbus 0.758s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/gateway 0.483s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/health 0.384s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/history 0.502s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity 0.631s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/loadsim 0.640s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/metrics 0.626s` + - `? lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability [no test files]` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/gb32960 0.622s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/jt808 0.371s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/protocol/yutongmqtt 0.384s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/realtime 0.496s` + - `ok lingniu-vehicle-ingest/go/vehicle-gateway/internal/stats 0.443s` + - `? lingniu-vehicle-ingest/go/vehicle-gateway/internal/topics [no test files]` diff --git a/go/vehicle-gateway/cmd/stats-backfill/main.go b/go/vehicle-gateway/cmd/stats-backfill/main.go index 1d6672cb..910643b8 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main.go @@ -579,7 +579,7 @@ func loadConfig() (config, error) { DateFrom: dateFrom, DateTo: dateTo, Protocols: protocols, - Method: env("BACKFILL_METHOD", "scan"), + Method: env("BACKFILL_METHOD", "last_diff"), Limit: envInt("BACKFILL_LIMIT", 0), DryRun: envBool("BACKFILL_DRY_RUN", true), Reset: envBool("BACKFILL_RESET", false), diff --git a/go/vehicle-gateway/cmd/stats-backfill/main_test.go b/go/vehicle-gateway/cmd/stats-backfill/main_test.go index 2429b62d..80a27e41 100644 --- a/go/vehicle-gateway/cmd/stats-backfill/main_test.go +++ b/go/vehicle-gateway/cmd/stats-backfill/main_test.go @@ -160,6 +160,16 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T) "2026-07-08", "JT808", ). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`). + WithArgs( + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + ). WillReturnResult(sqlmock.NewResult(0, 1)) written, err := writeAggregates(context.Background(), db, aggregates, 500) @@ -173,3 +183,19 @@ func TestWriteAggregatesClearsTargetRowsBeforeStaleCandidateUpsert(t *testing.T) t.Fatalf("sql expectations: %v", err) } } + +func TestLoadConfigDefaultsBackfillMethodToLastDiff(t *testing.T) { + t.Setenv("BACKFILL_METHOD", "") + t.Setenv("BACKFILL_DATE_FROM", "2026-07-08") + t.Setenv("BACKFILL_DATE_TO", "2026-07-08") + t.Setenv("BACKFILL_PROTOCOLS", "JT808") + t.Setenv("LOCAL_TZ", "Asia/Shanghai") + + cfg, err := loadConfig() + if err != nil { + t.Fatalf("loadConfig() error = %v", err) + } + if cfg.Method != "last_diff" { + t.Fatalf("method = %q, want last_diff", cfg.Method) + } +} diff --git a/go/vehicle-gateway/internal/stats/daily_metric.go b/go/vehicle-gateway/internal/stats/daily_metric.go index dcef0def..41be538c 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric.go +++ b/go/vehicle-gateway/internal/stats/daily_metric.go @@ -19,9 +19,11 @@ type Execer interface { type Writer struct { exec Execer + query Queryer loc *time.Location mu sync.Mutex lastTotalMileage map[string]float64 + baselineCache map[string]sourceBaselineCacheEntry } type MetricSample struct { @@ -43,7 +45,16 @@ func NewWriter(exec Execer, loc *time.Location) *Writer { if loc == nil { loc = time.FixedZone("Asia/Shanghai", 8*3600) } - return &Writer{exec: exec, loc: loc, lastTotalMileage: map[string]float64{}} + writer := &Writer{ + exec: exec, + loc: loc, + lastTotalMileage: map[string]float64{}, + baselineCache: map[string]sourceBaselineCacheEntry{}, + } + if query, ok := exec.(Queryer); ok { + writer.query = query + } + return writer } func (w *Writer) EnsureSchema(ctx context.Context) error { @@ -79,6 +90,9 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { return err } candidate := SourceMileageSampleFromMetric(sample, identity) + if err := w.applyRealtimeBaseline(ctx, &candidate); err != nil { + return err + } if err := UpsertSourceMileage(ctx, w.exec, candidate); err != nil { return err } @@ -92,8 +106,7 @@ func (w *Writer) Append(ctx context.Context, env envelope.FrameEnvelope) error { } func (w *Writer) seenSameMileage(sample MetricSample) bool { - prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol) - key := prefix + sample.StatDate + key := mileageCacheKey(sample) w.mu.Lock() defer w.mu.Unlock() if last, ok := w.lastTotalMileage[key]; ok && last == sample.TotalMileageKM { @@ -103,8 +116,8 @@ func (w *Writer) seenSameMileage(sample MetricSample) bool { } func (w *Writer) markMileageWritten(sample MetricSample) { - prefix := fmt.Sprintf("%s|%s|", sample.VIN, sample.Protocol) - key := prefix + sample.StatDate + prefix := mileageCachePrefix(sample) + key := mileageCacheKey(sample) w.mu.Lock() defer w.mu.Unlock() for existing := range w.lastTotalMileage { @@ -112,9 +125,77 @@ func (w *Writer) markMileageWritten(sample MetricSample) { delete(w.lastTotalMileage, existing) } } + for existing := range w.baselineCache { + if strings.HasPrefix(existing, prefix) && existing != key { + delete(w.baselineCache, existing) + } + } w.lastTotalMileage[key] = sample.TotalMileageKM } +func (w *Writer) applyRealtimeBaseline(ctx context.Context, candidate *SourceMileageSample) error { + if candidate == nil { + return nil + } + baseline, found, err := w.previousBaseline(ctx, *candidate) + if err != nil { + return err + } + if !found { + candidate.QualityStatus = QualityNoPreviousBaseline + candidate.QualityReason = "missing_previous_source" + 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" + if candidate.DailyKM < 0 || candidate.DailyKM > maxSelectedDailyMileageKM { + candidate.QualityStatus = QualityInvalidDelta + candidate.QualityReason = "outside_daily_range" + } + return nil +} + +func (w *Writer) previousBaseline(ctx context.Context, candidate SourceMileageSample) (sourceBaseline, bool, error) { + cacheKey := mileageCacheKey(MetricSample{ + VIN: candidate.VIN, + Protocol: candidate.Protocol, + StatDate: candidate.StatDate, + SourceKey: candidate.SourceKey, + }) + w.mu.Lock() + if cached, ok := w.baselineCache[cacheKey]; ok { + w.mu.Unlock() + return cached.baseline, cached.found, nil + } + w.mu.Unlock() + + baseline, found, err := lookupPreviousSourceBaseline(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() + return baseline, found, nil +} + +func mileageCachePrefix(sample MetricSample) string { + return fmt.Sprintf("%s|%s|%s|", sample.VIN, sample.Protocol, sample.SourceKey) +} + +func mileageCacheKey(sample MetricSample) string { + return mileageCachePrefix(sample) + sample.StatDate +} + +type sourceBaselineCacheEntry struct { + baseline sourceBaseline + found bool +} + func SamplesFromEnvelope(env envelope.FrameEnvelope, loc *time.Location) ([]MetricSample, error) { vin := strings.TrimSpace(env.VIN) if vin == "" { diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index efd9c070..6840b559 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -3,11 +3,14 @@ package stats import ( "context" "database/sql" + "database/sql/driver" "errors" + "math" "strings" "testing" "time" + "github.com/DATA-DOG/go-sqlmock" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) @@ -173,8 +176,8 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) { if err := writer.Append(context.Background(), event); err != nil { t.Fatalf("Append() error = %v", err) } - if len(exec.calls) != 5 { - t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark", len(exec.calls)) + if len(exec.calls) != 6 { + t.Fatalf("exec calls = %d, want source + candidate + clear + project + mark + cleanup", len(exec.calls)) } if !strings.Contains(exec.calls[0].query, "INSERT INTO vehicle_data_source") { t.Fatalf("first call should upsert source: %s", exec.calls[0].query) @@ -187,6 +190,216 @@ func TestWriterAppendWritesSourceCandidateAndProjection(t *testing.T) { } } +func TestWriterAppendDedupesRealtimeMileagePerSource(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec, time.FixedZone("Asia/Shanghai", 8*3600)) + base := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + EventTimeMS: time.Date(2026, 7, 8, 13, 20, 0, 0, time.FixedZone("Asia/Shanghai", 8*3600)).UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4123.9, + }, + } + + sourceA := base + sourceA.SourceEndpoint = "115.231.168.135:20215" + if err := writer.Append(context.Background(), sourceA); err != nil { + t.Fatalf("Append(sourceA) error = %v", err) + } + + sourceB := base + sourceB.SourceEndpoint = "115.159.85.149:28316" + sourceB.ReceivedAtMS = sourceB.EventTimeMS + 1000 + if err := writer.Append(context.Background(), sourceB); err != nil { + t.Fatalf("Append(sourceB) error = %v", err) + } + + sourceADuplicate := sourceA + sourceADuplicate.ReceivedAtMS = sourceA.EventTimeMS + 2000 + if err := writer.Append(context.Background(), sourceADuplicate); err != nil { + t.Fatalf("Append(sourceADuplicate) error = %v", err) + } + + if got := countExecQueries(exec.calls, "INSERT INTO vehicle_daily_mileage_source"); got != 2 { + t.Fatalf("candidate upserts = %d, want 2", got) + } + if got := countExecQueries(exec.calls, "INSERT INTO vehicle_data_source"); got != 2 { + t.Fatalf("source upserts = %d, want 2", got) + } +} + +func TestWriterAppendWritesNoPreviousBaselineCandidateAndCleansProjection(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) + eventTime := time.Date(2026, 7, 8, 13, 20, 0, 0, loc) + event := envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "LA9GG64L7PBAF4001", + Phone: "13307765812", + SourceEndpoint: "115.231.168.135:20215", + EventTimeMS: eventTime.UnixMilli(), + Fields: map[string]any{ + "jt808.location.total_mileage_km": 4123.9, + }, + } + + 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.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(4123.9), + float64(4123.9), + float64(0), + int64(1), + eventTime, + eventTime, + QualityNoPreviousBaseline, + "missing_previous_source", + ). + 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, 0)) + 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, 0)) + mock.ExpectExec(`DELETE FROM vehicle_daily_mileage`). + WithArgs( + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + "LA9GG64L7PBAF4001", + "2026-07-08", + "JT808", + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + + 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 TestWriterAppendUsesPreviousSourceBaselineForRealtimeCandidate(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) + previousTime := time.Date(2026, 7, 7, 23, 58, 0, 0, loc) + currentTime := time.Date(2026, 7, 8, 13, 20, 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": 4123.9, + }, + } + + 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"}). + AddRow(4100.8, previousTime)) + 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(4123.9), + approxFloat64{want: 23.1, tolerance: 0.000001}, + int64(1), + previousTime, + 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)) @@ -206,7 +419,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { t.Fatalf("Append() error = %v", err) } - if len(exec.calls) != 11 { + if len(exec.calls) != 12 { t.Fatalf("exec calls = %d", len(exec.calls)) } for i, want := range []string{ @@ -277,8 +490,8 @@ func TestWriterSkipsConsecutiveDuplicateMileageSamples(t *testing.T) { t.Fatalf("duplicate Append() error = %v", err) } - if len(exec.calls) != 3 { - t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + if len(exec.calls) != 4 { + t.Fatalf("exec calls = %d, want 4", len(exec.calls)) } } @@ -301,8 +514,8 @@ func TestWriterDoesNotCacheMileageWhenUpsertFails(t *testing.T) { t.Fatalf("retry Append() error = %v", err) } - if len(exec.calls) != 4 { - t.Fatalf("exec calls = %d, want 4", len(exec.calls)) + if len(exec.calls) != 5 { + t.Fatalf("exec calls = %d, want 5", len(exec.calls)) } } @@ -326,14 +539,24 @@ func TestWriterKeepsOnlyLatestDateInMileageCache(t *testing.T) { t.Fatalf("next-day Append() error = %v", err) } - if len(exec.calls) != 6 { - t.Fatalf("exec calls = %d, want 6", len(exec.calls)) + if len(exec.calls) != 8 { + t.Fatalf("exec calls = %d, want 8", len(exec.calls)) } if len(writer.lastTotalMileage) != 1 { t.Fatalf("cache entries = %d, want 1", len(writer.lastTotalMileage)) } } +func countExecQueries(calls []execCall, fragment string) int { + count := 0 + for _, call := range calls { + if strings.Contains(call.query, fragment) { + count++ + } + } + return count +} + type execCall struct { query string args []any @@ -353,3 +576,16 @@ func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any } return nil, nil } + +type approxFloat64 struct { + want float64 + tolerance float64 +} + +func (m approxFloat64) Match(value driver.Value) bool { + got, ok := value.(float64) + if !ok { + return false + } + return math.Abs(got-m.want) <= m.tolerance +} diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index 2a8cbf39..01976899 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -2,6 +2,7 @@ package stats import ( "context" + "database/sql" "strings" "time" @@ -127,6 +128,17 @@ func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate statDate, string(protocol), ) + if err != nil { + return err + } + _, err = exec.ExecContext(ctx, cleanupProjectedDailyMileageSQL, + vin, + statDate, + string(protocol), + vin, + statDate, + string(protocol), + ) return err } @@ -257,3 +269,70 @@ JOIN ( SET s.is_selected = 1 WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ? ` + +const cleanupProjectedDailyMileageSQL = ` +DELETE FROM vehicle_daily_mileage +WHERE vin = ? AND stat_date = ? AND protocol = ? + AND NOT EXISTS ( + SELECT 1 + FROM vehicle_daily_mileage_source + WHERE vin = ? AND stat_date = ? AND protocol = ? AND is_selected = 1 + ) +` + +type sourceBaseline struct { + LatestTotalKM float64 + LatestEventTime time.Time +} + +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) + 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 latestTotal sql.NullFloat64 + var latestEvent sql.NullTime + if err := rows.Scan(&latestTotal, &latestEvent); err != nil { + return sourceBaseline{}, false, err + } + if err := rows.Err(); err != nil { + return sourceBaseline{}, false, err + } + return sourceBaseline{ + LatestTotalKM: latestTotal.Float64, + LatestEventTime: latestEvent.Time, + }, latestTotal.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 protocol = ? + AND source_key = ? +ORDER BY latest_event_time DESC +LIMIT 1 +` diff --git a/go/vehicle-gateway/internal/stats/source_mileage_test.go b/go/vehicle-gateway/internal/stats/source_mileage_test.go index 4c49ce8e..d07232f5 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/source_mileage_test.go @@ -88,12 +88,13 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { if err != nil { t.Fatalf("ProjectDailyMileage() error = %v", err) } - if len(exec.calls) != 3 { - t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + if len(exec.calls) != 4 { + t.Fatalf("exec calls = %d, want 4", len(exec.calls)) } updateSources := exec.calls[0].query projectFinal := exec.calls[1].query markSelected := exec.calls[2].query + cleanupFinal := exec.calls[3].query if !strings.Contains(updateSources, "UPDATE vehicle_daily_mileage_source") || !strings.Contains(updateSources, "is_selected = 0") { t.Fatalf("first query should clear selected candidates: %s", updateSources) } @@ -142,4 +143,17 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { if got := exec.calls[2].args[3]; got != maxSelectedDailyMileageKM { t.Fatalf("mark query max mileage arg = %#v, want %d", got, maxSelectedDailyMileageKM) } + for _, want := range []string{ + "DELETE FROM vehicle_daily_mileage", + "NOT EXISTS (", + "FROM vehicle_daily_mileage_source", + "is_selected = 1", + } { + if !strings.Contains(cleanupFinal, want) { + t.Fatalf("cleanup query missing %q: %s", want, cleanupFinal) + } + } + if len(exec.calls[3].args) != 6 { + t.Fatalf("cleanup query args = %d, want 6", len(exec.calls[3].args)) + } }