diff --git a/go/vehicle-gateway/internal/identity/resolver.go b/go/vehicle-gateway/internal/identity/resolver.go index b82a097f..c30b4f2c 100644 --- a/go/vehicle-gateway/internal/identity/resolver.go +++ b/go/vehicle-gateway/internal/identity/resolver.go @@ -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() diff --git a/go/vehicle-gateway/internal/identity/resolver_test.go b/go/vehicle-gateway/internal/identity/resolver_test.go index 730c3da7..1972d42c 100644 --- a/go/vehicle-gateway/internal/identity/resolver_test.go +++ b/go/vehicle-gateway/internal/identity/resolver_test.go @@ -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()