feat: track jt808 registration identity

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"
+}

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()

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{}

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)