fix(go): flatten jt808 additional fields

go/vehicle-gateway/internal/realtime/repository.go

@@ -61,7 +61,7 @@ func (r *Repository) Update(ctx context.Context, env envelope.FrameEnvelope) err
 			protocolSnapshot.VIN = vin
 		}
 		mergeFields(&protocolSnapshot, env.Fields, eventMS)
-		protocolSnapshot.Parsed = mergeParsed(protocolSnapshot.Parsed, env.Parsed)
+		protocolSnapshot.Parsed = mergeParsedForProtocol(env.Protocol, protocolSnapshot.Parsed, env.Parsed)
 		if eventMS >= protocolSnapshot.EventTimeMS {
 			protocolSnapshot.EventTimeMS = eventMS
 			protocolSnapshot.EventID = env.StableEventID()
@@ -290,6 +290,27 @@ func mergeParsed(existing map[string]any, incoming map[string]any) map[string]an
 	return out
 }
 
+func mergeParsedForProtocol(protocol envelope.Protocol, existing map[string]any, incoming map[string]any) map[string]any {
+	out := mergeParsed(existing, incoming)
+	for _, key := range replaceParsedKeys(protocol, incoming) {
+		out[key] = cloneAny(incoming[key])
+	}
+	return out
+}
+
+func replaceParsedKeys(protocol envelope.Protocol, incoming map[string]any) []string {
+	keys := make([]string, 0, 2)
+	if _, ok := incoming["header"]; ok {
+		keys = append(keys, "header")
+	}
+	if protocol == envelope.ProtocolJT808 {
+		if _, ok := incoming["location"]; ok {
+			keys = append(keys, "location")
+		}
+	}
+	return keys
+}
+
 func mergeAny(existing any, incoming any) any {
 	switch incomingTyped := incoming.(type) {
 	case map[string]any:

go/vehicle-gateway/internal/realtime/repository_test.go

@@ -196,6 +196,72 @@ func TestRepositoryMergesNestedGB32960MotorSlicesBySerialNo(t *testing.T) {
 	}
 }
 
+func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) {
+	repo, closeFn := newTestRepository(t)
+	defer closeFn()
+	ctx := context.Background()
+
+	if err := repo.Update(ctx, envelope.FrameEnvelope{
+		Protocol:     envelope.ProtocolJT808,
+		VIN:          "VIN001",
+		EventTimeMS:  1000,
+		ReceivedAtMS: 1100,
+		MessageID:    "0x0200",
+		Parsed: map[string]any{
+			"header": map[string]any{
+				"message_id":    "0x0200",
+				"phone_raw":     "013307795425",
+				"phone_bcd_hex": "013307795425",
+			},
+			"location": map[string]any{
+				"additional": []any{
+					map[string]any{"id": "0x01", "parsed": map[string]any{"raw_value_tenth_km": 102412}},
+				},
+				"total_mileage_km": 10241.2,
+			},
+		},
+		Fields: map[string]any{envelope.FieldTotalMileageKM: 10241.2},
+	}); err != nil {
+		t.Fatalf("Update() initial error = %v", err)
+	}
+	if err := repo.Update(ctx, envelope.FrameEnvelope{
+		Protocol:     envelope.ProtocolJT808,
+		VIN:          "VIN001",
+		EventTimeMS:  2000,
+		ReceivedAtMS: 2100,
+		MessageID:    "0x0200",
+		Parsed: map[string]any{
+			"header": map[string]any{
+				"message_id": "0x0200",
+				"phone":      "13307795425",
+			},
+			"location": map[string]any{
+				"total_mileage_km":        10241.3,
+				"network_signal_strength": byte(31),
+			},
+		},
+		Fields: map[string]any{envelope.FieldTotalMileageKM: 10241.3},
+	}); err != nil {
+		t.Fatalf("Update() replacement error = %v", err)
+	}
+
+	realtimeRaw, err := repo.GetRealtimeRaw(ctx, "VIN001", envelope.ProtocolJT808)
+	if err != nil {
+		t.Fatalf("GetRealtimeRaw() error = %v", err)
+	}
+	header := realtimeRaw["header"].(map[string]any)
+	if _, ok := header["phone_raw"]; ok {
+		t.Fatalf("header should be replaced, got %#v", header)
+	}
+	location := realtimeRaw["location"].(map[string]any)
+	if _, ok := location["additional"]; ok {
+		t.Fatalf("jt808 location should be replaced, got %#v", location)
+	}
+	if location["total_mileage_km"] != float64(10241.3) {
+		t.Fatalf("mileage = %#v", location["total_mileage_km"])
+	}
+}
+
 func TestRepositoryOnlineStatus(t *testing.T) {
 	repo, closeFn := newTestRepository(t)
 	defer closeFn()