From 6fb6262d0dfbc55788af686e463f90b22be32bca Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 11:34:52 +0800 Subject: [PATCH] feat(go): add realtime kv projection --- go/vehicle-gateway/cmd/realtime-api/main.go | 2 + go/vehicle-gateway/internal/realtime/kv.go | 187 ++++++++++++++++++ .../internal/realtime/kv_test.go | 78 ++++++++ .../internal/realtime/mysql_query.go | 139 +++++++++++++ .../internal/realtime/mysql_query_test.go | 38 ++++ .../internal/realtime/openapi.go | 46 ++++- .../internal/realtime/repository.go | 31 +++ .../internal/realtime/repository_test.go | 42 ++++ .../internal/realtime/snapshot_writer.go | 70 ++++++- .../internal/realtime/snapshot_writer_test.go | 37 ++-- 10 files changed, 653 insertions(+), 17 deletions(-) create mode 100644 go/vehicle-gateway/internal/realtime/kv.go create mode 100644 go/vehicle-gateway/internal/realtime/kv_test.go diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index 228a6132..ad95e073 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -90,6 +90,7 @@ func main() { mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) mux.Handle("/api/realtime/snapshots", realtime.NewSnapshotQueryHandler(realtime.NewSnapshotQueryRepository(db))) mux.Handle("/api/realtime/locations", realtime.NewLocationQueryHandler(realtime.NewLocationQueryRepository(db))) + mux.Handle("/api/realtime/kv", realtime.NewKVQueryHandler(realtime.NewKVQueryRepository(db))) logger.Info("stats mysql query enabled") } else { mysqlUnavailable := func(w http.ResponseWriter, _ *http.Request) { @@ -100,6 +101,7 @@ func main() { mux.HandleFunc("/api/stats/daily-metrics", mysqlUnavailable) mux.HandleFunc("/api/realtime/snapshots", mysqlUnavailable) mux.HandleFunc("/api/realtime/locations", mysqlUnavailable) + mux.HandleFunc("/api/realtime/kv", mysqlUnavailable) logger.Warn("MYSQL_DSN is empty; stats query api disabled") } defer closeStats() diff --git a/go/vehicle-gateway/internal/realtime/kv.go b/go/vehicle-gateway/internal/realtime/kv.go new file mode 100644 index 00000000..b34a51b3 --- /dev/null +++ b/go/vehicle-gateway/internal/realtime/kv.go @@ -0,0 +1,187 @@ +package realtime + +import ( + "encoding/json" + "fmt" + "sort" + "strconv" + "strings" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type RealtimeKVField struct { + Protocol envelope.Protocol + VIN string + Domain string + Field string + Value string + ValueType string + EventTimeMS int64 + ReceivedAtMS int64 + EventID string +} + +func realtimeKVFields(env envelope.FrameEnvelope, parsed map[string]any) []RealtimeKVField { + vin := strings.TrimSpace(env.VIN) + if vin == "" { + return nil + } + eventID := env.StableEventID() + add := func(rows []RealtimeKVField, domain string, fields map[string]any) []RealtimeKVField { + flat := map[string]any{} + flattenKV("", fields, flat) + names := make([]string, 0, len(flat)) + for name := range flat { + names = append(names, name) + } + sort.Strings(names) + for _, name := range names { + value, valueType, ok := stringifyKVValue(flat[name]) + if !ok { + continue + } + rows = append(rows, RealtimeKVField{ + Protocol: env.Protocol, + VIN: vin, + Domain: domain, + Field: name, + Value: value, + ValueType: valueType, + EventTimeMS: env.EventTimeMS, + ReceivedAtMS: env.ReceivedAtMS, + EventID: eventID, + }) + } + return rows + } + + var rows []RealtimeKVField + switch env.Protocol { + case envelope.ProtocolGB32960: + units, ok := asAnySlice(parsed["data_units"]) + if !ok { + return rows + } + for _, unit := range units { + unitMap, ok := unit.(map[string]any) + if !ok { + continue + } + domain := strings.TrimSpace(strconvAny(unitMap["name"])) + if domain == "" { + domain = strings.TrimSpace(strconvAny(unitMap["type"])) + } + if domain == "" { + continue + } + value, ok := unitMap["value"].(map[string]any) + if !ok { + continue + } + rows = add(rows, domain, domainKVFields(domain, value)) + } + case envelope.ProtocolJT808: + fields := cloneFields(env.Fields) + if location, ok := parsed["location"].(map[string]any); ok { + for key, value := range location { + fields[key] = value + } + } + rows = add(rows, "location", fields) + case envelope.ProtocolYutongMQTT: + rows = add(rows, "vehicle", cloneFields(env.Fields)) + } + return rows +} + +func domainKVFields(domain string, value map[string]any) map[string]any { + out := cloneMap(value) + if domain != "gd_fc_stack" { + return out + } + summaries, ok := asAnySlice(out["summaries"]) + if !ok || len(summaries) != 1 { + return out + } + summary, ok := summaries[0].(map[string]any) + if !ok { + return out + } + delete(out, "summaries") + for key, value := range summary { + out[key] = value + } + return out +} + +func flattenKV(prefix string, value any, out map[string]any) { + switch typed := value.(type) { + case map[string]any: + keys := make([]string, 0, len(typed)) + for key := range typed { + keys = append(keys, key) + } + sort.Strings(keys) + for _, key := range keys { + next := key + if prefix != "" { + next = prefix + "." + key + } + flattenKV(next, typed[key], out) + } + case []any: + for index, item := range typed { + itemMap, ok := item.(map[string]any) + if !ok { + out[prefix] = typed + return + } + itemKey := strconv.Itoa(index) + if serial := strings.TrimSpace(strconvAny(itemMap["serial_no"])); serial != "" && serial != "" { + itemKey = serial + } + next := itemKey + if prefix != "" { + next = prefix + "." + itemKey + } + flattenKV(next, itemMap, out) + } + case []map[string]any: + items := make([]any, len(typed)) + for index, item := range typed { + items[index] = item + } + flattenKV(prefix, items, out) + default: + if prefix != "" && value != nil { + out[prefix] = value + } + } +} + +func stringifyKVValue(value any) (string, string, bool) { + switch typed := value.(type) { + case string: + text := strings.TrimSpace(typed) + return text, "string", text != "" + case bool: + return strconv.FormatBool(typed), "bool", true + case float64: + return strconv.FormatFloat(typed, 'f', -1, 64), "number", true + case float32: + return strconv.FormatFloat(float64(typed), 'f', -1, 64), "number", true + case int: + return strconv.Itoa(typed), "number", true + case int8, int16, int32, int64: + return fmt.Sprintf("%d", typed), "number", true + case uint, uint8, uint16, uint32, uint64: + return fmt.Sprintf("%d", typed), "number", true + default: + data, err := json.Marshal(typed) + if err != nil || string(data) == "null" { + return "", "", false + } + return string(data), "json", true + } +} diff --git a/go/vehicle-gateway/internal/realtime/kv_test.go b/go/vehicle-gateway/internal/realtime/kv_test.go new file mode 100644 index 00000000..ea7fa79c --- /dev/null +++ b/go/vehicle-gateway/internal/realtime/kv_test.go @@ -0,0 +1,78 @@ +package realtime + +import ( + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestRealtimeKVFieldsFromGB32960ParsedDomains(t *testing.T) { + rows := realtimeKVFields(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + EventID: "event-1", + }, map[string]any{ + "data_units": []any{ + map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}, + map[string]any{"type": "0x30", "name": "gd_fc_stack", "value": map[string]any{ + "stack_count": 1, + "summaries": []any{ + map[string]any{"stack_water_outlet_temp_c": 63, "hydrogen_inlet_pressure_kpa": 130}, + }, + }}, + }, + }) + + values := kvMap(rows) + if values["vehicle/soc_percent"] != "88" { + t.Fatalf("vehicle soc kv missing: %#v", values) + } + if values["gd_fc_stack/stack_water_outlet_temp_c"] != "63" { + t.Fatalf("stack temp kv missing: %#v", values) + } + if values["gd_fc_stack/hydrogen_inlet_pressure_kpa"] != "130" { + t.Fatalf("stack pressure kv missing: %#v", values) + } + if values["gd_fc_stack/summaries.0.stack_water_outlet_temp_c"] != "" { + t.Fatalf("stack single summary should be flattened onto domain, got %#v", values) + } +} + +func TestRealtimeKVFieldsFromJT808LocationAndMQTTFields(t *testing.T) { + jtRows := realtimeKVFields(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "VIN001", + Fields: map[string]any{ + envelope.FieldLongitude: 121.1, + envelope.FieldLatitude: 30.2, + envelope.FieldTotalMileageKM: 10241.2, + }, + }, map[string]any{}) + jtValues := kvMap(jtRows) + if jtValues["location/total_mileage_km"] != "10241.2" || jtValues["location/longitude"] != "121.1" { + t.Fatalf("jt808 location kv missing: %#v", jtValues) + } + + mqttRows := realtimeKVFields(envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + VIN: "VIN002", + Fields: map[string]any{ + envelope.FieldSOCPercent: 76, + "gear": 3, + }, + }, map[string]any{}) + mqttValues := kvMap(mqttRows) + if mqttValues["vehicle/soc_percent"] != "76" || mqttValues["vehicle/gear"] != "3" { + t.Fatalf("mqtt vehicle kv missing: %#v", mqttValues) + } +} + +func kvMap(rows []RealtimeKVField) map[string]string { + out := map[string]string{} + for _, row := range rows { + out[row.Domain+"/"+row.Field] = row.Value + } + return out +} diff --git a/go/vehicle-gateway/internal/realtime/mysql_query.go b/go/vehicle-gateway/internal/realtime/mysql_query.go index 8a0c5845..52db3acd 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query.go @@ -19,6 +19,8 @@ type RealtimeTableQuery struct { Protocol string VIN string Plate string + Domain string + Field string IncludeTotal bool Limit int Offset int @@ -56,6 +58,19 @@ type LocationRow struct { UpdatedAt string `json:"updated_at"` } +type KVRow struct { + Protocol string `json:"protocol"` + VIN string `json:"vin"` + Domain string `json:"domain"` + Field string `json:"field"` + Value string `json:"value"` + ValueType string `json:"value_type"` + EventTime string `json:"event_time,omitempty"` + ReceivedAt string `json:"received_at,omitempty"` + EventID string `json:"event_id"` + UpdatedAt string `json:"updated_at"` +} + type SnapshotQueryRepository struct { db mysqlQueryer } @@ -108,6 +123,10 @@ type LocationQueryRepository struct { db mysqlQueryer } +type KVQueryRepository struct { + db mysqlQueryer +} + func NewLocationQueryRepository(db mysqlQueryer) *LocationQueryRepository { if db == nil { panic("location query db must not be nil") @@ -115,6 +134,13 @@ func NewLocationQueryRepository(db mysqlQueryer) *LocationQueryRepository { return &LocationQueryRepository{db: db} } +func NewKVQueryRepository(db mysqlQueryer) *KVQueryRepository { + if db == nil { + panic("kv query db must not be nil") + } + return &KVQueryRepository{db: db} +} + func (r *LocationQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) { sqlText, args := buildRealtimeCountSQL("vehicle_realtime_location", normalizeRealtimeTableQuery(query)) return queryCount(ctx, r.db, sqlText, args) @@ -174,6 +200,49 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable return out, rows.Err() } +func (r *KVQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) { + where, args := buildRealtimeKVWhere(normalizeRealtimeTableQuery(query)) + sqlText := "SELECT COUNT(*) FROM vehicle_realtime_kv" + if len(where) > 0 { + sqlText += " WHERE " + strings.Join(where, " AND ") + } + return queryCount(ctx, r.db, sqlText, args) +} + +func (r *KVQueryRepository) Query(ctx context.Context, query RealtimeTableQuery) ([]KVRow, error) { + query = normalizeRealtimeTableQuery(query) + where, args := buildRealtimeKVWhere(query) + sqlText := "SELECT protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id, updated_at FROM vehicle_realtime_kv" + if len(where) > 0 { + sqlText += " WHERE " + strings.Join(where, " AND ") + } + sqlText += " ORDER BY updated_at DESC LIMIT ? OFFSET ?" + args = append(args, query.Limit, query.Offset) + rows, err := r.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + out := make([]KVRow, 0) + for rows.Next() { + var row KVRow + var eventTime, receivedAt, updatedAt scanSQLDateTime + var value sql.NullString + if err := rows.Scan(&row.Protocol, &row.VIN, &row.Domain, &row.Field, &value, &row.ValueType, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { + return nil, err + } + if value.Valid { + row.Value = value.String + } + row.EventTime = eventTime.String + row.ReceivedAt = receivedAt.String + row.UpdatedAt = updatedAt.String + out = append(out, row) + } + return out, rows.Err() +} + type SnapshotQueryHandler struct { repository *SnapshotQueryRepository } @@ -222,6 +291,10 @@ type LocationQueryHandler struct { repository *LocationQueryRepository } +type KVQueryHandler struct { + repository *KVQueryRepository +} + func NewLocationQueryHandler(repository *LocationQueryRepository) *LocationQueryHandler { if repository == nil { panic("location query repository must not be nil") @@ -229,6 +302,13 @@ func NewLocationQueryHandler(repository *LocationQueryRepository) *LocationQuery return &LocationQueryHandler{repository: repository} } +func NewKVQueryHandler(repository *KVQueryRepository) *KVQueryHandler { + if repository == nil { + panic("kv query repository must not be nil") + } + return &KVQueryHandler{repository: repository} +} + func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "method not allowed") @@ -262,6 +342,39 @@ func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) writePage(w, rows, total, query) } +func (h *KVQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if strings.Trim(r.URL.Path, "/") != "api/realtime/kv" { + writeError(w, http.StatusNotFound, "route not found") + return + } + query, err := parseRealtimeTableQuery(r) + if err != nil { + writeError(w, http.StatusBadRequest, err.Error()) + return + } + var total int64 + if query.IncludeTotal { + total, err = h.repository.Count(r.Context(), query) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + } + rows, err := h.repository.Query(r.Context(), query) + if err != nil { + writeError(w, http.StatusInternalServerError, err.Error()) + return + } + if !query.IncludeTotal { + total = int64(len(rows)) + } + writePage(w, rows, total, query) +} + func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) { values := r.URL.Query() limit, err := parseRealtimeBoundedInt(values.Get("limit"), 50, 1, 1000, "limit") @@ -276,6 +389,8 @@ func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) { Protocol: values.Get("protocol"), VIN: values.Get("vin"), Plate: values.Get("plate"), + Domain: values.Get("domain"), + Field: values.Get("field"), IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"), Limit: limit, Offset: offset, @@ -286,12 +401,36 @@ func normalizeRealtimeTableQuery(query RealtimeTableQuery) RealtimeTableQuery { query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) query.VIN = strings.TrimSpace(query.VIN) query.Plate = strings.TrimSpace(query.Plate) + query.Domain = strings.TrimSpace(query.Domain) + query.Field = strings.TrimSpace(query.Field) if query.Limit <= 0 { query.Limit = 50 } return query } +func buildRealtimeKVWhere(query RealtimeTableQuery) ([]string, []any) { + var where []string + var args []any + if query.Protocol != "" { + where = append(where, "protocol = ?") + args = append(args, query.Protocol) + } + if query.VIN != "" { + where = append(where, "vin = ?") + args = append(args, query.VIN) + } + if query.Domain != "" { + where = append(where, "domain_name = ?") + args = append(args, query.Domain) + } + if query.Field != "" { + where = append(where, "field_name = ?") + args = append(args, query.Field) + } + return where, args +} + func buildRealtimeCountSQL(table string, query RealtimeTableQuery) (string, []any) { where, args := buildRealtimeWhere(query) sqlText := "SELECT COUNT(*) FROM " + table diff --git a/go/vehicle-gateway/internal/realtime/mysql_query_test.go b/go/vehicle-gateway/internal/realtime/mysql_query_test.go index 17d5adbe..8f0ca4f3 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query_test.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query_test.go @@ -94,6 +94,44 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) { } } +func TestKVQueryHandlerReturnsRealtimeKVFields(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + + mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_kv"). + WithArgs("GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c"). + WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) + mock.ExpectQuery("SELECT protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id, updated_at FROM vehicle_realtime_kv"). + WithArgs("GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c", 10, 0). + WillReturnRows(sqlmock.NewRows([]string{ + "protocol", "vin", "domain_name", "field_name", "field_value", "value_type", "event_time", "received_at", "event_id", "updated_at", + }).AddRow( + "GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c", "63", "number", "2026-07-03 11:18:45.000", "2026-07-03 11:18:45.123", "evt-kv", "2026-07-03 11:18:46", + )) + + handler := NewKVQueryHandler(NewKVQueryRepository(db)) + request := httptest.NewRequest(http.MethodGet, "/api/realtime/kv?protocol=gb32960&vin=VIN001&domain=gd_fc_stack&field=stack_water_outlet_temp_c&includeTotal=true&limit=10", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"protocol":"GB32960"`, `"domain":"gd_fc_stack"`, `"field":"stack_water_outlet_temp_c"`, `"value":"63"`, `"value_type":"number"`, `"total":1`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestSnapshotQueryHandlerSkipsTotalCountByDefault(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { diff --git a/go/vehicle-gateway/internal/realtime/openapi.go b/go/vehicle-gateway/internal/realtime/openapi.go index f10c6b54..9457606d 100644 --- a/go/vehicle-gateway/internal/realtime/openapi.go +++ b/go/vehicle-gateway/internal/realtime/openapi.go @@ -38,7 +38,7 @@ func realtimeOpenAPISpec() map[string]any { "info": map[string]any{ "title": "Lingniu Vehicle Realtime API", "version": "1.0.0", - "description": "MySQL realtime query APIs for vehicle_realtime_snapshot and vehicle_realtime_location.", + "description": "MySQL realtime query APIs for vehicle_realtime_snapshot, vehicle_realtime_location, and vehicle_realtime_kv.", }, "paths": map[string]any{ "/api/realtime/snapshots": map[string]any{ @@ -79,11 +79,31 @@ func realtimeOpenAPISpec() map[string]any { }, }, }, + "/api/realtime/kv": map[string]any{ + "get": map[string]any{ + "summary": "查询车辆实时 KV 字段", + "description": "从 vehicle_realtime_kv 查询实时字段投影，支持协议、VIN、domain、field 过滤和分页。适合 UI 局部刷新、快速字段查询和统计输入。", + "tags": []string{"Realtime KV API"}, + "x-table": "vehicle_realtime_kv", + "parameters": kvRealtimeParameters(), + "responses": map[string]any{ + "200": map[string]any{ + "description": "KV page", + "content": map[string]any{ + "application/json": map[string]any{ + "schema": map[string]any{"$ref": "#/components/schemas/KVPage"}, + }, + }, + }, + }, + }, + }, }, "components": map[string]any{ "schemas": map[string]any{ "SnapshotPage": pageSchema("#/components/schemas/SnapshotRow"), "LocationPage": pageSchema("#/components/schemas/LocationRow"), + "KVPage": pageSchema("#/components/schemas/KVRow"), "SnapshotRow": map[string]any{ "type": "object", "properties": map[string]any{ @@ -117,11 +137,35 @@ func realtimeOpenAPISpec() map[string]any { "updated_at": stringSchema("2026-07-02 16:11:04"), }, }, + "KVRow": map[string]any{ + "type": "object", + "properties": map[string]any{ + "protocol": stringSchema("GB32960"), + "vin": stringSchema("LNXNEGRR0SR321372"), + "domain": stringSchema("gd_fc_stack"), + "field": stringSchema("stack_water_outlet_temp_c"), + "value": stringSchema("63"), + "value_type": stringSchema("number"), + "event_time": stringSchema("2026-07-03 11:18:45.000"), + "received_at": stringSchema("2026-07-03 11:18:45.123"), + "event_id": stringSchema("event id"), + "updated_at": stringSchema("2026-07-03 11:18:46"), + }, + }, }, }, } } +func kvRealtimeParameters() []map[string]any { + params := commonRealtimeParameters() + params = append(params, + map[string]any{"name": "domain", "in": "query", "schema": map[string]any{"type": "string", "example": "gd_fc_stack"}, "required": false}, + map[string]any{"name": "field", "in": "query", "schema": map[string]any{"type": "string", "example": "stack_water_outlet_temp_c"}, "required": false}, + ) + return params +} + func commonRealtimeParameters() []map[string]any { return []map[string]any{ {"name": "protocol", "in": "query", "schema": map[string]any{"type": "string", "enum": []string{"GB32960", "JT808", "YUTONG_MQTT"}}, "required": false}, diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index fdf75db8..bf03b638 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -81,6 +81,9 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err if err := r.setJSON(ctx, realtimeRawKey(vehicleKey, env.Protocol), protocolSnapshot.Parsed, r.cfg.ttl()); err != nil { return err } + if err := r.setKV(ctx, vin, env, protocolSnapshot.Parsed); err != nil { + return err + } protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol) if err != nil { @@ -180,6 +183,30 @@ func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl tim return r.client.Set(ctx, key, payload, ttl).Err() } +func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEnvelope, parsed map[string]any) error { + rows := realtimeKVFields(env, parsed) + for _, row := range rows { + key := realtimeKVKey(row.Protocol, vin, row.Domain) + values := map[string]any{ + row.Field: row.Value, + "_event_time_ms": strconv.FormatInt(row.EventTimeMS, 10), + "_received_at_ms": strconv.FormatInt(row.ReceivedAtMS, 10), + "_event_id": row.EventID, + "_protocol": string(row.Protocol), + "_vin": vin, + "_domain": row.Domain, + "_value_type:" + row.Field: row.ValueType, + } + if err := r.client.HSet(ctx, key, values).Err(); err != nil { + return err + } + if err := r.client.Expire(ctx, key, r.cfg.ttl()).Err(); err != nil { + return err + } + } + return nil +} + func (r *Repository) addProtocol(ctx context.Context, vehicleKey string, protocol envelope.Protocol) ([]envelope.Protocol, error) { key := protocolsKey(vehicleKey) if err := r.client.SAdd(ctx, key, string(protocol)).Err(); err != nil { @@ -480,6 +507,10 @@ func realtimeRawKey(vehicleKey string, protocol envelope.Protocol) string { return "vehicle:realtime-raw:" + string(protocol) + ":" + strings.TrimSpace(vehicleKey) } +func realtimeKVKey(protocol envelope.Protocol, vin string, domain string) string { + return "vehicle:rt-kv:" + string(protocol) + ":" + strings.TrimSpace(vin) + ":" + strings.TrimSpace(domain) +} + func onlineKey(vehicleKey string) string { return "vehicle:online:" + strings.TrimSpace(vehicleKey) } diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 2e5ccf33..f3e65cf5 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -271,6 +271,48 @@ func TestRepositoryCollapsesGB32960VendorStackFragmentsIntoSingleRealtimeSummary } } +func TestRepositoryWritesRealtimeKVHashes(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + if err := repo.Update(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}, + map[string]any{"type": "0x30", "name": "gd_fc_stack", "value": map[string]any{ + "stack_count": 1, + "summaries": []any{map[string]any{"stack_water_outlet_temp_c": 63}}, + }}, + }, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + vehicleKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:vehicle").Result() + if err != nil { + t.Fatalf("vehicle kv HGetAll error = %v", err) + } + if vehicleKV["soc_percent"] != "88" || vehicleKV["_event_id"] == "" || vehicleKV["_event_time_ms"] != "1000" { + t.Fatalf("vehicle kv = %#v", vehicleKV) + } + stackKV, err := repo.client.HGetAll(ctx, "vehicle:rt-kv:GB32960:VIN001:gd_fc_stack").Result() + if err != nil { + t.Fatalf("stack kv HGetAll error = %v", err) + } + if stackKV["stack_water_outlet_temp_c"] != "63" { + t.Fatalf("stack kv = %#v", stackKV) + } + if ttl := repo.client.TTL(ctx, "vehicle:rt-kv:GB32960:VIN001:vehicle").Val(); ttl <= 0 { + t.Fatalf("kv ttl should be set, got %v", ttl) + } +} + func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 41c70ec2..682eedfe 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -111,6 +111,9 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { return err } + if _, err := w.exec.ExecContext(ctx, realtimeKVTableSQL); err != nil { + return err + } return nil } @@ -129,7 +132,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) platformName := platformNameFromEnvelope(env) - parsedJSON, err := w.parsedJSONForEnvelope(ctx, env, vin) + parsed, err := w.parsedForEnvelope(ctx, env, vin) if err != nil { return err } @@ -139,13 +142,16 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) plate, platformName, env.SourceEndpoint, - parsedJSON, + marshalParsedJSON(parsed), eventTime, receivedAt, env.StableEventID(), ); err != nil { return err } + if err := w.upsertKV(ctx, env, parsed); err != nil { + return err + } location, ok := realtimeLocationFromEnvelope(env, vin, plate) if !ok { return nil @@ -170,7 +176,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) return err } -func (w *SnapshotWriter) parsedJSONForEnvelope(ctx context.Context, env envelope.FrameEnvelope, vin string) (any, error) { +func (w *SnapshotWriter) parsedForEnvelope(ctx context.Context, env envelope.FrameEnvelope, vin string) (map[string]any, error) { if len(env.Parsed) == 0 { return nil, nil } @@ -184,7 +190,32 @@ func (w *SnapshotWriter) parsedJSONForEnvelope(ctx context.Context, env envelope parsed = mergeParsedForProtocol(env.Protocol, existing, env.Parsed) } } - return marshalParsedJSON(parsed), nil + return parsed, nil +} + +func (w *SnapshotWriter) upsertKV(ctx context.Context, env envelope.FrameEnvelope, parsed map[string]any) error { + rows := realtimeKVFields(env, parsed) + if len(rows) == 0 { + return nil + } + placeholders := make([]string, 0, len(rows)) + args := make([]any, 0, len(rows)*9) + for _, row := range rows { + placeholders = append(placeholders, "(?, ?, ?, ?, ?, ?, ?, ?, ?)") + args = append(args, + string(row.Protocol), + row.VIN, + row.Domain, + row.Field, + row.Value, + row.ValueType, + nullableTime(row.EventTimeMS), + nullableTime(row.ReceivedAtMS), + row.EventID, + ) + } + _, err := w.exec.ExecContext(ctx, upsertRealtimeKVSQLPrefix+strings.Join(placeholders, ",")+upsertRealtimeKVSQLSuffix, args...) + return err } func platformNameFromEnvelope(env envelope.FrameEnvelope) string { @@ -499,6 +530,37 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo KEY idx_protocol_updated (protocol, updated_at) )` +const realtimeKVTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_kv ( + protocol VARCHAR(32) NOT NULL, + vin VARCHAR(32) NOT NULL, + domain_name VARCHAR(64) NOT NULL, + field_name VARCHAR(160) NOT NULL, + field_value TEXT NULL, + value_type VARCHAR(16) NOT NULL DEFAULT '', + event_time DATETIME(3) NULL, + received_at DATETIME(3) NULL, + event_id VARCHAR(64) NOT NULL DEFAULT '', + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (protocol, vin, domain_name, field_name), + KEY idx_vin_domain (vin, domain_name), + KEY idx_protocol_domain (protocol, domain_name) +)` + +const upsertRealtimeKVSQLPrefix = ` +INSERT INTO vehicle_realtime_kv + (protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id) +VALUES ` + +const upsertRealtimeKVSQLSuffix = ` +ON DUPLICATE KEY UPDATE + field_value = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(field_value), field_value), + value_type = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(value_type), value_type), + event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(event_time), event_time), + received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(received_at), received_at), + event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(event_id), event_id), + updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), CURRENT_TIMESTAMP, updated_at) +` + const upsertRealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location (protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 86e0fcd5..3a898c87 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -42,8 +42,8 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 6 { - t.Fatalf("exec calls = %d, want 6", len(exec.calls)) + if len(exec.calls) != 8 { + t.Fatalf("exec calls = %d, want 8", len(exec.calls)) } if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") { t.Fatalf("schema query = %s", exec.calls[0].query) @@ -51,6 +51,9 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if !strings.Contains(exec.calls[4].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { t.Fatalf("location schema query = %s", exec.calls[4].query) } + if !strings.Contains(exec.calls[5].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_kv") { + t.Fatalf("kv schema query = %s", exec.calls[5].query) + } for _, call := range []snapshotExecCall{exec.calls[0], exec.calls[4]} { if strings.Contains(call.query, "fields_json") { t.Fatalf("realtime schema should not contain fields_json: %s", call.query) @@ -72,7 +75,7 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if strings.Contains(exec.calls[4].query, "idx_location") { t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[4].query) } - upsert := exec.calls[5] + upsert := exec.calls[6] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } @@ -99,6 +102,10 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if len(upsert.args) != 9 { t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args)) } + kvUpsert := exec.calls[7] + if !strings.Contains(kvUpsert.query, "INSERT INTO vehicle_realtime_kv") || !strings.Contains(kvUpsert.query, "ON DUPLICATE KEY UPDATE") { + t.Fatalf("kv upsert query = %s", kvUpsert.query) + } } func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { @@ -119,6 +126,8 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location"). WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_kv"). + WillReturnResult(sqlmock.NewResult(0, 0)) if err := writer.EnsureSchema(context.Background()); err != nil { t.Fatalf("EnsureSchema() error = %v", err) @@ -160,10 +169,10 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", len(exec.calls)) } - locationUpsert := exec.calls[1] + locationUpsert := exec.calls[2] if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") { t.Fatalf("location upsert query = %s", locationUpsert.query) } @@ -226,6 +235,8 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("INSERT INTO vehicle_realtime_kv"). + WillReturnResult(sqlmock.NewResult(0, 2)) err = writer.Update(context.Background(), envelope.FrameEnvelope{ Protocol: envelope.ProtocolGB32960, @@ -272,6 +283,8 @@ func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *te sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) + mock.ExpectExec("INSERT INTO vehicle_realtime_kv"). + WillReturnResult(sqlmock.NewResult(0, 3)) err = writer.Update(context.Background(), envelope.FrameEnvelope{ Protocol: envelope.ProtocolGB32960, @@ -366,13 +379,13 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { if resolver.vin != "VIN001" { t.Fatalf("resolver vin = %q", resolver.vin) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", len(exec.calls)) } if got, want := exec.calls[0].args[2], "沪A12345"; got != want { t.Fatalf("snapshot plate arg = %#v, want %q", got, want) } - if got, want := exec.calls[1].args[2], "沪A12345"; got != want { + if got, want := exec.calls[2].args[2], "沪A12345"; got != want { t.Fatalf("location plate arg = %#v, want %q", got, want) } } @@ -428,10 +441,10 @@ func TestSnapshotWriterCachesBindingPlateByVIN(t *testing.T) { if resolver.calls != 1 { t.Fatalf("plate resolver calls = %d, want 1", resolver.calls) } - if len(exec.calls) != 4 { - t.Fatalf("exec calls = %d, want 4", len(exec.calls)) + if len(exec.calls) != 6 { + t.Fatalf("exec calls = %d, want 6", len(exec.calls)) } - for _, index := range []int{0, 2} { + for _, index := range []int{0, 3} { if got, want := exec.calls[index].args[2], "沪A12345"; got != want { t.Fatalf("snapshot %d plate arg = %#v, want %q", index, got, want) }