package identity import ( "context" "database/sql" "fmt" "strings" "time" ) type SnapshotRefreshResult struct { BindingEntries int IdentifierEntries int RegistrationEntries int SourceEntries int RefreshedAt time.Time } // identitySnapshot is immutable after atomic publication, so frame handling // performs only local map lookups and never waits for MySQL or a refresh lock. type identitySnapshot struct { bindings map[string]string identifiers map[string]vehicleIdentifierMatch registrations map[string]registrationCacheEntry sources map[string]sourceMetadata refreshedAt time.Time } func (r *MySQLResolver) RefreshSnapshot(ctx context.Context) (SnapshotRefreshResult, error) { if r == nil || r.db == nil { return SnapshotRefreshResult{}, fmt.Errorf("identity snapshot database is not configured") } r.snapshotRefreshMu.Lock() defer r.snapshotRefreshMu.Unlock() next := &identitySnapshot{ bindings: map[string]string{}, identifiers: map[string]vehicleIdentifierMatch{}, registrations: map[string]registrationCacheEntry{}, sources: map[string]sourceMetadata{}, } if err := r.loadBindingSnapshot(ctx, next); err != nil { return SnapshotRefreshResult{}, err } if err := r.loadIdentifierSnapshot(ctx, next); err != nil { return SnapshotRefreshResult{}, err } if err := r.loadRegistrationSnapshot(ctx, next); err != nil { return SnapshotRefreshResult{}, err } if err := r.loadSourceSnapshot(ctx, next); err != nil { return SnapshotRefreshResult{}, err } next.refreshedAt = time.Now() r.snapshot.Store(next) return SnapshotRefreshResult{ BindingEntries: len(next.bindings), IdentifierEntries: len(next.identifiers), RegistrationEntries: len(next.registrations), SourceEntries: len(next.sources), RefreshedAt: next.refreshedAt, }, nil } func (r *MySQLResolver) loadBindingSnapshot(ctx context.Context, target *identitySnapshot) error { rows, err := r.db.QueryContext(ctx, "SELECT vin, plate, phone FROM "+r.table+" WHERE vin IS NOT NULL AND TRIM(vin) <> ''") if err != nil { return fmt.Errorf("load identity binding snapshot: %w", err) } defer rows.Close() ambiguous := map[string]struct{}{} for rows.Next() { var vin, plate, phone sql.NullString if err := rows.Scan(&vin, &plate, &phone); err != nil { return fmt.Errorf("scan identity binding snapshot: %w", err) } vinValue := strings.TrimSpace(vin.String) if vinValue == "" { continue } addSnapshotBinding(target.bindings, ambiguous, bindingSnapshotKey("vin", vinValue), vinValue) if value := strings.TrimSpace(plate.String); value != "" { addSnapshotBinding(target.bindings, ambiguous, bindingSnapshotKey("plate", value), vinValue) } if value := normalizePhone(phone.String); value != "" { addSnapshotBinding(target.bindings, ambiguous, bindingSnapshotKey("phone", value), vinValue) } } if err := rows.Err(); err != nil { return fmt.Errorf("iterate identity binding snapshot: %w", err) } return nil } func (r *MySQLResolver) loadIdentifierSnapshot(ctx context.Context, target *identitySnapshot) error { rows, err := r.db.QueryContext(ctx, `SELECT protocol, source_code, identifier_type, identifier_value, vin, COALESCE(NULLIF(TRIM(oem), ''), source_code) AS platform_name FROM vehicle_identifier WHERE enabled = 1 AND vin IS NOT NULL AND TRIM(vin) <> '' AND identifier_value IS NOT NULL AND TRIM(identifier_value) <> ''`) if err != nil { return fmt.Errorf("load vehicle identifier snapshot: %w", err) } defer rows.Close() ambiguous := map[string]struct{}{} for rows.Next() { var protocol, sourceCode, identifierType, identifierValue, vin, platformName sql.NullString if err := rows.Scan(&protocol, &sourceCode, &identifierType, &identifierValue, &vin, &platformName); err != nil { return fmt.Errorf("scan vehicle identifier snapshot: %w", err) } protocolValue := strings.TrimSpace(protocol.String) typeValue := strings.ToUpper(strings.TrimSpace(identifierType.String)) value := normalizeIdentifierValue(typeValue, identifierValue.String) vinValue := strings.TrimSpace(vin.String) if protocolValue == "" || typeValue == "" || value == "" || vinValue == "" { continue } match := vehicleIdentifierMatch{ VIN: vinValue, SourceCode: strings.TrimSpace(sourceCode.String), PlatformName: strings.TrimSpace(platformName.String), } scopedKey := vehicleIdentifierSnapshotKey(protocolValue, match.SourceCode, typeValue, value) addSnapshotIdentifier(target.identifiers, ambiguous, scopedKey, match) globalKey := vehicleIdentifierSnapshotKey(protocolValue, "", typeValue, value) addSnapshotIdentifier(target.identifiers, ambiguous, globalKey, match) } if err := rows.Err(); err != nil { return fmt.Errorf("iterate vehicle identifier snapshot: %w", err) } return nil } func (r *MySQLResolver) loadRegistrationSnapshot(ctx context.Context, target *identitySnapshot) error { rows, err := r.db.QueryContext(ctx, `SELECT phone, vin, device_id, plate, auth_token FROM jt808_registration WHERE phone IS NOT NULL AND TRIM(phone) <> ''`) if err != nil { return fmt.Errorf("load jt808 registration snapshot: %w", err) } defer rows.Close() for rows.Next() { var phone, vin, deviceID, plate, authToken sql.NullString if err := rows.Scan(&phone, &vin, &deviceID, &plate, &authToken); err != nil { return fmt.Errorf("scan jt808 registration snapshot: %w", err) } phoneValue := normalizePhone(phone.String) if phoneValue == "" { continue } target.registrations[phoneValue] = registrationCacheEntry{ vin: strings.TrimSpace(vin.String), deviceID: strings.TrimSpace(deviceID.String), plate: strings.TrimSpace(plate.String), authToken: strings.TrimSpace(authToken.String), } } if err := rows.Err(); err != nil { return fmt.Errorf("iterate jt808 registration snapshot: %w", err) } return nil } func (r *MySQLResolver) loadSourceSnapshot(ctx context.Context, target *identitySnapshot) error { rows, err := r.db.QueryContext(ctx, `SELECT protocol, source_ip, source_code, platform_name, source_kind FROM vehicle_data_source WHERE enabled = 1 AND source_ip IS NOT NULL AND TRIM(source_ip) <> '' AND ( (source_code IS NOT NULL AND TRIM(source_code) <> '') OR source_kind IN ('PLATFORM', 'DIRECT') OR (platform_name IS NOT NULL AND TRIM(platform_name) <> '') )`) if err != nil { if isOptionalSourceCodeLookupError(err) { return nil } return fmt.Errorf("load vehicle data source snapshot: %w", err) } defer rows.Close() for rows.Next() { var protocol, sourceIP, sourceCode, platformName, sourceKind sql.NullString if err := rows.Scan(&protocol, &sourceIP, &sourceCode, &platformName, &sourceKind); err != nil { return fmt.Errorf("scan vehicle data source snapshot: %w", err) } key := sourceSnapshotKey(protocol.String, sourceIP.String) if key == "" { continue } target.sources[key] = sourceMetadata{ SourceCode: strings.TrimSpace(sourceCode.String), PlatformName: strings.TrimSpace(platformName.String), SourceKind: strings.TrimSpace(sourceKind.String), } } if err := rows.Err(); err != nil { return fmt.Errorf("iterate vehicle data source snapshot: %w", err) } return nil } func addSnapshotBinding(values map[string]string, ambiguous map[string]struct{}, key string, vin string) { if key == "" { return } if _, exists := ambiguous[key]; exists { return } if current, exists := values[key]; exists && !strings.EqualFold(current, vin) { delete(values, key) ambiguous[key] = struct{}{} return } values[key] = vin } func addSnapshotIdentifier(values map[string]vehicleIdentifierMatch, ambiguous map[string]struct{}, key string, match vehicleIdentifierMatch) { if key == "" { return } if _, exists := ambiguous[key]; exists { return } current, exists := values[key] if !exists { values[key] = match return } if !strings.EqualFold(current.VIN, match.VIN) { delete(values, key) ambiguous[key] = struct{}{} return } if !strings.EqualFold(current.SourceCode, match.SourceCode) { current.SourceCode = "" current.PlatformName = "" } else if current.PlatformName != match.PlatformName { current.PlatformName = "" } values[key] = current } func (r *MySQLResolver) snapshotBinding(column string, value string) (string, bool) { key := bindingSnapshotKey(column, value) if key == "" { return "", false } snapshot := r.snapshot.Load() if snapshot == nil { return "", false } vin, ok := snapshot.bindings[key] return vin, ok } func (r *MySQLResolver) snapshotIdentifier(protocolValue string, sourceCode string, identifierType string, value string) (vehicleIdentifierMatch, bool) { key := vehicleIdentifierSnapshotKey(protocolValue, sourceCode, identifierType, value) snapshot := r.snapshot.Load() if snapshot == nil { return vehicleIdentifierMatch{}, false } match, ok := snapshot.identifiers[key] return match, ok } func (r *MySQLResolver) snapshotRegistration(phone string) (registrationCacheEntry, bool) { phone = normalizePhone(phone) snapshot := r.snapshot.Load() if snapshot == nil { return registrationCacheEntry{}, false } entry, ok := snapshot.registrations[phone] return entry, ok } func (r *MySQLResolver) snapshotSource(protocolValue string, sourceIP string) (sourceMetadata, bool) { key := sourceSnapshotKey(protocolValue, sourceIP) snapshot := r.snapshot.Load() if snapshot == nil { return sourceMetadata{}, false } metadata, ok := snapshot.sources[key] return metadata, ok } // JT808AuthToken serves authentication from the same immutable snapshot used // by identity resolution. It deliberately never falls back to a per-frame SQL // query because authentication is on the protocol response hot path. func (r *MySQLResolver) JT808AuthToken(phone string) (string, bool) { entry, ok := r.snapshotRegistration(phone) token := strings.TrimSpace(entry.authToken) return token, ok && token != "" } func bindingSnapshotKey(column string, value string) string { column = strings.ToLower(strings.TrimSpace(column)) value = strings.TrimSpace(value) if column == "phone" { value = normalizePhone(value) } if value == "" || (column != "vin" && column != "plate" && column != "phone") { return "" } return column + "\x00" + value } func vehicleIdentifierSnapshotKey(protocolValue string, sourceCode string, identifierType string, value string) string { protocolValue = strings.TrimSpace(protocolValue) sourceCode = strings.TrimSpace(sourceCode) identifierType = strings.ToUpper(strings.TrimSpace(identifierType)) value = normalizeIdentifierValue(identifierType, value) if protocolValue == "" || identifierType == "" || value == "" { return "" } return "vehicle_identifier\x00" + protocolValue + "\x00" + sourceCode + "\x00" + identifierType + "\x00" + value } func sourceSnapshotKey(protocolValue string, sourceIP string) string { protocolValue = strings.TrimSpace(protocolValue) sourceIP = normalizeEndpointIP(sourceIP) if protocolValue == "" || sourceIP == "" { return "" } return protocolValue + "\x00" + sourceIP }