diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 660a9724..4cc8cdc5 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -5,6 +5,8 @@ import ( "database/sql" "encoding/json" "errors" + "math" + "net" "strings" "sync" "time" @@ -179,6 +181,12 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { return err } } + if _, err := w.exec.ExecContext(ctx, jt808RealtimeLocationSourceTableSQL); err != nil { + return err + } + if _, err := w.exec.ExecContext(ctx, vehicleLocationSourcePolicyTableSQL); err != nil { + return err + } for _, statement := range cleanupInvalidRealtimeTotalMileageSQL { if _, err := w.exec.ExecContext(ctx, statement); err != nil { return err @@ -230,7 +238,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) if !ok { return nil } - _, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL, + locationArgs := []any{ location.Protocol, location.VIN, location.Plate, @@ -247,10 +255,66 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) location.StatusFlag, location.ReceivedAt, location.EventID, - ) + } + if env.Protocol != envelope.ProtocolJT808 { + _, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL, locationArgs...) + return err + } + sourceKey := realtimeLocationSourceKey(env, vin) + sourceArgs := append([]any{sourceKey, strings.TrimSpace(env.Phone), strings.TrimSpace(env.DeviceID), strings.TrimSpace(env.SourceCode), normalizedSourceKind(env.SourceKind), strings.TrimSpace(env.SourceEndpoint)}, locationArgs...) + sourceArgs = append(sourceArgs, location.ReceivedAt) + if _, err = w.exec.ExecContext(ctx, upsertJT808RealtimeLocationSourceSQL, sourceArgs...); err != nil { + return err + } + _, err = w.exec.ExecContext(ctx, electJT808RealtimeLocationSQL, vin) return err } +func realtimeLocationSourceKey(env envelope.FrameEnvelope, vin string) string { + identity := strings.TrimSpace(env.Phone) + if identity == "" { + identity = strings.TrimSpace(env.DeviceID) + } + if identity == "" { + identity = strings.TrimSpace(vin) + } + scope := strings.TrimSpace(env.SourceCode) + if scope == "" { + scope = normalizedSourceKind(env.SourceKind) + } + if scope == "" || scope == "UNKNOWN" { + scope = sourceHost(env.SourceEndpoint) + } + if scope == "" { + scope = "unknown" + } + return string(env.Protocol) + ":" + identity + "@" + scope +} + +func normalizedSourceKind(value string) string { + value = strings.ToUpper(strings.TrimSpace(value)) + switch value { + case "DIRECT", "PLATFORM": + return value + default: + return "UNKNOWN" + } +} + +func sourceHost(endpoint string) string { + endpoint = strings.TrimSpace(endpoint) + if endpoint == "" { + return "" + } + if host, _, err := net.SplitHostPort(endpoint); err == nil { + return strings.TrimSpace(host) + } + if index := strings.LastIndex(endpoint, ":"); index > 0 && strings.Count(endpoint, ":") == 1 { + return endpoint[:index] + } + return endpoint +} + func snapshotFieldsForEnvelope(env envelope.FrameEnvelope) map[string]any { if len(env.ParsedFields) == 0 { return nil @@ -347,6 +411,11 @@ func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vin string, plate if !ok { return realtimeLocationRow{}, false } + if math.IsNaN(location.Latitude) || math.IsNaN(location.Longitude) || math.IsInf(location.Latitude, 0) || math.IsInf(location.Longitude, 0) || + location.Latitude < -90 || location.Latitude > 90 || location.Longitude < -180 || location.Longitude > 180 || + (math.Abs(location.Latitude) < 0.000001 && math.Abs(location.Longitude) < 0.000001) { + return realtimeLocationRow{}, false + } eventTimeMS, _ := envelope.NormalizedEventTimeMS(env) totalMileageKM, hasTotalMileage := telemetry.TotalMileageKM(env.Protocol, env.ParsedFields) if totalMileageKM <= 0 { @@ -554,8 +623,137 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo var realtimeLocationCompatibilitySQL = []string{ "ALTER TABLE vehicle_realtime_location ADD COLUMN total_mileage_event_time DATETIME(3) NULL AFTER total_mileage_km", + "ALTER TABLE vehicle_realtime_location ADD COLUMN source_key VARCHAR(256) NOT NULL DEFAULT '' AFTER plate", + "ALTER TABLE vehicle_realtime_location ADD COLUMN location_conflict TINYINT(1) NOT NULL DEFAULT 0 AFTER status_flag", + "ALTER TABLE vehicle_realtime_location ADD COLUMN location_conflict_distance_m DECIMAL(12,1) NULL AFTER location_conflict", } +const jt808RealtimeLocationSourceTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location_source ( + protocol VARCHAR(32) NOT NULL, + vin VARCHAR(32) NOT NULL, + source_key VARCHAR(256) NOT NULL, + phone VARCHAR(32) NOT NULL DEFAULT '', + device_id VARCHAR(64) NOT NULL DEFAULT '', + source_code VARCHAR(64) NOT NULL DEFAULT '', + source_kind VARCHAR(32) NOT NULL DEFAULT 'UNKNOWN', + source_endpoint VARCHAR(128) NOT NULL DEFAULT '', + plate VARCHAR(32) NOT NULL DEFAULT '', + event_time DATETIME(3) NULL, + latitude DECIMAL(12,6) NOT NULL, + longitude DECIMAL(12,6) NOT NULL, + speed_kmh DECIMAL(10,3) NULL, + total_mileage_km DECIMAL(18,3) NULL, + total_mileage_event_time DATETIME(3) NULL, + soc_percent DECIMAL(6,2) NULL, + altitude_m DECIMAL(10,3) NULL, + direction_deg DECIMAL(10,3) NULL, + alarm_flag BIGINT NULL, + status_flag BIGINT NULL, + received_at DATETIME(3) NULL, + event_id VARCHAR(64) NOT NULL DEFAULT '', + quality_status VARCHAR(32) NOT NULL DEFAULT 'OK', + quality_reason VARCHAR(64) NOT NULL DEFAULT '', + consecutive_good_samples INT NOT NULL DEFAULT 1, + latest_sample_received_at DATETIME(3) NULL, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (protocol, vin, source_key), + KEY idx_location_source_vehicle_fresh (vin, protocol, received_at), + KEY idx_location_source_quality (quality_status, updated_at) +)` + +const vehicleLocationSourcePolicyTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_location_source_policy ( + vin VARCHAR(32) NOT NULL, + protocol VARCHAR(32) NOT NULL, + source_key VARCHAR(256) NOT NULL, + enabled TINYINT(1) NOT NULL DEFAULT 1, + priority INT NOT NULL DEFAULT 100, + remark VARCHAR(255) NOT NULL DEFAULT '', + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (vin, protocol, source_key), + KEY idx_location_policy_enabled (vin, protocol, enabled, priority) +)` + +const jt808AcceptedLocationCondition = `VALUES(event_time) IS NOT NULL + AND (vehicle_realtime_location_source.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_location_source.event_time) + AND (vehicle_realtime_location_source.event_time IS NULL + OR ST_Distance_Sphere( + POINT(vehicle_realtime_location_source.longitude, vehicle_realtime_location_source.latitude), + POINT(VALUES(longitude), VALUES(latitude)) + ) <= GREATEST(500, GREATEST(1, TIMESTAMPDIFF(SECOND, vehicle_realtime_location_source.event_time, VALUES(event_time))) * 70))` + +var upsertJT808RealtimeLocationSourceSQL = ` +INSERT INTO vehicle_realtime_location_source + (source_key, phone, device_id, source_code, source_kind, source_endpoint, + protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, total_mileage_event_time, + soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, latest_sample_received_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +ON DUPLICATE KEY UPDATE + quality_status = IF(` + jt808AcceptedLocationCondition + `, 'OK', 'REJECTED'), + quality_reason = IF(quality_status = 'OK', '', 'impossible_jump'), + consecutive_good_samples = IF(quality_status = 'OK', IF(VALUES(event_id) <> event_id, consecutive_good_samples + 1, consecutive_good_samples), 0), + latest_sample_received_at = VALUES(latest_sample_received_at), + phone = IF(VALUES(phone) <> '', VALUES(phone), phone), + device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), + source_code = IF(VALUES(source_code) <> '', VALUES(source_code), source_code), + source_kind = IF(VALUES(source_kind) <> 'UNKNOWN', VALUES(source_kind), source_kind), + source_endpoint = IF(VALUES(source_endpoint) <> '', VALUES(source_endpoint), source_endpoint), + plate = IF(VALUES(plate) <> '', VALUES(plate), plate), + event_time = IF(quality_status = 'OK', VALUES(event_time), event_time), + latitude = IF(quality_status = 'OK', VALUES(latitude), latitude), + longitude = IF(quality_status = 'OK', VALUES(longitude), longitude), + speed_kmh = IF(quality_status = 'OK', COALESCE(VALUES(speed_kmh), speed_kmh), speed_kmh), + total_mileage_km = IF(quality_status = 'OK', COALESCE(VALUES(total_mileage_km), total_mileage_km), total_mileage_km), + total_mileage_event_time = IF(quality_status = 'OK' AND VALUES(total_mileage_km) IS NOT NULL, COALESCE(VALUES(total_mileage_event_time), total_mileage_event_time), total_mileage_event_time), + soc_percent = IF(quality_status = 'OK', COALESCE(VALUES(soc_percent), soc_percent), soc_percent), + altitude_m = IF(quality_status = 'OK', COALESCE(VALUES(altitude_m), altitude_m), altitude_m), + direction_deg = IF(quality_status = 'OK', COALESCE(VALUES(direction_deg), direction_deg), direction_deg), + alarm_flag = IF(quality_status = 'OK', COALESCE(VALUES(alarm_flag), alarm_flag), alarm_flag), + status_flag = IF(quality_status = 'OK', COALESCE(VALUES(status_flag), status_flag), status_flag), + received_at = IF(quality_status = 'OK', VALUES(received_at), received_at), + event_id = IF(quality_status = 'OK', VALUES(event_id), event_id) +` + +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) +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, + EXISTS(SELECT 1 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) + AND ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(other.longitude, other.latitude)) > 200), + (SELECT MAX(ST_Distance_Sphere(POINT(s.longitude, s.latitude), POINT(other.longitude, other.latitude))) + 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 +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 +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, + (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 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 +LIMIT 1 +ON DUPLICATE KEY UPDATE + plate = IF(VALUES(plate) <> '', VALUES(plate), vehicle_realtime_location.plate), source_key = VALUES(source_key), + event_time = VALUES(event_time), latitude = VALUES(latitude), longitude = VALUES(longitude), + speed_kmh = VALUES(speed_kmh), total_mileage_km = VALUES(total_mileage_km), + total_mileage_event_time = VALUES(total_mileage_event_time), soc_percent = VALUES(soc_percent), + 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 +` + var cleanupInvalidRealtimeTotalMileageSQL = []string{ `UPDATE vehicle_realtime_location SET total_mileage_km = NULL, diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 0c65fc97..78be1c03 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -46,8 +46,8 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 19 { - t.Fatalf("exec calls = %d, want 19", len(exec.calls)) + if len(exec.calls) != 24 { + t.Fatalf("exec calls = %d, want 24", len(exec.calls)) } if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") { t.Fatalf("schema query = %s", exec.calls[0].query) @@ -84,15 +84,18 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if !strings.Contains(exec.calls[12].query, "total_mileage_event_time") { t.Fatalf("location compatibility migration should add total mileage event time: %s", exec.calls[12].query) } - for _, call := range exec.calls[13:17] { + if !strings.Contains(exec.calls[16].query, "vehicle_realtime_location_source") || !strings.Contains(exec.calls[17].query, "vehicle_location_source_policy") { + t.Fatalf("source arbitration schema missing: %s / %s", exec.calls[16].query, exec.calls[17].query) + } + for _, call := range exec.calls[18:22] { if !strings.Contains(call.query, "total_mileage") { t.Fatalf("schema bootstrap should clean invalid realtime mileage: %s", call.query) } } - if !strings.Contains(exec.calls[17].query, "snapshot_backfill") { - t.Fatalf("access projection baseline should be backfilled explicitly: %s", exec.calls[17].query) + if !strings.Contains(exec.calls[22].query, "snapshot_backfill") { + t.Fatalf("access projection baseline should be backfilled explicitly: %s", exec.calls[22].query) } - upsert := exec.calls[18] + upsert := exec.calls[23] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } @@ -150,6 +153,14 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("ALTER TABLE vehicle_realtime_location ADD COLUMN total_mileage_event_time"). WillReturnResult(sqlmock.NewResult(0, 0)) + for _, column := range []string{"source_key", "location_conflict", "location_conflict_distance_m"} { + mock.ExpectExec("ALTER TABLE vehicle_realtime_location ADD COLUMN " + column). + WillReturnResult(sqlmock.NewResult(0, 0)) + } + mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location_source"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_location_source_policy"). + WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("UPDATE vehicle_realtime_location"). WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("UPDATE vehicle_realtime_snapshot"). @@ -235,47 +246,97 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", len(exec.calls)) } locationUpsert := exec.calls[1] - if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") { + if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location_source") { t.Fatalf("location upsert query = %s", locationUpsert.query) } if strings.Contains(locationUpsert.query, "fields_json") { t.Fatalf("location upsert should not write fields_json: %s", locationUpsert.query) } - for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { + for _, column := range []string{"message_id", "sequence_id"} { if strings.Contains(locationUpsert.query, column) { t.Fatalf("location upsert should not write %s: %s", column, locationUpsert.query) } } - if got, want := locationUpsert.args[0], "JT808"; got != want { + if got, want := locationUpsert.args[6], "JT808"; got != want { t.Fatalf("protocol arg = %#v, want %q", got, want) } - if got, want := locationUpsert.args[1], "VIN001"; got != want { + if got, want := locationUpsert.args[7], "VIN001"; got != want { t.Fatalf("vin arg = %#v, want %q", got, want) } - if got, want := locationUpsert.args[4], 30.590151; got != want { + if got, want := locationUpsert.args[10], 30.590151; got != want { t.Fatalf("latitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[5], 121.069881; got != want { + if got, want := locationUpsert.args[11], 121.069881; got != want { t.Fatalf("longitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[6], 23.0; got != want { + if got, want := locationUpsert.args[12], 23.0; got != want { t.Fatalf("speed arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[7], 10241.2; got != want { + if got, want := locationUpsert.args[13], 10241.2; got != want { t.Fatalf("mileage arg = %#v, want %v", got, want) } - if _, ok := locationUpsert.args[8].(time.Time); !ok { - t.Fatalf("total mileage event time arg = %#v, want time.Time", locationUpsert.args[8]) + if _, ok := locationUpsert.args[14].(time.Time); !ok { + t.Fatalf("total mileage event time arg = %#v, want time.Time", locationUpsert.args[14]) } - if got := locationUpsert.args[9]; got != nil { + if got := locationUpsert.args[15]; got != nil { t.Fatalf("JT808 location should not invent SOC, got %#v", got) } - if len(locationUpsert.args) != 16 { - t.Fatalf("location upsert args = %d, want 16", len(locationUpsert.args)) + if len(locationUpsert.args) != 23 { + t.Fatalf("location source upsert args = %d, want 23", len(locationUpsert.args)) + } + if got := locationUpsert.args[0]; got != "JT808:13307795425@1.2.3.4" { + t.Fatalf("stable source key = %#v", got) + } + if !strings.Contains(exec.calls[2].query, "cur.source_key") || !strings.Contains(exec.calls[2].query, "consecutive_good_samples < 3") { + t.Fatalf("canonical election must use sticky source hysteresis: %s", exec.calls[2].query) + } +} + +func TestRealtimeLocationSourceKeyIsStableAcrossConnectionPorts(t *testing.T) { + platform := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "013307795425", SourceCode: "g7s", SourceKind: "PLATFORM", SourceEndpoint: "1.2.3.4:40001"} + if got := realtimeLocationSourceKey(platform, "VIN001"); got != "JT808:013307795425@g7s" { + t.Fatalf("platform source key = %q", got) + } + platform.SourceEndpoint = "1.2.3.4:49999" + if got := realtimeLocationSourceKey(platform, "VIN001"); got != "JT808:013307795425@g7s" { + t.Fatalf("reconnected platform source key = %q", got) + } + direct := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "013307795425", SourceKind: "DIRECT", SourceEndpoint: "5.6.7.8:50001"} + if got := realtimeLocationSourceKey(direct, "VIN001"); got != "JT808:013307795425@DIRECT" { + t.Fatalf("direct source key = %q", got) + } +} + +func TestJT808LocationArbitrationGuardsJumpsAndSourceFlapping(t *testing.T) { + for _, want := range []string{"ST_Distance_Sphere", "GREATEST(500", "* 70", "impossible_jump", "consecutive_good_samples + 1"} { + if !strings.Contains(upsertJT808RealtimeLocationSourceSQL, want) { + 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"} { + if !strings.Contains(electJT808RealtimeLocationSQL, want) { + t.Fatalf("JT808 canonical election missing %q: %s", want, electJT808RealtimeLocationSQL) + } + } + if !strings.Contains(electJT808RealtimeLocationSQL, "vehicle_realtime_location.plate") { + t.Fatalf("canonical upsert must qualify the target plate across joined source tables: %s", electJT808RealtimeLocationSQL) + } +} + +func TestRealtimeLocationRejectsZeroCoordinate(t *testing.T) { + _, ok := realtimeLocationFromEnvelope(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + ParsedFields: map[string]any{ + "jt808.location.latitude": 0, + "jt808.location.longitude": 0, + }, + }, "VIN001", "") + if ok { + t.Fatal("zero coordinate must not enter realtime location state") } } @@ -299,7 +360,7 @@ func TestSnapshotWriterDropsNonPositiveTotalMileageFromRealtimeStores(t *testing t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 2 { + if len(exec.calls) != 3 { t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls)) } parsedJSON, ok := exec.calls[0].args[5].(string) @@ -312,11 +373,11 @@ func TestSnapshotWriterDropsNonPositiveTotalMileageFromRealtimeStores(t *testing if !strings.Contains(parsedJSON, "jt808.location.speed_kmh") { t.Fatalf("snapshot parsed_json should keep valid fields: %s", parsedJSON) } - if exec.calls[1].args[7] != nil { - t.Fatalf("location total mileage arg = %#v, want nil", exec.calls[1].args[7]) + if exec.calls[1].args[13] != nil { + t.Fatalf("location total mileage arg = %#v, want nil", exec.calls[1].args[13]) } - if exec.calls[1].args[8] != nil { - t.Fatalf("location total mileage time arg = %#v, want nil", exec.calls[1].args[8]) + if exec.calls[1].args[14] != nil { + t.Fatalf("location total mileage time arg = %#v, want nil", exec.calls[1].args[14]) } } @@ -342,17 +403,17 @@ func TestSnapshotWriterNormalizesFarFutureEventTime(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 2 { + if len(exec.calls) != 3 { t.Fatalf("exec calls = %d, want snapshot + location", len(exec.calls)) } if got, ok := exec.calls[0].args[6].(time.Time); !ok || !got.Equal(received) { t.Fatalf("snapshot event_time arg = %#v, want received %s", exec.calls[0].args[6], received) } - if got, ok := exec.calls[1].args[3].(time.Time); !ok || !got.Equal(received) { - t.Fatalf("location event_time arg = %#v, want received %s", exec.calls[1].args[3], received) + if got, ok := exec.calls[1].args[9].(time.Time); !ok || !got.Equal(received) { + t.Fatalf("location event_time arg = %#v, want received %s", exec.calls[1].args[9], received) } - if got, ok := exec.calls[1].args[8].(time.Time); !ok || !got.Equal(received) { - t.Fatalf("total mileage event time arg = %#v, want received %s", exec.calls[1].args[8], received) + if got, ok := exec.calls[1].args[14].(time.Time); !ok || !got.Equal(received) { + t.Fatalf("total mileage event time arg = %#v, want received %s", exec.calls[1].args[14], received) } } diff --git a/vehicle-data-platform/apps/api/internal/platform/model.go b/vehicle-data-platform/apps/api/internal/platform/model.go index fdfd33f6..7ac2ef23 100644 --- a/vehicle-data-platform/apps/api/internal/platform/model.go +++ b/vehicle-data-platform/apps/api/internal/platform/model.go @@ -50,6 +50,8 @@ type MonitorMapPoint struct { TotalMileageKm float64 `json:"totalMileageKm"` LastSeen string `json:"lastSeen"` ReportIntervalMs *int64 `json:"reportIntervalMs,omitempty"` + LocationSource string `json:"locationSource"` + LocationConflict bool `json:"locationConflict"` Status string `json:"status"` } @@ -1019,6 +1021,9 @@ type VehicleRealtimeRow struct { BindingStatus string `json:"bindingStatus"` ServiceStatus *VehicleServiceStatus `json:"serviceStatus,omitempty"` PrimaryProtocol string `json:"primaryProtocol"` + LocationSource string `json:"locationSource"` + LocationConflict bool `json:"locationConflict"` + ConflictDistanceM *float64 `json:"conflictDistanceM,omitempty"` Longitude float64 `json:"longitude"` Latitude float64 `json:"latitude"` SpeedKmh float64 `json:"speedKmh"` 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 65debf03..f0117164 100644 --- a/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go +++ b/vehicle-data-platform/apps/api/internal/platform/mysql_queries.go @@ -398,7 +398,14 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery { `WHERE ` + strings.Join(where, " AND ") + ` ` + `GROUP BY l.vin, b.plate, b.phone, b.oem ` + havingSQL - orderExpr := `l.updated_at DESC, l.protocol ASC` + // Keep one authoritative coordinate source while all protocols are healthy. + // 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 ` + + `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{ Text: `SELECT l.vin, ` + `COALESCE(NULLIF(MAX(NULLIF(l.plate, '')), ''), b.plate, '') AS plate, ` + @@ -410,6 +417,9 @@ func buildVehicleRealtimeSQL(query url.Values) SQLQuery { `CASE WHEN COUNT(DISTINCT CASE WHEN l.updated_at >= DATE_SUB(NOW(), INTERVAL 1 MINUTE) THEN l.protocol END) > 0 THEN 1 ELSE 0 END AS online, ` + `CASE WHEN MAX(CASE WHEN b.vin IS NOT NULL THEN 1 ELSE 0 END) = 1 THEN 'bound' ELSE 'unbound' END AS binding_status, ` + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(l.protocol ORDER BY ` + orderExpr + `), ',', 1), '') AS primary_protocol, ` + + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(NULLIF(l.source_key, ''), CONCAT(l.protocol, ':canonical')) ORDER BY ` + orderExpr + `), ',', 1), '') AS location_source, ` + + `COALESCE(CAST(SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(l.location_conflict, 0) ORDER BY ` + orderExpr + `), ',', 1) AS UNSIGNED), 0) AS location_conflict, ` + + `SUBSTRING_INDEX(GROUP_CONCAT(COALESCE(CAST(l.location_conflict_distance_m AS CHAR), '-1') ORDER BY ` + orderExpr + `), ',', 1) AS conflict_distance_m, ` + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.longitude AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS longitude, ` + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.latitude AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS latitude, ` + `COALESCE(SUBSTRING_INDEX(GROUP_CONCAT(CAST(l.speed_kmh AS CHAR) ORDER BY ` + orderExpr + `), ',', 1), '') AS speed_kmh, ` + diff --git a/vehicle-data-platform/apps/api/internal/platform/production_store.go b/vehicle-data-platform/apps/api/internal/platform/production_store.go index 7872bcb5..befc0ff8 100644 --- a/vehicle-data-platform/apps/api/internal/platform/production_store.go +++ b/vehicle-data-platform/apps/api/internal/platform/production_store.go @@ -401,13 +401,22 @@ func (s *ProductionStore) VehicleRealtime(ctx context.Context, query url.Values) var onlineProtocols string var online int var longitude, latitude, speed, soc, mileage string + var conflictDistance sql.NullString + var locationConflict int var reportIntervalMs int64 - if err := rows.Scan(&row.VIN, &row.Plate, &row.Phone, &row.OEM, &protocols, &onlineProtocols, &row.SourceCount, &row.OnlineSourceCount, &online, &row.BindingStatus, &row.PrimaryProtocol, &longitude, &latitude, &speed, &soc, &mileage, &row.LastSeen, &reportIntervalMs); err != nil { + if err := rows.Scan(&row.VIN, &row.Plate, &row.Phone, &row.OEM, &protocols, &onlineProtocols, &row.SourceCount, &row.OnlineSourceCount, &online, &row.BindingStatus, &row.PrimaryProtocol, &row.LocationSource, &locationConflict, &conflictDistance, &longitude, &latitude, &speed, &soc, &mileage, &row.LastSeen, &reportIntervalMs); err != nil { return Page[VehicleRealtimeRow]{}, err } row.Protocols = splitCSV(protocols) row.SourceStatus = buildVehicleCoverageSourceStatus(row.Protocols, splitCSV(onlineProtocols), row.LastSeen) row.Online = online == 1 + row.LocationConflict = locationConflict == 1 + if conflictDistance.Valid { + value := parseFloatString(conflictDistance.String) + if value >= 0 { + row.ConflictDistanceM = &value + } + } row.ServiceStatus = buildRealtimeServiceStatus(row) row.Longitude = parseFloatString(longitude) row.Latitude = parseFloatString(latitude) 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 e3cd02bb..f2b1679e 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 @@ -208,6 +208,20 @@ func TestBuildVehicleRealtimeSQL(t *testing.T) { if !strings.Contains(built.Text, "ORDER BY MAX(l.updated_at) DESC, l.vin ASC") { t.Fatalf("SQL should keep vehicle-level stable pagination order: %s", built.Text) } + for _, want := range []string{ + "INTERVAL 2 MINUTE", + "WHEN 'GB32960' THEN 10", + "WHEN 'YUTONG_MQTT' THEN 20", + "WHEN 'JT808' THEN 30", + "ELSE 100 END ELSE 100 END ASC, l.updated_at DESC", + "location_source", + "location_conflict", + "location_conflict_distance_m", + } { + if !strings.Contains(built.Text, want) { + t.Fatalf("realtime source arbitration missing %q: %s", want, built.Text) + } + } if len(built.Args) != 8 || built.Args[0] != "JT808" || built.Args[1] != "%粤A%" || built.Args[6] != 10 || built.Args[7] != 20 { t.Fatalf("args = %#v", built.Args) } diff --git a/vehicle-data-platform/apps/api/internal/platform/service.go b/vehicle-data-platform/apps/api/internal/platform/service.go index 1412b18b..a2e217cd 100644 --- a/vehicle-data-platform/apps/api/internal/platform/service.go +++ b/vehicle-data-platform/apps/api/internal/platform/service.go @@ -607,6 +607,8 @@ func buildMonitorMapResponse(vehicles Page[VehicleRealtimeRow], query url.Values Longitude: vehicle.Longitude, Latitude: vehicle.Latitude, SpeedKmh: vehicle.SpeedKmh, SOCPercent: vehicle.SOCPercent, TotalMileageKm: vehicle.TotalMileageKm, LastSeen: vehicle.LastSeen, ReportIntervalMs: vehicle.ReportIntervalMs, + LocationSource: vehicle.LocationSource, + LocationConflict: vehicle.LocationConflict, Status: monitorVehicleStatus(vehicle), }) } diff --git a/vehicle-data-platform/apps/web/src/api/types.ts b/vehicle-data-platform/apps/web/src/api/types.ts index 4e1e0240..f702f832 100644 --- a/vehicle-data-platform/apps/web/src/api/types.ts +++ b/vehicle-data-platform/apps/web/src/api/types.ts @@ -37,6 +37,8 @@ export interface MonitorMapPoint { totalMileageKm: number; lastSeen: string; reportIntervalMs?: number; + locationSource?: string; + locationConflict?: boolean; status: 'driving' | 'idle' | 'offline' | 'unknown'; } @@ -540,6 +542,9 @@ export interface VehicleRealtimeRow { bindingStatus: string; serviceStatus?: VehicleServiceStatus; primaryProtocol: string; + locationSource?: string; + locationConflict?: boolean; + conflictDistanceM?: number; longitude: number; latitude: number; speedKmh: number; diff --git a/vehicle-data-platform/apps/web/src/integrations/amap.ts b/vehicle-data-platform/apps/web/src/integrations/amap.ts index 2cf7c89d..14853dff 100644 --- a/vehicle-data-platform/apps/web/src/integrations/amap.ts +++ b/vehicle-data-platform/apps/web/src/integrations/amap.ts @@ -48,6 +48,7 @@ export type AMapMassPoint = { style: number; id: string; label: string; + sourceToken?: string; }; type AMapAddressComponent = { diff --git a/vehicle-data-platform/apps/web/src/v2/map/FleetMap.test.tsx b/vehicle-data-platform/apps/web/src/v2/map/FleetMap.test.tsx index e2d3561e..b60dd21d 100644 --- a/vehicle-data-platform/apps/web/src/v2/map/FleetMap.test.tsx +++ b/vehicle-data-platform/apps/web/src/v2/map/FleetMap.test.tsx @@ -2,7 +2,7 @@ import { act, cleanup, fireEvent, render, screen, waitFor } from '@testing-libra import { afterEach, expect, test, vi } from 'vitest'; import type { MonitorMapResponse } from '../../api/types'; import { wgs84ToGcj02, type AMapLike, type AMapMap, type AMapMassPoint } from '../../integrations/amap'; -import { advancePointMotions, FleetMap, interpolateMassPoints, massPointFingerprint, pointMoveDurationMs, pointMoveFrameMs } from './FleetMap'; +import { advancePointMotions, FleetMap, interpolateMassPoints, massPointFingerprint, movingPointIDs, pointMoveDurationMs, pointMoveFrameMs } from './FleetMap'; const setData = vi.fn<(data: AMapMassPoint[]) => void>(); const setStyle = vi.fn(); @@ -170,6 +170,15 @@ test('interpolates existing mass points at constant speed while keeping new poin expect(interpolateMassPoints(previous, target, 1)).toEqual(target); }); +test('does not interpolate across an authoritative location source switch', () => { + const previous: AMapMassPoint[] = [{ id: 'vehicle', label: 'A', style: 0, lnglat: [113.2, 23.1], sourceToken: 'JT808:terminal-a' }]; + const sameSource: AMapMassPoint[] = [{ id: 'vehicle', label: 'A', style: 0, lnglat: [113.3, 23.2], sourceToken: 'JT808:terminal-a' }]; + const switchedSource: AMapMassPoint[] = [{ id: 'vehicle', label: 'A', style: 0, lnglat: [113.4, 23.3], sourceToken: 'GB32960:canonical' }]; + + expect(movingPointIDs(previous, sameSource).has('vehicle')).toBe(true); + expect(movingPointIDs(previous, switchedSource).has('vehicle')).toBe(false); +}); + test('uses each protocol report interval with bounded fallbacks for point motion', () => { expect(pointMoveDurationMs(10_000)).toBe(10_000); expect(pointMoveDurationMs(30_000)).toBe(30_000); diff --git a/vehicle-data-platform/apps/web/src/v2/map/FleetMap.tsx b/vehicle-data-platform/apps/web/src/v2/map/FleetMap.tsx index 4ec621b4..6e1ff3e7 100644 --- a/vehicle-data-platform/apps/web/src/v2/map/FleetMap.tsx +++ b/vehicle-data-platform/apps/web/src/v2/map/FleetMap.tsx @@ -160,6 +160,7 @@ export function massPointFingerprint( fingerprint.add(point.status); fingerprint.add(point.plate || ''); fingerprint.add(point.reportIntervalMs ?? ''); + fingerprint.add(point.locationSource || point.protocol); } } else { fingerprint.add(points.length); @@ -170,6 +171,7 @@ export function massPointFingerprint( fingerprint.add(vehicleStatus(point)); fingerprint.add(point.plate || ''); fingerprint.add(point.reportIntervalMs ?? ''); + fingerprint.add(point.locationSource || point.primaryProtocol); } } return fingerprint.value(); @@ -225,12 +227,12 @@ export function pointMoveFrameMs(totalPoints: number, movingPoints: number) { return Math.round(Math.min(MAX_POINT_MOVE_FRAME_MS, MIN_POINT_MOVE_FRAME_MS + pressure * 120)); } -function movingPointIDs(previous: AMapMassPoint[], target: AMapMassPoint[]) { - const previousByID = new Map(previous.map((point) => [point.id, point.lnglat])); - const moving = new Set(); - for (const point of target) { - const start = previousByID.get(point.id); - if (start && (Math.abs(start[0] - point.lnglat[0]) >= POINT_MOVE_EPSILON || Math.abs(start[1] - point.lnglat[1]) >= POINT_MOVE_EPSILON)) moving.add(point.id); +export function movingPointIDs(previous: AMapMassPoint[], target: AMapMassPoint[]) { + const previousByID = new Map(previous.map((point) => [point.id, point])); + const moving = new Set(); + for (const point of target) { + const start = previousByID.get(point.id); + if (start && start.sourceToken === point.sourceToken && (Math.abs(start.lnglat[0] - point.lnglat[0]) >= POINT_MOVE_EPSILON || Math.abs(start.lnglat[1] - point.lnglat[1]) >= POINT_MOVE_EPSILON)) moving.add(point.id); } return moving; } @@ -486,9 +488,9 @@ export function FleetMap({ vehicles, selectedVin, onSelect, monitorMap, onSelect const clusterStyleIndexes = new Map(clusterCounts.map((count, index) => [count, COLORS.length + index])); const data: AMapMassPoint[] = monitorMode ? [ ...clusters.map((cluster) => ({ lnglat: wgs84ToGcj02(cluster.longitude, cluster.latitude), style: clusterStyleIndexes.get(cluster.count) ?? COLORS.length, id: cluster.id, label: `${cluster.count} 辆` })), - ...mapPoints.map((point) => ({ lnglat: wgs84ToGcj02(point.longitude, point.latitude), style: point.status === 'driving' ? 2 : point.status === 'idle' ? 0 : point.status === 'offline' ? 1 : 3, id: point.vin, label: point.plate || point.vin })) + ...mapPoints.map((point) => ({ lnglat: wgs84ToGcj02(point.longitude, point.latitude), style: point.status === 'driving' ? 2 : point.status === 'idle' ? 0 : point.status === 'offline' ? 1 : 3, id: point.vin, label: point.plate || point.vin, sourceToken: point.locationSource || point.protocol })) ] : fallbackPoints.map((vehicle) => ({ - lnglat: wgs84ToGcj02(vehicle.longitude, vehicle.latitude), style: styleIndex(vehicle), id: vehicle.vin, label: vehicle.plate || vehicle.vin + lnglat: wgs84ToGcj02(vehicle.longitude, vehicle.latitude), style: styleIndex(vehicle), id: vehicle.vin, label: vehicle.plate || vehicle.vin, sourceToken: vehicle.locationSource || vehicle.primaryProtocol })); const moveDurationByID = new Map((monitorMode ? mapPoints : fallbackPoints) .map((point) => [point.vin, pointMoveDurationMs(point.reportIntervalMs)]));