feat(go): flatten mysql realtime snapshots

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 {