diff --git a/go/vehicle-gateway/internal/gateway/tcp_server.go b/go/vehicle-gateway/internal/gateway/tcp_server.go index 6154b54c..5b4bd78f 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server.go @@ -32,6 +32,10 @@ type TCPProtocol struct { Respond FrameResponder } +type connectionState struct { + platformName string +} + type TCPServer struct { protocol TCPProtocol sink eventbus.Sink @@ -155,6 +159,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { readBuffer := make([]byte, s.readBufferSize) var pending []byte + state := &connectionState{} for { _ = conn.SetReadDeadline(time.Now().Add(s.idleTimeout)) n, err := conn.Read(readBuffer) @@ -167,7 +172,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { } pending = remainder for _, frame := range frames { - s.handleFrame(ctx, conn, frame, source) + s.handleFrame(ctx, conn, frame, source, state) } } if err != nil { @@ -188,7 +193,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) { } } -func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string) { +func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string, state *connectionState) { frameCtx, cancelFrame := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout) defer cancelFrame() @@ -217,6 +222,7 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } else { env = resolved } + enrichConnectionPlatform(&env, state) } s.recordFrameMetric(env.ParseStatus) @@ -254,6 +260,30 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, } } +func enrichConnectionPlatform(env *envelope.FrameEnvelope, state *connectionState) { + if env == nil || state == nil || env.ParseStatus == envelope.ParseBadFrame { + return + } + if env.Fields == nil { + env.Fields = map[string]any{} + } + if current := strings.TrimSpace(fmt.Sprint(env.Fields["platform_account"])); current != "" && current != "" { + state.platformName = current + } + if state.platformName == "" { + return + } + if strings.TrimSpace(fmt.Sprint(env.Fields["platform_account"])) == "" || fmt.Sprint(env.Fields["platform_account"]) == "" { + env.Fields["platform_account"] = state.platformName + } + if env.Parsed == nil { + env.Parsed = map[string]any{} + } + if _, ok := env.Parsed["platform_name"]; !ok { + env.Parsed["platform_name"] = state.platformName + } +} + func (s *TCPServer) recordFrameMetric(status envelope.ParseStatus) { if s.metrics == nil { return diff --git a/go/vehicle-gateway/internal/gateway/tcp_server_test.go b/go/vehicle-gateway/internal/gateway/tcp_server_test.go index 0f6dc049..a5ee19af 100644 --- a/go/vehicle-gateway/internal/gateway/tcp_server_test.go +++ b/go/vehicle-gateway/internal/gateway/tcp_server_test.go @@ -74,7 +74,7 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) { t.Fatalf("NewTCPServer() error = %v", err) } - server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808") + server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{}) text := registry.Render() for _, want := range []string{ @@ -226,7 +226,7 @@ func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) { t.Fatalf("NewTCPServer() error = %v", err) } - server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808") + server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808", &connectionState{}) if resolver.ctxErr != nil { t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr) @@ -262,7 +262,7 @@ func TestTCPServerPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) { }, sink) server.publishUnified = true - server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808") + server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{}) if len(sink.raw) != 1 || len(sink.unified) != 1 { t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) diff --git a/go/vehicle-gateway/internal/realtime/mysql_query.go b/go/vehicle-gateway/internal/realtime/mysql_query.go index d6406a7e..8a0c5845 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query.go @@ -25,13 +25,16 @@ type RealtimeTableQuery struct { } type SnapshotRow struct { - Protocol string `json:"protocol"` - VIN string `json:"vin"` - Plate string `json:"plate"` - EventTime string `json:"event_time,omitempty"` - ReceivedAt string `json:"received_at,omitempty"` - EventID string `json:"event_id"` - UpdatedAt string `json:"updated_at"` + Protocol string `json:"protocol"` + VIN string `json:"vin"` + Plate string `json:"plate"` + PlatformName string `json:"platform_name,omitempty"` + Peer string `json:"peer,omitempty"` + ParsedJSON string `json:"parsed_json,omitempty"` + EventTime string `json:"event_time,omitempty"` + ReceivedAt string `json:"received_at,omitempty"` + EventID string `json:"event_id"` + UpdatedAt string `json:"updated_at"` } type LocationRow struct { @@ -73,7 +76,7 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable query = normalizeRealtimeTableQuery(query) sqlText, args := buildRealtimeSelectSQL( "vehicle_realtime_snapshot", - "protocol, vin, plate, event_time, received_at, event_id, updated_at", + "protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at", query, ) rows, err := r.db.QueryContext(ctx, sqlText, args...) @@ -86,9 +89,13 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable for rows.Next() { var row SnapshotRow var eventTime, receivedAt, updatedAt scanSQLDateTime - if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { + var parsedJSON sql.NullString + if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &row.PlatformName, &row.Peer, &parsedJSON, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { return nil, err } + if parsedJSON.Valid { + row.ParsedJSON = parsedJSON.String + } row.EventTime = eventTime.String row.ReceivedAt = receivedAt.String row.UpdatedAt = updatedAt.String diff --git a/go/vehicle-gateway/internal/realtime/mysql_query_test.go b/go/vehicle-gateway/internal/realtime/mysql_query_test.go index abfb9d79..17d5adbe 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query_test.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query_test.go @@ -20,12 +20,12 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) { mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_snapshot"). WithArgs("GB32960", "LB9A32A21R0LS1707"). WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) - mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). + mock.ExpectQuery("SELECT protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). WithArgs("GB32960", "LB9A32A21R0LS1707", 20, 0). WillReturnRows(sqlmock.NewRows([]string{ - "protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at", + "protocol", "vin", "plate", "platform_name", "peer", "parsed_json", "event_time", "received_at", "event_id", "updated_at", }).AddRow( - "GB32960", "LB9A32A21R0LS1707", "浙A12345", "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04", + "GB32960", "LB9A32A21R0LS1707", "浙A12345", "YueJin", "117.160.0.65:50945", `{"header":{"command":"0x02"}}`, "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04", )) handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db)) @@ -38,7 +38,7 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) { t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) } body := response.Body.String() - for _, want := range []string{`"protocol":"GB32960"`, `"vin":"LB9A32A21R0LS1707"`, `"plate":"浙A12345"`, `"total":1`, `"limit":20`} { + for _, want := range []string{`"protocol":"GB32960"`, `"vin":"LB9A32A21R0LS1707"`, `"plate":"浙A12345"`, `"platform_name":"YueJin"`, `"peer":"117.160.0.65:50945"`, `"parsed_json":"{\"header\":{\"command\":\"0x02\"}}"`, `"total":1`, `"limit":20`} { if !strings.Contains(body, want) { t.Fatalf("response missing %s: %s", want, body) } @@ -101,12 +101,12 @@ func TestSnapshotQueryHandlerSkipsTotalCountByDefault(t *testing.T) { } defer db.Close() - mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). + mock.ExpectQuery("SELECT protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). WithArgs("GB32960", 1, 0). WillReturnRows(sqlmock.NewRows([]string{ - "protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at", + "protocol", "vin", "plate", "platform_name", "peer", "parsed_json", "event_time", "received_at", "event_id", "updated_at", }).AddRow( - "GB32960", "LB9A32A21R0LS1707", "浙A12345", "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04", + "GB32960", "LB9A32A21R0LS1707", "浙A12345", "", "", nil, "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04", )) handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db)) diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 80cd2cce..68133ce9 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -3,6 +3,7 @@ package realtime import ( "context" "database/sql" + "encoding/json" "errors" "strings" "sync" @@ -102,6 +103,11 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil { return err } + for _, statement := range realtimeSnapshotCompatibilitySQL { + if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) { + return err + } + } if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { return err } @@ -122,10 +128,15 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) } eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) + platformName := platformNameFromEnvelope(env) + parsedJSON := parsedJSONFromEnvelope(env) if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, string(env.Protocol), vin, plate, + platformName, + env.SourceEndpoint, + parsedJSON, eventTime, receivedAt, env.StableEventID(), @@ -156,6 +167,43 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) return err } +func platformNameFromEnvelope(env envelope.FrameEnvelope) string { + if env.Fields != nil { + if value := strings.TrimSpace(stringValue(env.Fields["platform_account"])); value != "" { + return value + } + } + if env.Parsed != nil { + if value := strings.TrimSpace(stringValue(env.Parsed["platform_name"])); value != "" { + return value + } + if login, ok := env.Parsed["platform_login"].(map[string]any); ok { + return strings.TrimSpace(stringValue(login["username"])) + } + } + return "" +} + +func stringValue(value any) string { + switch typed := value.(type) { + case string: + return typed + default: + return "" + } +} + +func parsedJSONFromEnvelope(env envelope.FrameEnvelope) any { + if len(env.Parsed) == 0 { + return nil + } + data, err := json.Marshal(env.Parsed) + if err != nil { + return nil + } + return string(data) +} + func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.FrameEnvelope) (string, error) { if plate := strings.TrimSpace(env.Plate); plate != "" { return plate, nil @@ -331,6 +379,9 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn protocol VARCHAR(32) NOT NULL, vin VARCHAR(32) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '', + platform_name VARCHAR(64) NOT NULL DEFAULT '', + peer VARCHAR(128) NOT NULL DEFAULT '', + parsed_json LONGTEXT NULL, event_time DATETIME(3) NULL, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', @@ -340,12 +391,26 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn KEY idx_protocol_updated (protocol, updated_at) )` +var realtimeSnapshotCompatibilitySQL = []string{ + "ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name VARCHAR(64) NOT NULL DEFAULT '' AFTER plate", + "ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer VARCHAR(128) NOT NULL DEFAULT '' AFTER platform_name", + "ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json LONGTEXT NULL AFTER peer", +} + +func isDuplicateColumnError(err error) bool { + text := strings.ToLower(err.Error()) + return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060") +} + const upsertRealtimeSnapshotSQL = ` INSERT INTO vehicle_realtime_snapshot - (protocol, vin, plate, event_time, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?) + (protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE plate = IF(VALUES(plate) <> '', VALUES(plate), plate), + platform_name = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(platform_name), platform_name), + peer = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(peer), peer), + parsed_json = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(parsed_json), parsed_json), event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_time), event_time), received_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), VALUES(received_at), received_at), event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_id), event_id), diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 3dc0ed92..2973bb00 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -40,18 +40,18 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 3 { - t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + if len(exec.calls) != 6 { + t.Fatalf("exec calls = %d, want 6", len(exec.calls)) } if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") { t.Fatalf("schema query = %s", exec.calls[0].query) } - if !strings.Contains(exec.calls[1].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { - t.Fatalf("location schema query = %s", exec.calls[1].query) + if !strings.Contains(exec.calls[4].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { + t.Fatalf("location schema query = %s", exec.calls[4].query) } - for _, call := range exec.calls[:2] { - if strings.Contains(call.query, "parsed_json") || strings.Contains(call.query, "fields_json") { - t.Fatalf("realtime schema should not contain json payload columns: %s", call.query) + for _, call := range []snapshotExecCall{exec.calls[0], exec.calls[4]} { + if strings.Contains(call.query, "fields_json") { + t.Fatalf("realtime schema should not contain fields_json: %s", call.query) } for _, column := range []string{"id BIGINT", "created_at", "message_id", "sequence_id", "source_endpoint"} { if strings.Contains(call.query, column) { @@ -62,15 +62,20 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query) } } - if strings.Contains(exec.calls[1].query, "idx_location") { - t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[1].query) + for _, want := range []string{"platform_name", "peer", "parsed_json"} { + if !strings.Contains(exec.calls[0].query, want) { + t.Fatalf("snapshot schema should contain %s: %s", want, exec.calls[0].query) + } } - upsert := exec.calls[2] + if strings.Contains(exec.calls[4].query, "idx_location") { + t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[4].query) + } + upsert := exec.calls[5] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } - if strings.Contains(upsert.query, "parsed_json") || strings.Contains(upsert.query, "fields_json") { - t.Fatalf("snapshot upsert should not write json payload columns: %s", upsert.query) + if !strings.Contains(upsert.query, "parsed_json") || !strings.Contains(upsert.query, "platform_name") || !strings.Contains(upsert.query, "peer") { + t.Fatalf("snapshot upsert should write realtime context columns: %s", upsert.query) } for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { if strings.Contains(upsert.query, column) { @@ -83,8 +88,14 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if got, want := upsert.args[1], "VIN001"; got != want { t.Fatalf("vin arg = %#v, want %q", got, want) } - if len(upsert.args) != 6 { - t.Fatalf("snapshot upsert args = %d, want 6", len(upsert.args)) + if got, want := upsert.args[4], "1.2.3.4:32960"; got != want { + t.Fatalf("peer arg = %#v, want %q", got, want) + } + if got := upsert.args[5]; !strings.Contains(got.(string), `"data_units"`) { + t.Fatalf("parsed json arg = %#v", got) + } + if len(upsert.args) != 9 { + t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args)) } } @@ -98,6 +109,12 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot"). WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer"). + WillReturnResult(sqlmock.NewResult(0, 0)) + mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json"). + WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location"). WillReturnResult(sqlmock.NewResult(0, 0))