diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 1307f2e7..ef7ea7d8 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -62,9 +62,6 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) string(env.Protocol), vin, plate, - strings.TrimSpace(env.MessageID), - env.Sequence, - strings.TrimSpace(env.SourceEndpoint), eventTime, receivedAt, env.StableEventID(), @@ -79,9 +76,6 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) location.Protocol, location.VIN, location.Plate, - location.MessageID, - location.Sequence, - location.SourceEndpoint, location.EventTime, location.Latitude, location.Longitude, @@ -136,10 +130,16 @@ func (w *SnapshotWriter) dropObsoleteColumns(ctx context.Context) error { {table: "vehicle_realtime_snapshot", name: "vehicle_key"}, {table: "vehicle_realtime_snapshot", name: "phone"}, {table: "vehicle_realtime_snapshot", name: "device_id"}, + {table: "vehicle_realtime_snapshot", name: "message_id"}, + {table: "vehicle_realtime_snapshot", name: "sequence_id"}, + {table: "vehicle_realtime_snapshot", name: "source_endpoint"}, {table: "vehicle_realtime_location", name: "fields_json"}, {table: "vehicle_realtime_location", name: "vehicle_key"}, {table: "vehicle_realtime_location", name: "phone"}, {table: "vehicle_realtime_location", name: "device_id"}, + {table: "vehicle_realtime_location", name: "message_id"}, + {table: "vehicle_realtime_location", name: "sequence_id"}, + {table: "vehicle_realtime_location", name: "source_endpoint"}, } { exists, err := columnExists(ctx, queryer, column.table, column.name) if err != nil { @@ -237,9 +237,6 @@ type realtimeLocationRow struct { Protocol string VIN string Plate string - MessageID string - Sequence uint16 - SourceEndpoint string EventTime any Latitude float64 Longitude float64 @@ -264,9 +261,6 @@ func realtimeLocationFromEnvelope(env envelope.FrameEnvelope, vin string, plate Protocol: string(env.Protocol), VIN: vin, Plate: plate, - MessageID: strings.TrimSpace(env.MessageID), - Sequence: env.Sequence, - SourceEndpoint: strings.TrimSpace(env.SourceEndpoint), EventTime: nullableTime(env.EventTimeMS), Latitude: latitude, Longitude: longitude, @@ -394,9 +388,6 @@ 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 '', - 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, received_at DATETIME(3) NULL, event_id VARCHAR(64) NOT NULL DEFAULT '', @@ -409,13 +400,10 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn const upsertRealtimeSnapshotSQL = ` INSERT INTO vehicle_realtime_snapshot - (protocol, vin, plate, message_id, sequence_id, source_endpoint, event_time, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + (protocol, vin, plate, event_time, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE 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), received_at = VALUES(received_at), event_id = VALUES(event_id), @@ -427,9 +415,6 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo protocol VARCHAR(32) NOT NULL, vin VARCHAR(32) 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, @@ -452,15 +437,11 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo const upsertRealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location - (protocol, vin, plate, message_id, sequence_id, source_endpoint, event_time, - latitude, longitude, speed_kmh, total_mileage_km, + (protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id) -VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE 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), diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 237ae8a2..0251c7e8 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -50,6 +50,11 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { 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 _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { + if strings.Contains(call.query, column) { + t.Fatalf("realtime schema should not contain %s: %s", column, call.query) + } + } } upsert := exec.calls[2] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { @@ -58,14 +63,19 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { 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) } + for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { + if strings.Contains(upsert.query, column) { + t.Fatalf("snapshot upsert should not write %s: %s", column, upsert.query) + } + } if got, want := upsert.args[0], "GB32960"; got != want { t.Fatalf("protocol arg = %#v, want %q", got, want) } if got, want := upsert.args[1], "VIN001"; got != want { t.Fatalf("vin arg = %#v, want %q", got, want) } - if len(upsert.args) != 9 { - t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args)) + if len(upsert.args) != 6 { + t.Fatalf("snapshot upsert args = %d, want 6", len(upsert.args)) } } @@ -111,29 +121,34 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) if strings.Contains(locationUpsert.query, "fields_json") { t.Fatalf("location upsert should not write fields_json: %s", locationUpsert.query) } + for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { + if strings.Contains(locationUpsert.query, column) { + t.Fatalf("location upsert should not write %s: %s", column, 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("vin arg = %#v, want %q", got, want) } - if got, want := locationUpsert.args[7], 30.590151; got != want { + if got, want := locationUpsert.args[4], 30.590151; got != want { t.Fatalf("latitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[8], 121.069881; got != want { + if got, want := locationUpsert.args[5], 121.069881; got != want { t.Fatalf("longitude arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[9], 23.0; got != want { + if got, want := locationUpsert.args[6], 23.0; got != want { t.Fatalf("speed arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[10], 10241.2; got != want { + if got, want := locationUpsert.args[7], 10241.2; got != want { t.Fatalf("mileage arg = %#v, want %v", got, want) } - if got, want := locationUpsert.args[11], 88.0; got != want { + if got, want := locationUpsert.args[8], 88.0; got != want { t.Fatalf("soc arg = %#v, want %v", got, want) } - if len(locationUpsert.args) != 18 { - t.Fatalf("location upsert args = %d, want 18", len(locationUpsert.args)) + if len(locationUpsert.args) != 15 { + t.Fatalf("location upsert args = %d, want 15", len(locationUpsert.args)) } }