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

feat(go): persist full raw payload chunks

Browse Source
This commit is contained in:
lingniu
2026-07-02 12:11:24 +08:00
parent 3a67f6e05f
commit 7a9a6f441a
7 changed files with 334 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -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

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

@@ -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 {

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

@@ -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,

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

@@ -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{

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

@@ -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,

2
go/vehicle-gateway/internal/protocol/jt808/parser.go
View File

@@ -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,

7
go/vehicle-gateway/internal/protocol/jt808/parser_test.go
View File

@@ -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)
}
}