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

fix(go): backfill realtime plate from binding

Browse Source
This commit is contained in:
lingniu
2026-07-02 16:36:52 +08:00
parent bc399d2819
commit 75b36f4011
3 changed files with 206 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
go/vehicle-gateway/cmd/realtime-api/main.go
View File

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

94
go/vehicle-gateway/internal/realtime/snapshot_writer.go
View File

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

115
go/vehicle-gateway/internal/realtime/snapshot_writer_test.go
View File

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