From 75b36f40116ea9498595f9f5d904327889df24cc Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 16:36:52 +0800 Subject: [PATCH] fix(go): backfill realtime plate from binding --- go/vehicle-gateway/cmd/realtime-api/main.go | 5 +- .../internal/realtime/snapshot_writer.go | 94 +++++++++++++- .../internal/realtime/snapshot_writer_test.go | 115 ++++++++++++++++++ 3 files changed, 206 insertions(+), 8 deletions(-) diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index c0e978bf..e387e11e 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -61,7 +61,8 @@ func main() { os.Exit(1) } closeStats = func() { _ = db.Close() } - snapshotWriter := realtime.NewSnapshotWriter(db) + bindingTable := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding") + snapshotWriter := realtime.NewSnapshotWriterWithPlateResolver(db, realtime.NewBindingPlateResolver(db, bindingTable)) if env("MYSQL_REALTIME_SNAPSHOT_ENABLED", "true") != "false" { if err := snapshotWriter.EnsureSchema(ctx); err != nil { _ = db.Close() @@ -69,7 +70,7 @@ func main() { os.Exit(1) } updater = compositeRealtimeUpdater{primary: repository, secondary: snapshotWriter} - logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot") + logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot", "plate_binding_table", bindingTable) } mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) logger.Info("stats mysql query enabled") diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index e979b87e..92867e08 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "encoding/json" + "errors" "strings" "time" @@ -14,15 +15,24 @@ type SnapshotExecer interface { ExecContext(context.Context, string, ...any) (sql.Result, error) } +type PlateResolver interface { + PlateByVIN(context.Context, string) (string, error) +} + type SnapshotWriter struct { - exec SnapshotExecer + exec SnapshotExecer + plateResolver PlateResolver } func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter { + return NewSnapshotWriterWithPlateResolver(exec, nil) +} + +func NewSnapshotWriterWithPlateResolver(exec SnapshotExecer, plateResolver PlateResolver) *SnapshotWriter { if exec == nil { panic("snapshot execer must not be nil") } - return &SnapshotWriter{exec: exec} + return &SnapshotWriter{exec: exec, plateResolver: plateResolver} } func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { @@ -41,6 +51,10 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) if !hasRealtimePayload(env) { return nil } + plate, err := w.plateForEnvelope(ctx, env) + if err != nil { + return err + } fieldsJSON, err := marshalObject(env.Fields) if err != nil { return err @@ -57,7 +71,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) strings.TrimSpace(env.VIN), strings.TrimSpace(env.Phone), strings.TrimSpace(env.DeviceID), - strings.TrimSpace(env.Plate), + plate, strings.TrimSpace(env.MessageID), env.Sequence, strings.TrimSpace(env.SourceEndpoint), @@ -69,7 +83,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) ); err != nil { return err } - location, ok := realtimeLocationFromEnvelope(env, vehicleKey, fieldsJSON) + location, ok := realtimeLocationFromEnvelope(env, vehicleKey, fieldsJSON, plate) if !ok { return nil } @@ -100,6 +114,27 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) return err } +func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.FrameEnvelope) (string, error) { + if plate := strings.TrimSpace(env.Plate); plate != "" { + return plate, nil + } + if w.plateResolver == nil { + return "", nil + } + vin := strings.TrimSpace(env.VIN) + if vin == "" { + return "", nil + } + plate, err := w.plateResolver.PlateByVIN(ctx, vin) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return "", nil + } + return "", err + } + return strings.TrimSpace(plate), nil +} + type realtimeLocationRow struct { Protocol string VehicleKey string @@ -125,7 +160,7 @@ type realtimeLocationRow struct { EventID string } -func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, fieldsJSON string) (realtimeLocationRow, bool) { +func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, fieldsJSON string, plate string) (realtimeLocationRow, bool) { latitude, okLat := numberField(env.Fields, envelope.FieldLatitude) longitude, okLon := numberField(env.Fields, envelope.FieldLongitude) if !okLat || !okLon { @@ -137,7 +172,7 @@ func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, VIN: strings.TrimSpace(env.VIN), Phone: strings.TrimSpace(env.Phone), DeviceID: strings.TrimSpace(env.DeviceID), - Plate: strings.TrimSpace(env.Plate), + Plate: plate, MessageID: strings.TrimSpace(env.MessageID), Sequence: env.Sequence, SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), @@ -200,6 +235,53 @@ func numberField(fields map[string]any, key string) (float64, bool) { } } +type BindingPlateResolver struct { + queryer Queryer + table string +} + +type Queryer interface { + QueryRowContext(context.Context, string, ...any) *sql.Row +} + +func NewBindingPlateResolver(queryer Queryer, table string) *BindingPlateResolver { + if queryer == nil { + panic("plate binding queryer must not be nil") + } + table = strings.TrimSpace(table) + if table == "" || !safeIdentifier(table) { + table = "vehicle_identity_binding" + } + return &BindingPlateResolver{queryer: queryer, table: table} +} + +func (r *BindingPlateResolver) PlateByVIN(ctx context.Context, vin string) (string, error) { + vin = strings.TrimSpace(vin) + if vin == "" { + return "", sql.ErrNoRows + } + query := "SELECT plate FROM " + r.table + " WHERE vin = ? AND plate IS NOT NULL AND plate <> '' ORDER BY updated_at DESC LIMIT 1" + var plate string + err := r.queryer.QueryRowContext(ctx, query, vin).Scan(&plate) + if err != nil { + return "", err + } + return strings.TrimSpace(plate), nil +} + +func safeIdentifier(value string) bool { + if value == "" { + return false + } + for _, r := range value { + if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' { + continue + } + return false + } + return true +} + func marshalObject(value map[string]any) (string, error) { if value == nil { value = map[string]any{} diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 5e07a1de..efcaadaa 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -3,6 +3,7 @@ package realtime import ( "context" "database/sql" + "errors" "strings" "testing" @@ -126,6 +127,109 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) } } +func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { + exec := &recordingSnapshotExec{} + resolver := &recordingPlateResolver{plate: "沪A12345"} + writer := NewSnapshotWriterWithPlateResolver(exec, resolver) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{"data_units": []any{}}, + Fields: map[string]any{ + envelope.FieldLatitude: 30.590151, + envelope.FieldLongitude: 121.069881, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + if resolver.vin != "VIN001" { + t.Fatalf("resolver vin = %q", resolver.vin) + } + 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 { + t.Fatalf("snapshot plate arg = %#v, want %q", got, want) + } + if got, want := exec.calls[1].args[5], "沪A12345"; got != want { + t.Fatalf("location plate arg = %#v, want %q", got, want) + } +} + +func TestSnapshotWriterKeepsEventPlateWhenPresent(t *testing.T) { + exec := &recordingSnapshotExec{} + resolver := &recordingPlateResolver{plate: "沪B99999"} + writer := NewSnapshotWriterWithPlateResolver(exec, resolver) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + VIN: "VIN001", + Plate: "沪A12345", + 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 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 { + t.Fatalf("snapshot plate arg = %#v, want %q", got, want) + } +} + +func TestSnapshotWriterIgnoresMissingBindingPlate(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriterWithPlateResolver(exec, &recordingPlateResolver{err: sql.ErrNoRows}) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{"data_units": []any{}}, + Fields: map[string]any{envelope.FieldSOCPercent: 90}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if got := exec.calls[0].args[5]; got != "" { + t.Fatalf("snapshot plate arg = %#v, want empty", got) + } +} + +func TestSnapshotWriterReturnsUnexpectedPlateLookupError(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriterWithPlateResolver(exec, &recordingPlateResolver{err: errors.New("db down")}) + + err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{"data_units": []any{}}, + Fields: map[string]any{envelope.FieldSOCPercent: 90}, + }) + if err == nil || !strings.Contains(err.Error(), "db down") { + t.Fatalf("Update() error = %v, want db down", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec) @@ -176,3 +280,14 @@ func (e *recordingSnapshotExec) ExecContext(_ context.Context, query string, arg e.calls = append(e.calls, snapshotExecCall{query: query, args: args}) return nil, nil } + +type recordingPlateResolver struct { + vin string + plate string + err error +} + +func (r *recordingPlateResolver) PlateByVIN(_ context.Context, vin string) (string, error) { + r.vin = vin + return r.plate, r.err +}