lingniu-vehicle-ingest/go/vehicle-gateway/internal/identity/resolver.go

package identity

import (
	"context"
	"database/sql"
	"errors"
	"fmt"
	"strings"

	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
)

type Resolver interface {
	Resolve(context.Context, envelope.FrameEnvelope) (envelope.FrameEnvelope, error)
}

type NoopResolver struct{}

func (NoopResolver) Resolve(_ context.Context, env envelope.FrameEnvelope) (envelope.FrameEnvelope, error) {
	return env, nil
}

type MySQLResolver struct {
	db    *sql.DB
	table string
}

func NewMySQLResolver(db *sql.DB, table string) *MySQLResolver {
	if db == nil {
		panic("identity db must not be nil")
	}
	table = strings.TrimSpace(table)
	if table == "" || !safeIdentifier(table) {
		table = "vehicle_identity_binding"
	}
	return &MySQLResolver{db: db, table: table}
}

func safeIdentifier(value string) bool {
	if value == "" {
		return false
	}
	for _, r := range value {
		if (r >= 'a' && r <= 'z') || (r >= 'A' && r <= 'Z') || (r >= '0' && r <= '9') || r == '_' {
			continue
		}
		return false
	}
	return true
}

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)
	for _, candidate := range candidates {
		vin, err := r.lookup(ctx, candidate.Column, candidate.Value)
		if err != nil {
			if errors.Is(err, sql.ErrNoRows) {
				continue
			}
			return env, err
		}
		if strings.TrimSpace(vin) != "" {
			env.VIN = strings.TrimSpace(vin)
			if env.Parsed == nil {
				env.Parsed = map[string]any{}
			}
			env.Parsed["identity"] = map[string]any{
				"resolved": true,
				"source":   candidate.Column,
				"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
}

func (r *MySQLResolver) lookup(ctx context.Context, column string, value string) (string, error) {
	query := "SELECT vin FROM " + r.table + " WHERE " + column + " = ? AND vin IS NOT NULL AND vin <> '' ORDER BY updated_at DESC LIMIT 1"
	var vin string
	err := r.db.QueryRowContext(ctx, query, value).Scan(&vin)
	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
}

func CandidateKeys(env envelope.FrameEnvelope) []CandidateKey {
	var out []CandidateKey
	add := func(column string, value string) {
		value = strings.TrimSpace(value)
		if value == "" || strings.EqualFold(value, "unknown") {
			return
		}
		for _, existing := range out {
			if existing.Column == column && existing.Value == value {
				return
			}
		}
		out = append(out, CandidateKey{Column: column, Value: value})
	}
	add("phone", env.Phone)
	if trimmed := strings.TrimLeft(strings.TrimSpace(env.Phone), "0"); trimmed != "" {
		add("phone", trimmed)
	}
	add("device_id", env.DeviceID)
	add("plate", env.Plate)
	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"
}