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

refactor(go): remove duplicate raw fields json

Browse Source
This commit is contained in:
lingniu
2026-07-02 19:47:32 +08:00
parent 0bbe87e6fc
commit 29cd044a57
7 changed files with 40 additions and 77 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
go/vehicle-gateway/internal/history/query.go
View File

@@ -43,7 +43,6 @@ type RawFrameRow struct {
RawHex string `json:"raw_hex,omitempty"`
RawText string `json:"raw_text,omitempty"`
ParsedJSON string `json:"parsed_json,omitempty"`
FieldsJSON string `json:"fields_json,omitempty"`
ParseStatus string `json:"parse_status"`
ParseError string `json:"parse_error,omitempty"`
SourceEndpoint string `json:"source_endpoint"`
@@ -188,7 +187,6 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
&row.RawHex,
&row.RawText,
&row.ParsedJSON,
&row.FieldsJSON,
&row.ParseStatus,
&row.ParseError,
&row.SourceEndpoint,
@@ -420,7 +418,7 @@ func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery {
func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) {
where := rawFrameWhere(query)
sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table
sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table
if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
@@ -460,7 +458,6 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra
{kind: "raw_hex", value: rows[index].RawHex},
{kind: "raw_text", value: rows[index].RawText},
{kind: "parsed_json", value: rows[index].ParsedJSON},
{kind: "fields_json", value: rows[index].FieldsJSON},
} {
manifest, ok := parsePayloadChunkManifest(candidate.value)
if !ok || manifest.PayloadKind != candidate.kind || manifest.ChunkCount <= 0 {
@@ -533,8 +530,6 @@ func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []Ra
rows[target.rowIndex].RawText = hydrated
case "parsed_json":
rows[target.rowIndex].ParsedJSON = hydrated
case "fields_json":
rows[target.rowIndex].FieldsJSON = hydrated
}
}
return nil

33
go/vehicle-gateway/internal/history/query_test.go
View File

@@ -18,17 +18,17 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
"go_frame", "event-1", 0x0200,
time.Date(2026, 7, 1, 23, 25, 36, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
time.Date(2026, 7, 1, 23, 25, 37, 0, time.FixedZone("Asia/Shanghai", 8*3600)),
64, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"speed_kmh":0}`,
64, "7E0200", "", `{"header":{"message_id":"0x0200"}}`,
"OK", "", "222.66.200.68:29646", "JT808", "013079963379", "LKLG7C4E3NA774736", "013079963379", "9963379",
))
@@ -65,15 +65,15 @@ func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) {
defer db.Close()
manifest := `{"chunked":true,"payload_kind":"parsed_json","event_id":"event-oversized","chunk_count":2}`
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
"2026-07-02 10:00:00", "go_frame", "event-oversized", 2,
"2026-07-02 10:00:00", "2026-07-02 10:00:00", 4096,
"2323", "", manifest, `{"speed_kmh":0}`, "OK", "", "8.134.95.166:53702",
"2323", "", manifest, "OK", "", "8.134.95.166:53702",
"GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "",
))
mock.ExpectQuery("SELECT event_id, payload_kind, chunk_index, chunk_text FROM lingniu_vehicle_ts.raw_frame_payload_chunks").
@@ -109,14 +109,14 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) {
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(38))
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
"2026-07-01 22:28:25", "go_frame", "event-2", 2, "2026-07-01 22:28:25", "2026-07-01 22:28:25",
128, "2323", "", `{"command":"REALTIME"}`, `{"total_mileage_km":53490.9}`,
128, "2323", "", `{"command":"REALTIME"}`,
"OK", "", "8.134.95.166:53702", "GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "",
))
@@ -135,6 +135,9 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) {
t.Fatalf("response missing %s: %s", want, body)
}
}
if strings.Contains(body, "fields_json") {
t.Fatalf("response should not expose duplicate fields_json: %s", body)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
@@ -151,11 +154,11 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) {
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
"2026-07-02 00:18:22", "go_frame", "event-3", 0x0200, "2026-07-02 00:18:22", "2026-07-02 00:22:43",
63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"total_mileage_km":8792.8}`,
63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`,
"OK", "", "115.231.168.135:22170", "JT808", "JT808:013307811350", "", "013307811350", "",
))
@@ -188,11 +191,11 @@ func TestRawFrameHandlerCanSkipTotalCountForFreshnessProbe(t *testing.T) {
mock.ExpectQuery("ORDER BY received_at DESC LIMIT 1 OFFSET 0").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
"2026-07-02 01:26:48", "go_frame", "event-4", 0x0200, "2026-07-02 01:20:46", "2026-07-02 01:26:48",
63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`, `{"speed_kmh":0}`,
63, "7E0200", "", `{"header":{"message_id":"0x0200"}}`,
"OK", "", "115.231.168.135:22170", "JT808", "JT808:013307811254", "", "013307811254", "",
))
@@ -224,10 +227,10 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) {
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(0))
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_json", "fields_json", "parse_status", "parse_error", "source_endpoint",
"raw_hex", "raw_text", "parsed_json", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}))

1
go/vehicle-gateway/internal/history/schema.go
View File

@@ -20,7 +20,6 @@ func SchemaStatements(database string) []string {
raw_hex BINARY(16374),
raw_text BINARY(16374),
parsed_json BINARY(16374),
fields_json BINARY(4096),
parse_status NCHAR(16),
parse_error BINARY(1024),
source_endpoint NCHAR(128)

9
go/vehicle-gateway/internal/history/writer.go
View File

@@ -73,17 +73,15 @@ func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope)
rawHex, rawHexChunks := chunkPayload(env, "raw_hex", env.RawHex)
rawText, rawTextChunks := chunkPayload(env, "raw_text", env.RawText)
parsedJSON, parsedChunks := chunkPayload(env, "parsed_json", jsonString(env.Parsed))
fieldsJSON, fieldsChunks := chunkPayload(env, "fields_json", jsonString(env.Fields))
_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
(ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint)
VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON, fieldsJSON))))
raw_hex, raw_text, parsed_json, parse_status, parse_error, source_endpoint)
VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON))))
if err != nil {
return err
}
chunks := append(rawHexChunks, rawTextChunks...)
chunks = append(chunks, parsedChunks...)
chunks = append(chunks, fieldsChunks...)
if len(chunks) == 0 {
return nil
}
@@ -147,7 +145,7 @@ func (w *Writer) ensureChild(ctx context.Context, table string, stable string, e
return nil
}
func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string, fieldsJSON string) []any {
func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string) []any {
received := millis(env.ReceivedAtMS)
eventTime := millis(env.EventTimeMS)
return []any{
@@ -161,7 +159,6 @@ func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsed
rawHex,
rawText,
parsedJSON,
fieldsJSON,
string(env.ParseStatus),
env.ParseError,
env.SourceEndpoint,

6
go/vehicle-gateway/internal/history/writer_test.go
View File

@@ -19,6 +19,9 @@ func TestSchemaStatementsCreateCoreStables(t *testing.T) {
if strings.Contains(statements, "vehicle_mileage_points") {
t.Fatalf("schema should not create duplicate mileage stable:\n%s", statements)
}
if strings.Contains(statements, "fields_json") {
t.Fatalf("raw schema should not create duplicate fields_json column:\n%s", statements)
}
}
func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
@@ -55,6 +58,9 @@ func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
if strings.Contains(rawInsert, "VALUES (?,") {
t.Fatalf("raw insert should not use placeholders for tdengine websocket plain insert: %s", rawInsert)
}
if strings.Contains(rawInsert, "fields_json") {
t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert)
}
}
func TestWriterNormalizesPhoneTagAndRefreshesExistingChildTags(t *testing.T) {