feat: cache protocol realtime raw

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)
 }