shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(go): simplify realtime mysql current tables

Browse Source
This commit is contained in:
lingniu
2026-07-03 07:11:40 +08:00
parent 7c41b81654
commit e89c9fd98f
7 changed files with 31 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
go/vehicle-gateway/internal/realtime/mysql_query.go
View File

@@ -30,7 +30,6 @@ type SnapshotRow struct {
EventTime string `json:"event_time,omitempty"`
ReceivedAt string `json:"received_at,omitempty"`
EventID string `json:"event_id"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
@@ -50,7 +49,6 @@ type LocationRow struct {
StatusFlag *int64 `json:"status_flag,omitempty"`
ReceivedAt string `json:"received_at,omitempty"`
EventID string `json:"event_id"`
CreatedAt string `json:"created_at"`
UpdatedAt string `json:"updated_at"`
}
@@ -74,7 +72,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, created_at, updated_at",
"protocol, vin, plate, event_time, received_at, event_id, updated_at",
query,
)
rows, err := r.db.QueryContext(ctx, sqlText, args...)
@@ -86,13 +84,12 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable
out := make([]SnapshotRow, 0)
for rows.Next() {
var row SnapshotRow
var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime
if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &createdAt, &updatedAt); err != nil {
var eventTime, receivedAt, updatedAt scanSQLDateTime
if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil {
return nil, err
}
row.EventTime = eventTime.String
row.ReceivedAt = receivedAt.String
row.CreatedAt = createdAt.String
row.UpdatedAt = updatedAt.String
out = append(out, row)
}
@@ -119,7 +116,7 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable
query = normalizeRealtimeTableQuery(query)
sqlText, args := buildRealtimeSelectSQL(
"vehicle_realtime_location",
"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, created_at, updated_at",
"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, updated_at",
query,
)
rows, err := r.db.QueryContext(ctx, sqlText, args...)
@@ -131,7 +128,7 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable
out := make([]LocationRow, 0)
for rows.Next() {
var row LocationRow
var eventTime, receivedAt, createdAt, updatedAt scanSQLDateTime
var eventTime, receivedAt, updatedAt scanSQLDateTime
var speed, mileage, soc, altitude, direction sql.NullFloat64
var alarm, status sql.NullInt64
if err := rows.Scan(
@@ -150,7 +147,6 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable
&status,
&receivedAt,
&row.EventID,
&createdAt,
&updatedAt,
); err != nil {
return nil, err
@@ -164,7 +160,6 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable
row.AlarmFlag = nullableInt(alarm)
row.StatusFlag = nullableInt(status)
row.ReceivedAt = receivedAt.String
row.CreatedAt = createdAt.String
row.UpdatedAt = updatedAt.String
out = append(out, row)
}

18
go/vehicle-gateway/internal/realtime/mysql_query_test.go
View File

@@ -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, created_at, updated_at FROM vehicle_realtime_snapshot").
mock.ExpectQuery("SELECT protocol, vin, plate, 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", "created_at", "updated_at",
"protocol", "vin", "plate", "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:03", "2026-07-02 16:01:04",
"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",
))
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))
@@ -43,6 +43,9 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) {
t.Fatalf("response missing %s: %s", want, body)
}
}
if strings.Contains(body, "created_at") {
t.Fatalf("snapshot response should not expose created_at: %s", body)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
@@ -58,14 +61,14 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) {
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_location").
WithArgs("JT808", "粤B98765").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1))
mock.ExpectQuery("SELECT 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, created_at, updated_at FROM vehicle_realtime_location").
mock.ExpectQuery("SELECT 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, updated_at FROM vehicle_realtime_location").
WithArgs("JT808", "粤B98765", 10, 10).
WillReturnRows(sqlmock.NewRows([]string{
"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", "created_at", "updated_at",
"soc_percent", "altitude_m", "direction_deg", "alarm_flag", "status_flag", "received_at", "event_id", "updated_at",
}).AddRow(
"JT808", "LKLG7C4E3NA774736", "粤B98765", "2026-07-02 16:11:02.000", 30.123456, 120.654321, 54.3, 48798.9,
nil, 19.0, 88.0, int64(0), int64(3), "2026-07-02 16:11:03.000", "evt-2", "2026-07-02 16:11:03", "2026-07-02 16:11:04",
nil, 19.0, 88.0, int64(0), int64(3), "2026-07-02 16:11:03.000", "evt-2", "2026-07-02 16:11:04",
))
handler := NewLocationQueryHandler(NewLocationQueryRepository(db))
@@ -83,6 +86,9 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) {
t.Fatalf("response missing %s: %s", want, body)
}
}
if strings.Contains(body, "created_at") {
t.Fatalf("location response should not expose created_at: %s", body)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}

2
go/vehicle-gateway/internal/realtime/openapi.go
View File

@@ -93,7 +93,6 @@ func realtimeOpenAPISpec() map[string]any {
"event_time": stringSchema("2026-07-02 16:01:02.123"),
"received_at": stringSchema("2026-07-02 16:01:03.456"),
"event_id": stringSchema("event id"),
"created_at": stringSchema("2026-07-02 16:01:03"),
"updated_at": stringSchema("2026-07-02 16:01:04"),
},
},
@@ -115,7 +114,6 @@ func realtimeOpenAPISpec() map[string]any {
"status_flag": integerSchema(3),
"received_at": stringSchema("2026-07-02 16:11:03.000"),
"event_id": stringSchema("event id"),
"created_at": stringSchema("2026-07-02 16:11:03"),
"updated_at": stringSchema("2026-07-02 16:11:04"),
},
},

8
go/vehicle-gateway/internal/realtime/snapshot_writer.go
View File

@@ -264,16 +264,14 @@ func hasRealtimePayload(env envelope.FrameEnvelope) bool {
}
const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
protocol VARCHAR(32) NOT NULL,
vin VARCHAR(32) NOT NULL DEFAULT '',
plate VARCHAR(32) NOT NULL DEFAULT '',
event_time DATETIME(3) 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_snapshot_vin (protocol, vin),
PRIMARY KEY (protocol, vin),
KEY idx_vin (vin),
KEY idx_protocol_updated (protocol, updated_at)
)`
@@ -291,7 +289,6 @@ ON DUPLICATE KEY UPDATE
`
const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_location (
id BIGINT PRIMARY KEY AUTO_INCREMENT,
protocol VARCHAR(32) NOT NULL,
vin VARCHAR(32) NOT NULL DEFAULT '',
plate VARCHAR(32) NOT NULL DEFAULT '',
@@ -307,9 +304,8 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo
status_flag BIGINT 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_vin (protocol, vin),
PRIMARY KEY (protocol, vin),
KEY idx_vin (vin),
KEY idx_protocol_updated (protocol, updated_at),
KEY idx_location (longitude, latitude)

5
go/vehicle-gateway/internal/realtime/snapshot_writer_test.go
View File

@@ -52,11 +52,14 @@ 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"} {
for _, column := range []string{"id BIGINT", "created_at", "message_id", "sequence_id", "source_endpoint"} {
if strings.Contains(call.query, column) {
t.Fatalf("realtime schema should not contain %s: %s", column, call.query)
}
}
if !strings.Contains(call.query, "PRIMARY KEY (protocol, vin)") {
t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query)
}
}
upsert := exec.calls[2]
if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {