From 378a5d5e5704e7731ed8ef4835bf7bc0e3904323 Mon Sep 17 00:00:00 2001 From: lingniu Date: Fri, 3 Jul 2026 17:38:50 +0800 Subject: [PATCH] feat(go): reuse parsed fields across projections --- docs/ops/go-vehicle-ingest-memory.md | 341 ++++++++++++++++++ .../internal/envelope/envelope.go | 40 +- .../internal/gateway/mqtt_client.go | 3 + .../internal/gateway/tcp_server.go | 3 + go/vehicle-gateway/internal/history/writer.go | 6 +- .../internal/history/writer_test.go | 28 +- go/vehicle-gateway/internal/realtime/kv.go | 127 +++++-- .../internal/realtime/repository.go | 43 ++- .../internal/realtime/repository_test.go | 31 ++ .../internal/realtime/snapshot_writer.go | 11 +- .../internal/realtime/snapshot_writer_test.go | 51 +++ 11 files changed, 620 insertions(+), 64 deletions(-) create mode 100644 docs/ops/go-vehicle-ingest-memory.md diff --git a/docs/ops/go-vehicle-ingest-memory.md b/docs/ops/go-vehicle-ingest-memory.md new file mode 100644 index 00000000..a9335085 --- /dev/null +++ b/docs/ops/go-vehicle-ingest-memory.md @@ -0,0 +1,341 @@ +# Go 车辆接入上下文记忆 + +更新时间：2026-07-03 + +这份文档用于在 Codex 上下文重载后快速恢复项目状态。更完整的运行排查步骤见： + +- `docs/ops/vehicle-ingest-runbook.md` +- `docs/architecture/production-data-plane-inventory.md` + +## 当前目标 + +Go 版本车辆数据接入链路已经作为生产主链路运行在 ECS `115.29.187.205`。 + +核心目标： + +1. 高性能可靠接收 GB32960、JT808、宇通 MQTT。 +2. 协议解析后先写 NATS，再桥接 Kafka。 +3. Kafka raw 供 TDengine 历史、Redis 实时、MySQL 当前态/统计消费。 +4. 尽量避免 Java 老链路、Xinda 相关链路和重复存储。 +5. 解析字段只投影一次，避免 TDengine/Redis/MySQL 各自重复 flatten。 + +## 服务和端口 + +ECS：`115.29.187.205` + +| 服务 | systemd | 端口 | 作用 | +| --- | --- | --- | --- | +| Gateway | `lingniu-go-gateway.service` | `808`、`32960`、`20211` | JT808、GB32960、宇通 MQTT 接入和解析 | +| NATS fast writer | `lingniu-go-nats-fast-writer.service` | 健康端口随 env | NATS 快速写 TDengine/Redis | +| NATS Kafka bridge | `lingniu-go-nats-kafka-bridge.service` | `20214` | NATS 到 Kafka | +| History writer | `lingniu-go-history-writer.service` | `20212` | Kafka raw 写 TDengine | +| Stat writer | `lingniu-go-stat-writer.service` | `20213` | Kafka raw 写每日里程 | +| Realtime API | `lingniu-go-realtime-api.service` | `20200` | Redis/MySQL 投影和 HTTP API | + +业务端口： + +- JT808：`0.0.0.0:808` +- GB32960：`0.0.0.0:32960` +- Swagger/API：`http://115.29.187.205:20200/swagger-ui/index.html` + +## 部署路径 + +Go 服务使用 systemd 裸机部署，不使用 Docker。 + +```bash +/opt/lingniu-go-native/current +/opt/lingniu-go-native/releases/* +/opt/lingniu-go-native/env/*.env +``` + +最近 release： + +```text +/opt/lingniu-go-native/releases/parsed-fields-once-20260703163821 +``` + +服务重启： + +```bash +systemctl restart lingniu-go-gateway.service +systemctl restart lingniu-go-nats-fast-writer.service +systemctl restart lingniu-go-nats-kafka-bridge.service +systemctl restart lingniu-go-history-writer.service +systemctl restart lingniu-go-stat-writer.service +systemctl restart lingniu-go-realtime-api.service +``` + +健康检查： + +```bash +for port in 20211 20212 20213 20214 20200; do + curl -fsS "http://127.0.0.1:${port}/readyz" + echo +done +``` + +## 中间件 + +Kafka / NATS ECS： + +- 公网：`114.55.58.251` +- 内网：`172.17.111.56` +- NATS：`172.17.111.56:4222` + +TDengine： + +- 内网：`172.17.111.57` +- WS：`172.17.111.57:6041` +- 数据库：`lingniu_vehicle_ts` + +MySQL RDS： + +- 内网：`rm-bp179zbv481rnw3e2.mysql.rds.aliyuncs.com:3306` +- 数据库：`lingniu_vehicle_data` + +Redis： + +- 内网：`r-bp1u741kij7e51i481.redis.rds.aliyuncs.com:6379` +- DB：`50` + +不要在最终回答中明文输出生产密码。 + +## Topic + +当前 Go raw topic： + +```text +vehicle.raw.go.gb32960.v1 +vehicle.raw.go.jt808.v1 +vehicle.raw.go.yutong-mqtt.v1 +``` + +fields topic 已存在： + +```text +vehicle.fields.go.gb32960.v1 +vehicle.fields.go.jt808.v1 +vehicle.fields.go.yutong-mqtt.v1 +``` + +当前核心链路仍以 raw topic 作为事实输入；raw envelope 里已经包含一次性预计算的 `parsed_fields`。 + +## 代码结构 + +主要目录： + +```text +go/vehicle-gateway/cmd/gateway +go/vehicle-gateway/cmd/history-writer +go/vehicle-gateway/cmd/realtime-api +go/vehicle-gateway/cmd/stat-writer +go/vehicle-gateway/cmd/nats-fast-writer +go/vehicle-gateway/cmd/nats-kafka-bridge +go/vehicle-gateway/internal/protocol/gb32960 +go/vehicle-gateway/internal/protocol/jt808 +go/vehicle-gateway/internal/protocol/yutongmqtt +go/vehicle-gateway/internal/realtime +go/vehicle-gateway/internal/history +go/vehicle-gateway/internal/identity +``` + +每次改代码后在本地跑： + +```bash +cd go/vehicle-gateway +go test ./... +``` + +## 解析和存储原则 + +### 一次解析字段 + +2026-07-03 已完成优化： + +- `FrameEnvelope` 增加： + - `parsed_fields` + - `parsed_field_types` +- Gateway 在发布 raw 前调用 `realtime.EnsureParsedFields(&env)`。 +- TDengine writer、Redis realtime KV、MySQL snapshot 优先使用 `env.ParsedFields`。 +- 旧消息没有 `parsed_fields` 时，保留 fallback，从 `Parsed` 展开一次。 + +相关文件： + +```text +go/vehicle-gateway/internal/envelope/envelope.go +go/vehicle-gateway/internal/realtime/kv.go +go/vehicle-gateway/internal/gateway/tcp_server.go +go/vehicle-gateway/internal/gateway/mqtt_client.go +go/vehicle-gateway/internal/history/writer.go +go/vehicle-gateway/internal/realtime/repository.go +go/vehicle-gateway/internal/realtime/snapshot_writer.go +``` + +### GB32960 多帧 + +GB32960 有两类实际模式： + +1. 单条 `0x02` 同时包含标准 data unit 和 vendor data unit。 +2. 标准帧加 vendor 栈分片帧。 + +例子： + +- `LB9A32A24R0LS1426`：单帧完整混合上报，单帧约 `774 bytes`，`gd_fc_stack.cell_count=108`。 +- `LNXNEGRR7SR318212`：多帧上报，`gd_fc_stack.cell_count=432`，按 `200 + 200 + 32` 分片。 + +TDengine raw_frames 必须保持单帧事实，不跨帧合并。 + +Redis/MySQL 实时视图可以按 `protocol + vin` 增量合并，但不能让后来的 stack-only 帧覆盖标准车辆字段。 + +### JT808 + +JT808 主入口端口是 `808`，不再使用 `8089`。 + +解析重点： + +- 包头手机号按 BCD 去前导 0。 +- `0x0100` 注册帧写 `jt808_registration`。 +- `0x0102` 鉴权帧尝试按 phone 查 registration/binding 关联 VIN。 +- `0x0200` 位置帧即使没有注册/鉴权，也会按 phone 降级查 `vehicle_identity_binding`，建立 session。 +- 位置附加信息 `0x01` 是 GPS 总里程，按差值用于每日里程。 + +## MySQL 身份表 + +`vehicle_identity_binding` 是外部维护事实表，服务只读，不允许运行时写入。 + +核心字段： + +```text +vin +plate +phone +oem +updated_at +``` + +用途： + +- JT808 用 phone 或 plate 找 VIN。 +- Realtime snapshot/location 用 VIN 反查 plate。 +- 后续导入 G7s、广安车联等来源时更新 phone/oem。 + +`jt808_registration` 是 808 注册鉴权运行状态表，以 phone 为主键，记录注册、鉴权、source_endpoint、关联 VIN 状态。 + +## 2026-07-03 数据维护记录 + +### G7s binding + +源文件： + +```text +/Users/lingniu/Library/Mobile Documents/com~apple~CloudDocs/rsync/2026_07_02_17_11_37转发配置列表-1(2).xlsx +``` + +处理规则： + +1. 删除原有 G7/G7s 的 `phone/oem`。 +2. 使用文件中的 `手机号` 列更新。 +3. 冲突的不更新，其他更新。 + +执行结果： + +- 清空旧 G7/G7s：`666` 行。 +- 更新安全 G7s：`561` 行。 +- 跳过：`72` 行。 + - 缺 plate 或 VIN：`14` + - 与非 G7/G7s 绑定冲突：`58` + +跳过清单： + +```text +outputs/g7s_binding_skipped_apply.csv +``` + +### JT808 unknown VIN 导出 + +导出条件： + +- `jt808_registration.vin` 为空、`unknown` 或 `unknow` +- 且 `phone` 不存在于 `vehicle_identity_binding.phone` + +结果： + +- `30` 个 phone + +导出文件： + +```text +outputs/jt808_unknown_vin_phone_not_in_binding.csv +``` + +## 常用查询 + +历史 raw frame： + +```bash +curl -s 'http://115.29.187.205:20200/api/history/raw-frames?protocol=GB32960&vin=LB9A32A24R0LS1426&limit=1&includeFields=true' +``` + +只取指定 parsed fields： + +```bash +curl -s 'http://115.29.187.205:20200/api/history/raw-frames?protocol=GB32960&vin=LB9A32A24R0LS1426&limit=1&includeFields=true&fields=gb32960.vehicle.speed_kmh&fields=gb32960.gd_fc_stack.stack_water_outlet_temp_c' +``` + +808 源 IP： + +```bash +journalctl -u lingniu-go-gateway.service --since '10 minutes ago' --no-pager +ss -tn sport = :808 +``` + +最近已观察到 808 源 IP： + +```text +115.159.85.149 +222.66.200.68 +122.152.221.156 +115.231.168.135 +``` + +## 部署步骤 + +本地构建： + +```bash +cd go/vehicle-gateway +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/gateway ./cmd/gateway +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/history-writer ./cmd/history-writer +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/realtime-api ./cmd/realtime-api +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/stat-writer ./cmd/stat-writer +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/nats-fast-writer ./cmd/nats-fast-writer +GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /tmp/lingniu-go-deploy/nats-kafka-bridge ./cmd/nats-kafka-bridge +``` + +上传部署： + +```bash +release="name-$(date +%Y%m%d%H%M%S)" +ssh root@115.29.187.205 "mkdir -p /opt/lingniu-go-native/releases/$release" +scp /tmp/lingniu-go-deploy/* root@115.29.187.205:/opt/lingniu-go-native/releases/$release/ +ssh root@115.29.187.205 "chmod 755 /opt/lingniu-go-native/releases/$release/* && ln -sfn /opt/lingniu-go-native/releases/$release /opt/lingniu-go-native/current" +``` + +部署后验证： + +```bash +ssh root@115.29.187.205 ' +systemctl is-active lingniu-go-gateway.service lingniu-go-nats-fast-writer.service lingniu-go-nats-kafka-bridge.service lingniu-go-history-writer.service lingniu-go-stat-writer.service lingniu-go-realtime-api.service +for port in 20211 20212 20213 20214 20200; do curl -fsS "http://127.0.0.1:${port}/readyz"; echo; done +journalctl -u lingniu-go-gateway.service --since "2 minutes ago" --no-pager | grep -Ei "error|failed|panic|fatal" || true +' +``` + +## 下次恢复上下文时先做 + +1. 读本文件。 +2. 读 `docs/ops/vehicle-ingest-runbook.md`。 +3. 读 `docs/architecture/production-data-plane-inventory.md`。 +4. 执行 `git status --short`。 +5. 如果涉及生产，先查 ECS 服务健康和最近日志。 + diff --git a/go/vehicle-gateway/internal/envelope/envelope.go b/go/vehicle-gateway/internal/envelope/envelope.go index 7d402198..6fb4ad33 100644 --- a/go/vehicle-gateway/internal/envelope/envelope.go +++ b/go/vehicle-gateway/internal/envelope/envelope.go @@ -33,25 +33,27 @@ const ( ) type FrameEnvelope struct { - EventID string `json:"event_id"` - TraceID string `json:"trace_id"` - Protocol Protocol `json:"protocol"` - MessageID string `json:"message_id"` - Sequence uint16 `json:"sequence"` - VIN string `json:"vin,omitempty"` - VehicleKeyHint string `json:"vehicle_key,omitempty"` - Phone string `json:"phone,omitempty"` - DeviceID string `json:"device_id,omitempty"` - Plate string `json:"plate,omitempty"` - SourceEndpoint string `json:"source_endpoint,omitempty"` - EventTimeMS int64 `json:"event_time_ms"` - ReceivedAtMS int64 `json:"received_at_ms"` - RawHex string `json:"raw_hex,omitempty"` - RawText string `json:"raw_text,omitempty"` - Parsed map[string]any `json:"parsed,omitempty"` - Fields map[string]any `json:"fields,omitempty"` - ParseStatus ParseStatus `json:"parse_status"` - ParseError string `json:"parse_error,omitempty"` + EventID string `json:"event_id"` + TraceID string `json:"trace_id"` + Protocol Protocol `json:"protocol"` + MessageID string `json:"message_id"` + Sequence uint16 `json:"sequence"` + VIN string `json:"vin,omitempty"` + VehicleKeyHint string `json:"vehicle_key,omitempty"` + Phone string `json:"phone,omitempty"` + DeviceID string `json:"device_id,omitempty"` + Plate string `json:"plate,omitempty"` + SourceEndpoint string `json:"source_endpoint,omitempty"` + EventTimeMS int64 `json:"event_time_ms"` + ReceivedAtMS int64 `json:"received_at_ms"` + RawHex string `json:"raw_hex,omitempty"` + RawText string `json:"raw_text,omitempty"` + Parsed map[string]any `json:"parsed,omitempty"` + ParsedFields map[string]any `json:"parsed_fields,omitempty"` + ParsedFieldTypes map[string]string `json:"parsed_field_types,omitempty"` + Fields map[string]any `json:"fields,omitempty"` + ParseStatus ParseStatus `json:"parse_status"` + ParseError string `json:"parse_error,omitempty"` } func (e FrameEnvelope) VehicleKey() string { diff --git a/go/vehicle-gateway/internal/gateway/mqtt_client.go b/go/vehicle-gateway/internal/gateway/mqtt_client.go index 84621cd0..81a6fa74 100644 --- a/go/vehicle-gateway/internal/gateway/mqtt_client.go +++ b/go/vehicle-gateway/internal/gateway/mqtt_client.go @@ -222,6 +222,9 @@ func (c *MQTTClient) handleMessage(ctx context.Context, topic string, payload [] } frameStatus = env.ParseStatus c.recordFrameMetric(env.ParseStatus) + if env.ParseStatus != envelope.ParseBadFrame { + realtime.EnsureParsedFields(&env) + } if err := c.cfg.Sink.PublishRaw(messageCtx, env); err != nil { c.recordPublishMetric("raw", "error") c.cfg.Logger.Error("publish mqtt raw failed", "topic", topic, "event_id", env.StableEventID(), "error", err) diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 58056670..8177a765 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -237,6 +237,9 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } frameStatus = env.ParseStatus s.recordFrameMetric(env.ParseStatus) + if env.ParseStatus != envelope.ParseBadFrame { + realtime.EnsureParsedFields(&env) + } if err := s.sink.PublishRaw(frameCtx, env); err != nil { s.recordPublishMetric("raw", "error") diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index cd87ef1c..aba25e25 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -318,11 +318,11 @@ func jsonString(value any) string { } func parsedFieldsJSONString(env envelope.FrameEnvelope) string { - fieldsEnv, ok := realtime.BuildFieldsEnvelope(env) - if !ok || len(fieldsEnv.Fields) == 0 { + fields, _, ok := realtime.ParsedFieldsForEnvelope(env) + if !ok || len(fields) == 0 { return "" } - return jsonString(fieldsEnv.Fields) + return jsonString(fields) } func floatField(env envelope.FrameEnvelope, key string) (float64, bool) { diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index 7d9cfd32..535f72c0 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -111,8 +111,8 @@ func TestWriterChunksOversizedParsedFields(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec) env := sampleEnvelope() - env.Parsed = map[string]any{ - "large_payload": strings.Repeat("x", 16_128), + env.ParsedFields = map[string]any{ + "jt808.location.large_payload": strings.Repeat("x", 16_128), } if err := writer.AppendRawFrame(context.Background(), env); err != nil { @@ -131,6 +131,30 @@ func TestWriterChunksOversizedParsedFields(t *testing.T) { } } +func TestWriterUsesEnvelopeParsedFieldsWithoutReflattening(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec) + env := sampleEnvelope() + env.Parsed = map[string]any{ + "location": map[string]any{"speed_kmh": 99}, + } + env.ParsedFields = map[string]any{ + "jt808.location.speed_kmh": "12.3", + } + + if err := writer.AppendRawFrame(context.Background(), env); err != nil { + t.Fatalf("AppendRawFrame() error = %v", err) + } + + rawInsert := findSQL(exec.calls, "INSERT INTO raw_") + if !strings.Contains(rawInsert, `"jt808.location.speed_kmh":"12.3"`) { + t.Fatalf("raw insert should use precomputed parsed fields: %s", rawInsert) + } + if strings.Contains(rawInsert, `"jt808.location.speed_kmh":"99"`) { + t.Fatalf("raw insert should not re-flatten parsed payload: %s", rawInsert) + } +} + func TestWriterLeavesParsedFieldsEmptyWhenNoParsedPayload(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec) diff --git a/go/vehicle-gateway/internal/realtime/kv.go b/go/vehicle-gateway/internal/realtime/kv.go index 1cbc13ef..5325dcc1 100644 --- a/go/vehicle-gateway/internal/realtime/kv.go +++ b/go/vehicle-gateway/internal/realtime/kv.go @@ -26,33 +26,27 @@ func BuildFieldsEnvelope(env envelope.FrameEnvelope) (envelope.FrameEnvelope, bo if !envelope.IsRealtimeTelemetryFrame(env) { return envelope.FrameEnvelope{}, false } - rows := realtimeKVFields(env, env.Parsed) - fields := make(map[string]any, len(rows)) - for _, row := range rows { - key := row.Domain - if strings.TrimSpace(row.Field) != "" { - key += "." + row.Field - } - fields[key] = row.Value - } - if len(fields) == 0 { + fields, fieldTypes, ok := ParsedFieldsForEnvelope(env) + if !ok { return envelope.FrameEnvelope{}, false } out := envelope.FrameEnvelope{ - EventID: env.StableEventID() + ":fields", - TraceID: env.TraceID, - Protocol: env.Protocol, - MessageID: env.MessageID, - Sequence: env.Sequence, - VIN: env.VIN, - VehicleKeyHint: env.VehicleKeyHint, - Phone: env.Phone, - DeviceID: env.DeviceID, - Plate: env.Plate, - SourceEndpoint: env.SourceEndpoint, - EventTimeMS: env.EventTimeMS, - ReceivedAtMS: env.ReceivedAtMS, - Fields: fields, + EventID: env.StableEventID() + ":fields", + TraceID: env.TraceID, + Protocol: env.Protocol, + MessageID: env.MessageID, + Sequence: env.Sequence, + VIN: env.VIN, + VehicleKeyHint: env.VehicleKeyHint, + Phone: env.Phone, + DeviceID: env.DeviceID, + Plate: env.Plate, + SourceEndpoint: env.SourceEndpoint, + EventTimeMS: env.EventTimeMS, + ReceivedAtMS: env.ReceivedAtMS, + Fields: fields, + ParsedFields: fields, + ParsedFieldTypes: fieldTypes, Parsed: map[string]any{ "source_event_id": env.StableEventID(), "field_mapping": realtimeFieldMappingVersion, @@ -62,6 +56,91 @@ func BuildFieldsEnvelope(env envelope.FrameEnvelope) (envelope.FrameEnvelope, bo return out, true } +func EnsureParsedFields(env *envelope.FrameEnvelope) bool { + if env == nil { + return false + } + if len(env.ParsedFields) > 0 { + if env.ParsedFieldTypes == nil { + env.ParsedFieldTypes = inferParsedFieldTypes(env.ParsedFields) + } + return true + } + fields, fieldTypes, ok := ParsedFieldsForEnvelope(*env) + if !ok { + return false + } + env.ParsedFields = fields + env.ParsedFieldTypes = fieldTypes + return true +} + +func ParsedFieldsForEnvelope(env envelope.FrameEnvelope) (map[string]any, map[string]string, bool) { + if len(env.ParsedFields) > 0 { + fields := cloneAnyMap(env.ParsedFields) + fieldTypes := cloneStringMap(env.ParsedFieldTypes) + if len(fieldTypes) == 0 { + fieldTypes = inferParsedFieldTypes(fields) + } + return fields, fieldTypes, true + } + rows := realtimeKVFields(env, env.Parsed) + if len(rows) == 0 { + return nil, nil, false + } + fields := make(map[string]any, len(rows)) + fieldTypes := make(map[string]string, len(rows)) + for _, row := range rows { + key := realtimeKVFieldPath(row.Domain, row.Field) + if key == "" { + continue + } + fields[key] = row.Value + fieldTypes[key] = row.ValueType + } + if len(fields) == 0 { + return nil, nil, false + } + return fields, fieldTypes, true +} + +func inferParsedFieldTypes(fields map[string]any) map[string]string { + if len(fields) == 0 { + return nil + } + out := make(map[string]string, len(fields)) + for key, value := range fields { + _, valueType, ok := stringifyKVValue(value) + if !ok { + continue + } + out[key] = valueType + } + return out +} + +func cloneAnyMap(in map[string]any) map[string]any { + if len(in) == 0 { + return nil + } + out := make(map[string]any, len(in)) + for key, value := range in { + out[key] = value + } + return out +} + +func cloneStringMap(in map[string]string) map[string]string { + if len(in) == 0 { + return nil + } + out := make(map[string]string, len(in)) + for key, value := range in { + out[key] = value + } + return out +} + type protocolFieldMapping struct { Prefix string GBDataUnits bool diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index 2f8f4f9c..f1d2ca73 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -348,20 +348,10 @@ func (r *Repository) setJSON(ctx context.Context, key string, value any, ttl tim } func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEnvelope, parsed map[string]any) error { - rows := realtimeKVFields(env, parsed) - if len(rows) == 0 { - return nil - } - values := make(map[string]any, len(rows)) - types := make(map[string]any, len(rows)) - for _, row := range rows { - field := realtimeKVFieldPath(row.Domain, row.Field) - if field == "" { - continue - } - values[field] = row.Value - types[field] = row.ValueType + if len(env.ParsedFields) == 0 && len(parsed) > 0 { + env.Parsed = parsed } + values, types := realtimeKVMapsForEnvelope(env) if len(values) == 0 { return nil } @@ -383,8 +373,7 @@ func (r *Repository) setKV(ctx context.Context, vin string, env envelope.FrameEn } func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, vin string, env envelope.FrameEnvelope) error { - rows := realtimeKVFields(env, env.Parsed) - values, types := realtimeKVMaps(rows) + values, types := realtimeKVMapsForEnvelope(env) eventTimeMS := eventTimeOrReceivedMS(env) offlineAfterMS := env.ReceivedAtMS + r.cfg.ttl().Milliseconds() online := OnlineStatus{ @@ -435,6 +424,30 @@ func (r *Repository) setFastProjection(ctx context.Context, vehicleKey string, v return err } +func realtimeKVMapsForEnvelope(env envelope.FrameEnvelope) (map[string]any, map[string]any) { + if fields, fieldTypes, ok := ParsedFieldsForEnvelope(env); ok { + values := make(map[string]any, len(fields)) + types := make(map[string]any, len(fields)) + for field, value := range fields { + if strings.TrimSpace(field) == "" { + continue + } + stringValue, valueType, ok := stringifyKVValue(value) + if !ok { + continue + } + values[field] = stringValue + if typed := strings.TrimSpace(fieldTypes[field]); typed != "" { + types[field] = typed + } else { + types[field] = valueType + } + } + return values, types + } + return nil, nil +} + func realtimeKVMaps(rows []RealtimeKVField) (map[string]any, map[string]any) { values := make(map[string]any, len(rows)) types := make(map[string]any, len(rows)) diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index 1f3a16cd..0822d738 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -646,6 +646,37 @@ func TestRepositoryOnlineStatus(t *testing.T) { } } +func TestRepositoryUsesEnvelopeParsedFieldsForRealtimeKV(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + ctx := context.Background() + + if err := repo.Update(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + VIN: "VIN001", + EventTimeMS: 1000, + ReceivedAtMS: 1100, + MessageID: "0x0200", + Parsed: map[string]any{ + "location": map[string]any{"speed_kmh": 99}, + }, + ParsedFields: map[string]any{ + "jt808.location.speed_kmh": "12.3", + }, + Fields: map[string]any{envelope.FieldSpeedKMH: 12.3}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + values, err := repo.client.HGetAll(ctx, realtimeKVValuesKey(envelope.ProtocolJT808, "VIN001")).Result() + if err != nil { + t.Fatalf("HGetAll() error = %v", err) + } + if got, want := values["jt808.location.speed_kmh"], "12.3"; got != want { + t.Fatalf("speed kv = %#v, want %q; all=%#v", got, want, values) + } +} + func TestRepositorySkipsFramesWithoutVIN(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 27e1dcb5..41299afa 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -187,7 +187,10 @@ func (w *SnapshotWriter) snapshotFieldsForEnvelope(ctx context.Context, env enve if len(existing) > 0 { if isStructuredSnapshotParsed(env.Protocol, existing) { parsed = mergeParsedForProtocol(env.Protocol, existing, env.Parsed) - return realtimeSnapshotFlatFields(env, parsed), nil + mergedEnv := env + mergedEnv.ParsedFields = nil + mergedEnv.ParsedFieldTypes = nil + return realtimeSnapshotFlatFields(mergedEnv, parsed), nil } return mergeRealtimeSnapshotFields(existing, incoming), nil } @@ -196,6 +199,12 @@ func (w *SnapshotWriter) snapshotFieldsForEnvelope(ctx context.Context, env enve } func realtimeSnapshotFlatFields(env envelope.FrameEnvelope, parsed map[string]any) map[string]any { + if len(env.ParsedFields) > 0 { + fields := cloneMap(env.ParsedFields) + if len(fields) > 0 { + return fields + } + } rows := realtimeKVFields(env, parsed) if len(rows) == 0 { return nil diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 05d413f2..8e8f1bbb 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -257,6 +257,57 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi } } +func TestSnapshotWriterUsesEnvelopeParsedFieldsWithoutReflattening(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + writer := NewSnapshotWriter(db) + + mock.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?"). + WithArgs("GB32960", "VIN001"). + WillReturnRows(sqlmock.NewRows([]string{"parsed_json"})) + mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot"). + WithArgs( + "GB32960", + "VIN001", + "", + "", + "", + jsonFlatFieldsArg{fields: map[string]string{ + "gb32960.vehicle.speed_kmh": "12.3", + }}, + sqlmock.AnyArg(), + sqlmock.AnyArg(), + sqlmock.AnyArg(), + ). + WillReturnResult(sqlmock.NewResult(0, 1)) + + err = writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"speed_kmh": 99}}, + }, + }, + ParsedFields: map[string]any{ + "gb32960.vehicle.speed_kmh": "12.3", + }, + Fields: map[string]any{envelope.FieldSpeedKMH: 12.3}, + }) + if err != nil { + t.Fatalf("Update() error = %v", err) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *testing.T) { db, mock, err := sqlmock.New() if err != nil {