diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index 66554e0f..13de8817 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -64,6 +64,7 @@ - 索引：不在 `vehicle_realtime_location` 维护经纬度组合索引；实时表只回答当前态，地理范围和轨迹类查询走 TDengine 历史位置。 - 查询：`/api/realtime/snapshots`、`/api/realtime/locations` 默认不执行 `COUNT(*)`，`total` 表示本页返回条数；只有需要精确总数时传 `includeTotal=true`。 - 写入：车牌从 `vehicle_identity_binding` 按 VIN 主键直查，并使用 VIN 级短 TTL 内存缓存，避免实时帧每条都打 MySQL；缓存不作为事实来源。 + - `vehicle_realtime_snapshot.parsed_json` 使用与 Redis KV 同口径的扁平字段对象，例如 `gb32960.vehicle.soc_percent`、`jt808.location.total_mileage_km`。 - 不写 MySQL KV 投影表；实时全量字段保留在 Redis KV，历史全量字段走 TDengine raw fields。 - 生产：项目上线前可直接重建这两张 MySQL 表，实时数据会从 Kafka 新消息继续投影。 diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index a4a82943..27e1dcb5 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -132,7 +132,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) platformName := platformNameFromEnvelope(env) - parsed, err := w.parsedForEnvelope(ctx, env, vin) + parsed, err := w.snapshotFieldsForEnvelope(ctx, env, vin) if err != nil { return err } @@ -173,21 +173,71 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) return err } -func (w *SnapshotWriter) parsedForEnvelope(ctx context.Context, env envelope.FrameEnvelope, vin string) (map[string]any, error) { +func (w *SnapshotWriter) snapshotFieldsForEnvelope(ctx context.Context, env envelope.FrameEnvelope, vin string) (map[string]any, error) { if len(env.Parsed) == 0 { return nil, nil } parsed := cloneMap(env.Parsed) + incoming := realtimeSnapshotFlatFields(env, parsed) if queryer, ok := w.exec.(Queryer); ok { existing, err := realtimeSnapshotParsedJSON(ctx, queryer, env.Protocol, vin) if err != nil { return nil, err } if len(existing) > 0 { - parsed = mergeParsedForProtocol(env.Protocol, existing, env.Parsed) + if isStructuredSnapshotParsed(env.Protocol, existing) { + parsed = mergeParsedForProtocol(env.Protocol, existing, env.Parsed) + return realtimeSnapshotFlatFields(env, parsed), nil + } + return mergeRealtimeSnapshotFields(existing, incoming), nil } } - return parsed, nil + return incoming, nil +} + +func realtimeSnapshotFlatFields(env envelope.FrameEnvelope, parsed map[string]any) map[string]any { + rows := realtimeKVFields(env, parsed) + if len(rows) == 0 { + return nil + } + fields := make(map[string]any, len(rows)) + for _, row := range rows { + key := realtimeKVFieldPath(row.Domain, row.Field) + if key == "" { + continue + } + fields[key] = row.Value + } + return fields +} + +func mergeRealtimeSnapshotFields(existing map[string]any, incoming map[string]any) map[string]any { + if len(existing) == 0 { + return cloneMap(incoming) + } + merged := cloneMap(existing) + for key, value := range incoming { + merged[key] = value + } + return merged +} + +func isStructuredSnapshotParsed(protocol envelope.Protocol, parsed map[string]any) bool { + if len(parsed) == 0 { + return false + } + if protocol == envelope.ProtocolGB32960 { + if _, ok := parsed["data_units"]; ok { + return true + } + } + mapping := realtimeMapping(protocol) + for key := range mapping.TopLevelName { + if _, ok := parsed[key]; ok { + return true + } + } + return false } func platformNameFromEnvelope(env envelope.FrameEnvelope) string { diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 595438f9..05d413f2 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -98,7 +98,7 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if got, want := upsert.args[4], "1.2.3.4:32960"; got != want { t.Fatalf("peer arg = %#v, want %q", got, want) } - if got := upsert.args[5]; !strings.Contains(got.(string), `"data_units"`) { + if got := upsert.args[5]; !strings.Contains(got.(string), `"gb32960.vehicle.soc_percent":"90"`) { t.Fatalf("parsed json arg = %#v", got) } if len(upsert.args) != 9 { @@ -225,7 +225,12 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi "", "", "", - jsonDataUnitsArg{types: []string{"0x01", "0x30"}}, + jsonFlatFieldsArg{fields: map[string]string{ + "gb32960.vehicle.soc_percent": "88", + "gb32960.vehicle.type": "0x01", + "gb32960.gd_fc_stack.stack_count": "1", + "gb32960.gd_fc_stack.type": "0x30", + }}, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), @@ -271,7 +276,17 @@ func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *te "", "", "", - jsonGDFCStackSummaryArg{}, + jsonFlatFieldsArg{ + fields: map[string]string{ + "gb32960.gd_fc_stack.stack_water_outlet_temp_c": "63", + }, + forbidden: []string{ + "gb32960.gd_fc_stack.frame_cell_start", + "gb32960.gd_fc_stack.frame_cell_count", + "gb32960.gd_fc_stack.frame_max_cell_voltage_v", + "gb32960.gd_fc_stack.frame_min_cell_voltage_v", + }, + }, sqlmock.AnyArg(), sqlmock.AnyArg(), sqlmock.AnyArg(), @@ -697,13 +712,12 @@ func (r *recordingPlateResolver) PlateByVIN(_ context.Context, vin string) (stri return r.plate, r.err } -type jsonDataUnitsArg struct { - types []string +type jsonFlatFieldsArg struct { + fields map[string]string + forbidden []string } -type jsonGDFCStackSummaryArg struct{} - -func (jsonGDFCStackSummaryArg) Match(value driver.Value) bool { +func (a jsonFlatFieldsArg) Match(value driver.Value) bool { text, ok := value.(string) if !ok { return false @@ -712,62 +726,13 @@ func (jsonGDFCStackSummaryArg) Match(value driver.Value) bool { if err := json.Unmarshal([]byte(text), &parsed); err != nil { return false } - units, ok := parsed["data_units"].([]any) - if !ok { - return false - } - for _, unit := range units { - unitMap, ok := unit.(map[string]any) - if !ok || unitMap["name"] != "gd_fc_stack" { - continue - } - valueMap, ok := unitMap["value"].(map[string]any) - if !ok { + for field, want := range a.fields { + if got, ok := parsed[field].(string); !ok || got != want { return false } - summaries, ok := valueMap["summaries"].([]any) - if !ok || len(summaries) != 1 { - return false - } - summary, ok := summaries[0].(map[string]any) - if !ok { - return false - } - for _, fragmentOnly := range []string{"frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v"} { - if _, ok := summary[fragmentOnly]; ok { - return false - } - } - return summary["stack_water_outlet_temp_c"] == float64(63) } - return false -} - -func (a jsonDataUnitsArg) Match(value driver.Value) bool { - text, ok := value.(string) - if !ok { - return false - } - var parsed map[string]any - if err := json.Unmarshal([]byte(text), &parsed); err != nil { - return false - } - units, ok := parsed["data_units"].([]any) - if !ok { - return false - } - seen := map[string]bool{} - for _, unit := range units { - unitMap, ok := unit.(map[string]any) - if !ok { - continue - } - if unitType, ok := unitMap["type"].(string); ok { - seen[unitType] = true - } - } - for _, unitType := range a.types { - if !seen[unitType] { + for _, field := range a.forbidden { + if _, ok := parsed[field]; ok { return false } }