feat(go): add realtime kv projection

2026-07-03 11:34:52 +08:00
parent b948a46e64
commit 6fb6262d0d
10 changed files with 653 additions and 17 deletions
--- 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()
--- a/go/vehicle-gateway/internal/realtime/kv.go
+++ 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 != "<nil>" {
+				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
+	}
+}
--- a/go/vehicle-gateway/internal/realtime/kv_test.go
+++ 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
+}
--- 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
--- 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 {
--- 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},
--- 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)
 }
--- 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()
--- 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,
--- 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)
 		}