From e27af63025c752d960b9139ac6528160db31d591 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 21:28:29 +0800 Subject: [PATCH] refactor(go): simplify history location identity --- .../production-data-plane-inventory.md | 2 +- docs/architecture/storage-minimal-contract.md | 9 +- go/vehicle-gateway/internal/history/query.go | 100 +++++------------- .../internal/history/query_test.go | 68 +++++++----- go/vehicle-gateway/internal/history/schema.go | 5 +- go/vehicle-gateway/internal/history/writer.go | 38 ++++++- .../internal/history/writer_test.go | 51 +++++++++ 7 files changed, 161 insertions(+), 112 deletions(-) diff --git a/docs/architecture/production-data-plane-inventory.md b/docs/architecture/production-data-plane-inventory.md index b1c0d01a..b197fecc 100644 --- a/docs/architecture/production-data-plane-inventory.md +++ b/docs/architecture/production-data-plane-inventory.md @@ -50,7 +50,7 @@ NATS 部署在 Kafka ECS 内网 `172.17.111.56:4222`。 | --- | --- | --- | --- | | `raw_frames` | `ts`、`frame_id`、`event_id`、`message_id`、`event_time`、`received_at`、`raw_size_bytes`、`raw_hex`、`raw_text`、`parsed_json`、`parse_status`、`parse_error`、`source_endpoint` | `protocol`、`vehicle_key`、`vin`、`phone`、`device_id` | RAW 证据层，保存完整解析 JSON | | `raw_frame_payload_chunks` | 分片 payload 列 | 协议和车辆标识 tags | 保存超长 raw/parsed payload | -| `vehicle_locations` | `ts`、`event_id`、`frame_id`、`received_at`、`longitude`、`latitude`、`altitude_m`、`speed_kmh`、`direction_deg`、`alarm_flag`、`status_flag`、`total_mileage_km` | `protocol`、`vehicle_key`、`vin`、`phone`、`device_id` | 高频历史位置和总里程查询 | +| `vehicle_locations` | `ts`、`event_id`、`frame_id`、`received_at`、`longitude`、`latitude`、`altitude_m`、`speed_kmh`、`direction_deg`、`alarm_flag`、`status_flag`、`total_mileage_km` | `protocol`、`vin` | 高频历史位置和总里程查询；只写 VIN 非空车辆 | 不应重新出现： diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index 2efe86dd..b81b5ade 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -20,7 +20,7 @@ | --- | --- | --- | --- | | TDengine | `raw_frames` | raw hex/text、完整 `parsed_json`、解析状态、协议标签、车辆标识标签 | `fields_json` 这类可从 `parsed_json`/业务表再得到的重复字段 | | TDengine | `raw_frame_payload_chunks` | 超长 raw/parsed payload 分片 | 业务查询字段 | -| TDengine | `vehicle_locations` | 时间、VIN/协议标签、经纬度、速度、方向、SOC、总里程等位置核心字段 | 完整协议 JSON、注册鉴权信息 | +| TDengine | `vehicle_locations` | 时间、VIN/协议标签、经纬度、速度、方向、SOC、总里程等位置核心字段 | 完整协议 JSON、注册鉴权信息、phone/device_id/vehicle_key 兜底标识 | | 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、每帧细节、位置点列表 | @@ -45,7 +45,12 @@ - 无 VIN 的 JT808 等数据保留在 raw/Redis/session 中用于排查和待绑定，不进入正式车辆指标。 - 生产：RDS 已在上线前删除泛化 `vehicle_daily_metric`，改为专用 `vehicle_daily_mileage`。 -4. identity 层只保留两张表。 +4. `vehicle_locations` 不再接收临时身份兜底字段。 + - 原因：位置历史是正式车辆时间序列，只按 `protocol + vin` 分表和查询。 + - 无 VIN 的位置帧仍保留在 `raw_frames.parsed_json`，待 identity/binding 修复后可从 raw 重放补写。 + - 生产：ECS TDengine `vehicle_locations` 已在上线前重建为 `protocol`、`vin` 两个 tag。 + +5. 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/history/query.go b/go/vehicle-gateway/internal/history/query.go index 19586968..606eddd3 100644 --- a/go/vehicle-gateway/internal/history/query.go +++ b/go/vehicle-gateway/internal/history/query.go @@ -54,15 +54,12 @@ type RawFrameRow struct { } type LocationQuery struct { - Protocol string - VehicleKey string - VIN string - Phone string - DeviceID string - DateFrom string - DateTo string - Limit int - Offset int + Protocol string + VIN string + DateFrom string + DateTo string + Limit int + Offset int } type LocationRow struct { @@ -79,22 +76,16 @@ type LocationRow struct { StatusFlag *int64 `json:"status_flag,omitempty"` TotalMileageKM *float64 `json:"total_mileage_km,omitempty"` Protocol string `json:"protocol"` - VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` - Phone string `json:"phone,omitempty"` - DeviceID string `json:"device_id,omitempty"` } type MileagePointQuery struct { - Protocol string - VehicleKey string - VIN string - Phone string - DeviceID string - DateFrom string - DateTo string - Limit int - Offset int + Protocol string + VIN string + DateFrom string + DateTo string + Limit int + Offset int } type MileagePointRow struct { @@ -107,10 +98,7 @@ type MileagePointRow struct { Longitude *float64 `json:"longitude,omitempty"` Latitude *float64 `json:"latitude,omitempty"` Protocol string `json:"protocol"` - VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` - Phone string `json:"phone,omitempty"` - DeviceID string `json:"device_id,omitempty"` } type RawFrameRepository struct { @@ -253,10 +241,7 @@ func (r *LocationRepository) Query(ctx context.Context, query LocationQuery) ([] &status, &mileage, &row.Protocol, - &row.VehicleKey, &row.VIN, - &row.Phone, - &row.DeviceID, ); err != nil { return nil, err } @@ -306,10 +291,7 @@ func (r *MileagePointRepository) Query(ctx context.Context, query MileagePointQu &longitude, &latitude, &row.Protocol, - &row.VehicleKey, &row.VIN, - &row.Phone, - &row.DeviceID, ); err != nil { return nil, err } @@ -390,10 +372,7 @@ func normalizeRawFrameQuery(query RawFrameQuery) RawFrameQuery { func normalizeLocationQuery(query LocationQuery) LocationQuery { query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) - query.VehicleKey = strings.TrimSpace(query.VehicleKey) query.VIN = strings.TrimSpace(query.VIN) - query.Phone = strings.TrimSpace(query.Phone) - query.DeviceID = strings.TrimSpace(query.DeviceID) query.DateFrom = normalizeDateTimeLiteral(query.DateFrom) query.DateTo = normalizeDateTimeLiteral(query.DateTo) if query.Limit <= 0 { @@ -404,10 +383,7 @@ func normalizeLocationQuery(query LocationQuery) LocationQuery { func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery { query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) - query.VehicleKey = strings.TrimSpace(query.VehicleKey) query.VIN = strings.TrimSpace(query.VIN) - query.Phone = strings.TrimSpace(query.Phone) - query.DeviceID = strings.TrimSpace(query.DeviceID) query.DateFrom = normalizeDateTimeLiteral(query.DateFrom) query.DateTo = normalizeDateTimeLiteral(query.DateTo) if query.Limit <= 0 { @@ -556,7 +532,7 @@ func quotedList(values []string) string { func buildLocationSQL(table string, query LocationQuery) (string, []any) { where := locationWhere(query) - sqlText := `SELECT ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, protocol, vehicle_key, vin, phone, device_id FROM ` + table + sqlText := `SELECT ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, protocol, vin FROM ` + table if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") } @@ -574,7 +550,7 @@ func buildLocationCountSQL(table string, query LocationQuery) (string, []any) { func buildMileagePointSQL(table string, query MileagePointQuery) (string, []any) { where := mileagePointWhere(query) - sqlText := `SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vehicle_key, vin, phone, device_id FROM ` + table + sqlText := `SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vin FROM ` + table if len(where) > 0 { sqlText += " WHERE " + strings.Join(where, " AND ") } @@ -649,18 +625,9 @@ func locationWhere(query LocationQuery) []string { if query.Protocol != "" { add("protocol = '" + quote(query.Protocol) + "'") } - if query.VehicleKey != "" { - add("vehicle_key = '" + quote(query.VehicleKey) + "'") - } if query.VIN != "" { add("vin = '" + quote(query.VIN) + "'") } - if query.Phone != "" { - add("phone = '" + quote(query.Phone) + "'") - } - if query.DeviceID != "" { - add("device_id = '" + quote(query.DeviceID) + "'") - } if query.DateFrom != "" { add("ts >= '" + quote(normalizeDateTimeLiteral(query.DateFrom)) + "'") } @@ -678,18 +645,9 @@ func mileagePointWhere(query MileagePointQuery) []string { if query.Protocol != "" { add("protocol = '" + quote(query.Protocol) + "'") } - if query.VehicleKey != "" { - add("vehicle_key = '" + quote(query.VehicleKey) + "'") - } if query.VIN != "" { add("vin = '" + quote(query.VIN) + "'") } - if query.Phone != "" { - add("phone = '" + quote(query.Phone) + "'") - } - if query.DeviceID != "" { - add("device_id = '" + quote(query.DeviceID) + "'") - } if query.DateFrom != "" { add("ts >= '" + quote(normalizeDateTimeLiteral(query.DateFrom)) + "'") } @@ -883,15 +841,12 @@ func parseMileagePointQuery(r *http.Request) (MileagePointQuery, error) { return MileagePointQuery{}, err } query := MileagePointQuery{ - Protocol: values.Get("protocol"), - VehicleKey: values.Get("vehicleKey"), - VIN: values.Get("vin"), - Phone: values.Get("phone"), - DeviceID: values.Get("deviceId"), - DateFrom: values.Get("dateFrom"), - DateTo: values.Get("dateTo"), - Limit: limit, - Offset: offset, + Protocol: values.Get("protocol"), + VIN: values.Get("vin"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, } if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) { return MileagePointQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss") @@ -910,15 +865,12 @@ func parseLocationQuery(r *http.Request) (LocationQuery, error) { return LocationQuery{}, err } query := LocationQuery{ - Protocol: values.Get("protocol"), - VehicleKey: values.Get("vehicleKey"), - VIN: values.Get("vin"), - Phone: values.Get("phone"), - DeviceID: values.Get("deviceId"), - DateFrom: values.Get("dateFrom"), - DateTo: values.Get("dateTo"), - Limit: limit, - Offset: offset, + Protocol: values.Get("protocol"), + VIN: values.Get("vin"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, } if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) { return LocationQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss") diff --git a/go/vehicle-gateway/internal/history/query_test.go b/go/vehicle-gateway/internal/history/query_test.go index bbde2346..8a15b99f 100644 --- a/go/vehicle-gateway/internal/history/query_test.go +++ b/go/vehicle-gateway/internal/history/query_test.go @@ -252,7 +252,7 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) { } } -func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) { +func TestLocationHandlerReturnsLocationsByVIN(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) @@ -260,18 +260,18 @@ func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) { defer db.Close() mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_locations"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(17)) - mock.ExpectQuery("vehicle_key = 'JT808:013307811350'"). + mock.ExpectQuery("vin = 'LKLG7C4E3NA774736'"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "event_id", "frame_id", "received_at", "longitude", "latitude", "altitude_m", "speed_kmh", - "direction_deg", "alarm_flag", "status_flag", "total_mileage_km", "protocol", "vehicle_key", "vin", "phone", "device_id", + "direction_deg", "alarm_flag", "status_flag", "total_mileage_km", "protocol", "vin", }).AddRow( "2026-07-02 00:18:22", "event-3", "go_frame", "2026-07-02 00:22:43", 121.07764, 30.585928, 11.0, 8.0, 171, 0, 4718595, 8792.8, - "JT808", "JT808:013307811350", "", "013307811350", "", + "JT808", "LKLG7C4E3NA774736", )) handler := NewLocationHandler(NewLocationRepository(db, "lingniu_vehicle_ts")) - request := httptest.NewRequest(http.MethodGet, "/api/history/locations?vehicleKey=JT808:013307811350&protocol=JT808&limit=1", nil) + request := httptest.NewRequest(http.MethodGet, "/api/history/locations?vin=LKLG7C4E3NA774736&protocol=JT808&limit=1", nil) response := httptest.NewRecorder() handler.ServeHTTP(response, request) @@ -280,17 +280,22 @@ func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) { t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) } body := response.Body.String() - for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"longitude":121.07764`, `"total_mileage_km":8792.8`, `"total":17`} { + for _, want := range []string{`"vin":"LKLG7C4E3NA774736"`, `"longitude":121.07764`, `"total_mileage_km":8792.8`, `"total":17`} { if !strings.Contains(body, want) { t.Fatalf("response missing %s: %s", want, body) } } + for _, legacy := range []string{"vehicle_key", "phone", "device_id"} { + if strings.Contains(body, legacy) { + t.Fatalf("location response should not expose %s: %s", legacy, body) + } + } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } } -func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) { +func TestMileagePointHandlerReturnsMileageByVIN(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { t.Fatalf("sqlmock.New() error = %v", err) @@ -298,18 +303,18 @@ func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) { defer db.Close() mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_locations"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(19)) - mock.ExpectQuery("SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.vehicle_locations"). + mock.ExpectQuery("SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vin FROM lingniu_vehicle_ts.vehicle_locations"). WillReturnRows(sqlmock.NewRows([]string{ "ts", "event_id", "frame_id", "received_at", "total_mileage_km", "speed_kmh", "longitude", "latitude", - "protocol", "vehicle_key", "vin", "phone", "device_id", + "protocol", "vin", }).AddRow( "2026-07-02 00:18:22", "event-3", "go_frame", "2026-07-02 00:22:43", 8792.8, 8.0, 121.07764, 30.585928, - "JT808", "JT808:013307811350", "", "013307811350", "", + "JT808", "LKLG7C4E3NA774736", )) handler := NewMileagePointHandler(NewMileagePointRepository(db, "lingniu_vehicle_ts")) - request := httptest.NewRequest(http.MethodGet, "/api/history/mileage-points?vehicleKey=JT808:013307811350&protocol=JT808&limit=1", nil) + request := httptest.NewRequest(http.MethodGet, "/api/history/mileage-points?vin=LKLG7C4E3NA774736&protocol=JT808&limit=1", nil) response := httptest.NewRecorder() handler.ServeHTTP(response, request) @@ -318,11 +323,16 @@ func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) { t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) } body := response.Body.String() - for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"total_mileage_km":8792.8`, `"speed_kmh":8`, `"total":19`} { + for _, want := range []string{`"vin":"LKLG7C4E3NA774736"`, `"total_mileage_km":8792.8`, `"speed_kmh":8`, `"total":19`} { if !strings.Contains(body, want) { t.Fatalf("response missing %s: %s", want, body) } } + for _, legacy := range []string{"vehicle_key", "phone", "device_id"} { + if strings.Contains(body, legacy) { + t.Fatalf("mileage response should not expose %s: %s", legacy, body) + } + } if err := mock.ExpectationsWereMet(); err != nil { t.Fatalf("sql expectations: %v", err) } @@ -380,12 +390,12 @@ func TestParseMessageIDSupportsDecimalAndHex(t *testing.T) { func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) { sqlText, args := buildMileagePointSQL("lingniu_vehicle_ts.vehicle_locations", MileagePointQuery{ - Protocol: "JT808", - VehicleKey: "JT808:013307811350", - DateFrom: "2026-07-02 00:00:00", - DateTo: "2026-07-02 23:59:59", - Limit: 20, - Offset: 5, + Protocol: "JT808", + VIN: "LKLG7C4E3NA774736", + DateFrom: "2026-07-02 00:00:00", + DateTo: "2026-07-02 23:59:59", + Limit: 20, + Offset: 5, }) if len(args) != 0 { t.Fatalf("expected no query args for TDengine, got %#v", args) @@ -394,7 +404,7 @@ func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) { "FROM lingniu_vehicle_ts.vehicle_locations", "total_mileage_km IS NOT NULL", "protocol = 'JT808'", - "vehicle_key = 'JT808:013307811350'", + "vin = 'LKLG7C4E3NA774736'", "ts >= '2026-07-01 16:00:00'", "LIMIT 20 OFFSET 5", } { @@ -402,23 +412,26 @@ func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) { t.Fatalf("sql missing %s: %s", want, sqlText) } } + if strings.Contains(sqlText, "vehicle_key") || strings.Contains(sqlText, "phone") || strings.Contains(sqlText, "device_id") { + t.Fatalf("mileage point sql should use vin-only identity filters: %s", sqlText) + } } func TestBuildLocationSQLUsesLiteralsForTDengine(t *testing.T) { sqlText, args := buildLocationSQL("lingniu_vehicle_ts.vehicle_locations", LocationQuery{ - Protocol: "JT808", - VehicleKey: "JT808:013307811350", - DateFrom: "2026-07-02 00:00:00", - DateTo: "2026-07-02 23:59:59", - Limit: 20, - Offset: 5, + Protocol: "JT808", + VIN: "LKLG7C4E3NA774736", + DateFrom: "2026-07-02 00:00:00", + DateTo: "2026-07-02 23:59:59", + Limit: 20, + Offset: 5, }) if len(args) != 0 { t.Fatalf("expected no query args for TDengine, got %#v", args) } for _, want := range []string{ "protocol = 'JT808'", - "vehicle_key = 'JT808:013307811350'", + "vin = 'LKLG7C4E3NA774736'", "ts >= '2026-07-01 16:00:00'", "LIMIT 20 OFFSET 5", } { @@ -426,6 +439,9 @@ func TestBuildLocationSQLUsesLiteralsForTDengine(t *testing.T) { t.Fatalf("sql missing %s: %s", want, sqlText) } } + if strings.Contains(sqlText, "vehicle_key") || strings.Contains(sqlText, "phone") || strings.Contains(sqlText, "device_id") { + t.Fatalf("location sql should use vin-only identity filters: %s", sqlText) + } } func TestBuildRawFrameSQLUsesLiteralsForTDengine(t *testing.T) { diff --git a/go/vehicle-gateway/internal/history/schema.go b/go/vehicle-gateway/internal/history/schema.go index ed9cace7..6240f8c3 100644 --- a/go/vehicle-gateway/internal/history/schema.go +++ b/go/vehicle-gateway/internal/history/schema.go @@ -61,10 +61,7 @@ func SchemaStatements(database string) []string { total_mileage_km DOUBLE ) TAGS ( protocol NCHAR(32), - vehicle_key NCHAR(64), - vin NCHAR(32), - phone NCHAR(32), - device_id NCHAR(64) + vin NCHAR(32) )`, } } diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index 840e0770..b768f81b 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -67,7 +67,7 @@ func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) erro func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) error { table := tableName("raw", env) - if err := w.ensureChild(ctx, table, "raw_frames", env); err != nil { + if err := w.ensureRawChild(ctx, table, "raw_frames", env); err != nil { return err } rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex) @@ -86,7 +86,7 @@ VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON)))) return nil } chunkTable := tableName("chunk", env) - if err := w.ensureChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil { + if err := w.ensureRawChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil { return err } for _, chunk := range chunks { @@ -100,13 +100,16 @@ VALUES (%s)`, chunkTable, joinLiterals(chunkValues(env, chunk)))); err != nil { } func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope) error { + if strings.TrimSpace(env.VIN) == "" { + return nil + } longitude, okLon := floatField(env, envelope.FieldLongitude) latitude, okLat := floatField(env, envelope.FieldLatitude) if !okLon || !okLat { return nil } - table := tableName("loc", env) - if err := w.ensureChild(ctx, table, "vehicle_locations", env); err != nil { + table := locationTableName(env) + if err := w.ensureLocationChild(ctx, table, env); err != nil { return err } _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s @@ -116,7 +119,7 @@ VALUES (%s)`, table, joinLiterals(locationValues(env, longitude, latitude)))) return err } -func (w *Writer) ensureChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error { +func (w *Writer) ensureRawChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error { key := stable + "." + table w.cache.mu.Lock() _, ok := w.cache.seen[key] @@ -145,6 +148,27 @@ func (w *Writer) ensureChild(ctx context.Context, table string, stable string, e return nil } +func (w *Writer) ensureLocationChild(ctx context.Context, table string, env envelope.FrameEnvelope) error { + key := "vehicle_locations." + table + w.cache.mu.Lock() + _, ok := w.cache.seen[key] + w.cache.mu.Unlock() + if ok { + return nil + } + statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING vehicle_locations TAGS ('%s', '%s')", + table, + quote(string(env.Protocol)), + quote(strings.TrimSpace(env.VIN))) + if _, err := w.exec.ExecContext(ctx, statement); err != nil { + return err + } + w.cache.mu.Lock() + w.cache.seen[key] = struct{}{} + w.cache.mu.Unlock() + return nil +} + func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string) []any { received := millis(env.ReceivedAtMS) eventTime := millis(env.EventTimeMS) @@ -246,6 +270,10 @@ func tableName(prefix string, env envelope.FrameEnvelope) string { return prefix + "_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(env.VehicleKey()) } +func locationTableName(env envelope.FrameEnvelope) string { + return "loc_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(strings.TrimSpace(env.VIN)) +} + func normalizePhoneTag(phone string) string { trimmed := strings.TrimLeft(strings.TrimSpace(phone), "0") if trimmed == "" { diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index 51ff209e..2f226907 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -22,6 +22,12 @@ func TestSchemaStatementsCreateCoreStables(t *testing.T) { if strings.Contains(statements, "fields_json") { t.Fatalf("raw schema should not create duplicate fields_json column:\n%s", statements) } + locationStable := between(statements, "CREATE STABLE IF NOT EXISTS vehicle_locations", "}") + for _, field := range []string{"vehicle_key", "phone", "device_id"} { + if strings.Contains(locationStable, field) { + t.Fatalf("location stable should only keep protocol/vin identity tags, found %s:\n%s", field, locationStable) + } + } } func TestWriterAppendsRawAndLocationOnly(t *testing.T) { @@ -61,6 +67,17 @@ func TestWriterAppendsRawAndLocationOnly(t *testing.T) { if strings.Contains(rawInsert, "fields_json") { t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert) } + locationChild := findSQL(exec.calls, "USING vehicle_locations") + for _, want := range []string{"'JT808'", "'LNBVIN00000000001'"} { + if !strings.Contains(locationChild, want) { + t.Fatalf("location child should tag by protocol and vin only: %s", locationChild) + } + } + for _, field := range []string{"'JT808:013307795425'", "'13307795425'", "device_id"} { + if strings.Contains(locationChild, field) { + t.Fatalf("location child should not carry raw identity fallback tag %s: %s", field, locationChild) + } + } } func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) { @@ -130,6 +147,27 @@ func TestWriterSkipsSparseDerivedRows(t *testing.T) { } } +func TestWriterSkipsLocationWhenVINIsMissing(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec) + env := sampleEnvelope() + env.VIN = "" + + if err := writer.AppendAll(context.Background(), env); err != nil { + t.Fatalf("AppendAll() error = %v", err) + } + + if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 { + t.Fatalf("raw insert count = %d", got) + } + if got := countSQL(exec.calls, "USING vehicle_locations"); got != 0 { + t.Fatalf("location child create count = %d", got) + } + if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 0 { + t.Fatalf("location insert count = %d", got) + } +} + func TestRawSizeBytesUsesRawTextWhenHexIsEmpty(t *testing.T) { env := envelope.FrameEnvelope{RawText: `{"code":"0F80","data":{"speed":12}}`} if got, want := rawSizeBytes(env), len([]byte(env.RawText)); got != want { @@ -195,6 +233,19 @@ func findSQL(calls []execCall, pattern string) string { return "" } +func between(value string, start string, end string) string { + startIndex := strings.Index(value, start) + if startIndex < 0 { + return "" + } + value = value[startIndex:] + endIndex := strings.Index(value, end) + if endIndex < 0 { + return value + } + return value[:endIndex] +} + type execCall struct { query string args []any