diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 68133ce9..1e00c8b4 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -119,7 +119,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) if vin == "" { return nil } - if !hasRealtimePayload(env) { + if !isRealtimeSnapshotEvent(env) { return nil } plate, err := w.plateForEnvelope(ctx, env) @@ -365,16 +365,35 @@ func nullableTime(ms int64) any { return time.UnixMilli(ms) } -func hasRealtimePayload(env envelope.FrameEnvelope) bool { - if len(env.Fields) > 0 { - return true +func isRealtimeSnapshotEvent(env envelope.FrameEnvelope) bool { + switch env.Protocol { + case envelope.ProtocolGB32960: + return env.MessageID == "0x02" && hasGB32960RealtimeDataUnits(env) + case envelope.ProtocolJT808: + return env.MessageID == "0x0200" && hasLocationFields(env) + case envelope.ProtocolYutongMQTT: + return len(env.Fields) > 0 + default: + return false } +} + +func hasGB32960RealtimeDataUnits(env envelope.FrameEnvelope) bool { if units, ok := env.Parsed["data_units"].([]any); ok && len(units) > 0 { return true } + if units, ok := env.Parsed["data_units"].([]map[string]any); ok && len(units) > 0 { + return true + } return false } +func hasLocationFields(env envelope.FrameEnvelope) bool { + _, okLat := numberField(env.Fields, envelope.FieldLatitude) + _, okLon := numberField(env.Fields, envelope.FieldLongitude) + return okLat && okLon +} + const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot ( 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 2973bb00..f0e1c7ca 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -234,7 +234,7 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { VIN: "VIN001", EventTimeMS: 1782918600000, ReceivedAtMS: 1782918601000, - Parsed: map[string]any{"data_units": []any{}}, + Parsed: sampleGB32960RealtimeParsed(), Fields: map[string]any{ envelope.FieldLatitude: 30.590151, envelope.FieldLongitude: 121.069881, @@ -289,6 +289,7 @@ func TestSnapshotWriterCachesBindingPlateByVIN(t *testing.T) { VIN: "VIN001", EventTimeMS: 1782918600000, ReceivedAtMS: 1782918601000, + Parsed: sampleGB32960RealtimeParsed(), Fields: map[string]any{ envelope.FieldLatitude: 30.590151, envelope.FieldLongitude: 121.069881, @@ -355,7 +356,7 @@ func TestSnapshotWriterIgnoresMissingBindingPlate(t *testing.T) { VIN: "VIN001", EventTimeMS: 1782918600000, ReceivedAtMS: 1782918601000, - Parsed: map[string]any{"data_units": []any{}}, + Parsed: sampleGB32960RealtimeParsed(), Fields: map[string]any{envelope.FieldSOCPercent: 90}, }); err != nil { t.Fatalf("Update() error = %v", err) @@ -375,7 +376,7 @@ func TestSnapshotWriterReturnsUnexpectedPlateLookupError(t *testing.T) { VIN: "VIN001", EventTimeMS: 1782918600000, ReceivedAtMS: 1782918601000, - Parsed: map[string]any{"data_units": []any{}}, + Parsed: sampleGB32960RealtimeParsed(), Fields: map[string]any{envelope.FieldSOCPercent: 90}, }) if err == nil || !strings.Contains(err.Error(), "db down") { @@ -445,6 +446,78 @@ func TestSnapshotWriterSkipsGB32960HeartbeatWithoutRealtimePayload(t *testing.T) } } +func TestSnapshotWriterSkipsGB32960NonRealtimeEvenWithConnectionMetadata(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x07", + VIN: "ABCDE600000000009", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "header": map[string]any{"command": "0x07", "vin": "ABCDE600000000009"}, + "platform_name": "YueJin", + }, + Fields: map[string]any{ + "platform_account": "YueJin", + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + +func TestSnapshotWriterSkipsJT808NonLocationEvenWithIdentityFields(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0100", + VIN: "VIN001", + Plate: "粤A12345", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "registration": map[string]any{"plate": "粤A12345"}, + }, + Fields: map[string]any{ + "plate": "粤A12345", + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + +func TestSnapshotWriterSkipsMQTTWithoutActualDataFields(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolYutongMQTT, + MessageID: "MQTT", + VIN: "VIN001", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "topic": "/vehicle/VIN001/state", + "data": map[string]any{}, + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + func TestSnapshotWriterSkipsEmptyDataUnitsWithoutCoreFields(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec) @@ -468,6 +541,14 @@ type recordingSnapshotExec struct { calls []snapshotExecCall } +func sampleGB32960RealtimeParsed() map[string]any { + return map[string]any{ + "data_units": []any{ + map[string]any{"type": "0x01", "name": "vehicle", "value": map[string]any{"soc_percent": 90}}, + }, + } +} + type snapshotExecCall struct { query string args []any