diff --git a/docs/architecture/production-data-plane-inventory.md b/docs/architecture/production-data-plane-inventory.md index 0a675304..b1c0d01a 100644 --- a/docs/architecture/production-data-plane-inventory.md +++ b/docs/architecture/production-data-plane-inventory.md @@ -82,7 +82,7 @@ Redis 使用 DB 50,定位为实时缓存,不作为历史事实来源。 | Key 族 | 用途 | | --- | --- | | `vehicle:latest:{vehicleKey}` | 跨协议合并后的最新核心字段快照,不重复保存完整 parsed | -| `vehicle:latest:{vehicleKey}:{protocol}` | 单协议最新快照 | +| `vehicle:latest:{vehicleKey}:{protocol}` | 单协议最新轻量快照,保留核心 fields 和时间信息,不重复保存完整 parsed | | `vehicle:realtime-raw:{protocol}:{vehicleKey}` | 单协议最新完整 parsed 状态,是实时完整协议字段的唯一 Redis 副本 | | `vehicle:online:{vehicleKey}` | 在线状态和 TTL | | `vehicle:protocols:{vehicleKey}` | 当前车辆最近出现过的协议集合 | diff --git a/docs/go-version-data-flow.html b/docs/go-version-data-flow.html index 91133eb3..fe14df52 100644 --- a/docs/go-version-data-flow.html +++ b/docs/go-version-data-flow.html @@ -587,7 +587,7 @@ Redis vehicle:latest:{vehicle_key}:{protocol} realtime-api Kafka consumer - 单协议实时快照 + 单协议实时轻量快照 区分 32960/808/MQTT 的实时状态 diff --git a/go/vehicle-gateway/internal/realtime/model.go b/go/vehicle-gateway/internal/realtime/model.go index 342280d3..d5b978f0 100644 --- a/go/vehicle-gateway/internal/realtime/model.go +++ b/go/vehicle-gateway/internal/realtime/model.go @@ -21,6 +21,11 @@ type Snapshot struct { UpdatedAtMS int64 `json:"updated_at_ms"` } +func (s Snapshot) Lightweight() Snapshot { + s.Parsed = nil + return s +} + type OnlineStatus struct { VehicleKey string `json:"vehicle_key"` VIN string `json:"vin"` diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index 4019026d..596188ff 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -55,13 +55,16 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err if err != nil && !errors.Is(err, redis.Nil) { return err } + existingParsed, err := r.GetRealtimeRaw(ctx, vehicleKey, env.Protocol) + if err != nil && !errors.Is(err, redis.Nil) { + return err + } if existingProtocol.VehicleKey != "" { protocolSnapshot = existingProtocol if protocolSnapshot.VIN == "" && vin != "" { protocolSnapshot.VIN = vin } mergeFields(&protocolSnapshot, env.Fields, eventMS) - protocolSnapshot.Parsed = mergeParsedForProtocol(env.Protocol, protocolSnapshot.Parsed, env.Parsed) if eventMS >= protocolSnapshot.EventTimeMS { protocolSnapshot.EventTimeMS = eventMS protocolSnapshot.EventID = env.StableEventID() @@ -70,7 +73,8 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err } protocolSnapshot.UpdatedAtMS = nowMS } - if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil { + protocolSnapshot.Parsed = mergeParsedForProtocol(env.Protocol, existingParsed, env.Parsed) + if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot.Lightweight(), r.cfg.ttl()); err != nil { return err } if err := r.setJSON(ctx, realtimeRawKey(vehicleKey, env.Protocol), protocolSnapshot.Parsed, r.cfg.ttl()); err != nil { diff --git a/go/vehicle-gateway/internal/realtime/repository_test.go b/go/vehicle-gateway/internal/realtime/repository_test.go index b807e94e..a9970529 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -67,7 +67,7 @@ func TestRepositoryUpdatesMergedAndProtocolSnapshots(t *testing.T) { } } -func TestRepositoryStoresFullParsedProtocolSnapshotAndMergesGB32960Units(t *testing.T) { +func TestRepositoryStoresFullParsedOnlyInRealtimeRawAndMergesGB32960Units(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() ctx := context.Background() @@ -110,9 +110,12 @@ func TestRepositoryStoresFullParsedProtocolSnapshotAndMergesGB32960Units(t *test if err != nil { t.Fatalf("GetRealtimeRaw() error = %v", err) } - units := protocol.Parsed["data_units"].([]any) - if len(units) != 2 { - t.Fatalf("expected merged gb32960 units, got %#v", protocol.Parsed) + protocolJSON, err := json.Marshal(protocol) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(protocolJSON), `"parsed"`) { + t.Fatalf("protocol snapshot should not duplicate full parsed data: %s", string(protocolJSON)) } if len(realtimeRaw["data_units"].([]any)) != 2 { t.Fatalf("realtime-raw did not keep merged parsed fields: %#v", realtimeRaw)