diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 4cc8cdc5..73c0b589 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -717,7 +717,7 @@ const electJT808RealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location (protocol, vin, plate, source_key, event_time, latitude, longitude, speed_kmh, total_mileage_km, total_mileage_event_time, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, location_conflict, location_conflict_distance_m, - received_at, event_id) + received_at, event_id, updated_at) SELECT s.protocol, s.vin, s.plate, s.source_key, s.event_time, s.latitude, s.longitude, s.speed_kmh, s.total_mileage_km, s.total_mileage_event_time, s.soc_percent, s.altitude_m, s.direction_deg, s.alarm_flag, s.status_flag, @@ -729,7 +729,7 @@ SELECT s.protocol, s.vin, s.plate, s.source_key, s.event_time, s.latitude, s.lon FROM vehicle_realtime_location_source other WHERE other.vin = s.vin AND other.protocol = s.protocol AND other.source_key <> s.source_key AND other.quality_status = 'OK' AND other.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE)), - s.received_at, s.event_id + s.received_at, s.event_id, COALESCE(s.received_at, s.event_time, CURRENT_TIMESTAMP) FROM vehicle_realtime_location_source s LEFT JOIN vehicle_location_source_policy policy ON policy.vin = s.vin AND policy.protocol = s.protocol AND policy.source_key = s.source_key @@ -737,8 +737,22 @@ LEFT JOIN vehicle_realtime_location cur ON cur.vin = s.vin AND cur.protocol = s.protocol WHERE s.vin = ? AND s.protocol = 'JT808' AND s.quality_status = 'OK' AND COALESCE(policy.enabled, 1) = 1 ORDER BY - CASE WHEN s.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN 0 ELSE 1 END ASC, + CASE WHEN s.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) + OR (cur.source_key = s.source_key AND EXISTS( + SELECT 1 FROM vehicle_realtime_location_source conflicting + WHERE conflicting.vin = s.vin AND conflicting.protocol = s.protocol + AND conflicting.source_key <> s.source_key AND conflicting.quality_status = 'OK' + AND conflicting.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) + AND ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(conflicting.longitude, conflicting.latitude)) > 200 + )) THEN 0 ELSE 1 END ASC, (COALESCE(policy.priority, CASE s.source_kind WHEN 'DIRECT' THEN 20 WHEN 'PLATFORM' THEN 30 ELSE 40 END) + + CASE WHEN cur.source_key = s.source_key AND EXISTS( + SELECT 1 FROM vehicle_realtime_location_source conflicting + WHERE conflicting.vin = s.vin AND conflicting.protocol = s.protocol + AND conflicting.source_key <> s.source_key AND conflicting.quality_status = 'OK' + AND conflicting.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) + AND ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(conflicting.longitude, conflicting.latitude)) > 200 + ) THEN -10000 ELSE 0 END + CASE WHEN cur.source_key = s.source_key AND cur.received_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN -5 ELSE 0 END + CASE WHEN cur.source_key <> '' AND cur.source_key <> s.source_key AND s.consecutive_good_samples < 3 THEN 1000 ELSE 0 END) ASC, s.received_at DESC, s.source_key ASC @@ -751,7 +765,7 @@ ON DUPLICATE KEY UPDATE altitude_m = VALUES(altitude_m), direction_deg = VALUES(direction_deg), alarm_flag = VALUES(alarm_flag), status_flag = VALUES(status_flag), location_conflict = VALUES(location_conflict), location_conflict_distance_m = VALUES(location_conflict_distance_m), received_at = VALUES(received_at), - event_id = VALUES(event_id), updated_at = CURRENT_TIMESTAMP + event_id = VALUES(event_id), updated_at = VALUES(updated_at) ` var cleanupInvalidRealtimeTotalMileageSQL = []string{ diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 78be1c03..2ca82dcf 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -317,7 +317,7 @@ func TestJT808LocationArbitrationGuardsJumpsAndSourceFlapping(t *testing.T) { t.Fatalf("JT808 source guard missing %q: %s", want, upsertJT808RealtimeLocationSourceSQL) } } - for _, want := range []string{"vehicle_location_source_policy", "cur.source_key = s.source_key", "consecutive_good_samples < 3", "INTERVAL 2 MINUTE", "> 200"} { + for _, want := range []string{"vehicle_location_source_policy", "cur.source_key = s.source_key", "consecutive_good_samples < 3", "INTERVAL 2 MINUTE", "> 200", "THEN -10000", "updated_at = VALUES(updated_at)"} { if !strings.Contains(electJT808RealtimeLocationSQL, want) { t.Fatalf("JT808 canonical election missing %q: %s", want, electJT808RealtimeLocationSQL) } diff --git a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go index f0117164..1758a9fd 100644 --- a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go +++ b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go @@ -402,8 +402,8 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery { // Ordering only by updated_at makes the selected point flap whenever protocols // report at different cadences. When every source is stale, recency remains the // safest fallback so an old high-priority source cannot mask newer evidence. - orderExpr := `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN 0 ELSE 1 END ASC, ` + - `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 2 MINUTE) THEN CASE l.protocol ` + + orderExpr := `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 10 MINUTE) THEN 0 ELSE 1 END ASC, ` + + `CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 10 MINUTE) THEN CASE l.protocol ` + `WHEN 'GB32960' THEN 10 WHEN 'YUTONG_MQTT' THEN 20 WHEN 'JT808' THEN 30 ELSE 100 END ELSE 100 END ASC, ` + `l.updated_at DESC, l.protocol ASC` return SQLQuery{ diff --git a/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go b/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go index f2b1679e..9ecb7c8b 100644 --- a/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go +++ b/vehicle-data-platform/apps/api/internal/platform/query_builders_test.go @@ -209,7 +209,7 @@ func TestBuildVehicleRealtimeSQL(t *testing.T) { t.Fatalf("SQL should keep vehicle-level stable pagination order: %s", built.Text) } for _, want := range []string{ - "INTERVAL 2 MINUTE", + "INTERVAL 10 MINUTE", "WHEN 'GB32960' THEN 10", "WHEN 'YUTONG_MQTT' THEN 20", "WHEN 'JT808' THEN 30",