diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 4dfbb39e..41c70ec2 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -129,7 +129,10 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) platformName := platformNameFromEnvelope(env) - parsedJSON := parsedJSONFromEnvelope(env) + parsedJSON, err := w.parsedJSONForEnvelope(ctx, env, vin) + if err != nil { + return err + } if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, string(env.Protocol), vin, @@ -167,6 +170,23 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) return err } +func (w *SnapshotWriter) parsedJSONForEnvelope(ctx context.Context, env envelope.FrameEnvelope, vin string) (any, error) { + if len(env.Parsed) == 0 { + return nil, nil + } + parsed := cloneMap(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) + } + } + return marshalParsedJSON(parsed), nil +} + func platformNameFromEnvelope(env envelope.FrameEnvelope) string { if env.Fields != nil { if value := strings.TrimSpace(stringValue(env.Fields["platform_account"])); value != "" { @@ -193,11 +213,30 @@ func stringValue(value any) string { } } -func parsedJSONFromEnvelope(env envelope.FrameEnvelope) any { - if len(env.Parsed) == 0 { +func realtimeSnapshotParsedJSON(ctx context.Context, queryer Queryer, protocol envelope.Protocol, vin string) (map[string]any, error) { + var raw sql.NullString + err := queryer.QueryRowContext(ctx, selectRealtimeSnapshotParsedJSONSQL, string(protocol), vin).Scan(&raw) + if err != nil { + if errors.Is(err, sql.ErrNoRows) { + return nil, nil + } + return nil, err + } + if !raw.Valid || strings.TrimSpace(raw.String) == "" { + return nil, nil + } + var parsed map[string]any + if err := json.Unmarshal([]byte(raw.String), &parsed); err != nil { + return nil, nil + } + return parsed, nil +} + +func marshalParsedJSON(parsed map[string]any) any { + if len(parsed) == 0 { return nil } - data, err := json.Marshal(env.Parsed) + data, err := json.Marshal(parsed) if err != nil { return nil } @@ -436,6 +475,8 @@ ON DUPLICATE KEY UPDATE updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), CURRENT_TIMESTAMP, updated_at) ` +const selectRealtimeSnapshotParsedJSONSQL = `SELECT parsed_json FROM vehicle_realtime_snapshot WHERE protocol = ? AND vin = ?` + const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location ( protocol VARCHAR(32) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index de296169..8f793ef1 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -3,6 +3,8 @@ package realtime import ( "context" "database/sql" + "database/sql/driver" + "encoding/json" "errors" "strings" "testing" @@ -199,6 +201,52 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) } } +func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatal(err) + } + defer db.Close() + writer := NewSnapshotWriter(db) + + existing := `{"data_units":[{"type":"0x01","name":"vehicle","value":{"soc_percent":88}}]}` + 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", + "", + "", + "", + jsonDataUnitsArg{types: []string{"0x01", "0x30"}}, + 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}}, + }, + }, + }) + 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 @@ -588,3 +636,38 @@ func (r *recordingPlateResolver) PlateByVIN(_ context.Context, vin string) (stri r.vin = vin return r.plate, r.err } + +type jsonDataUnitsArg struct { + types []string +} + +func (a jsonDataUnitsArg) 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 + } + seen := map[string]bool{} + for _, unit := range units { + unitMap, ok := unit.(map[string]any) + if !ok { + continue + } + if unitType, ok := unitMap["type"].(string); ok { + seen[unitType] = true + } + } + for _, unitType := range a.types { + if !seen[unitType] { + return false + } + } + return true +}