From e89c9fd98fb8feecebcaa72ac0659a5843a32e84 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 07:11:40 +0800 Subject: [PATCH] refactor(go): simplify realtime mysql current tables --- .../production-data-plane-inventory.md | 2 ++ docs/architecture/storage-minimal-contract.md | 7 ++++++- .../internal/realtime/mysql_query.go | 15 +++++---------- .../internal/realtime/mysql_query_test.go | 18 ++++++++++++------ .../internal/realtime/openapi.go | 2 -- .../internal/realtime/snapshot_writer.go | 8 ++------ .../internal/realtime/snapshot_writer_test.go | 5 ++++- 7 files changed, 31 insertions(+), 26 deletions(-) diff --git a/docs/architecture/production-data-plane-inventory.md b/docs/architecture/production-data-plane-inventory.md index b197fecc..a4867b5b 100644 --- a/docs/architecture/production-data-plane-inventory.md +++ b/docs/architecture/production-data-plane-inventory.md @@ -69,6 +69,8 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。 | `vehicle_realtime_location` | `protocol`、`vin`、`plate`、经纬度、速度、总里程、SOC、事件时间 | Realtime API/projector | 每协议每 VIN 最新位置业务缓存 | | `vehicle_daily_mileage` | `vin`、`stat_date`、`protocol`、`daily_mileage_km`、首末总里程、样本数 | Stat writer | 基于总里程差值的每日里程；无 VIN 数据不进入正式统计表 | +两张实时当前态表均使用 `(protocol, vin)` 作为业务主键，不保留自增 `id` 和 `created_at`；最新更新时间使用 `updated_at`。 + 不应重新出现： - `vehicle_daily_metric` diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index eda18a67..1f85da1e 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -50,7 +50,12 @@ - 无 VIN 的位置帧仍保留在 `raw_frames.parsed_json`，待 identity/binding 修复后可从 raw 重放补写。 - 生产：ECS TDengine `vehicle_locations` 已在上线前重建为 `protocol`、`vin` 两个 tag。 -5. identity 层只保留两张表。 +5. MySQL realtime 当前态表不保留代理主键和创建时间。 + - 原因：`vehicle_realtime_snapshot`、`vehicle_realtime_location` 都是每协议每 VIN 一行的当前态投影，业务主键就是 `(protocol, vin)`。 + - 状态：schema 使用 `PRIMARY KEY (protocol, vin)`，不保留自增 `id` 和 `created_at`，对外只暴露最新 `updated_at`。 + - 生产：项目上线前可直接重建这两张 MySQL 表，实时数据会从 Kafka 新消息继续投影。 + +6. identity 层只保留两张表。 - `vehicle_identity_binding`：VIN 与 plate/phone/device_id 的映射，供 808 等协议反查 VIN。 - `jt808_registration`：808 注册、鉴权、最新活跃和 VIN 匹配状态。 - 生产：RDS 已在上线前删除旧 `vehicle_identity_binding_registration`、`vehicle_identity_bindings`，gateway 启动会自动创建最小 schema。 diff --git a/go/vehicle-gateway/internal/realtime/mysql_query.go b/go/vehicle-gateway/internal/realtime/mysql_query.go index 43f833f8..57588fe5 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query.go @@ -30,7 +30,6 @@ type SnapshotRow struct { EventTime string `json:"event_time,omitempty"` ReceivedAt string `json:"received_at,omitempty"` EventID string `json:"event_id"` - CreatedAt string `json:"created_at"` UpdatedAt string `json:"updated_at"` } @@ -50,7 +49,6 @@ type LocationRow struct { StatusFlag *int64 `json:"status_flag,omitempty"` ReceivedAt string `json:"received_at,omitempty"` EventID string `json:"event_id"` - CreatedAt string `json:"created_at"` UpdatedAt string `json:"updated_at"` } @@ -74,7 +72,7 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable query = normalizeRealtimeTableQuery(query) sqlText, args := buildRealtimeSelectSQL( "vehicle_realtime_snapshot", - "protocol, vin, plate, event_time, received_at, event_id, created_at, updated_at", + "protocol, vin, plate, event_time, received_at, event_id, updated_at", query, ) rows, err := r.db.QueryContext(ctx, sqlText, args...) @@ -86,13 +84,12 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable out := make([]SnapshotRow, 0) for rows.Next() { var row SnapshotRow - var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime - if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &createdAt, &updatedAt); err != nil { + var eventTime, receivedAt, updatedAt scanSQLDateTime + if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { return nil, err } row.EventTime = eventTime.String row.ReceivedAt = receivedAt.String - row.CreatedAt = createdAt.String row.UpdatedAt = updatedAt.String out = append(out, row) } @@ -119,7 +116,7 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable query = normalizeRealtimeTableQuery(query) sqlText, args := buildRealtimeSelectSQL( "vehicle_realtime_location", - "protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, created_at, updated_at", + "protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, updated_at", query, ) rows, err := r.db.QueryContext(ctx, sqlText, args...) @@ -131,7 +128,7 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable out := make([]LocationRow, 0) for rows.Next() { var row LocationRow - var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime + var eventTime, receivedAt, updatedAt scanSQLDateTime var speed, mileage, soc, altitude, direction sql.NullFloat64 var alarm, status sql.NullInt64 if err := rows.Scan( @@ -150,7 +147,6 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable &status, &receivedAt, &row.EventID, - &createdAt, &updatedAt, ); err != nil { return nil, err @@ -164,7 +160,6 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable row.AlarmFlag = nullableInt(alarm) row.StatusFlag = nullableInt(status) row.ReceivedAt = receivedAt.String - row.CreatedAt = createdAt.String row.UpdatedAt = updatedAt.String out = append(out, row) } diff --git a/go/vehicle-gateway/internal/realtime/mysql_query_test.go b/go/vehicle-gateway/internal/realtime/mysql_query_test.go index 24690f70..f5978ca2 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query_test.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query_test.go @@ -20,12 +20,12 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) { mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_snapshot"). WithArgs("GB32960", "LB9A32A21R0LS1707"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) - mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, created_at, updated_at FROM vehicle_realtime_snapshot"). + mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). WithArgs("GB32960", "LB9A32A21R0LS1707", 20, 0). WillReturnRows(sqlmock.NewRows([]string{ - "protocol", "vin", "plate", "event_time", "received_at", "event_id", "created_at", "updated_at", + "protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at", }).AddRow( - "GB32960", "LB9A32A21R0LS1707", "浙A12345", "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:03", "2026-07-02 16:01:04", + "GB32960", "LB9A32A21R0LS1707", "浙A12345", "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04", )) handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db)) @@ -43,6 +43,9 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) { t.Fatalf("response missing %s: %s", want, body) } } + if strings.Contains(body, "created_at") { + t.Fatalf("snapshot response should not expose created_at: %s", body) + } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } @@ -58,14 +61,14 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) { mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_location"). WithArgs("JT808", "粤B98765"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) - mock.ExpectQuery("SELECT protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, created_at, updated_at FROM vehicle_realtime_location"). + mock.ExpectQuery("SELECT protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, updated_at FROM vehicle_realtime_location"). WithArgs("JT808", "粤B98765", 10, 10). WillReturnRows(sqlmock.NewRows([]string{ "protocol", "vin", "plate", "event_time", "latitude", "longitude", "speed_kmh", "total_mileage_km", - "soc_percent", "altitude_m", "direction_deg", "alarm_flag", "status_flag", "received_at", "event_id", "created_at", "updated_at", + "soc_percent", "altitude_m", "direction_deg", "alarm_flag", "status_flag", "received_at", "event_id", "updated_at", }).AddRow( "JT808", "LKLG7C4E3NA774736", "粤B98765", "2026-07-02 16:11:02.000", 30.123456, 120.654321, 54.3, 48798.9, - nil, 19.0, 88.0, int64(0), int64(3), "2026-07-02 16:11:03.000", "evt-2", "2026-07-02 16:11:03", "2026-07-02 16:11:04", + nil, 19.0, 88.0, int64(0), int64(3), "2026-07-02 16:11:03.000", "evt-2", "2026-07-02 16:11:04", )) handler := NewLocationQueryHandler(NewLocationQueryRepository(db)) @@ -83,6 +86,9 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) { t.Fatalf("response missing %s: %s", want, body) } } + if strings.Contains(body, "created_at") { + t.Fatalf("location response should not expose created_at: %s", body) + } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } diff --git a/go/vehicle-gateway/internal/realtime/openapi.go b/go/vehicle-gateway/internal/realtime/openapi.go index 2fe9b0ae..6eebceec 100644 --- a/go/vehicle-gateway/internal/realtime/openapi.go +++ b/go/vehicle-gateway/internal/realtime/openapi.go @@ -93,7 +93,6 @@ func realtimeOpenAPISpec() map[string]any { "event_time": stringSchema("2026-07-02 16:01:02.123"), "received_at": stringSchema("2026-07-02 16:01:03.456"), "event_id": stringSchema("event id"), - "created_at": stringSchema("2026-07-02 16:01:03"), "updated_at": stringSchema("2026-07-02 16:01:04"), }, }, @@ -115,7 +114,6 @@ func realtimeOpenAPISpec() map[string]any { "status_flag": integerSchema(3), "received_at": stringSchema("2026-07-02 16:11:03.000"), "event_id": stringSchema("event id"), - "created_at": stringSchema("2026-07-02 16:11:03"), "updated_at": stringSchema("2026-07-02 16:11:04"), }, }, diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 1c27a8dc..ada9dad6 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -264,16 +264,14 @@ func hasRealtimePayload(env envelope.FrameEnvelope) bool { } const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot ( - id BIGINT PRIMARY KEY AUTO_INCREMENT, protocol VARCHAR(32) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '', event_time DATETIME(3) NULL, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_realtime_snapshot_vin (protocol, vin), + PRIMARY KEY (protocol, vin), KEY idx_vin (vin), KEY idx_protocol_updated (protocol, updated_at) )` @@ -291,7 +289,6 @@ ON DUPLICATE KEY UPDATE ` const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location ( - id BIGINT PRIMARY KEY AUTO_INCREMENT, protocol VARCHAR(32) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '', @@ -307,9 +304,8 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo status_flag BIGINT NULL, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', - created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - UNIQUE KEY uk_realtime_location_vin (protocol, vin), + PRIMARY KEY (protocol, vin), KEY idx_vin (vin), KEY idx_protocol_updated (protocol, updated_at), KEY idx_location (longitude, latitude) diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 17f66bec..74e92555 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -52,11 +52,14 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if strings.Contains(call.query, "parsed_json") || strings.Contains(call.query, "fields_json") { t.Fatalf("realtime schema should not contain json payload columns: %s", call.query) } - for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { + for _, column := range []string{"id BIGINT", "created_at", "message_id", "sequence_id", "source_endpoint"} { if strings.Contains(call.query, column) { t.Fatalf("realtime schema should not contain %s: %s", column, call.query) } } + if !strings.Contains(call.query, "PRIMARY KEY (protocol, vin)") { + t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query) + } } upsert := exec.calls[2] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {