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

fix(go): resolve jt808 locations through registration

Browse Source
This commit is contained in:
lingniu
2026-07-03 10:30:20 +08:00
parent cabdac1fe2
commit 02075b5ea0
2 changed files with 167 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

113
go/vehicle-gateway/internal/identity/resolver.go
View File

@@ -29,6 +29,7 @@ type MySQLResolver struct {
lookupCacheTTL time.Duration
lookupMu sync.Mutex
lookupCache map[string]lookupCacheEntry
registrationCache map[string]registrationCacheEntry
touchMu sync.Mutex
locationTouches map[string]time.Time
}
@@ -44,6 +45,14 @@ type lookupCacheEntry struct {
expiresAt time.Time
}
type registrationCacheEntry struct {
vin string
deviceID string
plate string
notFound bool
expiresAt time.Time
}
func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver {
return NewMySQLResolverWithOptions(db, table, MySQLResolverOptions{})
}
@@ -70,6 +79,7 @@ func NewMySQLResolverWithOptions(db *sql.DB, table string, opts MySQLResolverOpt
locationTouchInterval: interval,
lookupCacheTTL: lookupTTL,
lookupCache: map[string]lookupCacheEntry{},
registrationCache: map[string]registrationCacheEntry{},
locationTouches: map[string]time.Time{},
}
}
@@ -170,12 +180,115 @@ func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope)
return env, nil
}
}
resolved, ok, err := r.resolveFromJT808Registration(ctx, env)
if err != nil {
return env, err
}
if ok {
env = resolved
if err := r.trackJT808(ctx, env); err != nil {
return env, err
}
return env, nil
}
if err := r.trackJT808(ctx, env); err != nil {
return env, err
}
return env, nil
}
func (r *MySQLResolver) resolveFromJT808Registration(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, bool, error) {
if env.Protocol != envelope.ProtocolJT808 || env.MessageID != "0x0200" || strings.TrimSpace(env.Phone) == "" {
return env, false, nil
}
vin, deviceID, plate, err := r.lookupJT808Registration(ctx, env.Phone)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return env, false, nil
}
return env, false, err
}
env.DeviceID = firstNonEmpty(env.DeviceID, deviceID)
env.Plate = firstNonEmpty(env.Plate, plate)
if strings.TrimSpace(vin) != "" && !strings.EqualFold(strings.TrimSpace(vin), "unknown") {
env.VIN = strings.TrimSpace(vin)
env.EventID = env.StableEventID()
addIdentityMetadata(env.Parsed, "jt808_registration.vin", env.Phone)
return env, true, nil
}
for _, candidate := range []CandidateKey{
{Column: "device_id", Value: env.DeviceID},
{Column: "plate", Value: env.Plate},
} {
value := strings.TrimSpace(candidate.Value)
if value == "" {
continue
}
vin, err := r.lookup(ctx, candidate.Column, value)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
continue
}
return env, false, err
}
if strings.TrimSpace(vin) == "" {
continue
}
env.VIN = strings.TrimSpace(vin)
env.EventID = env.StableEventID()
addIdentityMetadata(env.Parsed, "jt808_registration."+candidate.Column, value)
return env, true, nil
}
return env, false, nil
}
func (r *MySQLResolver) lookupJT808Registration(ctx context.Context, phone string) (vin string, deviceID string, plate string, err error) {
phone = normalizePhone(phone)
now := time.Now()
r.lookupMu.Lock()
cached, ok := r.registrationCache[phone]
if ok && now.Before(cached.expiresAt) {
r.lookupMu.Unlock()
if cached.notFound {
return "", "", "", sql.ErrNoRows
}
return cached.vin, cached.deviceID, cached.plate, nil
}
r.lookupMu.Unlock()
var vinValue, deviceValue, plateValue sql.NullString
err = r.db.QueryRowContext(ctx, "SELECT vin, device_id, plate FROM jt808_registration WHERE phone = ?", phone).
Scan(&vinValue, &deviceValue, &plateValue)
if err != nil && !errors.Is(err, sql.ErrNoRows) {
return "", "", "", err
}
entry := registrationCacheEntry{
vin: strings.TrimSpace(vinValue.String),
deviceID: strings.TrimSpace(deviceValue.String),
plate: strings.TrimSpace(plateValue.String),
notFound: errors.Is(err, sql.ErrNoRows),
expiresAt: now.Add(r.lookupCacheTTL),
}
r.lookupMu.Lock()
r.registrationCache[phone] = entry
r.lookupMu.Unlock()
if entry.notFound {
return "", "", "", sql.ErrNoRows
}
return entry.vin, entry.deviceID, entry.plate, nil
}
func addIdentityMetadata(parsed map[string]any, source string, value string) {
if parsed == nil {
return
}
parsed["identity"] = map[string]any{
"resolved": true,
"source": source,
"value": value,
}
}
func (r *MySQLResolver) lookup(ctx context.Context, column string, value string) (string, error) {
cacheKey := column + "\x00" + value
now := time.Now()

54
go/vehicle-gateway/internal/identity/resolver_test.go
View File

@@ -187,6 +187,9 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) {
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("13307795425").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT vin, device_id, plate FROM jt808_registration WHERE phone = \\?").
WithArgs("13307795425").
WillReturnError(sql.ErrNoRows)
mock.ExpectExec("INSERT INTO jt808_registration").
WillReturnResult(sqlmock.NewResult(1, 1))
@@ -208,6 +211,57 @@ func TestMySQLResolverTracksFirstJT808LocationThenThrottles(t *testing.T) {
}
}
func TestMySQLResolverUsesJT808RegistrationPlateForLocationVIN(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
WithArgs("40692934322").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT vin, device_id, plate FROM jt808_registration WHERE phone = \\?").
WithArgs("40692934322").
WillReturnRows(sqlmock.NewRows([]string{"vin", "device_id", "plate"}).AddRow("unknown", "18285", "粤AG18285"))
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE device_id = \\?").
WithArgs("18285").
WillReturnError(sql.ErrNoRows)
mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE plate = \\?").
WithArgs("粤AG18285").
WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LNXNEGRR7SR318212"))
mock.ExpectExec("INSERT INTO jt808_registration").
WillReturnResult(sqlmock.NewResult(1, 1))
mock.ExpectExec("UPDATE IGNORE vehicle_identity_binding").
WithArgs("40692934322", "40692934322", "18285", "18285", "粤AG18285", "粤AG18285", "LNXNEGRR7SR318212").
WillReturnResult(sqlmock.NewResult(0, 1))
resolver := NewMySQLResolver(db, "vehicle_identity_binding")
env, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
Protocol: envelope.ProtocolJT808,
MessageID: "0x0200",
Phone: "40692934322",
SourceEndpoint: "115.231.168.135:47822",
Parsed: map[string]any{},
Fields: map[string]any{
envelope.FieldLatitude: 30.1,
envelope.FieldLongitude: 120.1,
},
})
if err != nil {
t.Fatalf("Resolve() error = %v", err)
}
if env.VIN != "LNXNEGRR7SR318212" {
t.Fatalf("vin = %q", env.VIN)
}
if env.DeviceID != "18285" || env.Plate != "粤AG18285" {
t.Fatalf("registration identity not copied: device=%q plate=%q", env.DeviceID, env.Plate)
}
identity, ok := env.Parsed["identity"].(map[string]any)
if !ok || identity["source"] != "jt808_registration.plate" {
t.Fatalf("identity metadata = %#v", env.Parsed["identity"])
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestMySQLResolverCachesIdentityMissesForHighFrequencyFrames(t *testing.T) {
db, mock := newMockDB(t)
defer db.Close()