diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 92963373..38f11689 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -1836,7 +1836,7 @@ func TestWriterEnsuresSchemaAndUpsertsDailyMileage(t *testing.T) { if !strings.Contains(projectCall.query, "ds.id") { t.Fatalf("project sql should write source id: %s", projectCall.query) } - for _, forbidden := range []string{"trusted_source_key", "trusted_phone", "trusted_source_endpoint", "first_total_mileage_km", "sample_count)"} { + for _, forbidden := range []string{"trusted_source_key", "trusted_phone", "trusted_source_endpoint", "first_total_mileage_km"} { if strings.Contains(projectCall.query, forbidden) { t.Fatalf("project sql should not write final-table field %q: %s", forbidden, projectCall.query) } diff --git a/go/vehicle-gateway/internal/stats/source_mileage.go b/go/vehicle-gateway/internal/stats/source_mileage.go index a883c2a2..75f149b9 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage.go +++ b/go/vehicle-gateway/internal/stats/source_mileage.go @@ -620,6 +620,32 @@ WHERE s.stat_date = ? ORDER BY s.vin ` +// The same physical terminal can be forwarded by more than one platform. +// Preserve each forwarding source for audit, but rank the terminal as one +// evidence group: accumulated samples choose the established terminal and the +// freshest forwarding source keeps its projected mileage current. +const projectSourceIdentitySampleCountSQL = `( + SELECT COALESCE(SUM(identity_source.sample_count), 0) + FROM vehicle_daily_mileage_source identity_source + WHERE identity_source.vin = s.vin + AND identity_source.stat_date = s.stat_date + AND identity_source.protocol = s.protocol + AND identity_source.quality_status = '` + QualityOK + `' + AND COALESCE(NULLIF(TRIM(identity_source.phone), ''), NULLIF(TRIM(identity_source.device_id), ''), identity_source.source_key) + = COALESCE(NULLIF(TRIM(s.phone), ''), NULLIF(TRIM(s.device_id), ''), s.source_key) +)` + +const selectedSourceIdentitySampleCountSQL = `( + SELECT COALESCE(SUM(identity_source.sample_count), 0) + FROM vehicle_daily_mileage_source identity_source + WHERE identity_source.vin = s2.vin + AND identity_source.stat_date = s2.stat_date + AND identity_source.protocol = s2.protocol + AND identity_source.quality_status = '` + QualityOK + `' + AND COALESCE(NULLIF(TRIM(identity_source.phone), ''), NULLIF(TRIM(identity_source.device_id), ''), identity_source.source_key) + = COALESCE(NULLIF(TRIM(s2.phone), ''), NULLIF(TRIM(s2.device_id), ''), s2.source_key) +)` + const projectDailyMileageSQL = ` INSERT INTO vehicle_daily_mileage (vin, stat_date, protocol, source_id, daily_mileage_km, latest_total_mileage_km) @@ -650,8 +676,9 @@ ORDER BY CASE COALESCE(NULLIF(NULLIF(TRIM(ds.source_kind), ''), 'UNKNOWN'), CASE ELSE 3 END, COALESCE(ds.trust_priority, 100000), - s.sample_count DESC, + ` + projectSourceIdentitySampleCountSQL + ` DESC, s.latest_event_time DESC, + s.sample_count DESC, s.source_key ASC LIMIT 1 ON DUPLICATE KEY UPDATE @@ -689,8 +716,9 @@ LEFT JOIN ( ELSE 3 END, COALESCE(ds.trust_priority, 100000), - s2.sample_count DESC, + ` + selectedSourceIdentitySampleCountSQL + ` DESC, s2.latest_event_time DESC, + s2.sample_count DESC, s2.source_key ASC LIMIT 1 ) selected_source diff --git a/go/vehicle-gateway/internal/stats/source_mileage_test.go b/go/vehicle-gateway/internal/stats/source_mileage_test.go index 43014eff..f1c1cd5f 100644 --- a/go/vehicle-gateway/internal/stats/source_mileage_test.go +++ b/go/vehicle-gateway/internal/stats/source_mileage_test.go @@ -516,6 +516,9 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { "AND (ds.platform_name IS NULL OR TRIM(ds.platform_name) = '')", "s.daily_mileage_km BETWEEN 0 AND", "DATEDIFF(DATE(s.latest_event_time), DATE(s.first_event_time))", + "SUM(identity_source.sample_count)", + "identity_source.vin = s.vin", + "COALESCE(NULLIF(TRIM(identity_source.phone), ''), NULLIF(TRIM(identity_source.device_id), ''), identity_source.source_key)", } { if !strings.Contains(projectFinal, want) { t.Fatalf("project query missing %q: %s", want, projectFinal) @@ -526,7 +529,6 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { "trusted_phone", "trusted_source_endpoint", "first_total_mileage_km", - "sample_count)", "ds.latest_source_endpoint", } { if strings.Contains(projectFinal, forbidden) { @@ -556,6 +558,9 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { "WHEN 'DIRECT' THEN 1", "WHEN 'UNKNOWN' THEN 2", "ds.trust_priority", + "SUM(identity_source.sample_count)", + "identity_source.vin = s2.vin", + "COALESCE(NULLIF(TRIM(identity_source.phone), ''), NULLIF(TRIM(identity_source.device_id), ''), identity_source.source_key)", "LIMIT 1", } { if !strings.Contains(markSelected, want) { @@ -573,6 +578,17 @@ func TestProjectDailyMileageSelectsCandidateAndMarksSource(t *testing.T) { t.Fatalf("source selection should prefer classified DIRECT before UNKNOWN, found %q", forbidden) } } + for queryName, query := range map[string]string{ + "project": projectFinal, + "mark": markSelected, + } { + identityScore := strings.Index(query, "SUM(identity_source.sample_count)") + freshness := strings.Index(query, "latest_event_time DESC") + rowSamples := strings.LastIndex(query, "sample_count DESC") + if identityScore < 0 || freshness < identityScore || rowSamples < freshness { + t.Fatalf("%s query should rank terminal evidence before forwarding freshness and row samples: %s", queryName, query) + } + } if got := exec.calls[1].args[3]; got != maxSelectedDailyMileageKM { t.Fatalf("mark query max mileage arg = %#v, want %d", got, maxSelectedDailyMileageKM) }