diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index 3d226762..428669ac 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -12,6 +12,7 @@ const ( QualityOK = "OK" QualityNoPreviousBaseline = "NO_PREVIOUS_BASELINE" QualityInvalidDelta = "INVALID_DELTA" + maxSelectedDailyMileageKM = 1000 ) type SourceMileageSample struct { @@ -99,6 +100,28 @@ func UpsertSourceMileage(ctx context.Context, exec Execer, sample SourceMileageS return err } +func ProjectDailyMileage(ctx context.Context, exec Execer, vin string, statDate string, protocol envelope.Protocol) error { + if exec == nil { + panic("stats execer must not be nil") + } + if strings.TrimSpace(vin) == "" || strings.TrimSpace(statDate) == "" { + return nil + } + if _, err := exec.ExecContext(ctx, clearSelectedSourceSQL, vin, statDate, string(protocol)); err != nil { + return err + } + if _, err := exec.ExecContext(ctx, projectDailyMileageSQL, + vin, + statDate, + string(protocol), + maxSelectedDailyMileageKM, + ); err != nil { + return err + } + _, err := exec.ExecContext(ctx, markSelectedSourceSQL, vin, statDate, string(protocol)) + return err +} + const upsertSourceMileageSQL = ` INSERT INTO vehicle_daily_mileage_source (vin, stat_date, protocol, source_key, source_ip, source_endpoint, phone, device_id, platform_name, @@ -148,3 +171,61 @@ VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) quality_reason = VALUES(quality_reason), updated_at = CURRENT_TIMESTAMP ` + +const clearSelectedSourceSQL = ` +UPDATE vehicle_daily_mileage_source +SET is_selected = 0 +WHERE vin = ? AND stat_date = ? AND protocol = ? +` + +const projectDailyMileageSQL = ` +INSERT INTO vehicle_daily_mileage + (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) +SELECT + s.vin, + s.stat_date, + s.protocol, + s.daily_mileage_km, + s.first_total_mileage_km, + s.latest_total_mileage_km, + s.source_key, + s.phone, + s.source_endpoint, + s.sample_count +FROM vehicle_daily_mileage_source s +LEFT JOIN vehicle_data_source ds + ON ds.protocol = s.protocol AND ds.source_ip = s.source_ip +WHERE s.vin = ? + AND s.stat_date = ? + AND s.protocol = ? + AND s.quality_status = 'OK' + AND s.daily_mileage_km BETWEEN 0 AND ? + AND COALESCE(ds.enabled, 1) = 1 +ORDER BY COALESCE(ds.trust_priority, 100), + s.sample_count DESC, + s.latest_event_time DESC, + s.source_key ASC +LIMIT 1 +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 +` + +const markSelectedSourceSQL = ` +UPDATE vehicle_daily_mileage_source s +JOIN vehicle_daily_mileage m + ON m.vin = s.vin + AND m.stat_date = s.stat_date + AND m.protocol = s.protocol + AND m.trusted_source_key = s.source_key +SET s.is_selected = 1 +WHERE s.vin = ? AND s.stat_date = ? AND s.protocol = ? +` diff --git a/go/vehicle-gateway/internal/stats/source_mileage_test.go b/go/vehicle-gateway/internal/stats/source_mileage_test.go index 0de32b1b..66b0ad52 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/source_mileage_test.go @@ -81,3 +81,35 @@ func TestUpsertSourceMileageSkipsBlankSourceIP(t *testing.T) { t.Fatalf("exec calls = %d, want 0", len(exec.calls)) } } + +func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { + exec := &recordingExec{} + err := ProjectDailyMileage(context.Background(), exec, "LA9GG64L7PBAF4001", "2026-07-08", envelope.ProtocolJT808) + if err != nil { + t.Fatalf("ProjectDailyMileage() error = %v", err) + } + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + } + updateSources := exec.calls[0].query + projectFinal := exec.calls[1].query + markSelected := exec.calls[2].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) + } + for _, want := range []string{ + "INSERT INTO vehicle_daily_mileage", + "FROM vehicle_daily_mileage_source s", + "LEFT JOIN vehicle_data_source ds", + "ORDER BY COALESCE(ds.trust_priority, 100)", + "s.quality_status = 'OK'", + "s.daily_mileage_km BETWEEN 0 AND", + } { + if !strings.Contains(projectFinal, want) { + t.Fatalf("project query missing %q: %s", want, projectFinal) + } + } + if !strings.Contains(markSelected, "UPDATE vehicle_daily_mileage_source s") || !strings.Contains(markSelected, "SET s.is_selected = 1") { + t.Fatalf("mark query should flag the elected source: %s", markSelected) + } +}