feat(go): make identity binding read only

2026-07-03 15:45:47 +08:00
parent 1d79dc25e0
commit f292d4c0d6
5 changed files with 51 additions and 67 deletions
--- a/docs/architecture/iot-data-platform-principles.md
+++ b/docs/architecture/iot-data-platform-principles.md
@@ -50,7 +50,7 @@ flowchart LR
 | MySQL | `vehicle_realtime_snapshot` | Latest per-protocol vehicle heartbeat | protocol, vin, plate, event time, received time | phone, device, message sequence, parsed JSON |
 | MySQL | `vehicle_realtime_location` | Latest business location cache | vin, plate, location, speed, mileage, SOC, event time | full raw payload and protocol internals |
 | MySQL | `vehicle_daily_mileage` | Queryable daily mileage | vin, date, protocol, daily mileage, source mileage | temporary vehicle keys, generic metric key/value rows, per-frame raw details |
-| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, device id | registration history |
+| MySQL | `vehicle_identity_binding` | Manual identity mapping | vin, plate, phone, oem | registration history, device id |
 | MySQL | `jt808_registration` | JT808 registration and auth trace | phone, device id, plate, auth code, vin match state, first/latest seen | GB32960 or MQTT records |
 | Redis | `vehicle:latest:{vin}` | Latest merged realtime state | cross-protocol latest fields only for VIN-bound vehicles | full parsed payloads, temporary identities, and historical data |
 | Redis | `vehicle:realtime-raw:{protocol}:{vin}` | Latest full realtime protocol state | latest protocol parsed payload for VIN-bound vehicles | historical data |
--- a/docs/architecture/production-data-plane-inventory.md
+++ b/docs/architecture/production-data-plane-inventory.md
@@ -63,13 +63,13 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。

 | 表 | 核心字段 | 写入方 | 职责 |
 | --- | --- | --- | --- |
-| `vehicle_identity_binding` | `vin`、`plate`、`phone`、`device_id` | 导入/人工维护，Gateway 读取 | 通过车牌、手机号、设备号反查 VIN |
+| `vehicle_identity_binding` | `vin`、`plate`、`phone`、`oem` | 导入/人工维护，Gateway 只读 | 通过车牌、手机号反查 VIN |
 | `jt808_registration` | `phone`、`device_id`、`plate`、`vin`、`manufacturer`、`auth_token`、`source_endpoint`、首次/最新注册鉴权时间 | Gateway | JT808 注册、鉴权和 VIN 匹配状态 |
 | `vehicle_realtime_snapshot` | `protocol`、`vin`、`plate`、`event_time`、`received_at`、`event_id` | Realtime API/projector | 每协议每 VIN 最新事件轻量快照 |
 | `vehicle_realtime_location` | `protocol`、`vin`、`plate`、经纬度、速度、总里程、SOC、事件时间 | Realtime API/projector | 每协议每 VIN 最新位置业务缓存 |
 | `vehicle_daily_mileage` | `vin`、`stat_date`、`protocol`、`daily_mileage_km`、首末总里程、样本数 | Stat writer | 基于总里程差值的每日里程；无 VIN 数据不进入正式统计表 |

-身份表使用业务主键：`vehicle_identity_binding` 以 `vin` 为主键，`jt808_registration` 以 `phone` 为主键；两者不保留代理自增主键和 `created_at`。
+身份表使用业务主键：`vehicle_identity_binding` 以 `vin` 为主键，`jt808_registration` 以 `phone` 为主键；两者不保留代理自增主键和 `created_at`。`vehicle_identity_binding` 是外部维护的事实表，服务运行时不回写。

 日里程表使用 `(vin, stat_date, protocol)` 作为业务主键，不保留自增 `id` 和 `created_at`；首末总里程和样本数保留用于解释差值计算。

--- a/docs/architecture/storage-minimal-contract.md
+++ b/docs/architecture/storage-minimal-contract.md
@@ -24,7 +24,7 @@
 | MySQL | `vehicle_realtime_snapshot` | 每个协议+VIN 的最新事件时间、接收时间、车牌、事件 ID | phone、device、source endpoint、完整 JSON |
 | MySQL | `vehicle_realtime_location` | 每个协议+VIN 的最新位置核心字段 | raw、parsed JSON、消息头内部字段 |
 | MySQL | `vehicle_daily_mileage` | 日期、VIN、协议、日里程、首末总里程、样本数 | 临时 vehicle key、泛化 metric key/value、自增 id、created_at、每帧细节、位置点列表 |
-| MySQL | `vehicle_identity_binding` | 人工维护或导入的 VIN、车牌、phone、device_id 映射 | 注册历史、协议状态、自增 id、created_at |
+| MySQL | `vehicle_identity_binding` | 人工维护或导入的 VIN、车牌、phone、oem 映射 | 注册历史、协议状态、device_id、自增 id、created_at |
 | MySQL | `jt808_registration` | JT808 phone 主键下的注册、鉴权、VIN 匹配状态、来源端点 | GB32960/MQTT 注册信息、位置历史、created_at |
 | Redis | `vehicle:realtime-raw:{protocol}:{vin}` | 每协议每 VIN 最新完整 parsed 状态 | 无 VIN 临时身份、历史数据、统计结果 |

@@ -72,10 +72,10 @@
   - 绑定补齐 VIN 后，后续新帧自然进入实时态；历史补偿需要从 raw/Kafka 回放。

 8. identity 层只保留两张表。
-   - `vehicle_identity_binding`：VIN 与 plate/phone/device_id 的映射，供 808 等协议反查 VIN。
+   - `vehicle_identity_binding`：VIN 与 plate/phone/oem 的映射，供 808 等协议反查 VIN。该表只允许外部导入或人工维护，服务运行时只读。
   - `jt808_registration`：808 注册、鉴权、最新活跃和 VIN 匹配状态。
   - 状态：`vehicle_identity_binding` 使用 VIN 主键；`jt808_registration` 使用 phone 主键；两张表都不保留代理自增主键和 `created_at`。
-   - 查询：`vehicle_identity_binding` 的 phone/device_id/plate 都是唯一键，接入侧按唯一键直查 VIN，不做无意义排序。
+   - 查询：`vehicle_identity_binding` 的 phone/plate 都是唯一键，接入侧按唯一键直查 VIN，不做无意义排序。
   - 写入链路：gateway 对 binding 查 VIN 使用短 TTL 内存缓存，包含未命中缓存，避免无 VIN 高频位置帧每条都打 MySQL。
   - 生产：RDS 已在上线前删除旧 `vehicle_identity_binding_registration`、`vehicle_identity_bindings`，gateway 启动会自动创建最小 schema。

--- a/go/vehicle-gateway/internal/identity/resolver.go
+++ b/go/vehicle-gateway/internal/identity/resolver.go
@@ -90,7 +90,7 @@ func (r *MySQLResolver) EnsureSchema(ctx context.Context) error {
 	}
 	for _, statement := range identityBindingAlterSQL(r.table) {
 		if _, err := r.db.ExecContext(ctx, statement); err != nil {
-			if !isDuplicateColumnError(err) {
+			if !isIgnoredIdentityBindingAlterError(err) {
 				return err
 			}
 		}
@@ -107,12 +107,10 @@ func identityBindingTableSQL(table string) string {
  vin VARCHAR(32) PRIMARY KEY,
  plate VARCHAR(32) NULL,
  phone VARCHAR(32) NULL,
-  device_id VARCHAR(64) NULL,
  oem VARCHAR(64) NULL,
  updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  UNIQUE KEY uk_identity_plate (plate),
-  UNIQUE KEY uk_identity_phone (phone),
-  UNIQUE KEY uk_identity_device (device_id)
+  UNIQUE KEY uk_identity_phone (phone)
 )`
 }

@@ -121,13 +119,19 @@ func identityBindingAlterSQL(table string) []string {
 		table = "vehicle_identity_binding"
 	}
 	return []string{
-		`ALTER TABLE ` + table + ` ADD COLUMN oem VARCHAR(64) NULL AFTER device_id`,
+		`ALTER TABLE ` + table + ` ADD COLUMN oem VARCHAR(64) NULL AFTER phone`,
+		`ALTER TABLE ` + table + ` DROP INDEX uk_identity_device`,
+		`ALTER TABLE ` + table + ` DROP COLUMN device_id`,
 	}
 }

-func isDuplicateColumnError(err error) bool {
+func isIgnoredIdentityBindingAlterError(err error) bool {
 	text := strings.ToLower(strings.TrimSpace(fmt.Sprint(err)))
-	return strings.Contains(text, "duplicate column") || strings.Contains(text, "error 1060")
+	return strings.Contains(text, "duplicate column") ||
+		strings.Contains(text, "error 1060") ||
+		strings.Contains(text, "can't drop") ||
+		strings.Contains(text, "check that column/key exists") ||
+		strings.Contains(text, "error 1091")
 }

 const jt808RegistrationTableSQL = `CREATE TABLE IF NOT EXISTS jt808_registration (
@@ -238,30 +242,24 @@ func (r *MySQLResolver) resolveFromJT808Registration(ctx context.Context, env en
 		addIdentityMetadata(env.Parsed, "jt808_registration.vin", env.Phone)
 		return env, true, nil
 	}
-	for _, candidate := range []CandidateKey{
-		{Column: "device_id", Value: env.DeviceID},
-		{Column: "plate", Value: env.Plate},
-	} {
-		value := strings.TrimSpace(candidate.Value)
-		if value == "" {
-			continue
-		}
-		vin, err := r.lookup(ctx, candidate.Column, value)
-		if err != nil {
-			if errors.Is(err, sql.ErrNoRows) {
-				continue
-			}
-			return env, false, err
-		}
-		if strings.TrimSpace(vin) == "" {
-			continue
-		}
-		env.VIN = strings.TrimSpace(vin)
-		env.EventID = env.StableEventID()
-		addIdentityMetadata(env.Parsed, "jt808_registration."+candidate.Column, value)
-		return env, true, nil
+	value := strings.TrimSpace(env.Plate)
+	if value == "" {
+		return env, false, nil
 	}
-	return env, false, nil
+	vin, err = r.lookup(ctx, "plate", value)
+	if err != nil {
+		if errors.Is(err, sql.ErrNoRows) {
+			return env, false, nil
+		}
+		return env, false, err
+	}
+	if strings.TrimSpace(vin) == "" {
+		return env, false, nil
+	}
+	env.VIN = strings.TrimSpace(vin)
+	env.EventID = env.StableEventID()
+	addIdentityMetadata(env.Parsed, "jt808_registration.plate", value)
+	return env, true, nil
 }

 func (r *MySQLResolver) lookupJT808Registration(ctx context.Context, phone string) (vin string, deviceID string, plate string, err error) {
@@ -407,16 +405,7 @@ ON DUPLICATE KEY UPDATE
 	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
+	return nil
 }

 func (r *MySQLResolver) shouldTouchJT808Location(env envelope.FrameEnvelope) bool {
@@ -458,7 +447,6 @@ func CandidateKeys(env envelope.FrameEnvelope) []CandidateKey {
 		out = append(out, CandidateKey{Column: column, Value: value})
 	}
 	add("phone", normalizePhone(env.Phone))
-	add("device_id", env.DeviceID)
 	add("plate", env.Plate)
 	add("vin", env.VehicleKeyHint)
 	return out
--- a/go/vehicle-gateway/internal/identity/resolver_test.go
+++ b/go/vehicle-gateway/internal/identity/resolver_test.go
@@ -13,7 +13,7 @@ import (
 	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
 )

-func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) {
+func TestCandidateKeysExcludeDeviceIDForBindingLookup(t *testing.T) {
 	keys := CandidateKeys(envelope.FrameEnvelope{
 		Phone:          "013307795425",
 		DeviceID:       "D1",
@@ -24,7 +24,7 @@ func TestCandidateKeysPrioritizesPhoneDevicePlate(t *testing.T) {
 	for _, key := range keys {
 		got = append(got, key.Column+"="+key.Value)
 	}
-	want := []string{"phone=13307795425", "device_id=D1", "plate=豫A12345", "vin=D1"}
+	want := []string{"phone=13307795425", "plate=豫A12345", "vin=D1"}
 	if len(got) != len(want) {
 		t.Fatalf("candidate count = %d got=%#v", len(got), got)
 	}
@@ -106,15 +106,12 @@ func TestMySQLResolverLooksUpVINByUniqueKeyWithoutSort(t *testing.T) {
 	}
 }

-func TestMySQLResolverFallsBackToDeviceID(t *testing.T) {
+func TestMySQLResolverDoesNotLookupBindingByDeviceID(t *testing.T) {
 	db, mock := newMockDB(t)
 	defer db.Close()
 	mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE phone = \\?").
 		WithArgs("13307795425").
 		WillReturnError(sql.ErrNoRows)
-	mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE device_id = \\?").
-		WithArgs("D1").
-		WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LNBVIN00000000002"))

 	resolver := NewMySQLResolver(db, "vehicle_identity_binding")
 	env, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
@@ -125,7 +122,7 @@ func TestMySQLResolverFallsBackToDeviceID(t *testing.T) {
 	if err != nil {
 		t.Fatalf("Resolve() error = %v", err)
 	}
-	if env.VIN != "LNBVIN00000000002" {
+	if env.VIN != "" {
 		t.Fatalf("vin = %q", env.VIN)
 	}
 	if err := mock.ExpectationsWereMet(); err != nil {
@@ -133,23 +130,17 @@ func TestMySQLResolverFallsBackToDeviceID(t *testing.T) {
 	}
 }

-func TestMySQLResolverTracksJT808RegistrationAndBackfillsBinding(t *testing.T) {
+func TestMySQLResolverTracksJT808RegistrationWithoutWritingBinding(t *testing.T) {
 	db, mock := newMockDB(t)
 	defer db.Close()
 	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("13079963379", "13079963379", "DEV0001", "DEV0001", "TEST123", "TEST123", "LKLG7C4E3NA774736").
-		WillReturnResult(sqlmock.NewResult(0, 1))

 	resolver := NewMySQLResolver(db, "vehicle_identity_binding")
 	env, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
@@ -221,17 +212,11 @@ func TestMySQLResolverUsesJT808RegistrationPlateForLocationVIN(t *testing.T) {
 	mock.ExpectQuery("SELECT vin, device_id, plate FROM jt808_registration WHERE phone = \\?").
 		WithArgs("40692934322").
 		WillReturnRows(sqlmock.NewRows([]string{"vin", "device_id", "plate"}).AddRow("unknown", "18285", "粤AG18285"))
-	mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE device_id = \\?").
-		WithArgs("18285").
-		WillReturnError(sql.ErrNoRows)
 	mock.ExpectQuery("SELECT vin FROM vehicle_identity_binding WHERE plate = \\?").
 		WithArgs("粤AG18285").
 		WillReturnRows(sqlmock.NewRows([]string{"vin"}).AddRow("LNXNEGRR7SR318212"))
 	mock.ExpectExec("INSERT INTO jt808_registration").
 		WillReturnResult(sqlmock.NewResult(1, 1))
-	mock.ExpectExec("UPDATE IGNORE vehicle_identity_binding").
-		WithArgs("40692934322", "40692934322", "18285", "18285", "粤AG18285", "粤AG18285", "LNXNEGRR7SR318212").
-		WillReturnResult(sqlmock.NewResult(0, 1))

 	resolver := NewMySQLResolver(db, "vehicle_identity_binding")
 	env, err := resolver.Resolve(context.Background(), envelope.FrameEnvelope{
@@ -296,6 +281,10 @@ func TestMySQLResolverEnsuresMinimalIdentitySchema(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mock.ExpectExec("ALTER TABLE vehicle_identity_binding ADD COLUMN oem").
 		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec("ALTER TABLE vehicle_identity_binding DROP INDEX uk_identity_device").
+		WillReturnResult(sqlmock.NewResult(0, 0))
+	mock.ExpectExec("ALTER TABLE vehicle_identity_binding DROP COLUMN device_id").
+		WillReturnResult(sqlmock.NewResult(0, 0))
 	mock.ExpectExec("CREATE TABLE IF NOT EXISTS jt808_registration").
 		WillReturnResult(sqlmock.NewResult(0, 0))

@@ -315,6 +304,10 @@ func TestMySQLResolverIgnoresExistingOEMColumn(t *testing.T) {
 		WillReturnResult(sqlmock.NewResult(0, 0))
 	mock.ExpectExec("ALTER TABLE vehicle_identity_binding ADD COLUMN oem").
 		WillReturnError(errors.New("Error 1060 (42S21): Duplicate column name 'oem'"))
+	mock.ExpectExec("ALTER TABLE vehicle_identity_binding DROP INDEX uk_identity_device").
+		WillReturnError(errors.New("Error 1091 (42000): Can't DROP 'uk_identity_device'; check that column/key exists"))
+	mock.ExpectExec("ALTER TABLE vehicle_identity_binding DROP COLUMN device_id").
+		WillReturnError(errors.New("Error 1091 (42000): Can't DROP 'device_id'; check that column/key exists"))
 	mock.ExpectExec("CREATE TABLE IF NOT EXISTS jt808_registration").
 		WillReturnResult(sqlmock.NewResult(0, 0))

@@ -340,6 +333,9 @@ func TestIdentitySchemaUsesBusinessKeysOnly(t *testing.T) {
 	if !strings.Contains(binding, "vin VARCHAR(32) PRIMARY KEY") {
 		t.Fatalf("binding table should key by vin:\n%s", binding)
 	}
+	if strings.Contains(binding, "device_id") {
+		t.Fatalf("binding table should not contain device_id:\n%s", binding)
+	}
 	if !strings.Contains(registration, "phone VARCHAR(32) PRIMARY KEY") {
 		t.Fatalf("registration table should key by phone:\n%s", registration)
 	}