feat(stats): project elected mileage source

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 = ?
+`

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)
+	}
+}