From e7d024405ae90ee77e23b02ccae871b695896236 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 10:24:44 +0800 Subject: [PATCH] fix: throttle jt808 registration touches --- go/vehicle-gateway/cmd/gateway/main.go | 5 +- .../internal/gateway/tcp_server_test.go | 2 +- .../internal/identity/resolver.go | 64 +++++++++++++++++-- .../internal/identity/resolver_test.go | 46 +++++++++---- .../internal/protocol/jt808/parser.go | 17 +++-- .../internal/protocol/jt808/parser_test.go | 6 +- 6 files changed, 111 insertions(+), 29 deletions(-) diff --git a/go/vehicle-gateway/cmd/gateway/main.go b/go/vehicle-gateway/cmd/gateway/main.go index c59c2479..246c4f4d 100644 --- a/go/vehicle-gateway/cmd/gateway/main.go +++ b/go/vehicle-gateway/cmd/gateway/main.go @@ -138,7 +138,10 @@ func buildIdentityResolver(ctx context.Context, logger *slog.Logger) (identity.R } table := env("VEHICLE_IDENTITY_TABLE", "vehicle_identity_binding") logger.Info("identity mysql resolver enabled", "table", table) - return identity.NewMySQLResolver(db, table), func() { _ = db.Close() }, nil + locationTouchInterval := time.Duration(envInt("JT808_REGISTRATION_LOCATION_TOUCH_INTERVAL_SECONDS", 600)) * time.Second + return identity.NewMySQLResolverWithOptions(db, table, identity.MySQLResolverOptions{ + LocationTouchInterval: locationTouchInterval, + }), func() { _ = db.Close() }, nil } func buildSink(ctx context.Context, logger *slog.Logger) (eventbus.Sink, error) { diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index 82d59dab..f9b1e647 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -37,7 +37,7 @@ func TestTCPServerPublishesGoodFrameToRawAndUnified(t *testing.T) { if len(sink.raw) != 1 || len(sink.unified) != 1 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) } - if sink.raw[0].Phone != "013307795425" { + if sink.raw[0].Phone != "13307795425" { t.Fatalf("phone = %q", sink.raw[0].Phone) } if sink.raw[0].Fields[envelope.FieldTotalMileageKM] != 10241.2 { diff --git a/go/vehicle-gateway/internal/identity/resolver.go b/go/vehicle-gateway/internal/identity/resolver.go index d9a425c6..31135d1d 100644 --- a/go/vehicle-gateway/internal/identity/resolver.go +++ b/go/vehicle-gateway/internal/identity/resolver.go @@ -6,6 +6,8 @@ import ( "errors" "fmt" "strings" + "sync" + "time" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) @@ -21,11 +23,22 @@ func (NoopResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (enve } type MySQLResolver struct { - db *sql.DB - table string + db *sql.DB + table string + locationTouchInterval time.Duration + touchMu sync.Mutex + locationTouches map[string]time.Time +} + +type MySQLResolverOptions struct { + LocationTouchInterval time.Duration } func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver { + return NewMySQLResolverWithOptions(db, table, MySQLResolverOptions{}) +} + +func NewMySQLResolverWithOptions(db *sql.DB, table string, opts MySQLResolverOptions) *MySQLResolver { if db == nil { panic("identity db must not be nil") } @@ -33,7 +46,16 @@ func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver { if table == "" || !safeIdentifier(table) { table = "vehicle_identity_binding" } - return &MySQLResolver{db: db, table: table} + interval := opts.LocationTouchInterval + if interval <= 0 { + interval = 10 * time.Minute + } + return &MySQLResolver{ + db: db, + table: table, + locationTouchInterval: interval, + locationTouches: map[string]time.Time{}, + } } func safeIdentifier(value string) bool { @@ -50,6 +72,7 @@ func safeIdentifier(value string) bool { } func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) { + env.Phone = normalizePhone(env.Phone) if strings.TrimSpace(env.VIN) != "" { if err := r.trackJT808(ctx, env); err != nil { return env, err @@ -99,6 +122,9 @@ func (r *MySQLResolver) trackJT808(ctx context.Context, env envelope.FrameEnvelo if env.Protocol != envelope.ProtocolJT808 || strings.TrimSpace(env.Phone) == "" || strings.TrimSpace(env.MessageID) == "" { return nil } + if env.MessageID != "0x0100" && env.MessageID != "0x0102" && !r.shouldTouchJT808Location(env) { + return nil + } registration := mapValue(env.Parsed, "registration") authentication := mapValue(env.Parsed, "authentication") deviceID := firstNonEmpty(env.DeviceID, textValue(registration, "device_id")) @@ -170,6 +196,25 @@ WHERE vin = ?`, env.Phone, env.Phone, deviceID, deviceID, plate, plate, vin) return err } +func (r *MySQLResolver) shouldTouchJT808Location(env envelope.FrameEnvelope) bool { + if env.MessageID != "0x0200" { + return false + } + phone := normalizePhone(env.Phone) + if phone == "" { + return false + } + now := time.Now() + r.touchMu.Lock() + defer r.touchMu.Unlock() + last, ok := r.locationTouches[phone] + if ok && now.Sub(last) < r.locationTouchInterval { + return false + } + r.locationTouches[phone] = now + return true +} + type CandidateKey struct { Column string Value string @@ -189,16 +234,21 @@ func CandidateKeys(env envelope.FrameEnvelope) []CandidateKey { } out = append(out, CandidateKey{Column: column, Value: value}) } - add("phone", env.Phone) - if trimmed := strings.TrimLeft(strings.TrimSpace(env.Phone), "0"); trimmed != "" { - add("phone", trimmed) - } + add("phone", normalizePhone(env.Phone)) add("device_id", env.DeviceID) add("plate", env.Plate) add("vin", env.VehicleKeyHint) return out } +func normalizePhone(phone string) string { + trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0") + if trimmed == "" { + return strings.TrimSpace(phone) + } + return trimmed +} + func mapValue(values map[string]any, key string) map[string]any { if values == nil { return nil diff --git a/go/vehicle-gateway/internal/identity/resolver_test.go b/go/vehicle-gateway/internal/identity/resolver_test.go index 151e6219..085d4d39 100644 --- a/go/vehicle-gateway/internal/identity/resolver_test.go +++ b/go/vehicle-gateway/internal/identity/resolver_test.go @@ -21,7 +21,7 @@ func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) { for _, key := range keys { got = append(got, key.Column+"="+key.Value) } - want := []string{"phone=013307795425", "phone=13307795425", "device_id=D1", "plate=豫A12345", "vin=D1"} + want := []string{"phone=13307795425", "device_id=D1", "plate=豫A12345", "vin=D1"} if len(got) != len(want) { t.Fatalf("candidate count = %d got=%#v", len(got), got) } @@ -32,7 +32,7 @@ func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) { } } -func TestCandidateKeysAddsTrimmedPhoneVariant(t *testing.T) { +func TestCandidateKeysNormalizesPhone(t *testing.T) { keys := CandidateKeys(envelope.FrameEnvelope{ Phone: "013079963379", }) @@ -40,7 +40,7 @@ func TestCandidateKeysAddsTrimmedPhoneVariant(t *testing.T) { for _, key := range keys { got = append(got, key.Column+"="+key.Value) } - want := []string{"phone=013079963379", "phone=13079963379"} + want := []string{"phone=13079963379"} if len(got) != len(want) { t.Fatalf("candidate count = %d got=%#v", len(got), got) } @@ -55,7 +55,7 @@ func TestMySQLResolverFillsVINFromPhone(t *testing.T) { db, mock := newMockDB(t) defer db.Close() mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). - WithArgs("013307795425"). + WithArgs("13307795425"). WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LNBVIN00000000001")) resolver := NewMySQLResolver(db, "vehicle_identity_binding") @@ -82,9 +82,6 @@ func TestMySQLResolverFillsVINFromPhone(t *testing.T) { func TestMySQLResolverFallsBackToDeviceID(t *testing.T) { db, mock := newMockDB(t) defer db.Close() - mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). - WithArgs("013307795425"). - WillReturnError(sql.ErrNoRows) mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). WithArgs("13307795425"). WillReturnError(sql.ErrNoRows) @@ -112,9 +109,6 @@ func TestMySQLResolverFallsBackToDeviceID(t *testing.T) { func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) { db, mock := newMockDB(t) defer db.Close() - mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). - WithArgs("013079963379"). - WillReturnError(sql.ErrNoRows) mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). WithArgs("13079963379"). WillReturnError(sql.ErrNoRows) @@ -127,7 +121,7 @@ func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) { mock.ExpectExec("INSERT INTO jt808_registration"). WillReturnResult(sqlmock.NewResult(1, 1)) mock.ExpectExec("UPDATE IGNORE vehicle_identity_binding"). - WithArgs("013079963379", "013079963379", "DEV0001", "DEV0001", "TEST123", "TEST123", "LKLG7C4E3NA774736"). + WithArgs("13079963379", "13079963379", "DEV0001", "DEV0001", "TEST123", "TEST123", "LKLG7C4E3NA774736"). WillReturnResult(sqlmock.NewResult(0, 1)) resolver := NewMySQLResolver(db, "vehicle_identity_binding") @@ -161,6 +155,36 @@ func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) { } } +func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) { + db, mock := newMockDB(t) + defer db.Close() + mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). + WithArgs("13307795425"). + WillReturnError(sql.ErrNoRows) + mock.ExpectExec("INSERT INTO jt808_registration"). + WillReturnResult(sqlmock.NewResult(1, 1)) + mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?"). + WithArgs("13307795425"). + WillReturnError(sql.ErrNoRows) + + resolver := NewMySQLResolver(db, "vehicle_identity_binding") + for i := 0; i < 2; i++ { + _, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Phone: "013307795425", + SourceEndpoint: "115.231.168.135:43625", + Parsed: map[string]any{}, + }) + if err != nil { + t.Fatalf("Resolve() error = %v", err) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func newMockDB(t *testing.T) (*sql.DB, sqlmock.Sqlmock) { t.Helper() db, mock, err := sqlmock.New() diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser.go b/go/vehicle-gateway/internal/protocol/jt808/parser.go index 6516a648..b6c5c9a4 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser.go @@ -112,11 +112,8 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope } phoneBCD := raw[phoneStart:phoneEnd] - phone := bcdString(phoneBCD) - phoneTrimZero := strings.TrimLeft(phone, "0") - if versioned && phoneTrimZero != "" { - phone = phoneTrimZero - } + phoneRaw := bcdString(phoneBCD) + phone := normalizePhone(phoneRaw) body := raw[headerLen : headerLen+int(bodySize)] parsed := map[string]any{ "header": map[string]any{ @@ -127,7 +124,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope "body_size": bodySize, "phone_bcd_hex": strings.ToUpper(hex.EncodeToString(phoneBCD)), "phone": phone, - "phone_trim_zero": phoneTrimZero, + "phone_raw": phoneRaw, "sequence": binary.BigEndian.Uint16(raw[serialStart : serialStart+2]), "subpackage": subpackage, "package_info": packageInfo, @@ -230,6 +227,14 @@ func parseRegistration(body []byte, versioned bool) (map[string]any, error) { return registration, nil } +func normalizePhone(phone string) string { + trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0") + if trimmed == "" { + return strings.TrimSpace(phone) + } + return trimmed +} + func parseAuthentication(body []byte, versioned bool) map[string]any { if versioned { if out, ok := parseVersionedAuthentication(body); ok { diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go index 6e0152f2..69518039 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go @@ -39,7 +39,7 @@ func TestExtractFramesUnescapesAndParsesLocationWithTotalMileage(t *testing.T) { if env.MessageID != "0x0200" { t.Fatalf("message id = %q", env.MessageID) } - if env.Phone != "013307795425" { + if env.Phone != "13307795425" { t.Fatalf("phone = %q", env.Phone) } if env.Sequence != 1 { @@ -140,7 +140,7 @@ func TestParseFrameParsesCommonLocationAdditionalItems(t *testing.T) { if err != nil { t.Fatalf("ParseFrame() error = %v", err) } - if env.Phone != "012345678901" { + if env.Phone != "12345678901" { t.Fatalf("phone = %q", env.Phone) } assertFloatField(t, env, envelope.FieldTotalMileageKM, 1.1) @@ -223,7 +223,7 @@ func TestParseFrameParsesRegistrationIdentity(t *testing.T) { if env.MessageID != "0x0100" { t.Fatalf("message id = %q", env.MessageID) } - if env.Phone != "013079963379" { + if env.Phone != "13079963379" { t.Fatalf("phone = %q", env.Phone) } if env.DeviceID != "DEV0001" {