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

fix(stats): elect freshest forwarding source per terminal

Browse Source
This commit is contained in:
lingniu
2026-07-19 17:33:53 +08:00
parent a13ee8fc30
commit 263107d47a
3 changed files with 48 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

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

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

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

18
go/vehicle-gateway/internal/stats/source_mileage_test.go
View File

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