From 7a9a6f441a33899b31f489e34caa4c0daa026566 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 12:11:24 +0800 Subject: [PATCH] feat(go): persist full raw payload chunks --- go/vehicle-gateway/internal/history/query.go | 138 +++++++++++++++++- .../internal/history/query_test.go | 44 ++++++ go/vehicle-gateway/internal/history/schema.go | 16 ++ go/vehicle-gateway/internal/history/writer.go | 106 +++++++++++++- .../internal/history/writer_test.go | 33 +++++ .../internal/protocol/jt808/parser.go | 2 - .../internal/protocol/jt808/parser_test.go | 7 +- 7 files changed, 334 insertions(+), 12 deletions(-) diff --git a/go/vehicle-gateway/internal/history/query.go b/go/vehicle-gateway/internal/history/query.go index 1f28a765..7dfb0358 100644 --- a/go/vehicle-gateway/internal/history/query.go +++ b/go/vehicle-gateway/internal/history/query.go @@ -206,7 +206,13 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([] row.MessageIDHex = fmt.Sprintf("0x%04X", row.MessageID) out = append(out, row) } - return out, rows.Err() + if err := rows.Err(); err != nil { + return nil, err + } + if err := r.hydratePayloadChunks(ctx, out); err != nil { + return nil, err + } + return out, nil } func (r *RawFrameRepository) Count(ctx context.Context, query RawFrameQuery) (int64, error) { @@ -347,6 +353,13 @@ func (r *RawFrameRepository) tableName() string { return r.database + ".raw_frames" } +func (r *RawFrameRepository) chunkTableName() string { + if r.database == "" { + return "raw_frame_payload_chunks" + } + return r.database + ".raw_frame_payload_chunks" +} + func (r *LocationRepository) tableName() string { if r.database == "" { return "vehicle_locations" @@ -423,6 +436,129 @@ func buildRawFrameCountSQL(table string, query RawFrameQuery) (string, []any) { return sqlText, nil } +type payloadChunkManifest struct { + Chunked bool `json:"chunked"` + PayloadKind string `json:"payload_kind"` + EventID string `json:"event_id"` + ChunkCount int `json:"chunk_count"` +} + +func (r *RawFrameRepository) hydratePayloadChunks(ctx context.Context, rows []RawFrameRow) error { + type target struct { + rowIndex int + kind string + eventID string + count int + } + targets := make([]target, 0) + eventSeen := map[string]struct{}{} + for index := range rows { + for _, candidate := range []struct { + kind string + value string + }{ + {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 { + continue + } + eventID := strings.TrimSpace(manifest.EventID) + if eventID == "" { + eventID = rows[index].EventID + } + targets = append(targets, target{ + rowIndex: index, + kind: candidate.kind, + eventID: eventID, + count: manifest.ChunkCount, + }) + eventSeen[eventID] = struct{}{} + } + } + if len(targets) == 0 { + return nil + } + + eventIDs := make([]string, 0, len(eventSeen)) + for eventID := range eventSeen { + eventIDs = append(eventIDs, eventID) + } + sqlText := `SELECT event_id, payload_kind, chunk_index, chunk_text FROM ` + r.chunkTableName() + + " WHERE event_id IN (" + quotedList(eventIDs) + ") ORDER BY event_id, payload_kind, chunk_index" + chunkRows, err := r.db.QueryContext(ctx, sqlText) + if err != nil { + return err + } + defer chunkRows.Close() + + chunks := map[string]map[string][]string{} + for chunkRows.Next() { + var eventID string + var kind string + var index int + var text string + if err := chunkRows.Scan(&eventID, &kind, &index, &text); err != nil { + return err + } + if chunks[eventID] == nil { + chunks[eventID] = map[string][]string{} + } + current := chunks[eventID][kind] + if index >= len(current) { + next := make([]string, index+1) + copy(next, current) + current = next + } + current[index] = text + chunks[eventID][kind] = current + } + if err := chunkRows.Err(); err != nil { + return err + } + + for _, target := range targets { + parts := chunks[target.eventID][target.kind] + if len(parts) != target.count { + continue + } + hydrated := strings.Join(parts, "") + switch target.kind { + case "raw_hex": + rows[target.rowIndex].RawHex = hydrated + case "raw_text": + rows[target.rowIndex].RawText = hydrated + case "parsed_json": + rows[target.rowIndex].ParsedJSON = hydrated + case "fields_json": + rows[target.rowIndex].FieldsJSON = hydrated + } + } + return nil +} + +func parsePayloadChunkManifest(value string) (payloadChunkManifest, bool) { + var manifest payloadChunkManifest + if !strings.Contains(value, `"chunked"`) { + return manifest, false + } + if err := json.Unmarshal([]byte(value), &manifest); err != nil { + return manifest, false + } + return manifest, manifest.Chunked +} + +func quotedList(values []string) string { + out := make([]string, 0, len(values)) + for _, value := range values { + out = append(out, "'"+quote(value)+"'") + } + return strings.Join(out, ", ") +} + func buildLocationSQL(table string, query LocationQuery) (string, []any) { where := locationWhere(query) sqlText := `SELECT ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, protocol, vehicle_key, vin, phone, device_id FROM ` + table diff --git a/go/vehicle-gateway/internal/history/query_test.go b/go/vehicle-gateway/internal/history/query_test.go index f804edc3..641826ee 100644 --- a/go/vehicle-gateway/internal/history/query_test.go +++ b/go/vehicle-gateway/internal/history/query_test.go @@ -57,6 +57,50 @@ func TestRawFrameRepositoryQueriesRawFramesWithFilters(t *testing.T) { } } +func TestRawFrameRepositoryHydratesChunkedParsedJSON(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + 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"). + 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", + "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", + "GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "", + )) + mock.ExpectQuery("SELECT event_id, payload_kind, chunk_index, chunk_text FROM lingniu_vehicle_ts.raw_frame_payload_chunks"). + WillReturnRows(sqlmock.NewRows([]string{"event_id", "payload_kind", "chunk_index", "chunk_text"}). + AddRow("event-oversized", "parsed_json", 0, `{"data_units":[`). + AddRow("event-oversized", "parsed_json", 1, `{"field":"value"}]}`)) + + repository := NewRawFrameRepository(db, "lingniu_vehicle_ts") + rows, err := repository.Query(context.Background(), RawFrameQuery{ + Protocol: "GB32960", + VIN: "LB9A32A21R0LS1707", + Limit: 1, + }) + if err != nil { + t.Fatalf("Query() error = %v", err) + } + if len(rows) != 1 { + t.Fatalf("row count = %d", len(rows)) + } + if want := `{"data_units":[{"field":"value"}]}`; rows[0].ParsedJSON != want { + t.Fatalf("hydrated parsed_json = %q, want %q", rows[0].ParsedJSON, want) + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestRawFrameHandlerReturnsRawFrames(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { diff --git a/go/vehicle-gateway/internal/history/schema.go b/go/vehicle-gateway/internal/history/schema.go index 3b48970e..e84cf54b 100644 --- a/go/vehicle-gateway/internal/history/schema.go +++ b/go/vehicle-gateway/internal/history/schema.go @@ -30,6 +30,22 @@ func SchemaStatements(database string) []string { vin NCHAR(32), phone NCHAR(32), device_id NCHAR(64) +)`, + `CREATE STABLE IF NOT EXISTS raw_frame_payload_chunks ( + ts TIMESTAMP, + event_id NCHAR(64), + frame_id NCHAR(64), + received_at TIMESTAMP, + payload_kind NCHAR(32), + chunk_index INT, + chunk_count INT, + chunk_text BINARY(16000) +) TAGS ( + protocol NCHAR(32), + vehicle_key NCHAR(64), + vin NCHAR(32), + phone NCHAR(32), + device_id NCHAR(64) )`, `CREATE STABLE IF NOT EXISTS vehicle_locations ( ts TIMESTAMP, diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go index e54b4553..c22a5dbe 100644 --- a/go/vehicle-gateway/internal/history/writer.go +++ b/go/vehicle-gateway/internal/history/writer.go @@ -11,6 +11,7 @@ import ( "strings" "sync" "time" + "unicode/utf8" "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" ) @@ -24,6 +25,18 @@ type Writer struct { cache tableCache } +const ( + rawFramePayloadInlineLimit = 12_000 + rawFramePayloadChunkSize = 16_000 +) + +type payloadChunk struct { + Kind string + Index int + Count int + Text string +} + type tableCache struct { mu sync.Mutex seen map[string]struct{} @@ -60,11 +73,35 @@ func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) if err := w.ensureChild(ctx, table, "raw_frames", env); err != nil { return err } + 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)))) - return err +VALUES (%s)`, table, joinLiterals(rawValues(env, rawHex, rawText, parsedJSON, fieldsJSON)))) + if err != nil { + return err + } + chunks := append(rawHexChunks, rawTextChunks...) + chunks = append(chunks, parsedChunks...) + chunks = append(chunks, fieldsChunks...) + if len(chunks) == 0 { + return nil + } + chunkTable := tableName("chunk", env) + if err := w.ensureChild(ctx, chunkTable, "raw_frame_payload_chunks", env); err != nil { + return err + } + for _, chunk := range chunks { + if _, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s + (ts, event_id, frame_id, received_at, payload_kind, chunk_index, chunk_count, chunk_text) +VALUES (%s)`, chunkTable, joinLiterals(chunkValues(env, chunk)))); err != nil { + return err + } + } + return nil } func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope) error { @@ -123,7 +160,7 @@ func (w *Writer) ensureChild(ctx context.Context, table string, stable string, e return nil } -func rawValues(env envelope.FrameEnvelope) []any { +func rawValues(env envelope.FrameEnvelope, rawHex string, rawText string, parsedJSON string, fieldsJSON string) []any { received := millis(env.ReceivedAtMS) eventTime := millis(env.EventTimeMS) return []any{ @@ -134,16 +171,71 @@ func rawValues(env envelope.FrameEnvelope) []any { eventTime, received, rawSizeBytes(env), - env.RawHex, - env.RawText, - jsonString(env.Parsed), - jsonString(env.Fields), + rawHex, + rawText, + parsedJSON, + fieldsJSON, string(env.ParseStatus), env.ParseError, env.SourceEndpoint, } } +func chunkValues(env envelope.FrameEnvelope, chunk payloadChunk) []any { + received := millis(env.ReceivedAtMS) + return []any{ + received, + env.StableEventID(), + frameID(env), + received, + chunk.Kind, + chunk.Index, + chunk.Count, + chunk.Text, + } +} + +func chunkPayload(env envelope.FrameEnvelope, kind string, value string) (string, []payloadChunk) { + if len(value) <= rawFramePayloadInlineLimit { + return value, nil + } + chunks := make([]payloadChunk, 0, (len(value)+rawFramePayloadChunkSize-1)/rawFramePayloadChunkSize) + for index, start := 0, 0; start < len(value); index++ { + end := safeChunkEnd(value, start, rawFramePayloadChunkSize) + chunks = append(chunks, payloadChunk{ + Kind: kind, + Index: index, + Text: value[start:end], + }) + start = end + } + count := len(chunks) + for index := range chunks { + chunks[index].Count = count + } + manifest := map[string]any{ + "chunked": true, + "payload_kind": kind, + "event_id": env.StableEventID(), + "chunk_count": count, + } + return jsonString(manifest), chunks +} + +func safeChunkEnd(value string, start int, maxBytes int) int { + end := start + maxBytes + if end >= len(value) { + return len(value) + } + for end > start && !utf8.RuneStart(value[end]) { + end-- + } + if end == start { + return start + maxBytes + } + return end +} + func locationValues(env envelope.FrameEnvelope, longitude float64, latitude float64) []any { received := millis(env.ReceivedAtMS) return []any{ diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go index 255d94e5..eff30475 100644 --- a/go/vehicle-gateway/internal/history/writer_test.go +++ b/go/vehicle-gateway/internal/history/writer_test.go @@ -54,6 +54,30 @@ func TestWriterAppendsRawLocationAndMileage(t *testing.T) { } } +func TestWriterChunksOversizedParsedJSON(t *testing.T) { + exec := &recordingExec{} + writer := NewWriter(exec) + env := sampleEnvelope() + env.Parsed = map[string]any{ + "large_payload": strings.Repeat("x", 16_128), + } + + if err := writer.AppendRawFrame(context.Background(), env); err != nil { + t.Fatalf("AppendRawFrame() error = %v", err) + } + + rawInsert := findSQL(exec.calls, "INSERT INTO raw_") + if !strings.Contains(rawInsert, `"chunked":true`) || !strings.Contains(rawInsert, `"payload_kind":"parsed_json"`) { + t.Fatalf("raw insert should store a parsed_json chunk manifest: %s", rawInsert) + } + if got := countSQL(exec.calls, "USING raw_frame_payload_chunks"); got != 1 { + t.Fatalf("chunk child create count = %d", got) + } + if got := countSQL(exec.calls, "INSERT INTO chunk_"); got < 2 { + t.Fatalf("chunk insert count = %d, calls=%v", got, exec.calls) + } +} + func TestWriterSkipsSparseDerivedRows(t *testing.T) { exec := &recordingExec{} writer := NewWriter(exec) @@ -87,6 +111,15 @@ func TestRawSizeBytesUsesRawTextWhenHexIsEmpty(t *testing.T) { } } +func TestSchemaStatementsCreateRawFramePayloadChunks(t *testing.T) { + statements := strings.Join(SchemaStatements("test_ts"), "\n") + for _, want := range []string{"raw_frame_payload_chunks", "payload_kind NCHAR(32)", "chunk_index INT", "chunk_text BINARY(16000)"} { + if !strings.Contains(statements, want) { + t.Fatalf("schema missing %q:\n%s", want, statements) + } + } +} + func sampleEnvelope() envelope.FrameEnvelope { return envelope.FrameEnvelope{ Protocol: envelope.ProtocolJT808, diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser.go b/go/vehicle-gateway/internal/protocol/jt808/parser.go index b6c5c9a4..f8d10ee9 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser.go @@ -122,9 +122,7 @@ func ParseFrame(raw []byte, receivedAtMS int64, sourceEndpoint string) (envelope "versioned": versioned, "properties": props, "body_size": bodySize, - "phone_bcd_hex": strings.ToUpper(hex.EncodeToString(phoneBCD)), "phone": phone, - "phone_raw": phoneRaw, "sequence": binary.BigEndian.Uint16(raw[serialStart : serialStart+2]), "subpackage": subpackage, "package_info": packageInfo, diff --git a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go index 69518039..8b9323a5 100644 --- a/go/vehicle-gateway/internal/protocol/jt808/parser_test.go +++ b/go/vehicle-gateway/internal/protocol/jt808/parser_test.go @@ -118,8 +118,11 @@ func TestParseFrameSupportsJT8082019VersionedHeader(t *testing.T) { if header["versioned"] != true || header["protocol_version"] != byte(1) { t.Fatalf("unexpected versioned header: %#v", header) } - if header["phone_bcd_hex"] != "00000000014894135060" { - t.Fatalf("phone bcd = %#v", header["phone_bcd_hex"]) + if _, ok := header["phone_bcd_hex"]; ok { + t.Fatalf("phone_bcd_hex should not be exposed: %#v", header) + } + if _, ok := header["phone_raw"]; ok { + t.Fatalf("phone_raw should not be exposed: %#v", header) } }