From 12c15896d03aa2195a584c481b0db3a2fa450d93 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 16:47:29 +0800 Subject: [PATCH] refactor(go): key realtime mysql tables by vin --- .../internal/realtime/snapshot_writer.go | 126 ++++++++++++------ .../internal/realtime/snapshot_writer_test.go | 52 +++++--- 2 files changed, 123 insertions(+), 55 deletions(-) diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 76e8df6d..1307f2e7 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -45,8 +45,8 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { } func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error { - vehicleKey := strings.TrimSpace(env.VehicleKey()) - if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { + vin := strings.TrimSpace(env.VIN) + if vin == "" { return nil } if !hasRealtimePayload(env) { @@ -60,10 +60,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) receivedAt := nullableTime(env.ReceivedAtMS) if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, string(env.Protocol), - vehicleKey, - strings.TrimSpace(env.VIN), - strings.TrimSpace(env.Phone), - strings.TrimSpace(env.DeviceID), + vin, plate, strings.TrimSpace(env.MessageID), env.Sequence, @@ -74,16 +71,13 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) ); err != nil { return err } - location, ok := realtimeLocationFromEnvelope(env, vehicleKey, plate) + location, ok := realtimeLocationFromEnvelope(env, vin, plate) if !ok { return nil } _, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL, location.Protocol, - location.VehicleKey, location.VIN, - location.Phone, - location.DeviceID, location.Plate, location.MessageID, location.Sequence, @@ -130,13 +124,22 @@ func (w *SnapshotWriter) dropObsoleteColumns(ctx context.Context) error { if !ok { return nil } + if err := w.ensureRealtimeVINIndexes(ctx, queryer); err != nil { + return err + } for _, column := range []struct { table string name string }{ {table: "vehicle_realtime_snapshot", name: "parsed_json"}, {table: "vehicle_realtime_snapshot", name: "fields_json"}, + {table: "vehicle_realtime_snapshot", name: "vehicle_key"}, + {table: "vehicle_realtime_snapshot", name: "phone"}, + {table: "vehicle_realtime_snapshot", name: "device_id"}, {table: "vehicle_realtime_location", name: "fields_json"}, + {table: "vehicle_realtime_location", name: "vehicle_key"}, + {table: "vehicle_realtime_location", name: "phone"}, + {table: "vehicle_realtime_location", name: "device_id"}, } { exists, err := columnExists(ctx, queryer, column.table, column.name) if err != nil { @@ -152,6 +155,56 @@ func (w *SnapshotWriter) dropObsoleteColumns(ctx context.Context) error { return nil } +func (w *SnapshotWriter) ensureRealtimeVINIndexes(ctx context.Context, queryer Queryer) error { + for _, item := range []struct { + table string + oldIndex string + newIndex string + }{ + {table: "vehicle_realtime_snapshot", oldIndex: "uk_realtime_snapshot_vehicle", newIndex: "uk_realtime_snapshot_vin"}, + {table: "vehicle_realtime_location", oldIndex: "uk_realtime_location_vehicle", newIndex: "uk_realtime_location_vin"}, + } { + if err := w.normalizeVINRows(ctx, item.table); err != nil { + return err + } + exists, err := indexExists(ctx, queryer, item.table, item.newIndex) + if err != nil { + return err + } + if !exists { + if _, err := w.exec.ExecContext(ctx, "ALTER TABLE "+item.table+" ADD UNIQUE KEY "+item.newIndex+" (protocol, vin)"); err != nil { + return err + } + } + exists, err = indexExists(ctx, queryer, item.table, item.oldIndex) + if err != nil { + return err + } + if exists { + if _, err := w.exec.ExecContext(ctx, "ALTER TABLE "+item.table+" DROP INDEX "+item.oldIndex); err != nil { + return err + } + } + } + return nil +} + +func (w *SnapshotWriter) normalizeVINRows(ctx context.Context, table string) error { + if _, err := w.exec.ExecContext(ctx, "DELETE FROM "+table+" WHERE vin IS NULL OR vin = ''"); err != nil { + return err + } + _, err := w.exec.ExecContext(ctx, `DELETE newer_duplicate +FROM `+table+` newer_duplicate +JOIN `+table+` latest + ON newer_duplicate.protocol = latest.protocol + AND newer_duplicate.vin = latest.vin + AND ( + newer_duplicate.updated_at < latest.updated_at + OR (newer_duplicate.updated_at = latest.updated_at AND newer_duplicate.id < latest.id) + )`) + return err +} + func columnExists(ctx context.Context, queryer Queryer, table string, column string) (bool, error) { var count int err := queryer.QueryRowContext(ctx, ` @@ -166,12 +219,23 @@ WHERE table_schema = DATABASE() return count > 0, nil } +func indexExists(ctx context.Context, queryer Queryer, table string, index string) (bool, error) { + var count int + err := queryer.QueryRowContext(ctx, ` +SELECT COUNT(*) +FROM information_schema.statistics +WHERE table_schema = DATABASE() + AND table_name = ? + AND index_name = ?`, table, index).Scan(&count) + if err != nil { + return false, err + } + return count > 0, nil +} + type realtimeLocationRow struct { Protocol string - VehicleKey string VIN string - Phone string - DeviceID string Plate string MessageID string Sequence uint16 @@ -190,7 +254,7 @@ type realtimeLocationRow struct { EventID string } -func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, plate string) (realtimeLocationRow, bool) { +func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vin string, plate string) (realtimeLocationRow, bool) { latitude, okLat := numberField(env.Fields, envelope.FieldLatitude) longitude, okLon := numberField(env.Fields, envelope.FieldLongitude) if !okLat || !okLon { @@ -198,10 +262,7 @@ func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, } return realtimeLocationRow{ Protocol: string(env.Protocol), - VehicleKey: vehicleKey, - VIN: strings.TrimSpace(env.VIN), - Phone: strings.TrimSpace(env.Phone), - DeviceID: strings.TrimSpace(env.DeviceID), + VIN: vin, Plate: plate, MessageID: strings.TrimSpace(env.MessageID), Sequence: env.Sequence, @@ -331,10 +392,7 @@ func hasRealtimePayload(env envelope.FrameEnvelope) bool { const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot ( id BIGINT PRIMARY KEY AUTO_INCREMENT, protocol VARCHAR(32) NOT NULL, - vehicle_key VARCHAR(96) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', - phone VARCHAR(32) NOT NULL DEFAULT '', - device_id VARCHAR(96) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '', message_id VARCHAR(32) NOT NULL DEFAULT '', sequence_id INT NOT NULL DEFAULT 0, @@ -344,21 +402,16 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn event_id VARCHAR(64) NOT NULL DEFAULT '', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_realtime_snapshot_vehicle (protocol, vehicle_key), - KEY idx_vehicle_key (vehicle_key), + UNIQUE KEY uk_realtime_snapshot_vin (protocol, vin), KEY idx_vin (vin), KEY idx_protocol_updated (protocol, updated_at) )` const upsertRealtimeSnapshotSQL = ` INSERT INTO vehicle_realtime_snapshot - (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, - source_endpoint, event_time, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + (protocol, vin, plate, message_id, sequence_id, source_endpoint, event_time, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE - vin = IF(VALUES(vin) <> '', VALUES(vin), vin), - phone = IF(VALUES(phone) <> '', VALUES(phone), phone), - device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), plate = IF(VALUES(plate) <> '', VALUES(plate), plate), message_id = VALUES(message_id), sequence_id = VALUES(sequence_id), @@ -372,10 +425,7 @@ ON DUPLICATE KEY UPDATE const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location ( id BIGINT PRIMARY KEY AUTO_INCREMENT, protocol VARCHAR(32) NOT NULL, - vehicle_key VARCHAR(96) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', - phone VARCHAR(32) NOT NULL DEFAULT '', - device_id VARCHAR(96) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '', message_id VARCHAR(32) NOT NULL DEFAULT '', sequence_id INT NOT NULL DEFAULT 0, @@ -394,8 +444,7 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo event_id VARCHAR(64) NOT NULL DEFAULT '', created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_realtime_location_vehicle (protocol, vehicle_key), - KEY idx_vehicle_key (vehicle_key), + UNIQUE KEY uk_realtime_location_vin (protocol, vin), KEY idx_vin (vin), KEY idx_protocol_updated (protocol, updated_at), KEY idx_location (longitude, latitude) @@ -403,14 +452,11 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo const upsertRealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location - (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, - source_endpoint, event_time, latitude, longitude, speed_kmh, total_mileage_km, + (protocol, vin, plate, message_id, sequence_id, source_endpoint, event_time, + latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE - vin = IF(VALUES(vin) <> '', VALUES(vin), vin), - phone = IF(VALUES(phone) <> '', VALUES(phone), phone), - device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), plate = IF(VALUES(plate) <> '', VALUES(plate), plate), message_id = VALUES(message_id), sequence_id = VALUES(sequence_id), diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 95538857..237ae8a2 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -62,10 +62,10 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("protocol arg = %#v, want %q", got, want) } if got, want := upsert.args[1], "VIN001"; got != want { - t.Fatalf("vehicle key arg = %#v, want %q", got, want) + t.Fatalf("vin arg = %#v, want %q", got, want) } - if len(upsert.args) != 12 { - t.Fatalf("snapshot upsert args = %d, want 12", len(upsert.args)) + if len(upsert.args) != 9 { + t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args)) } } @@ -115,25 +115,25 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) t.Fatalf("protocol arg = %#v, want %q", got, want) } if got, want := locationUpsert.args[1], "VIN001"; got != want { - t.Fatalf("vehicle key arg = %#v, want %q", got, want) + t.Fatalf("vin arg = %#v, want %q", got, want) } - if got, want := locationUpsert.args[10], 30.590151; got != want { + if got, want := locationUpsert.args[7], 30.590151; got != want { t.Fatalf("latitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[11], 121.069881; got != want { + if got, want := locationUpsert.args[8], 121.069881; got != want { t.Fatalf("longitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[12], 23.0; got != want { + if got, want := locationUpsert.args[9], 23.0; got != want { t.Fatalf("speed arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[13], 10241.2; got != want { + if got, want := locationUpsert.args[10], 10241.2; got != want { t.Fatalf("mileage arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[14], 88.0; got != want { + if got, want := locationUpsert.args[11], 88.0; got != want { t.Fatalf("soc arg = %#v, want %v", got, want) } - if len(locationUpsert.args) != 21 { - t.Fatalf("location upsert args = %d, want 21", len(locationUpsert.args)) + if len(locationUpsert.args) != 18 { + t.Fatalf("location upsert args = %d, want 18", len(locationUpsert.args)) } } @@ -163,10 +163,10 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { if len(exec.calls) != 2 { t.Fatalf("exec calls = %d, want 2", len(exec.calls)) } - if got, want := exec.calls[0].args[5], "沪A12345"; got != want { + if got, want := exec.calls[0].args[2], "沪A12345"; got != want { t.Fatalf("snapshot plate arg = %#v, want %q", got, want) } - if got, want := exec.calls[1].args[5], "沪A12345"; got != want { + if got, want := exec.calls[1].args[2], "沪A12345"; got != want { t.Fatalf("location plate arg = %#v, want %q", got, want) } } @@ -194,7 +194,7 @@ func TestSnapshotWriterKeepsEventPlateWhenPresent(t *testing.T) { if resolver.vin != "" { t.Fatalf("resolver should not be called, got vin=%q", resolver.vin) } - if got, want := exec.calls[0].args[5], "沪A12345"; got != want { + if got, want := exec.calls[0].args[2], "沪A12345"; got != want { t.Fatalf("snapshot plate arg = %#v, want %q", got, want) } } @@ -214,7 +214,7 @@ func TestSnapshotWriterIgnoresMissingBindingPlate(t *testing.T) { }); err != nil { t.Fatalf("Update() error = %v", err) } - if got := exec.calls[0].args[5]; got != "" { + if got := exec.calls[0].args[2]; got != "" { t.Fatalf("snapshot plate arg = %#v, want empty", got) } } @@ -254,6 +254,28 @@ func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) { } } +func TestSnapshotWriterSkipsEmptyVINEvenWhenPhoneExists(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Phone: "13307795425", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Fields: map[string]any{ + envelope.FieldLatitude: 30.590151, + envelope.FieldLongitude: 121.069881, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + func TestSnapshotWriterSkipsGB32960HeartbeatWithoutRealtimePayload(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec)