diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index d1da9c78..e979b87e 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -26,7 +26,10 @@ func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter { } func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { - _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL) + if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil { + return err + } + _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL) return err } @@ -48,7 +51,7 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) } eventTime := nullableTime(env.EventTimeMS) receivedAt := nullableTime(env.ReceivedAtMS) - _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, + if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, string(env.Protocol), vehicleKey, strings.TrimSpace(env.VIN), @@ -63,10 +66,140 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) fieldsJSON, receivedAt, env.StableEventID(), + ); err != nil { + return err + } + location, ok := realtimeLocationFromEnvelope(env, vehicleKey, fieldsJSON) + if !ok { + return nil + } + _, err = w.exec.ExecContext(ctx, upsertRealtimeLocationSQL, + location.Protocol, + location.VehicleKey, + location.VIN, + location.Phone, + location.DeviceID, + location.Plate, + location.MessageID, + location.Sequence, + location.SourceEndpoint, + location.EventTime, + location.Latitude, + location.Longitude, + location.SpeedKMH, + location.TotalMileageKM, + location.SOCPercent, + location.AltitudeM, + location.DirectionDeg, + location.AlarmFlag, + location.StatusFlag, + location.FieldsJSON, + location.ReceivedAt, + location.EventID, ) return err } +type realtimeLocationRow struct { + Protocol string + VehicleKey string + VIN string + Phone string + DeviceID string + Plate string + MessageID string + Sequence uint16 + SourceEndpoint string + EventTime any + Latitude float64 + Longitude float64 + SpeedKMH any + TotalMileageKM any + SOCPercent any + AltitudeM any + DirectionDeg any + AlarmFlag any + StatusFlag any + FieldsJSON string + ReceivedAt any + EventID string +} + +func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vehicleKey string, fieldsJSON string) (realtimeLocationRow, bool) { + latitude, okLat := numberField(env.Fields, envelope.FieldLatitude) + longitude, okLon := numberField(env.Fields, envelope.FieldLongitude) + if !okLat || !okLon { + return realtimeLocationRow{}, false + } + return realtimeLocationRow{ + Protocol: string(env.Protocol), + VehicleKey: vehicleKey, + VIN: strings.TrimSpace(env.VIN), + Phone: strings.TrimSpace(env.Phone), + DeviceID: strings.TrimSpace(env.DeviceID), + Plate: strings.TrimSpace(env.Plate), + MessageID: strings.TrimSpace(env.MessageID), + Sequence: env.Sequence, + SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), + EventTime: nullableTime(env.EventTimeMS), + Latitude: latitude, + Longitude: longitude, + SpeedKMH: nullableNumberField(env.Fields, envelope.FieldSpeedKMH), + TotalMileageKM: nullableNumberField(env.Fields, envelope.FieldTotalMileageKM), + SOCPercent: nullableNumberField(env.Fields, envelope.FieldSOCPercent), + AltitudeM: nullableNumberField(env.Fields, "altitude_m"), + DirectionDeg: nullableNumberField(env.Fields, "direction_deg"), + AlarmFlag: nullableNumberField(env.Fields, "alarm_flag"), + StatusFlag: nullableNumberField(env.Fields, "status_flag"), + FieldsJSON: fieldsJSON, + ReceivedAt: nullableTime(env.ReceivedAtMS), + EventID: env.StableEventID(), + }, true +} + +func nullableNumberField(fields map[string]any, key string) any { + value, ok := numberField(fields, key) + if !ok { + return nil + } + return value +} + +func numberField(fields map[string]any, key string) (float64, bool) { + value, ok := fields[key] + if !ok { + return 0, false + } + switch typed := value.(type) { + case float64: + return typed, true + case float32: + return float64(typed), true + case int: + return float64(typed), true + case int8: + return float64(typed), true + case int16: + return float64(typed), true + case int32: + return float64(typed), true + case int64: + return float64(typed), true + case uint: + return float64(typed), true + case uint8: + return float64(typed), true + case uint16: + return float64(typed), true + case uint32: + return float64(typed), true + case uint64: + return float64(typed), true + default: + return 0, false + } +} + func marshalObject(value map[string]any) (string, error) { if value == nil { value = map[string]any{} @@ -139,3 +272,67 @@ ON DUPLICATE KEY UPDATE event_id = VALUES(event_id), updated_at = CURRENT_TIMESTAMP ` + +const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + protocol VARCHAR(32) NOT NULL, + vehicle_key VARCHAR(96) NOT NULL, + vin VARCHAR(32) NOT NULL DEFAULT '', + phone VARCHAR(32) NOT NULL DEFAULT '', + device_id VARCHAR(96) NOT NULL DEFAULT '', + plate VARCHAR(32) NOT NULL DEFAULT '', + message_id VARCHAR(32) NOT NULL DEFAULT '', + sequence_id INT NOT NULL DEFAULT 0, + source_endpoint VARCHAR(128) NOT NULL DEFAULT '', + event_time DATETIME(3) NULL, + latitude DECIMAL(12,6) NOT NULL, + longitude DECIMAL(12,6) NOT NULL, + speed_kmh DECIMAL(10,3) NULL, + total_mileage_km DECIMAL(18,3) NULL, + soc_percent DECIMAL(6,2) NULL, + altitude_m DECIMAL(10,3) NULL, + direction_deg DECIMAL(10,3) NULL, + alarm_flag BIGINT NULL, + status_flag BIGINT NULL, + fields_json JSON NOT NULL, + received_at DATETIME(3) NULL, + event_id VARCHAR(64) NOT NULL DEFAULT '', + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_realtime_location_vehicle (protocol, vehicle_key), + KEY idx_vehicle_key (vehicle_key), + KEY idx_vin (vin), + KEY idx_protocol_updated (protocol, updated_at), + KEY idx_location (longitude, latitude) +)` + +const upsertRealtimeLocationSQL = ` +INSERT INTO vehicle_realtime_location + (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, + source_endpoint, event_time, latitude, longitude, speed_kmh, total_mileage_km, + soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, fields_json, + received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), ?, ?) +ON DUPLICATE KEY UPDATE + vin = IF(VALUES(vin) <> '', VALUES(vin), vin), + phone = IF(VALUES(phone) <> '', VALUES(phone), phone), + device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), + plate = IF(VALUES(plate) <> '', VALUES(plate), plate), + message_id = VALUES(message_id), + sequence_id = VALUES(sequence_id), + source_endpoint = VALUES(source_endpoint), + event_time = VALUES(event_time), + latitude = VALUES(latitude), + longitude = VALUES(longitude), + speed_kmh = VALUES(speed_kmh), + total_mileage_km = VALUES(total_mileage_km), + soc_percent = VALUES(soc_percent), + altitude_m = VALUES(altitude_m), + direction_deg = VALUES(direction_deg), + alarm_flag = VALUES(alarm_flag), + status_flag = VALUES(status_flag), + fields_json = VALUES(fields_json), + received_at = VALUES(received_at), + event_id = VALUES(event_id), + updated_at = CURRENT_TIMESTAMP +` diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 48374c39..5e07a1de 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -36,13 +36,16 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 2 { - t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + if len(exec.calls) != 3 { + t.Fatalf("exec calls = %d, want 3", 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) } - upsert := exec.calls[1] + 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) + } + upsert := exec.calls[2] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } @@ -61,6 +64,68 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) { } } +func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + MessageID: "0x0200", + Sequence: 9, + VIN: "VIN001", + Phone: "13307795425", + DeviceID: "device-1", + Plate: "沪A12345", + SourceEndpoint: "1.2.3.4:808", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + EventID: "event-1", + Parsed: map[string]any{"location": map[string]any{"latitude": 30.590151, "longitude": 121.069881}}, + Fields: map[string]any{ + envelope.FieldLatitude: 30.590151, + envelope.FieldLongitude: 121.069881, + envelope.FieldSpeedKMH: 23.0, + envelope.FieldTotalMileageKM: 10241.2, + envelope.FieldSOCPercent: 88, + "altitude_m": uint16(5), + "direction_deg": uint16(79), + "alarm_flag": uint32(0), + "status_flag": uint32(786435), + }, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + } + locationUpsert := exec.calls[1] + if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") { + t.Fatalf("location upsert query = %s", locationUpsert.query) + } + if got, want := locationUpsert.args[0], "JT808"; got != want { + t.Fatalf("protocol arg = %#v, want %q", got, want) + } + if got, want := locationUpsert.args[1], "VIN001"; got != want { + t.Fatalf("vehicle key arg = %#v, want %q", got, want) + } + if got, want := locationUpsert.args[10], 30.590151; got != want { + t.Fatalf("latitude arg = %#v, want %v", got, want) + } + if got, want := locationUpsert.args[11], 121.069881; got != want { + t.Fatalf("longitude arg = %#v, want %v", got, want) + } + if got, want := locationUpsert.args[12], 23.0; got != want { + t.Fatalf("speed arg = %#v, want %v", got, want) + } + if got, want := locationUpsert.args[13], 10241.2; got != want { + t.Fatalf("mileage arg = %#v, want %v", got, want) + } + if got, want := locationUpsert.args[14], 88.0; got != want { + t.Fatalf("soc arg = %#v, want %v", got, want) + } +} + func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) { exec := &recordingSnapshotExec{} writer := NewSnapshotWriter(exec)