diff --git a/go/vehicle-gateway/internal/identity/resolver.go b/go/vehicle-gateway/internal/identity/resolver.go index a8dad1cf..d9a425c6 100644 --- a/go/vehicle-gateway/internal/identity/resolver.go +++ b/go/vehicle-gateway/internal/identity/resolver.go @@ -4,6 +4,7 @@ import ( "context" "database/sql" "errors" + "fmt" "strings" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" @@ -50,6 +51,9 @@ func safeIdentifier(value string) bool { func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) { if strings.TrimSpace(env.VIN) != "" { + if err := r.trackJT808(ctx, env); err != nil { + return env, err + } return env, nil } candidates := CandidateKeys(env) @@ -72,9 +76,15 @@ func (r *MySQLResolver) Resolve(ctx context.Context, env envelope.FrameEnvelope) "value": candidate.Value, } env.EventID = env.StableEventID() + 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 } @@ -85,6 +95,81 @@ func (r *MySQLResolver) lookup(ctx context.Context, column string, value string) return vin, err } +func (r *MySQLResolver) trackJT808(ctx context.Context, env envelope.FrameEnvelope) error { + if env.Protocol != envelope.ProtocolJT808 || strings.TrimSpace(env.Phone) == "" || strings.TrimSpace(env.MessageID) == "" { + return nil + } + registration := mapValue(env.Parsed, "registration") + authentication := mapValue(env.Parsed, "authentication") + deviceID := firstNonEmpty(env.DeviceID, textValue(registration, "device_id")) + plate := firstNonEmpty(env.Plate, textValue(registration, "plate")) + vin := strings.TrimSpace(env.VIN) + if vin == "" { + vin = "unknown" + } + firstRegisteredAt := nilTime() + latestRegisteredAt := nilTime() + if env.MessageID == "0x0100" { + firstRegisteredAt = "CURRENT_TIMESTAMP" + latestRegisteredAt = "CURRENT_TIMESTAMP" + } + latestAuthenticatedAt := nilTime() + if env.MessageID == "0x0102" { + latestAuthenticatedAt = "CURRENT_TIMESTAMP" + } + + query := `INSERT INTO jt808_registration + (phone, device_id, plate, vin, province, city, manufacturer, device_type, plate_color, + auth_token, auth_imei, auth_software_version, source_endpoint, + first_registered_at, latest_registered_at, latest_authenticated_at) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ` + firstRegisteredAt + `, ` + latestRegisteredAt + `, ` + latestAuthenticatedAt + `) +ON DUPLICATE KEY UPDATE + device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), + plate = IF(VALUES(plate) <> '', VALUES(plate), plate), + vin = IF(VALUES(vin) <> 'unknown', VALUES(vin), vin), + province = IF(VALUES(province) <> '', VALUES(province), province), + city = IF(VALUES(city) <> '', VALUES(city), city), + manufacturer = IF(VALUES(manufacturer) <> '', VALUES(manufacturer), manufacturer), + device_type = IF(VALUES(device_type) <> '', VALUES(device_type), device_type), + plate_color = IF(VALUES(plate_color) <> '', VALUES(plate_color), plate_color), + auth_token = IF(VALUES(auth_token) <> '', VALUES(auth_token), auth_token), + auth_imei = IF(VALUES(auth_imei) <> '', VALUES(auth_imei), auth_imei), + auth_software_version = IF(VALUES(auth_software_version) <> '', VALUES(auth_software_version), auth_software_version), + source_endpoint = IF(VALUES(source_endpoint) <> '', VALUES(source_endpoint), source_endpoint), + latest_seen_at = CURRENT_TIMESTAMP, + first_registered_at = COALESCE(first_registered_at, VALUES(first_registered_at)), + latest_registered_at = COALESCE(VALUES(latest_registered_at), latest_registered_at), + latest_authenticated_at = COALESCE(VALUES(latest_authenticated_at), latest_authenticated_at)` + _, err := r.db.ExecContext(ctx, query, + env.Phone, + deviceID, + plate, + vin, + textValue(registration, "province"), + textValue(registration, "city"), + textValue(registration, "manufacturer"), + textValue(registration, "device_type"), + textValue(registration, "plate_color"), + textValue(authentication, "token"), + textValue(authentication, "imei"), + textValue(authentication, "software_version"), + env.SourceEndpoint, + ) + if err != nil { + return err + } + if vin == "unknown" { + return nil + } + _, err = r.db.ExecContext(ctx, `UPDATE IGNORE `+r.table+` +SET + phone = CASE WHEN (phone IS NULL OR phone = '') AND ? <> '' THEN ? ELSE phone END, + device_id = CASE WHEN (device_id IS NULL OR device_id = '') AND ? <> '' THEN ? ELSE device_id END, + plate = CASE WHEN (plate IS NULL OR plate = '') AND ? <> '' THEN ? ELSE plate END +WHERE vin = ?`, env.Phone, env.Phone, deviceID, deviceID, plate, plate, vin) + return err +} + type CandidateKey struct { Column string Value string @@ -113,3 +198,40 @@ func CandidateKeys(env envelope.FrameEnvelope) []CandidateKey { add("vin", env.VehicleKeyHint) return out } + +func mapValue(values map[string]any, key string) map[string]any { + if values == nil { + return nil + } + nested, _ := values[key].(map[string]any) + return nested +} + +func textValue(values map[string]any, key string) string { + if values == nil { + return "" + } + value, ok := values[key] + if !ok || value == nil { + return "" + } + switch typed := value.(type) { + case string: + return strings.TrimSpace(typed) + default: + return strings.TrimSpace(fmt.Sprint(typed)) + } +} + +func firstNonEmpty(values ...string) string { + for _, value := range values { + if trimmed := strings.TrimSpace(value); trimmed != "" { + return trimmed + } + } + return "" +} + +func nilTime() string { + return "NULL" +} diff --git a/go/vehicle-gateway/internal/identity/resolver_test.go b/go/vehicle-gateway/internal/identity/resolver_test.go index 285f8199..151e6219 100644 --- a/go/vehicle-gateway/internal/identity/resolver_test.go +++ b/go/vehicle-gateway/internal/identity/resolver_test.go @@ -109,6 +109,58 @@ 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) + mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE device_id = \\?"). + WithArgs("DEV0001"). + WillReturnError(sql.ErrNoRows) + mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE plate = \\?"). + WithArgs("TEST123"). + WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LKLG7C4E3NA774736")) + 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"). + WillReturnResult(sqlmock.NewResult(0, 1)) + + resolver := NewMySQLResolver(db, "vehicle_identity_binding") + env, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0100", + Phone: "013079963379", + DeviceID: "DEV0001", + Plate: "TEST123", + SourceEndpoint: "115.231.168.135:43625", + Parsed: map[string]any{ + "registration": map[string]any{ + "province": uint16(16), + "city": uint16(32), + "manufacturer": "YUTNG", + "device_type": "ZK6105CHEVNPG4", + "device_id": "DEV0001", + "plate_color": uint8(2), + "plate": "TEST123", + }, + }, + }) + if err != nil { + t.Fatalf("Resolve() error = %v", err) + } + if env.VIN != "LKLG7C4E3NA774736" { + t.Fatalf("vin = %q", env.VIN) + } + 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 db927e6c..28fd49a9 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser.go @@ -151,6 +151,24 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope if location.TotalMileageKM != nil { fields[envelope.FieldTotalMileageKM] = *location.TotalMileageKM } + } else if messageID == 0x0100 { + registration, err := parseRegistration(body) + if err != nil { + return envelope.FrameEnvelope{}, err + } + parsed["registration"] = registration + if deviceID, ok := registration["device_id"].(string); ok { + fields["device_id"] = deviceID + } + if plate, ok := registration["plate"].(string); ok { + fields["plate"] = plate + } + } else if messageID == 0x0102 { + authentication := parseAuthentication(body) + parsed["authentication"] = authentication + if token, ok := authentication["token"].(string); ok { + fields["auth_token"] = token + } } env := envelope.FrameEnvelope{ @@ -166,10 +184,36 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope Fields: fields, ParseStatus: envelope.ParseOK, } + if registration, ok := parsed["registration"].(map[string]any); ok { + env.DeviceID, _ = registration["device_id"].(string) + env.Plate, _ = registration["plate"].(string) + } env.EventID = env.StableEventID() return env, nil } +func parseRegistration(body []byte) (map[string]any, error) { + if len(body) < 37 { + return nil, fmt.Errorf("%w: registration body len=%d", ErrFrameTooShort, len(body)) + } + registration := map[string]any{ + "province": binary.BigEndian.Uint16(body[0:2]), + "city": binary.BigEndian.Uint16(body[2:4]), + "manufacturer": fixedText(body[4:9]), + "device_type": fixedText(body[9:29]), + "device_id": fixedText(body[29:36]), + "plate_color": body[36], + "plate": freeText(body[37:]), + } + return registration, nil +} + +func parseAuthentication(body []byte) map[string]any { + token := freeText(body) + out := map[string]any{"token": token} + return out +} + func parseLocation(body []byte) (Location, error) { if len(body) < 28 { return Location{}, fmt.Errorf("%w: location body len=%d", ErrFrameTooShort, len(body)) @@ -269,6 +313,14 @@ func bcdString(data []byte) string { return string(out) } +func fixedText(data []byte) string { + return strings.Trim(freeText(data), "\x00 ") +} + +func freeText(data []byte) string { + return strings.TrimRight(strings.TrimSpace(string(data)), "\x00") +} + func parseBCDTime(data []byte) time.Time { if len(data) != 6 { return time.Time{} diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go index 3ceaa58e..2e084bf1 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go @@ -1,6 +1,7 @@ package jt808 import ( + "encoding/binary" "encoding/hex" "testing" @@ -114,6 +115,61 @@ func TestParseFrameSupportsJT8082019VersionedHeader(t *testing.T) { } } +func TestParseFrameParsesRegistrationIdentity(t *testing.T) { + body := make([]byte, 0, 46) + body = binary.BigEndian.AppendUint16(body, 16) + body = binary.BigEndian.AppendUint16(body, 32) + body = appendFixedASCII(body, "YUTNG", 5) + body = appendFixedASCII(body, "ZK6105CHEVNPG4", 20) + body = appendFixedASCII(body, "DEV0001", 7) + body = append(body, 2) + body = append(body, []byte("TEST123")...) + + env, err := ParseFrame(buildFrame(0x0100, "013079963379", 7, body), 1782918600000, "115.231.168.135:43625") + if err != nil { + t.Fatalf("ParseFrame() error = %v", err) + } + if env.MessageID != "0x0100" { + t.Fatalf("message id = %q", env.MessageID) + } + if env.Phone != "013079963379" { + t.Fatalf("phone = %q", env.Phone) + } + if env.DeviceID != "DEV0001" { + t.Fatalf("device id = %q", env.DeviceID) + } + if env.Plate != "TEST123" { + t.Fatalf("plate = %q", env.Plate) + } + registration, ok := env.Parsed["registration"].(map[string]any) + if !ok { + t.Fatalf("registration missing: %#v", env.Parsed) + } + if registration["manufacturer"] != "YUTNG" || registration["device_type"] != "ZK6105CHEVNPG4" { + t.Fatalf("registration fields = %#v", registration) + } + if registration["plate_color"] != uint8(2) { + t.Fatalf("plate color = %#v", registration["plate_color"]) + } +} + +func TestParseFrameParsesAuthenticationToken(t *testing.T) { + env, err := ParseFrame(buildFrame(0x0102, "064646848757", 8, []byte("g7gps")), 1782918600000, "115.231.168.135:43625") + if err != nil { + t.Fatalf("ParseFrame() error = %v", err) + } + if env.MessageID != "0x0102" { + t.Fatalf("message id = %q", env.MessageID) + } + auth, ok := env.Parsed["authentication"].(map[string]any) + if !ok { + t.Fatalf("authentication missing: %#v", env.Parsed) + } + if auth["token"] != "g7gps" { + t.Fatalf("token = %#v", auth["token"]) + } +} + func TestExtractFramesHandlesEscapedPayload(t *testing.T) { payload := []byte{0x02, 0x00, 0x00, 0x02, 0x01, 0x33, 0x07, 0x79, 0x54, 0x25, 0x00, 0x01, 0x7e, 0x7d} frame := append([]byte{0x7e}, escape(append(payload, checksum(payload)))...) @@ -131,6 +187,41 @@ func TestExtractFramesHandlesEscapedPayload(t *testing.T) { } } +func buildFrame(messageID uint16, phone string, sequence uint16, body []byte) []byte { + payload := make([]byte, 0, 12+len(body)+1) + payload = binary.BigEndian.AppendUint16(payload, messageID) + payload = binary.BigEndian.AppendUint16(payload, uint16(len(body))) + payload = append(payload, encodeBCD(phone, 6)...) + payload = binary.BigEndian.AppendUint16(payload, sequence) + payload = append(payload, body...) + payload = append(payload, checksum(payload)) + return payload +} + +func appendFixedASCII(out []byte, value string, size int) []byte { + start := len(out) + out = append(out, make([]byte, size)...) + copy(out[start:], []byte(value)) + return out +} + +func encodeBCD(value string, size int) []byte { + out := make([]byte, size) + if len(value)%2 == 1 { + value = "0" + value + } + if len(value) > size*2 { + value = value[len(value)-size*2:] + } + for len(value) < size*2 { + value = "0" + value + } + for i := 0; i < size; i++ { + out[i] = (value[i*2]-'0')<<4 | (value[i*2+1] - '0') + } + return out +} + func assertFloatField(t *testing.T, env envelope.FrameEnvelope, key string, want float64) { t.Helper() got, ok := env.Fields[key].(float64)