fix(go): normalize gb32960 realtime vendor fragments

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

@@ -294,6 +294,9 @@ func mergeParsedForProtocol(protocol envelope.Protocol, existing map[string]any,
 	for _, key := range replaceParsedKeys(protocol, incoming) {
 		out[key] = cloneAny(incoming[key])
 	}
+	if protocol == envelope.ProtocolGB32960 {
+		normalizeGB32960RealtimeParsed(out)
+	}
 	return out
 }
 
@@ -404,6 +407,59 @@ func strconvAny(value any) string {
 	}
 }
 
+func normalizeGB32960RealtimeParsed(parsed map[string]any) {
+	units, ok := asAnySlice(parsed["data_units"])
+	if !ok {
+		return
+	}
+	for _, unit := range units {
+		unitMap, ok := unit.(map[string]any)
+		if !ok || strings.TrimSpace(strconvAny(unitMap["name"])) != "gd_fc_stack" {
+			continue
+		}
+		value, ok := unitMap["value"].(map[string]any)
+		if !ok {
+			continue
+		}
+		collapseGDFCStackValue(value)
+	}
+}
+
+func collapseGDFCStackValue(value map[string]any) {
+	summaries, ok := asAnySlice(value["summaries"])
+	if !ok || len(summaries) == 0 {
+		return
+	}
+	collapsed := map[string]any{}
+	for _, item := range summaries {
+		summary, ok := item.(map[string]any)
+		if !ok {
+			continue
+		}
+		for key, fieldValue := range summary {
+			if isGDFCStackFragmentField(key) {
+				continue
+			}
+			collapsed[key] = cloneAny(fieldValue)
+		}
+	}
+	if len(collapsed) == 0 {
+		delete(value, "summaries")
+		return
+	}
+	value["stack_count"] = 1
+	value["summaries"] = []any{collapsed}
+}
+
+func isGDFCStackFragmentField(key string) bool {
+	switch key {
+	case "frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v":
+		return true
+	default:
+		return false
+	}
+}
+
 func fieldTimes(fields map[string]any, eventMS int64) map[string]int64 {
 	out := make(map[string]int64, len(fields))
 	for key := range fields {

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

@@ -200,6 +200,77 @@ func TestRepositoryMergesNestedGB32960MotorSlicesBySerialNo(t *testing.T) {
 	}
 }
 
+func TestRepositoryCollapsesGB32960VendorStackFragmentsIntoSingleRealtimeSummary(t *testing.T) {
+	repo, closeFn := newTestRepository(t)
+	defer closeFn()
+	ctx := context.Background()
+
+	first := map[string]any{
+		"engine_work_state":           1,
+		"stack_water_outlet_temp_c":   65,
+		"hydrogen_inlet_pressure_kpa": 130,
+		"cell_count":                  432,
+		"avg_cell_voltage_v":          0.78,
+		"frame_cell_start":            201,
+		"frame_cell_count":            200,
+		"frame_max_cell_voltage_v":    1.0,
+		"frame_min_cell_voltage_v":    0.78,
+	}
+	second := map[string]any{
+		"engine_work_state":           1,
+		"stack_water_outlet_temp_c":   63,
+		"hydrogen_inlet_pressure_kpa": 130,
+		"cell_count":                  432,
+		"avg_cell_voltage_v":          0.80,
+		"frame_cell_start":            401,
+		"frame_cell_count":            32,
+		"frame_max_cell_voltage_v":    1.0,
+		"frame_min_cell_voltage_v":    1.0,
+	}
+	for i, summary := range []map[string]any{first, second} {
+		if err := repo.Update(ctx, envelope.FrameEnvelope{
+			Protocol:     envelope.ProtocolGB32960,
+			VIN:          "VIN001",
+			EventTimeMS:  int64(1000 + i),
+			ReceivedAtMS: int64(1100 + i),
+			Parsed: map[string]any{
+				"data_units": []any{
+					map[string]any{
+						"type": "0x30",
+						"name": "gd_fc_stack",
+						"value": map[string]any{
+							"stack_count": 1,
+							"summaries":   []any{summary},
+						},
+					},
+				},
+			},
+		}); err != nil {
+			t.Fatalf("Update() error = %v", err)
+		}
+	}
+
+	realtimeRaw, err := repo.GetRealtimeRaw(ctx, "VIN001", envelope.ProtocolGB32960)
+	if err != nil {
+		t.Fatalf("GetRealtimeRaw() error = %v", err)
+	}
+	unit := realtimeRaw["data_units"].([]any)[0].(map[string]any)
+	value := unit["value"].(map[string]any)
+	summaries := value["summaries"].([]any)
+	if len(summaries) != 1 {
+		t.Fatalf("stack summaries should be a current stack state, not appended fragments: %#v", summaries)
+	}
+	summary := summaries[0].(map[string]any)
+	for _, fragmentOnly := range []string{"frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v"} {
+		if _, ok := summary[fragmentOnly]; ok {
+			t.Fatalf("stack summary should not expose fragment-only field %s: %#v", fragmentOnly, summary)
+		}
+	}
+	if summary["stack_water_outlet_temp_c"] != float64(63) {
+		t.Fatalf("summary should keep latest stack-level values: %#v", summary)
+	}
+}
+
 func TestRepositoryReplacesJT808LocationWhenUpdatingRealtimeRaw(t *testing.T) {
 	repo, closeFn := newTestRepository(t)
 	defer closeFn()

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

@@ -247,6 +247,61 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi
 	}
 }
 
+func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *testing.T) {
+	db, mock, err := sqlmock.New()
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	writer := NewSnapshotWriter(db)
+
+	existing := `{"data_units":[{"type":"0x30","name":"gd_fc_stack","value":{"stack_count":1,"summaries":[{"cell_count":432,"stack_water_outlet_temp_c":65,"frame_cell_start":201,"frame_cell_count":200}]}}]}`
+	mock.ExpectQuery("SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = \\? AND vin = \\?").
+		WithArgs("GB32960", "VIN001").
+		WillReturnRows(sqlmock.NewRows([]string{"parsed_json"}).AddRow(existing))
+	mock.ExpectExec("INSERT INTO vehicle_realtime_snapshot").
+		WithArgs(
+			"GB32960",
+			"VIN001",
+			"",
+			"",
+			"",
+			jsonGDFCStackSummaryArg{},
+			sqlmock.AnyArg(),
+			sqlmock.AnyArg(),
+			sqlmock.AnyArg(),
+		).
+		WillReturnResult(sqlmock.NewResult(0, 1))
+
+	err = writer.Update(context.Background(), envelope.FrameEnvelope{
+		Protocol:     envelope.ProtocolGB32960,
+		MessageID:    "0x02",
+		VIN:          "VIN001",
+		EventTimeMS:  1782918600000,
+		ReceivedAtMS: 1782918601000,
+		Parsed: map[string]any{
+			"data_units": []any{
+				map[string]any{
+					"type": "0x30",
+					"name": "gd_fc_stack",
+					"value": map[string]any{
+						"stack_count": 1,
+						"summaries": []any{
+							map[string]any{"cell_count": 432, "stack_water_outlet_temp_c": 63, "frame_cell_start": 401, "frame_cell_count": 32},
+						},
+					},
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Update() error = %v", err)
+	}
+	if err := mock.ExpectationsWereMet(); err != nil {
+		t.Fatalf("sql expectations: %v", err)
+	}
+}
+
 func TestRealtimeUpsertsDoNotOverwriteWithOlderEventTime(t *testing.T) {
 	for _, tc := range []struct {
 		name  string
@@ -641,6 +696,48 @@ type jsonDataUnitsArg struct {
 	types []string
 }
 
+type jsonGDFCStackSummaryArg struct{}
+
+func (jsonGDFCStackSummaryArg) Match(value driver.Value) bool {
+	text, ok := value.(string)
+	if !ok {
+		return false
+	}
+	var parsed map[string]any
+	if err := json.Unmarshal([]byte(text), &parsed); err != nil {
+		return false
+	}
+	units, ok := parsed["data_units"].([]any)
+	if !ok {
+		return false
+	}
+	for _, unit := range units {
+		unitMap, ok := unit.(map[string]any)
+		if !ok || unitMap["name"] != "gd_fc_stack" {
+			continue
+		}
+		valueMap, ok := unitMap["value"].(map[string]any)
+		if !ok {
+			return false
+		}
+		summaries, ok := valueMap["summaries"].([]any)
+		if !ok || len(summaries) != 1 {
+			return false
+		}
+		summary, ok := summaries[0].(map[string]any)
+		if !ok {
+			return false
+		}
+		for _, fragmentOnly := range []string{"frame_cell_start", "frame_cell_count", "frame_max_cell_voltage_v", "frame_min_cell_voltage_v"} {
+			if _, ok := summary[fragmentOnly]; ok {
+				return false
+			}
+		}
+		return summary["stack_water_outlet_temp_c"] == float64(63)
+	}
+	return false
+}
+
 func (a jsonDataUnitsArg) Match(value driver.Value) bool {
 	text, ok := value.(string)
 	if !ok {