feat: persist jt808 registration identity

README.md

@@ -37,7 +37,7 @@ lingniu-vehicle-ingest/
 │   │   ├── ingest-codec-common/     公共编解码工具（BCD/CRC/BCC/bit utils）
 │   │   ├── ingest-core/             Pipeline / Dispatcher / Disruptor / Session 桥
 │   │   ├── session-core/            设备会话 + 鉴权 + Token + Redis/Memory SessionStore
-│   │   ├── vehicle-identity/        跨协议车辆身份解析 + 外部标识绑定（memory/file）
+│   │   ├── vehicle-identity/        跨协议车辆身份解析 + 外部标识绑定（memory/file/mysql）
 │   │   └── observability/           metrics / tracing / health
 │   ├── protocols/
 │   │   ├── protocol-gb32960/        GB/T 32960

deploy/local/launchctl/README.md

@@ -48,7 +48,7 @@ done
 
 如果生产配置由 Nacos 下发，可以把模板里的 `NACOS_CONFIG_ENABLED` 改成 `true`，并补充 `NACOS_SERVER_ADDR`、`NACOS_NAMESPACE`、`NACOS_GROUP`、`NACOS_USERNAME`、`NACOS_PASSWORD`。端口、Kafka topic、archive root、TDengine 连接建议仍保留为启动环境变量，便于 Portainer、launchctl 和临时压测保持一致。
 
-JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 表并按 phone/deviceId/plate 维护 VIN 绑定。
+JT808 身份绑定默认使用内存或文件。需要把注册信息维护到 MySQL 时，将 `com.lingniu.jt808.plist.template` 里的 `VEHICLE_IDENTITY_STORE` 改为 `mysql`，并提供 `VEHICLE_IDENTITY_MYSQL_*` 连接参数。服务会自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表：前者保存已确认 VIN 的外部标识绑定，后者保存 0x0100 注册字段；外部系统可把真实 VIN 反写到 registration 表，服务按刷新周期加载后，后续 raw/event 会带真实 VIN。
 
 ## 启停
 

docs/operations/vehicle-ingest-tdengine-verification.md

@@ -114,9 +114,10 @@ JT808 注册身份绑定：
 - 默认可以使用 `VEHICLE_IDENTITY_STORE=file` 或 `memory`。
 - 生产需要把 0x0100 注册信息维护到 MySQL 时，设置 `VEHICLE_IDENTITY_STORE=mysql`。
 - 需要提供 `VEHICLE_IDENTITY_MYSQL_JDBC_URL`、`VEHICLE_IDENTITY_MYSQL_USERNAME`、`VEHICLE_IDENTITY_MYSQL_PASSWORD`。
-- 服务启动时自动创建 `vehicle_identity_binding` 表，按 `protocol + identifier_type + identifier_value` 维护 VIN 绑定。
+- 服务启动时自动创建 `vehicle_identity_binding` 和 `vehicle_identity_binding_registration` 两张表。`vehicle_identity_binding` 按 `protocol + identifier_type + identifier_value` 维护已确认 VIN 绑定；`vehicle_identity_binding_registration` 按 `protocol + phone` 保存 JT808 0x0100 注册字段。
-- MySQL 绑定在启动时加载到内存索引，注册/补写时写穿 MySQL 并刷新内存索引；帧解析热路径只查内存，不按帧访问 MySQL。外部系统更新绑定后，后台按 `VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL` 周期刷新，默认 `60s`。
+- 注册帧无真实 VIN 时也会写入 registration 表，`vin` 默认为 `unknown`。外部系统识别车辆后可反写真实 VIN，例如：`UPDATE vehicle_identity_binding_registration SET vin = 'LNVIN000000000001' WHERE protocol = 'JT808' AND phone = '13079960003';`
-- 绑定类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定 VIN。
+- MySQL 绑定在启动时加载到内存索引，registration 表反写 VIN 后会按 `VEHICLE_IDENTITY_MYSQL_REFRESH_INTERVAL` 周期刷新，默认 `60s`；帧解析热路径只查内存，不按帧访问 MySQL。
+- 绑定/注册类型包括 `PHONE`、`DEVICE_ID`、`PLATE`；后续同一终端上报 raw/event 时会优先解析成已绑定或已反写 VIN。
 
 健康检查：
 
@@ -295,4 +296,4 @@ curl -sS 'http://127.0.0.1:20200/api/event-history/gb32960/snapshots/fields?vin=
 - 当前本机使用 TDengine WebSocket JDBC，例如 `jdbc:TAOS-WS://<tdengine-host>:6041/vehicle_ts`。
 - `KAFKA_CONSUMER_AUTO_OFFSET_RESET=latest` 适合生产接入新流量；如果要回放历史 Kafka 数据，需要切换 consumer group 或重置 offset。
 - `vehicle-history-app` raw 和 event 使用不同 consumer binding；raw consumer 必须开启，否则 `raw_frames` 不会持续增长。
-- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:<phone>`；启用 MySQL identity store 后，注册维护的 phone/deviceId/plate 绑定会用于后续 raw/event 的 VIN 解析。
+- JT808 设备没有 VIN 映射时，`vehicle_key` 使用 `jt808:<phone>`；启用 MySQL identity store 后，registration 表中已反写 VIN 的 phone/deviceId/plate 会用于后续 raw/event 的 VIN 解析。

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

@@ -22,6 +22,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
 
     private final DataSource dataSource;
     private final String table;
+    private final String registrationTable;
     private final ScheduledExecutorService refresher;
     private volatile InMemoryVehicleIdentityService index;
 
@@ -35,6 +36,7 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
         }
         this.dataSource = dataSource;
         this.table = sanitizeTable(table);
+        this.registrationTable = this.table + "_registration";
         initializeSchema();
         this.index = loadIndex();
         this.refresher = startRefresher(refreshInterval);
@@ -51,6 +53,22 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
         index.bind(binding);
     }
 
+    @Override
+    public void register(VehicleRegistrationBinding registration) {
+        if (registration == null) {
+            return;
+        }
+        upsertRegistration(registration);
+        if (registration.hasResolvedVin()) {
+            bind(new VehicleIdentityBinding(
+                    registration.protocol(),
+                    registration.vin(),
+                    registration.phone(),
+                    registration.deviceId(),
+                    registration.plate()));
+        }
+    }
+
     @Override
     public VehicleIdentity resolve(VehicleIdentityLookup lookup) {
         if (lookup == null) {
@@ -83,9 +101,30 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
                   KEY idx_vehicle_identity_vin (vin)
                 )
                 """.formatted(table);
+        String registrationSql = """
+                CREATE TABLE IF NOT EXISTS %s (
+                  protocol VARCHAR(32) NOT NULL,
+                  phone VARCHAR(64) NOT NULL,
+                  vin VARCHAR(64) NOT NULL DEFAULT 'unknown',
+                  device_id VARCHAR(128) NOT NULL DEFAULT '',
+                  plate VARCHAR(128) NOT NULL DEFAULT '',
+                  province INT NULL,
+                  city INT NULL,
+                  maker VARCHAR(64) NOT NULL DEFAULT '',
+                  device_type VARCHAR(128) NOT NULL DEFAULT '',
+                  plate_color INT NULL,
+                  last_registered_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+                  created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                  PRIMARY KEY (protocol, phone),
+                  KEY idx_vehicle_registration_vin (vin),
+                  KEY idx_vehicle_registration_device (protocol, device_id),
+                  KEY idx_vehicle_registration_plate (protocol, plate)
+                )
+                """.formatted(registrationTable);
         try (Connection connection = dataSource.getConnection();
              Statement statement = connection.createStatement()) {
             statement.execute(sql);
+            statement.execute(registrationSql);
         } catch (SQLException e) {
             throw new IllegalStateException("vehicle identity mysql schema initialize failed: " + table, e);
         }
@@ -105,12 +144,29 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
                         resultSet.getString("identifier_value"),
                         resultSet.getString("vin"));
             }
+            loadRegistrationIndex(loaded, connection);
             return loaded;
         } catch (SQLException e) {
             throw new IllegalStateException("vehicle identity mysql index load failed: " + table, e);
         }
     }
 
+    private void loadRegistrationIndex(InMemoryVehicleIdentityService loaded, Connection connection) throws SQLException {
+        String sql = "SELECT protocol, phone, device_id, plate, vin FROM " + registrationTable
+                + " WHERE vin IS NOT NULL AND vin <> '' AND LOWER(vin) <> 'unknown'";
+        try (PreparedStatement statement = connection.prepareStatement(sql);
+             ResultSet resultSet = statement.executeQuery()) {
+            while (resultSet.next()) {
+                loaded.bind(new VehicleIdentityBinding(
+                        protocol(resultSet.getString("protocol")),
+                        resultSet.getString("vin"),
+                        resultSet.getString("phone"),
+                        resultSet.getString("device_id"),
+                        resultSet.getString("plate")));
+            }
+        }
+    }
+
     private void bindLoadedRow(InMemoryVehicleIdentityService loaded,
                                ProtocolId protocol,
                                String identifierType,
@@ -173,6 +229,46 @@ public final class MySqlVehicleIdentityService implements VehicleIdentityResolve
         }
     }
 
+    private void upsertRegistration(VehicleRegistrationBinding registration) {
+        String phone = normalize(registration.phone());
+        if (phone.isBlank()) {
+            return;
+        }
+        String vin = registration.hasResolvedVin() ? registration.vin().trim() : "unknown";
+        String sql = "INSERT INTO " + registrationTable
+                + " (protocol, phone, vin, device_id, plate, province, city, maker, device_type, plate_color) "
+                + "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
+                + "ON DUPLICATE KEY UPDATE "
+                + "vin = CASE WHEN VALUES(vin) <> 'unknown' THEN VALUES(vin) ELSE vin END, "
+                + "device_id = VALUES(device_id), plate = VALUES(plate), province = VALUES(province), "
+                + "city = VALUES(city), maker = VALUES(maker), device_type = VALUES(device_type), "
+                + "plate_color = VALUES(plate_color), last_registered_at = CURRENT_TIMESTAMP";
+        try (Connection connection = dataSource.getConnection();
+             PreparedStatement statement = connection.prepareStatement(sql)) {
+            statement.setString(1, protocolName(registration.protocol()));
+            statement.setString(2, phone);
+            statement.setString(3, vin);
+            statement.setString(4, normalize(registration.deviceId()));
+            statement.setString(5, normalize(registration.plate()));
+            setInteger(statement, 6, registration.province());
+            setInteger(statement, 7, registration.city());
+            statement.setString(8, normalize(registration.maker()));
+            statement.setString(9, normalize(registration.deviceType()));
+            setInteger(statement, 10, registration.plateColor());
+            statement.executeUpdate();
+        } catch (SQLException e) {
+            throw new IllegalStateException("vehicle identity mysql registration bind failed: " + registrationTable, e);
+        }
+    }
+
+    private static void setInteger(PreparedStatement statement, int index, Integer value) throws SQLException {
+        if (value == null) {
+            statement.setObject(index, null);
+        } else {
+            statement.setInt(index, value);
+        }
+    }
+
     private static String sanitizeTable(String table) {
         String value = table == null || table.isBlank() ? "vehicle_identity_binding" : table.trim();
         if (!value.matches("[A-Za-z0-9_]+")) {

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

@@ -3,4 +3,7 @@ package com.lingniu.ingest.identity;
 public interface VehicleIdentityRegistry {
 
     void bind(VehicleIdentityBinding binding);
+
+    default void register(VehicleRegistrationBinding registration) {
+    }
 }

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

@@ -0,0 +1,36 @@
+package com.lingniu.ingest.identity;
+
+import com.lingniu.ingest.api.ProtocolId;
+
+public record VehicleRegistrationBinding(
+        ProtocolId protocol,
+        String vin,
+        String phone,
+        String deviceId,
+        String plate,
+        Integer province,
+        Integer city,
+        String maker,
+        String deviceType,
+        Integer plateColor
+) {
+    public VehicleRegistrationBinding {
+        vin = normalize(vin);
+        phone = normalize(phone);
+        deviceId = normalize(deviceId);
+        plate = normalize(plate);
+        maker = normalize(maker);
+        deviceType = normalize(deviceType);
+        if (phone.isBlank() && deviceId.isBlank() && plate.isBlank()) {
+            throw new IllegalArgumentException("at least one registration identifier must not be blank");
+        }
+    }
+
+    public boolean hasResolvedVin() {
+        return !vin.isBlank() && !"unknown".equalsIgnoreCase(vin);
+    }
+
+    private static String normalize(String value) {
+        return value == null ? "" : value.trim();
+    }
+}

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

@@ -39,8 +39,8 @@ class MySqlVehicleIdentityServiceTest {
         VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup(
                 ProtocolId.JT808, "", "", "", "粤B00001"));
 
-        assertThat(jdbc.createdTables).singleElement()
+        assertThat(jdbc.createdTables)
-                .satisfies(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding"));
+                .anySatisfy(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();
@@ -105,9 +105,50 @@ class MySqlVehicleIdentityServiceTest {
         assertThat(jdbc.lookupSelects).isZero();
     }
 
+    @Test
+    void storesRegistrationWithoutVinAndResolvesAfterVinIsWrittenBack() {
+        RecordingIdentityJdbc jdbc = new RecordingIdentityJdbc();
+        MySqlVehicleIdentityService service = new MySqlVehicleIdentityService(jdbc.dataSource(), "vehicle_identity_binding");
+
+        service.register(new VehicleRegistrationBinding(
+                ProtocolId.JT808,
+                "unknown",
+                "13079960003",
+                "DEV003",
+                "粤B00003",
+                44,
+                4401,
+                "MAKER",
+                "TYPE-A",
+                1));
+
+        VehicleIdentity before = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "13079960003", "", ""));
+
+        jdbc.writeBackRegistrationVin("JT808", "13079960003", "LNVIN000000000789");
+        service.refresh();
+        VehicleIdentity byPhone = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "13079960003", "", ""));
+        VehicleIdentity byDevice = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "", "DEV003", ""));
+        VehicleIdentity byPlate = service.resolve(new VehicleIdentityLookup(
+                ProtocolId.JT808, "", "", "", "粤B00003"));
+
+        assertThat(jdbc.createdTables)
+                .anySatisfy(sql -> assertThat(sql).contains("CREATE TABLE IF NOT EXISTS vehicle_identity_binding_registration"));
+        assertThat(jdbc.registrations).containsKey("JT808|13079960003");
+        assertThat(jdbc.bindings).isEmpty();
+        assertThat(before.resolved()).isFalse();
+        assertThat(byPhone.vin()).isEqualTo("LNVIN000000000789");
+        assertThat(byPhone.source()).isEqualTo(VehicleIdentitySource.BOUND_PHONE);
+        assertThat(byDevice.vin()).isEqualTo("LNVIN000000000789");
+        assertThat(byPlate.vin()).isEqualTo("LNVIN000000000789");
+    }
+
     private static final class RecordingIdentityJdbc {
         private final List<String> createdTables = new ArrayList<>();
         private final Map<String, String> bindings = new HashMap<>();
+        private final Map<String, RegistrationRow> registrations = new HashMap<>();
         private int loadAllSelects;
         private int lookupSelects;
 
@@ -115,6 +156,14 @@ class MySqlVehicleIdentityServiceTest {
             bindings.put(protocol + "|" + identifierType + "|" + normalize(identifierValue), vin);
         }
 
+        private void writeBackRegistrationVin(String protocol, String phone, String vin) {
+            RegistrationRow row = registrations.get(protocol + "|" + normalize(phone));
+            if (row == null) {
+                throw new IllegalArgumentException("missing registration " + phone);
+            }
+            row.vin = vin;
+        }
+
         private DataSource dataSource() {
             return new DataSource() {
                 @Override public Connection getConnection() { return connection(); }
@@ -162,12 +211,16 @@ class MySqlVehicleIdentityServiceTest {
             List<Object> params = new ArrayList<>();
             InvocationHandler handler = (proxy, method, args) -> {
                 switch (method.getName()) {
-                    case "setString" -> {
+                    case "setString", "setInt", "setObject" -> {
                         set(params, (Integer) args[0], args[1]);
                         return null;
                     }
                     case "executeUpdate" -> {
-                        upsert(params);
+                        if (sql.startsWith("INSERT INTO vehicle_identity_binding_registration ")) {
+                            upsertRegistration(params);
+                        } else {
+                            upsert(params);
+                        }
                         return 1;
                     }
                     case "executeQuery" -> {
@@ -175,6 +228,9 @@ class MySqlVehicleIdentityServiceTest {
                             loadAllSelects++;
                             return allBindingsResultSet();
                         }
+                        if (sql.startsWith("SELECT protocol, phone, device_id, plate, vin FROM ")) {
+                            return registrationResultSet();
+                        }
                         lookupSelects++;
                         return resultSet(resolve(params));
                     }
@@ -198,6 +254,21 @@ class MySqlVehicleIdentityServiceTest {
             bindings.put(protocol + "|" + identifierType + "|" + identifierValue, vin);
         }
 
+        private void upsertRegistration(List<Object> params) {
+            String protocol = (String) params.get(0);
+            String phone = normalize((String) params.get(1));
+            String vin = (String) params.get(2);
+            String key = protocol + "|" + phone;
+            RegistrationRow existing = registrations.get(key);
+            String effectiveVin = existing != null && "unknown".equalsIgnoreCase(vin) ? existing.vin : vin;
+            registrations.put(key, new RegistrationRow(
+                    protocol,
+                    phone,
+                    effectiveVin,
+                    normalize((String) params.get(3)),
+                    normalize((String) params.get(4))));
+        }
+
         private String resolve(List<Object> params) {
             String protocol = (String) params.get(0);
             String identifierType = (String) params.get(1);
@@ -239,6 +310,41 @@ class MySqlVehicleIdentityServiceTest {
             };
         }
 
+        private ResultSet registrationResultSet() {
+            List<RegistrationRow> rows = registrations.values().stream()
+                    .filter(row -> row.vin != null && !row.vin.isBlank() && !"unknown".equalsIgnoreCase(row.vin))
+                    .toList();
+            InvocationHandler handler = new InvocationHandler() {
+                private int index = -1;
+
+                @Override
+                public Object invoke(Object proxy, java.lang.reflect.Method method, Object[] args) {
+                    return switch (method.getName()) {
+                        case "next" -> {
+                            index++;
+                            yield index < rows.size();
+                        }
+                        case "getString" -> registrationColumn(rows.get(index), args[0]);
+                        case "close" -> null;
+                        default -> defaultValue(method.getReturnType());
+                    };
+                }
+            };
+            return (ResultSet) Proxy.newProxyInstance(
+                    ResultSet.class.getClassLoader(), new Class<?>[]{ResultSet.class}, handler);
+        }
+
+        private static String registrationColumn(RegistrationRow row, Object column) {
+            return switch (column.toString()) {
+                case "protocol" -> row.protocol;
+                case "phone" -> row.phone;
+                case "device_id" -> row.deviceId;
+                case "plate" -> row.plate;
+                case "vin" -> row.vin;
+                default -> throw new IllegalArgumentException("unsupported column " + column);
+            };
+        }
+
         private static ResultSet resultSet(String vin) {
             InvocationHandler handler = new InvocationHandler() {
                 private boolean read;
@@ -283,5 +389,21 @@ class MySqlVehicleIdentityServiceTest {
             if (type == Byte.TYPE) return (byte) 0;
             return null;
         }
+
+        private static final class RegistrationRow {
+            private final String protocol;
+            private final String phone;
+            private String vin;
+            private final String deviceId;
+            private final String plate;
+
+            private RegistrationRow(String protocol, String phone, String vin, String deviceId, String plate) {
+                this.protocol = protocol;
+                this.phone = phone;
+                this.vin = vin;
+                this.deviceId = deviceId;
+                this.plate = plate;
+            }
+        }
     }
 }

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

@@ -9,6 +9,7 @@ import com.lingniu.ingest.identity.VehicleIdentityLookup;
 import com.lingniu.ingest.identity.VehicleIdentityRegistry;
 import com.lingniu.ingest.identity.VehicleIdentityResolver;
 import com.lingniu.ingest.identity.VehicleIdentitySource;
+import com.lingniu.ingest.identity.VehicleRegistrationBinding;
 import com.lingniu.ingest.protocol.jt808.codec.Jt808FrameEncoder;
 import com.lingniu.ingest.protocol.jt808.codec.Jt808MalformedFrame;
 import com.lingniu.ingest.protocol.jt808.codec.Jt808MessageDecoder;
@@ -286,7 +287,18 @@ public class Jt808ChannelHandler extends SimpleChannelInboundHandler<Object> {
                         Instant.now(), Instant.now(),
                         Map.of("protocolVersion", msg.header().version().name())));
         session = session.withVin(identity.vin());
-        if (msg.body() instanceof Jt808Body.Register reg && reg.plate() != null) {
+        if (msg.body() instanceof Jt808Body.Register reg) {
+            identityRegistry.register(new VehicleRegistrationBinding(
+                    ProtocolId.JT808,
+                    session.vin(),
+                    phone,
+                    reg.deviceId(),
+                    reg.plate(),
+                    reg.province(),
+                    reg.city(),
+                    reg.maker(),
+                    reg.deviceType(),
+                    reg.plateColor()));
             if (identity.resolved() && session.vin() != null
                     && !session.vin().isBlank()
                     && !"unknown".equalsIgnoreCase(session.vin())) {