From 3a67f6e05f363bacb73a148370f6a130cd07c50c Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 11:19:46 +0800 Subject: [PATCH] feat: cache protocol realtime raw --- .../internal/envelope/envelope.go | 6 +- .../internal/envelope/envelope_test.go | 6 +- .../internal/eventbus/kafka_sink_test.go | 4 +- go/vehicle-gateway/internal/realtime/http.go | 3 + go/vehicle-gateway/internal/realtime/model.go | 24 +-- .../internal/realtime/repository.go | 183 +++++++++++++++++- .../internal/realtime/repository_test.go | 101 +++++++++- .../internal/stats/daily_metric_test.go | 4 +- 8 files changed, 304 insertions(+), 27 deletions(-) diff --git a/go/vehicle-gateway/internal/envelope/envelope.go b/go/vehicle-gateway/internal/envelope/envelope.go index bbd89b24..7d402198 100644 --- a/go/vehicle-gateway/internal/envelope/envelope.go +++ b/go/vehicle-gateway/internal/envelope/envelope.go @@ -58,12 +58,12 @@ func (e FrameEnvelope) VehicleKey() string { if key := strings.TrimSpace(e.VIN); key != "" { return key } - if key := strings.TrimSpace(e.VehicleKeyHint); key != "" { - return key - } if key := strings.TrimSpace(e.Phone); key != "" { return string(e.Protocol) + ":" + key } + if key := strings.TrimSpace(e.VehicleKeyHint); key != "" { + return key + } if key := strings.TrimSpace(e.DeviceID); key != "" { return string(e.Protocol) + ":" + key } diff --git a/go/vehicle-gateway/internal/envelope/envelope_test.go b/go/vehicle-gateway/internal/envelope/envelope_test.go index 934d3bf8..520430fa 100644 --- a/go/vehicle-gateway/internal/envelope/envelope_test.go +++ b/go/vehicle-gateway/internal/envelope/envelope_test.go @@ -6,15 +6,15 @@ import ( ) func TestFrameEnvelopeVehicleKeyPrefersVIN(t *testing.T) { - e := FrameEnvelope{Protocol: ProtocolJT808, VIN: "LNBVIN00000000001", Phone: "013307795425"} + e := FrameEnvelope{Protocol: ProtocolJT808, VIN: "LNBVIN00000000001", Phone: "13307795425"} if got := e.VehicleKey(); got != "LNBVIN00000000001" { t.Fatalf("VehicleKey() = %q", got) } } func TestFrameEnvelopeVehicleKeyFallsBackToPhone(t *testing.T) { - e := FrameEnvelope{Protocol: ProtocolJT808, Phone: "013307795425"} - if got := e.VehicleKey(); got != "JT808:013307795425" { + e := FrameEnvelope{Protocol: ProtocolJT808, Phone: "13307795425"} + if got := e.VehicleKey(); got != "JT808:13307795425" { t.Fatalf("VehicleKey() = %q", got) } } diff --git a/go/vehicle-gateway/internal/eventbus/kafka_sink_test.go b/go/vehicle-gateway/internal/eventbus/kafka_sink_test.go index a27fce47..bc02d25d 100644 --- a/go/vehicle-gateway/internal/eventbus/kafka_sink_test.go +++ b/go/vehicle-gateway/internal/eventbus/kafka_sink_test.go @@ -14,7 +14,7 @@ func TestKafkaSinkRoutesRawTopics(t *testing.T) { writer := &recordingWriter{} sink := newKafkaSinkWithWriter(writer, KafkaConfig{}) - env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "013307795425", MessageID: "0x0200"} + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolJT808, Phone: "13307795425", MessageID: "0x0200"} if err := sink.PublishRaw(context.Background(), env); err != nil { t.Fatalf("PublishRaw() error = %v", err) } @@ -25,7 +25,7 @@ func TestKafkaSinkRoutesRawTopics(t *testing.T) { if msg.Topic != "vehicle.raw.jt808.v1" { t.Fatalf("topic = %q", msg.Topic) } - if string(msg.Key) != "JT808:013307795425" { + if string(msg.Key) != "JT808:13307795425" { t.Fatalf("key = %q", string(msg.Key)) } var decoded envelope.FrameEnvelope diff --git a/go/vehicle-gateway/internal/realtime/http.go b/go/vehicle-gateway/internal/realtime/http.go index e6e5cfef..5c710131 100644 --- a/go/vehicle-gateway/internal/realtime/http.go +++ b/go/vehicle-gateway/internal/realtime/http.go @@ -44,6 +44,9 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) { case len(parts) == 3 && parts[1] == "protocols": snapshot, err := h.repository.GetProtocol(r.Context(), vin, envelope.Protocol(strings.ToUpper(parts[2]))) writeResult(w, snapshot, err) + case len(parts) == 3 && parts[1] == "realtime-raw": + payload, err := h.repository.GetRealtimeRaw(r.Context(), vin, envelope.Protocol(strings.ToUpper(parts[2]))) + writeResult(w, payload, err) default: writeError(w, http.StatusNotFound, "route not found") } diff --git a/go/vehicle-gateway/internal/realtime/model.go b/go/vehicle-gateway/internal/realtime/model.go index 747bd5ce..3474f4af 100644 --- a/go/vehicle-gateway/internal/realtime/model.go +++ b/go/vehicle-gateway/internal/realtime/model.go @@ -7,17 +7,19 @@ import ( ) type Snapshot struct { - VehicleKey string `json:"vehicle_key"` - VIN string `json:"vin"` - Protocol envelope.Protocol `json:"protocol,omitempty"` - Protocols []envelope.Protocol `json:"protocols,omitempty"` - EventID string `json:"event_id,omitempty"` - EventTimeMS int64 `json:"event_time_ms"` - ReceivedAtMS int64 `json:"received_at_ms"` - SourceEndpoint string `json:"source_endpoint,omitempty"` - Fields map[string]any `json:"fields,omitempty"` - FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"` - UpdatedAtMS int64 `json:"updated_at_ms"` + VehicleKey string `json:"vehicle_key"` + VIN string `json:"vin"` + Protocol envelope.Protocol `json:"protocol,omitempty"` + Protocols []envelope.Protocol `json:"protocols,omitempty"` + EventID string `json:"event_id,omitempty"` + EventTimeMS int64 `json:"event_time_ms"` + ReceivedAtMS int64 `json:"received_at_ms"` + SourceEndpoint string `json:"source_endpoint,omitempty"` + Fields map[string]any `json:"fields,omitempty"` + FieldTimesMS map[string]int64 `json:"field_times_ms,omitempty"` + Parsed map[string]any `json:"parsed,omitempty"` + ProtocolData map[envelope.Protocol]map[string]any `json:"protocol_data,omitempty"` + UpdatedAtMS int64 `json:"updated_at_ms"` } type OnlineStatus struct { diff --git a/go/vehicle-gateway/internal/realtime/repository.go b/go/vehicle-gateway/internal/realtime/repository.go index d4adab3b..6d629779 100644 --- a/go/vehicle-gateway/internal/realtime/repository.go +++ b/go/vehicle-gateway/internal/realtime/repository.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "errors" + "fmt" "sort" "strconv" "strings" @@ -47,11 +48,34 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err SourceEndpoint: env.SourceEndpoint, Fields: cloneFields(env.Fields), FieldTimesMS: fieldTimes(env.Fields, eventMS), + Parsed: cloneMap(env.Parsed), UpdatedAtMS: nowMS, } + existingProtocol, err := r.GetProtocol(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 = mergeParsed(protocolSnapshot.Parsed, env.Parsed) + if eventMS >= protocolSnapshot.EventTimeMS { + protocolSnapshot.EventTimeMS = eventMS + protocolSnapshot.EventID = env.StableEventID() + protocolSnapshot.ReceivedAtMS = env.ReceivedAtMS + protocolSnapshot.SourceEndpoint = env.SourceEndpoint + } + protocolSnapshot.UpdatedAtMS = nowMS + } if err := r.setJSON(ctx, protocolKey(vehicleKey, env.Protocol), protocolSnapshot, r.cfg.ttl()); err != nil { return err } + if err := r.setJSON(ctx, realtimeRawKey(vehicleKey, env.Protocol), protocolSnapshot.Parsed, r.cfg.ttl()); err != nil { + return err + } protocols, err := r.addProtocol(ctx, vehicleKey, env.Protocol) if err != nil { @@ -63,7 +87,13 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err return err } if merged.VehicleKey == "" { - merged = Snapshot{VehicleKey: vehicleKey, VIN: vin, Fields: map[string]any{}, FieldTimesMS: map[string]int64{}} + merged = Snapshot{ + VehicleKey: vehicleKey, + VIN: vin, + Fields: map[string]any{}, + FieldTimesMS: map[string]int64{}, + ProtocolData: map[envelope.Protocol]map[string]any{}, + } } else if merged.VIN == "" && vin != "" { merged.VIN = vin } @@ -75,6 +105,10 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err merged.SourceEndpoint = env.SourceEndpoint } merged.Protocols = protocols + if merged.ProtocolData == nil { + merged.ProtocolData = map[envelope.Protocol]map[string]any{} + } + merged.ProtocolData[env.Protocol] = cloneMap(protocolSnapshot.Parsed) merged.UpdatedAtMS = nowMS if err := r.setJSON(ctx, mergedKey(vehicleKey), merged, r.cfg.ttl()); err != nil { return err @@ -102,6 +136,15 @@ func (r *Repository) GetProtocol(ctx context.Context, vehicleKey string, protoco return r.getSnapshot(ctx, protocolKey(vehicleKey, protocol)) } +func (r *Repository) GetRealtimeRaw(ctx context.Context, vehicleKey string, protocol envelope.Protocol) (map[string]any, error) { + payload, err := r.client.Get(ctx, realtimeRawKey(vehicleKey, protocol)).Bytes() + if err != nil { + return nil, err + } + var out map[string]any + return out, json.Unmarshal(payload, &out) +} + func (r *Repository) IsOnline(ctx context.Context, vehicleKey string) (OnlineStatus, error) { vehicleKey = strings.TrimSpace(vehicleKey) var status OnlineStatus @@ -207,6 +250,140 @@ func cloneFields(fields map[string]any) map[string]any { return out } +func cloneMap(values map[string]any) map[string]any { + if len(values) == 0 { + return map[string]any{} + } + out := make(map[string]any, len(values)) + for key, value := range values { + out[key] = cloneAny(value) + } + return out +} + +func cloneAny(value any) any { + switch typed := value.(type) { + case map[string]any: + return cloneMap(typed) + case []any: + out := make([]any, len(typed)) + for i, item := range typed { + out[i] = cloneAny(item) + } + return out + case []map[string]any: + out := make([]any, len(typed)) + for i, item := range typed { + out[i] = cloneMap(item) + } + return out + default: + return value + } +} + +func mergeParsed(existing map[string]any, incoming map[string]any) map[string]any { + out := cloneMap(existing) + for key, value := range incoming { + out[key] = mergeAny(out[key], value) + } + return out +} + +func mergeAny(existing any, incoming any) any { + switch incomingTyped := incoming.(type) { + case map[string]any: + existingMap, _ := existing.(map[string]any) + return mergeParsed(existingMap, incomingTyped) + case []any: + if merged, ok := mergeMapSlice(existing, incomingTyped); ok { + return merged + } + return cloneAny(incomingTyped) + case []map[string]any: + items := make([]any, len(incomingTyped)) + for i, item := range incomingTyped { + items[i] = item + } + if merged, ok := mergeMapSlice(existing, items); ok { + return merged + } + return cloneAny(incomingTyped) + default: + return cloneAny(incoming) + } +} + +func mergeMapSlice(existing any, incoming []any) ([]any, bool) { + existingItems, ok := asAnySlice(existing) + if !ok { + return nil, false + } + merged := make([]any, 0, len(existingItems)+len(incoming)) + index := map[string]int{} + add := func(item any) { + itemMap, ok := item.(map[string]any) + if !ok { + merged = append(merged, cloneAny(item)) + return + } + key := mergeSliceKey(itemMap) + if key == "" { + merged = append(merged, cloneMap(itemMap)) + return + } + if pos, exists := index[key]; exists { + merged[pos] = mergeAny(merged[pos], itemMap) + return + } + index[key] = len(merged) + merged = append(merged, cloneMap(itemMap)) + } + for _, item := range existingItems { + add(item) + } + for _, item := range incoming { + add(item) + } + return merged, true +} + +func asAnySlice(value any) ([]any, bool) { + switch typed := value.(type) { + case []any: + return typed, true + case []map[string]any: + out := make([]any, len(typed)) + for i, item := range typed { + out[i] = item + } + return out, true + default: + return nil, false + } +} + +func mergeSliceKey(item map[string]any) string { + for _, key := range []string{"name", "type", "id"} { + if value, ok := item[key]; ok { + text := strings.TrimSpace(strconvAny(value)) + if text != "" { + return key + ":" + text + } + } + } + return "" +} + +func strconvAny(value any) string { + switch typed := value.(type) { + case string: + return typed + default: + return fmt.Sprint(typed) + } +} + func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 { out := make(map[string]int64, len(fields)) for key := range fields { @@ -223,6 +400,10 @@ func protocolKey(vehicleKey string, protocol envelope.Protocol) string { return "vehicle:latest:" + strings.TrimSpace(vehicleKey) + ":" + string(protocol) } +func realtimeRawKey(vehicleKey string, protocol envelope.Protocol) string { + return "vehicle:realtime-raw:" + string(protocol) + ":" + strings.TrimSpace(vehicleKey) +} + 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 169ce4f1..7734f1f8 100644 --- a/go/vehicle-gateway/internal/realtime/repository_test.go +++ b/go/vehicle-gateway/internal/realtime/repository_test.go @@ -66,6 +66,69 @@ func TestRepositoryUpdatesMergedAndProtocolSnapshots(t *testing.T) { } } +func TestRepositoryStoresFullParsedProtocolSnapshotAndMergesGB32960Units(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{ + "header": map[string]any{"command": "0x02"}, + "data_units": []any{ + map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 88.0}}, + }, + }, + Fields: map[string]any{envelope.FieldSOCPercent: 88.0}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if err := repo.Update(ctx, envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + VIN: "VIN001", + EventTimeMS: 1100, + ReceivedAtMS: 1200, + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{"type": "0x05", "name": "position", "value": map[string]any{"longitude": 121.1, "latitude": 30.2}}, + }, + }, + Fields: map[string]any{envelope.FieldLongitude: 121.1, envelope.FieldLatitude: 30.2}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + protocol, err := repo.GetProtocol(ctx, "VIN001", envelope.ProtocolGB32960) + if err != nil { + t.Fatalf("GetProtocol() error = %v", err) + } + realtimeRaw, err := repo.GetRealtimeRaw(ctx, "VIN001", envelope.ProtocolGB32960) + 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) + } + if len(realtimeRaw["data_units"].([]any)) != 2 { + t.Fatalf("realtime-raw did not keep merged parsed fields: %#v", realtimeRaw) + } + merged, err := repo.GetMerged(ctx, "VIN001") + if err != nil { + t.Fatalf("GetMerged() error = %v", err) + } + gb, ok := merged.ProtocolData[envelope.ProtocolGB32960] + if !ok { + t.Fatalf("merged protocol data missing: %#v", merged.ProtocolData) + } + if len(gb["data_units"].([]any)) != 2 { + t.Fatalf("merged protocol data did not keep full parsed fields: %#v", gb) + } +} + func TestRepositoryOnlineStatus(t *testing.T) { repo, closeFn := newTestRepository(t) defer closeFn() @@ -104,7 +167,7 @@ func TestRepositoryUpdatesPhoneOnlyVehicleKey(t *testing.T) { if err := repo.Update(ctx, envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, - Phone: "013307811170", + Phone: "13307811170", EventTimeMS: 1000, ReceivedAtMS: 1100, Fields: map[string]any{ @@ -115,7 +178,7 @@ func TestRepositoryUpdatesPhoneOnlyVehicleKey(t *testing.T) { t.Fatalf("Update() error = %v", err) } - vehicleKey := "JT808:013307811170" + vehicleKey := "JT808:13307811170" merged, err := repo.GetMerged(ctx, vehicleKey) if err != nil { t.Fatalf("GetMerged() error = %v", err) @@ -205,7 +268,7 @@ func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { defer closeFn() if err := repo.Update(context.Background(), envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, - Phone: "013307811170", + Phone: "13307811170", EventTimeMS: 1000, ReceivedAtMS: 1100, Fields: map[string]any{envelope.FieldSpeedKMH: 12.3}, @@ -213,13 +276,41 @@ func TestHandlerReturnsPhoneOnlyVehicleKeySnapshot(t *testing.T) { t.Fatalf("Update() error = %v", err) } - req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles/JT808:013307811170", nil) + req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles/JT808:13307811170", nil) rec := httptest.NewRecorder() NewHandler(repo).ServeHTTP(rec, req) if rec.Code != http.StatusOK { t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) } - if !stringsContains(rec.Body.String(), `"vehicle_key":"JT808:013307811170"`) { + if !stringsContains(rec.Body.String(), `"vehicle_key":"JT808:13307811170"`) { + t.Fatalf("unexpected body: %s", rec.Body.String()) + } +} + +func TestHandlerReturnsRealtimeRaw(t *testing.T) { + repo, closeFn := newTestRepository(t) + defer closeFn() + if err := repo.Update(context.Background(), 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": 90.0}}, + }, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + req := httptest.NewRequest(http.MethodGet, "/api/realtime/vehicles/VIN001/realtime-raw/GB32960", nil) + rec := httptest.NewRecorder() + NewHandler(repo).ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", rec.Code, rec.Body.String()) + } + if !stringsContains(rec.Body.String(), `"soc_percent":90`) { t.Fatalf("unexpected body: %s", rec.Body.String()) } } diff --git a/go/vehicle-gateway/internal/stats/daily_metric_test.go b/go/vehicle-gateway/internal/stats/daily_metric_test.go index 8a4f2046..63e1168b 100644 --- a/go/vehicle-gateway/internal/stats/daily_metric_test.go +++ b/go/vehicle-gateway/internal/stats/daily_metric_test.go @@ -46,7 +46,7 @@ func TestSamplesFromEnvelopeUsesVehicleKeyWhenVINIsMissing(t *testing.T) { loc := time.FixedZone("Asia/Shanghai", 8*3600) samples, err := SamplesFromEnvelope(envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, - Phone: "013307811254", + Phone: "13307811254", EventTimeMS: time.Date(2026, 7, 2, 0, 3, 0, 0, loc).UnixMilli(), Fields: map[string]any{ envelope.FieldTotalMileageKM: 10985.7, @@ -58,7 +58,7 @@ func TestSamplesFromEnvelopeUsesVehicleKeyWhenVINIsMissing(t *testing.T) { if len(samples) != 2 { t.Fatalf("sample count = %d", len(samples)) } - if samples[0].VehicleKey != "JT808:013307811254" { + if samples[0].VehicleKey != "JT808:13307811254" { t.Fatalf("vehicle key = %q", samples[0].VehicleKey) } if samples[0].VIN != "" {