feat: add mysql vehicle identity store

modules/apps/jt808-ingest-app/src/main/resources/application.yml

@@ -51,6 +51,12 @@ lingniu:
       store: ${VEHICLE_IDENTITY_STORE:file}
       file:
         path: ${VEHICLE_IDENTITY_FILE:./data/vehicle-identity.jsonl}
+      mysql:
+        table: ${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}
+        jdbc-url: ${VEHICLE_IDENTITY_MYSQL_JDBC_URL:}
+        username: ${VEHICLE_IDENTITY_MYSQL_USERNAME:}
+        password: ${VEHICLE_IDENTITY_MYSQL_PASSWORD:}
+        driver-class-name: ${VEHICLE_IDENTITY_MYSQL_DRIVER_CLASS_NAME:com.mysql.cj.jdbc.Driver}
     sink:
       mq:
         enabled: ${KAFKA_ENABLED:true}

modules/apps/jt808-ingest-app/src/test/java/com/lingniu/ingest/jt808app/Jt808IngestAppDefaultsTest.java

@@ -28,6 +28,10 @@ class Jt808IngestAppDefaultsTest {
                 .containsEntry("lingniu.ingest.sink.mq.topics.realtime", "${KAFKA_TOPIC_JT808_EVENT:vehicle.event.jt808.v1}")
                 .containsEntry("lingniu.ingest.sink.mq.topics.raw-archive", "${KAFKA_TOPIC_JT808_RAW:vehicle.raw.jt808.v1}")
                 .containsEntry("lingniu.ingest.sink.mq.topics.dlq", "${KAFKA_TOPIC_JT808_DLQ:vehicle.dlq.jt808.v1}")
+                .containsEntry("lingniu.ingest.identity.store", "${VEHICLE_IDENTITY_STORE:file}")
+                .containsEntry("lingniu.ingest.identity.mysql.table", "${VEHICLE_IDENTITY_MYSQL_TABLE:vehicle_identity_binding}")
+                .containsEntry("lingniu.ingest.identity.mysql.jdbc-url", "${VEHICLE_IDENTITY_MYSQL_JDBC_URL:}")
+                .containsEntry("lingniu.ingest.identity.mysql.username", "${VEHICLE_IDENTITY_MYSQL_USERNAME:}")
                 .containsEntry("lingniu.ingest.sink.archive.enabled", "${SINK_ARCHIVE_ENABLED:true}")
                 .containsEntry("lingniu.ingest.sink.archive.path", "${SINK_ARCHIVE_PATH:./archive/}")
                 .containsEntry("lingniu.ingest.event-file-store.enabled", false)

modules/core/vehicle-identity/pom.xml

@@ -28,6 +28,11 @@
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-databind</artifactId>
         </dependency>
+        <dependency>
+            <groupId>com.mysql</groupId>
+            <artifactId>mysql-connector-j</artifactId>
+            <scope>runtime</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/FileVehicleIdentityService.java

@@ -40,6 +40,9 @@ public final class FileVehicleIdentityService implements VehicleIdentityResolver
 
     @Override
     public void bind(VehicleIdentityBinding binding) {
+        if (binding == null || !binding.hasResolvedVin()) {
+            return;
+        }
         delegate.bind(binding);
         synchronized (writeLock) {
             try {

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/InMemoryVehicleIdentityService.java

@@ -14,6 +14,9 @@ public final class InMemoryVehicleIdentityService implements VehicleIdentityReso
 
     @Override
     public void bind(VehicleIdentityBinding binding) {
+        if (binding == null || !binding.hasResolvedVin()) {
+            return;
+        }
         if (!binding.phone().isBlank()) {
             phoneToVin.put(key(binding.protocol(), binding.phone()), binding.vin());
         }

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/MySqlVehicleIdentityService.java

@@ -0,0 +1,145 @@
+package com.lingniu.ingest.identity;
+
+import com.lingniu.ingest.api.ProtocolId;
+
+import javax.sql.DataSource;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.Locale;
+
+public final class MySqlVehicleIdentityService implements VehicleIdentityResolver, VehicleIdentityRegistry {
+
+    private final DataSource dataSource;
+    private final String table;
+
+    public MySqlVehicleIdentityService(DataSource dataSource, String table) {
+        if (dataSource == null) {
+            throw new IllegalArgumentException("dataSource must not be null");
+        }
+        this.dataSource = dataSource;
+        this.table = sanitizeTable(table);
+        initializeSchema();
+    }
+
+    @Override
+    public void bind(VehicleIdentityBinding binding) {
+        if (binding == null || !binding.hasResolvedVin()) {
+            return;
+        }
+        upsert(binding.protocol(), "PHONE", binding.phone(), binding.vin());
+        upsert(binding.protocol(), "DEVICE_ID", binding.deviceId(), binding.vin());
+        upsert(binding.protocol(), "PLATE", binding.plate(), binding.vin());
+    }
+
+    @Override
+    public VehicleIdentity resolve(VehicleIdentityLookup lookup) {
+        if (lookup == null) {
+            return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN);
+        }
+        if (!lookup.vin().isBlank()) {
+            return new VehicleIdentity(lookup.vin(), true, VehicleIdentitySource.EXPLICIT_VIN);
+        }
+        VehicleIdentity phone = resolveBound(lookup.protocol(), "PHONE", lookup.phone(), VehicleIdentitySource.BOUND_PHONE);
+        if (phone != null) return phone;
+        VehicleIdentity device = resolveBound(lookup.protocol(), "DEVICE_ID", lookup.deviceId(), VehicleIdentitySource.BOUND_DEVICE_ID);
+        if (device != null) return device;
+        VehicleIdentity plate = resolveBound(lookup.protocol(), "PLATE", lookup.plate(), VehicleIdentitySource.BOUND_PLATE);
+        if (plate != null) return plate;
+        if (!lookup.deviceId().isBlank()) {
+            return new VehicleIdentity(lookup.deviceId(), false, VehicleIdentitySource.FALLBACK_DEVICE_ID);
+        }
+        if (!lookup.phone().isBlank()) {
+            return new VehicleIdentity(lookup.phone(), false, VehicleIdentitySource.FALLBACK_PHONE);
+        }
+        if (!lookup.plate().isBlank()) {
+            return new VehicleIdentity(lookup.plate(), false, VehicleIdentitySource.FALLBACK_PLATE);
+        }
+        return new VehicleIdentity("unknown", false, VehicleIdentitySource.UNKNOWN);
+    }
+
+    private void initializeSchema() {
+        String sql = """
+                CREATE TABLE IF NOT EXISTS %s (
+                  protocol VARCHAR(32) NOT NULL,
+                  identifier_type VARCHAR(32) NOT NULL,
+                  identifier_value VARCHAR(128) NOT NULL,
+                  vin VARCHAR(64) NOT NULL,
+                  updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+                  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                  PRIMARY KEY (protocol, identifier_type, identifier_value),
+                  KEY idx_vehicle_identity_vin (vin)
+                )
+                """.formatted(table);
+        try (Connection connection = dataSource.getConnection();
+             Statement statement = connection.createStatement()) {
+            statement.execute(sql);
+        } catch (SQLException e) {
+            throw new IllegalStateException("vehicle identity mysql schema initialize failed: " + table, e);
+        }
+    }
+
+    private void upsert(ProtocolId protocol, String identifierType, String identifierValue, String vin) {
+        String normalizedIdentifier = normalize(identifierValue);
+        if (normalizedIdentifier.isBlank()) {
+            return;
+        }
+        String sql = "INSERT INTO " + table + " (protocol, identifier_type, identifier_value, vin) "
+                + "VALUES (?, ?, ?, ?) "
+                + "ON DUPLICATE KEY UPDATE vin = VALUES(vin), updated_at = CURRENT_TIMESTAMP";
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setString(1, protocolName(protocol));
+            statement.setString(2, identifierType);
+            statement.setString(3, normalizedIdentifier);
+            statement.setString(4, vin.trim());
+            statement.executeUpdate();
+        } catch (SQLException e) {
+            throw new IllegalStateException("vehicle identity mysql bind failed: " + table, e);
+        }
+    }
+
+    private VehicleIdentity resolveBound(ProtocolId protocol,
+                                         String identifierType,
+                                         String identifierValue,
+                                         VehicleIdentitySource source) {
+        String normalizedIdentifier = normalize(identifierValue);
+        if (normalizedIdentifier.isBlank()) {
+            return null;
+        }
+        String sql = "SELECT vin FROM " + table
+                + " WHERE protocol = ? AND identifier_type = ? AND identifier_value = ? LIMIT 1";
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setString(1, protocolName(protocol));
+            statement.setString(2, identifierType);
+            statement.setString(3, normalizedIdentifier);
+            try (ResultSet resultSet = statement.executeQuery()) {
+                if (resultSet.next()) {
+                    return new VehicleIdentity(resultSet.getString(1), true, source);
+                }
+            }
+        } catch (SQLException e) {
+            throw new IllegalStateException("vehicle identity mysql resolve failed: " + table, e);
+        }
+        return null;
+    }
+
+    private static String sanitizeTable(String table) {
+        String value = table == null || table.isBlank() ? "vehicle_identity_binding" : table.trim();
+        if (!value.matches("[A-Za-z0-9_]+")) {
+            throw new IllegalArgumentException("vehicle identity mysql table must contain only letters, digits, and underscore");
+        }
+        return value;
+    }
+
+    private static String protocolName(ProtocolId protocol) {
+        return protocol == null ? "UNKNOWN" : protocol.name();
+    }
+
+    private static String normalize(String value) {
+        return value == null ? "" : value.trim().toUpperCase(Locale.ROOT);
+    }
+}

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/VehicleIdentityBinding.java

@@ -19,6 +19,10 @@ public record VehicleIdentityBinding(
         }
     }
 
+    public boolean hasResolvedVin() {
+        return !"unknown".equalsIgnoreCase(vin);
+    }
+
     private static String normalize(String value) {
         return value == null ? "" : value.trim();
     }

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfiguration.java

@@ -3,6 +3,7 @@ package com.lingniu.ingest.identity.config;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.lingniu.ingest.identity.FileVehicleIdentityService;
 import com.lingniu.ingest.identity.InMemoryVehicleIdentityService;
+import com.lingniu.ingest.identity.MySqlVehicleIdentityService;
 import com.lingniu.ingest.identity.VehicleIdentityRegistry;
 import com.lingniu.ingest.identity.VehicleIdentityResolver;
 import org.springframework.boot.autoconfigure.AutoConfiguration;
@@ -11,7 +12,14 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 
+import javax.sql.DataSource;
+import java.io.PrintWriter;
 import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.util.logging.Logger;
 
 @AutoConfiguration
 @EnableConfigurationProperties(VehicleIdentityProperties.class)
@@ -32,6 +40,23 @@ public class VehicleIdentityAutoConfiguration {
         return new FileVehicleIdentityService(Path.of(properties.getFile().getPath()), objectMapper);
     }
 
+    @Bean
+    @ConditionalOnMissingBean(DataSource.class)
+    @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "mysql")
+    public DataSource vehicleIdentityDataSource(VehicleIdentityProperties properties) {
+        VehicleIdentityProperties.Mysql mysql = properties.getMysql();
+        return new DriverManagerVehicleIdentityDataSource(
+                mysql.getDriverClassName(), mysql.getJdbcUrl(), mysql.getUsername(), mysql.getPassword());
+    }
+
+    @Bean
+    @ConditionalOnMissingBean({VehicleIdentityResolver.class, VehicleIdentityRegistry.class})
+    @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "mysql")
+    public MySqlVehicleIdentityService mySqlVehicleIdentityService(VehicleIdentityProperties properties,
+                                                                   DataSource dataSource) {
+        return new MySqlVehicleIdentityService(dataSource, properties.getMysql().getTable());
+    }
+
     @Bean
     @ConditionalOnMissingBean({VehicleIdentityResolver.class, VehicleIdentityRegistry.class})
     @ConditionalOnProperty(prefix = "lingniu.ingest.identity", name = "store", havingValue = "memory",
@@ -39,4 +64,61 @@ public class VehicleIdentityAutoConfiguration {
     public InMemoryVehicleIdentityService vehicleIdentityService() {
         return new InMemoryVehicleIdentityService();
     }
+
+    private static final class DriverManagerVehicleIdentityDataSource implements DataSource {
+        private final String jdbcUrl;
+        private final String username;
+        private final String password;
+        private PrintWriter logWriter;
+        private int loginTimeout;
+
+        private DriverManagerVehicleIdentityDataSource(String driverClassName,
+                                                       String jdbcUrl,
+                                                       String username,
+                                                       String password) {
+            if (jdbcUrl == null || jdbcUrl.isBlank()) {
+                throw new IllegalArgumentException("lingniu.ingest.identity.mysql.jdbc-url is required");
+            }
+            if (driverClassName != null && !driverClassName.isBlank()) {
+                try {
+                    Class.forName(driverClassName.trim());
+                } catch (ClassNotFoundException e) {
+                    throw new IllegalArgumentException("vehicle identity mysql driver class not found: "
+                            + driverClassName, e);
+                }
+            }
+            this.jdbcUrl = jdbcUrl;
+            this.username = username == null ? "" : username;
+            this.password = password == null ? "" : password;
+        }
+
+        @Override
+        public Connection getConnection() throws SQLException {
+            DriverManager.setLoginTimeout(loginTimeout);
+            return DriverManager.getConnection(jdbcUrl, username, password);
+        }
+
+        @Override
+        public Connection getConnection(String username, String password) throws SQLException {
+            DriverManager.setLoginTimeout(loginTimeout);
+            return DriverManager.getConnection(jdbcUrl, username, password);
+        }
+
+        @Override public PrintWriter getLogWriter() { return logWriter; }
+        @Override public void setLogWriter(PrintWriter out) { this.logWriter = out; }
+        @Override public void setLoginTimeout(int seconds) { this.loginTimeout = seconds; }
+        @Override public int getLoginTimeout() { return loginTimeout; }
+        @Override public Logger getParentLogger() throws SQLFeatureNotSupportedException {
+            throw new SQLFeatureNotSupportedException();
+        }
+        @Override public <T> T unwrap(Class<T> iface) throws SQLException {
+            if (iface.isInstance(this)) {
+                return iface.cast(this);
+            }
+            throw new SQLException("not a wrapper for " + iface);
+        }
+        @Override public boolean isWrapperFor(Class<?> iface) {
+            return iface.isInstance(this);
+        }
+    }
 }

modules/core/vehicle-identity/src/main/java/com/lingniu/ingest/identity/config/VehicleIdentityProperties.java

@@ -7,11 +7,14 @@ public class VehicleIdentityProperties {
 
     private String store = "memory";
     private File file = new File();
+    private Mysql mysql = new Mysql();
 
     public String getStore() { return store; }
     public void setStore(String store) { this.store = store; }
     public File getFile() { return file; }
     public void setFile(File file) { this.file = file; }
+    public Mysql getMysql() { return mysql; }
+    public void setMysql(Mysql mysql) { this.mysql = mysql; }
 
     public static class File {
         private String path = "./data/vehicle-identity.jsonl";
@@ -19,4 +22,23 @@ public class VehicleIdentityProperties {
         public String getPath() { return path; }
         public void setPath(String path) { this.path = path; }
     }
+
+    public static class Mysql {
+        private String table = "vehicle_identity_binding";
+        private String jdbcUrl = "";
+        private String username = "";
+        private String password = "";
+        private String driverClassName = "com.mysql.cj.jdbc.Driver";
+
+        public String getTable() { return table; }
+        public void setTable(String table) { this.table = table; }
+        public String getJdbcUrl() { return jdbcUrl; }
+        public void setJdbcUrl(String jdbcUrl) { this.jdbcUrl = jdbcUrl; }
+        public String getUsername() { return username; }
+        public void setUsername(String username) { this.username = username; }
+        public String getPassword() { return password; }
+        public void setPassword(String password) { this.password = password; }
+        public String getDriverClassName() { return driverClassName; }
+        public void setDriverClassName(String driverClassName) { this.driverClassName = driverClassName; }
+    }
 }

modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/MySqlVehicleIdentityServiceTest.java

@@ -0,0 +1,199 @@
+package com.lingniu.ingest.identity;
+
+import com.lingniu.ingest.api.ProtocolId;
+import org.junit.jupiter.api.Test;
+
+import javax.sql.DataSource;
+import java.io.PrintWriter;
+import java.lang.reflect.InvocationHandler;
+import java.lang.reflect.Proxy;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.SQLFeatureNotSupportedException;
+import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.logging.Logger;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+class MySqlVehicleIdentityServiceTest {
+
+    @Test
+    void bindsAndResolvesIdentityThroughJdbcStore() {
+        RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc();
+        MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding");
+
+        service.bind(new VehicleIdentityBinding(
+                ProtocolId.JT808, "LNVIN000000000123", "13079960001", "DEV001", "粤B00001"));
+
+        VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "13079960001", "", ""));
+        VehicleIdentity byDevice = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "", "DEV001", ""));
+        VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "", "", "粤B00001"));
+
+        assertThat(jdbc.createdTables).singleElement()
+                .satisfies(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding"));
+        assertThat(jdbc.bindings).containsKey("JT808|PHONE|13079960001");
+        assertThat(byPhone.vin()).isEqualTo("LNVIN000000000123");
+        assertThat(byPhone.resolved()).isTrue();
+        assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE);
+        assertThat(byDevice.vin()).isEqualTo("LNVIN000000000123");
+        assertThat(byDevice.source()).isEqualTo(VehicleIdentitySource.BOUND_DEVICE_ID);
+        assertThat(byPlate.vin()).isEqualTo("LNVIN000000000123");
+        assertThat(byPlate.source()).isEqualTo(VehicleIdentitySource.BOUND_PLATE);
+    }
+
+    @Test
+    void ignoresUnknownVinBinding() {
+        RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc();
+        MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding");
+
+        service.bind(new VehicleIdentityBinding(
+                ProtocolId.JT808, "unknown", "13079960001", "DEV001", "粤B00001"));
+
+        assertThat(jdbc.bindings).isEmpty();
+    }
+
+    private static final class RecordingIdentityJdbc {
+        private final List<String> createdTables = new ArrayList<>();
+        private final Map<String, String> bindings = new HashMap<>();
+
+        private DataSource dataSource() {
+            return new DataSource() {
+                @Override public Connection getConnection() { return connection(); }
+                @Override public Connection getConnection(String username, String password) { return connection(); }
+                @Override public PrintWriter getLogWriter() { return null; }
+                @Override public void setLogWriter(PrintWriter out) {}
+                @Override public void setLoginTimeout(int seconds) {}
+                @Override public int getLoginTimeout() { return 0; }
+                @Override public Logger getParentLogger() throws SQLFeatureNotSupportedException { throw new SQLFeatureNotSupportedException(); }
+                @Override public <T> T unwrap(Class<T> iface) { throw new UnsupportedOperationException(); }
+                @Override public boolean isWrapperFor(Class<?> iface) { return false; }
+            };
+        }
+
+        private Connection connection() {
+            InvocationHandler handler = (proxy, method, args) -> switch (method.getName()) {
+                case "createStatement" -> statement();
+                case "prepareStatement" -> preparedStatement((String) args[0]);
+                case "close" -> null;
+                case "isClosed" -> false;
+                case "unwrap" -> throw new UnsupportedOperationException();
+                case "isWrapperFor" -> false;
+                default -> defaultValue(method.getReturnType());
+            };
+            return (Connection) Proxy.newProxyInstance(
+                    Connection.class.getClassLoader(), new Class<?>[]{Connection.class}, handler);
+        }
+
+        private Statement statement() {
+            InvocationHandler handler = (proxy, method, args) -> {
+                if ("execute".equals(method.getName())) {
+                    createdTables.add((String) args[0]);
+                    return false;
+                }
+                if ("close".equals(method.getName())) {
+                    return null;
+                }
+                return defaultValue(method.getReturnType());
+            };
+            return (Statement) Proxy.newProxyInstance(
+                    Statement.class.getClassLoader(), new Class<?>[]{Statement.class}, handler);
+        }
+
+        private PreparedStatement preparedStatement(String sql) {
+            List<Object> params = new ArrayList<>();
+            InvocationHandler handler = (proxy, method, args) -> {
+                switch (method.getName()) {
+                    case "setString" -> {
+                        set(params, (Integer) args[0], args[1]);
+                        return null;
+                    }
+                    case "executeUpdate" -> {
+                        upsert(params);
+                        return 1;
+                    }
+                    case "executeQuery" -> {
+                        return resultSet(resolve(params));
+                    }
+                    case "close" -> {
+                        return null;
+                    }
+                    default -> {
+                        return defaultValue(method.getReturnType());
+                    }
+                }
+            };
+            return (PreparedStatement) Proxy.newProxyInstance(
+                    PreparedStatement.class.getClassLoader(), new Class<?>[]{PreparedStatement.class}, handler);
+        }
+
+        private void upsert(List<Object> params) {
+            String protocol = (String) params.get(0);
+            String identifierType = (String) params.get(1);
+            String identifierValue = normalize((String) params.get(2));
+            String vin = (String) params.get(3);
+            bindings.put(protocol + "|" + identifierType + "|" + identifierValue, vin);
+        }
+
+        private String resolve(List<Object> params) {
+            String protocol = (String) params.get(0);
+            String identifierType = (String) params.get(1);
+            String identifierValue = normalize((String) params.get(2));
+            return bindings.get(protocol + "|" + identifierType + "|" + identifierValue);
+        }
+
+        private static ResultSet resultSet(String vin) {
+            InvocationHandler handler = new InvocationHandler() {
+                private boolean read;
+
+                @Override
+                public Object invoke(Object proxy, java.lang.reflect.Method method, Object[] args) {
+                    return switch (method.getName()) {
+                        case "next" -> {
+                            boolean hasNext = !read && vin != null;
+                            read = true;
+                            yield hasNext;
+                        }
+                        case "getString" -> vin;
+                        case "close" -> null;
+                        default -> defaultValue(method.getReturnType());
+                    };
+                }
+            };
+            return (ResultSet) Proxy.newProxyInstance(
+                    ResultSet.class.getClassLoader(), new Class<?>[]{ResultSet.class}, handler);
+        }
+
+        private static void set(List<Object> params, int oneBasedIndex, Object value) {
+            while (params.size() < oneBasedIndex) {
+                params.add(null);
+            }
+            params.set(oneBasedIndex - 1, value);
+        }
+
+        private static String normalize(String value) {
+            return value == null ? "" : value.trim().toUpperCase(Locale.ROOT);
+        }
+
+        private static Object defaultValue(Class<?> type) {
+            if (type == Void.TYPE) return null;
+            if (type == Boolean.TYPE) return false;
+            if (type == Integer.TYPE) return 0;
+            if (type == Long.TYPE) return 0L;
+            if (type == Double.TYPE) return 0D;
+            if (type == Float.TYPE) return 0F;
+            if (type == Short.TYPE) return (short) 0;
+            if (type == Byte.TYPE) return (byte) 0;
+            return null;
+        }
+    }
+}

modules/core/vehicle-identity/src/test/java/com/lingniu/ingest/identity/config/VehicleIdentityAutoConfigurationTest.java

@@ -7,7 +7,12 @@ import org.junit.jupiter.api.io.TempDir;
 import org.springframework.boot.autoconfigure.AutoConfigurations;
 import org.springframework.boot.test.context.runner.ApplicationContextRunner;
 
+import javax.sql.DataSource;
 import java.nio.file.Path;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.sql.SQLException;
+import java.lang.reflect.Proxy;
 
 import static org.assertj.core.api.Assertions.assertThat;
 
@@ -41,4 +46,59 @@ class VehicleIdentityAutoConfigurationTest {
                             .isEqualTo("FileVehicleIdentityService");
                 });
     }
+
+    @Test
+    void createsMysqlIdentityServiceWhenConfigured() {
+        runner
+                .withBean(DataSource.class, StubDataSource::new)
+                .withPropertyValues(
+                        "lingniu.ingest.identity.store=mysql",
+                        "lingniu.ingest.identity.mysql.table=vehicle_identity_binding")
+                .run(context -> {
+                    assertThat(context).hasSingleBean(VehicleIdentityResolver.class);
+                    assertThat(context.getBean(VehicleIdentityResolver.class).getClass().getSimpleName())
+                            .isEqualTo("MySqlVehicleIdentityService");
+                    assertThat(context.getBean(VehicleIdentityResolver.class))
+                            .isSameAs(context.getBean(VehicleIdentityRegistry.class));
+                });
+    }
+
+    private static final class StubDataSource implements DataSource {
+        @Override public Connection getConnection() { return connection(); }
+        @Override public Connection getConnection(String username, String password) { return connection(); }
+        @Override public java.io.PrintWriter getLogWriter() { return null; }
+        @Override public void setLogWriter(java.io.PrintWriter out) {}
+        @Override public void setLoginTimeout(int seconds) {}
+        @Override public int getLoginTimeout() { return 0; }
+        @Override public java.util.logging.Logger getParentLogger() throws java.sql.SQLFeatureNotSupportedException { throw new java.sql.SQLFeatureNotSupportedException(); }
+        @Override public <T> T unwrap(Class<T> iface) { throw new UnsupportedOperationException(); }
+        @Override public boolean isWrapperFor(Class<?> iface) { return false; }
+
+        private static Connection connection() {
+            return (Connection) Proxy.newProxyInstance(Connection.class.getClassLoader(), new Class<?>[]{Connection.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "createStatement" -> statement();
+                        case "close" -> null;
+                        case "isClosed" -> false;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private static Statement statement() {
+            return (Statement) Proxy.newProxyInstance(Statement.class.getClassLoader(), new Class<?>[]{Statement.class},
+                    (proxy, method, args) -> switch (method.getName()) {
+                        case "execute" -> false;
+                        case "close" -> null;
+                        default -> defaultValue(method.getReturnType());
+                    });
+        }
+
+        private static Object defaultValue(Class<?> type) {
+            if (type == Void.TYPE) return null;
+            if (type == Boolean.TYPE) return false;
+            if (type == Integer.TYPE) return 0;
+            if (type == Long.TYPE) return 0L;
+            return null;
+        }
+    }
 }

modules/protocols/protocol-jt808/src/main/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandler.java

@@ -287,7 +287,9 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
                         Map.of("protocolVersion", msg.header().version().name())));
         session = session.withVin(identity.vin());
         if (msg.body() instanceof Jt808Body.Register reg && reg.plate() != null) {
-            if (session.vin() != null && !session.vin().isBlank()) {
+            if (identity.resolved() && session.vin() != null
+                    && !session.vin().isBlank()
+                    && !"unknown".equalsIgnoreCase(session.vin())) {
                 identityRegistry.bind(new VehicleIdentityBinding(
                         ProtocolId.JT808, session.vin(), phone, reg.deviceId(), reg.plate()));
             }

modules/protocols/protocol-jt808/src/test/java/com/lingniu/ingest/protocol/jt808/inbound/Jt808ChannelHandlerTest.java

@@ -14,6 +14,8 @@ import com.lingniu.ingest.core.dispatcher.HandlerRegistry;
 import com.lingniu.ingest.core.pipeline.InterceptorChain;
 import com.lingniu.ingest.identity.InMemoryVehicleIdentityService;
 import com.lingniu.ingest.identity.VehicleIdentityBinding;
+import com.lingniu.ingest.identity.VehicleIdentityLookup;
+import com.lingniu.ingest.identity.VehicleIdentitySource;
 import com.lingniu.ingest.protocol.jt808.codec.BodyParserRegistry;
 import com.lingniu.ingest.protocol.jt808.codec.Jt808MalformedFrame;
 import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder;
@@ -82,6 +84,45 @@ class Jt808ChannelHandlerTest {
         eventBus.close();
     }
 
+    @Test
+    void unresolvedRegisterDoesNotBindUnknownVinToExternalIdentifiers() {
+        InMemorySessionStore sessions = new InMemorySessionStore();
+        InMemoryVehicleIdentityService identity = new InMemoryVehicleIdentityService();
+        DisruptorEventBus eventBus = new DisruptorEventBus(1024, "blocking", List.of());
+        AsyncBatchExecutor batchExecutor = new AsyncBatchExecutor(eventBus::publish);
+        Dispatcher dispatcher = new Dispatcher(
+                new HandlerRegistry(),
+                new InterceptorChain(List.of()),
+                new HandlerInvoker(),
+                eventBus,
+                batchExecutor);
+        Jt808ChannelHandler handler = new Jt808ChannelHandler(
+                new Jt808MessageDecoder(new BodyParserRegistry(List.of(new RegisterBodyParser()))),
+                dispatcher,
+                sessions,
+                identity,
+                new Jt808ChannelRegistry(),
+                new Jt808PendingRequests());
+        EmbeddedChannel channel = new EmbeddedChannel(handler);
+
+        channel.writeInbound(buildFrame(
+                Jt808MessageId.TERMINAL_REGISTER,
+                "123456789012",
+                1,
+                buildRegisterBody("DEV808", "B80808")));
+
+        assertThat(identity.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "123456789012", "", "")))
+                .satisfies(resolved -> {
+                    assertThat(resolved.vin()).isEqualTo("123456789012");
+                    assertThat(resolved.resolved()).isFalse();
+                    assertThat(resolved.source()).isEqualTo(VehicleIdentitySource.FALLBACK_PHONE);
+                });
+
+        batchExecutor.close();
+        eventBus.close();
+    }
+
     @Test
     void authSessionUsesResolvedInternalVinFromImei() {
         InMemorySessionStore sessions = new InMemorySessionStore();