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

feat(go): enrich realtime snapshot context

Browse Source
This commit is contained in:
lingniu
2026-07-03 10:17:10 +08:00
parent 75fbd92de9
commit cabdac1fe2
6 changed files with 156 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
go/vehicle-gateway/internal/gateway/tcp_server.go
View File

@@ -32,6 +32,10 @@ type TCPProtocol struct {
Respond FrameResponder Respond FrameResponder
} }
type connectionState struct {
platformName string
}
type TCPServer struct { type TCPServer struct {
protocol TCPProtocol protocol TCPProtocol
sink eventbus.Sink sink eventbus.Sink
@@ -155,6 +159,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) {
readBuffer := make([]byte, s.readBufferSize) readBuffer := make([]byte, s.readBufferSize)
var pending []byte var pending []byte
state := &connectionState{}
for { for {
_ = conn.SetReadDeadline(time.Now().Add(s.idleTimeout)) _ = conn.SetReadDeadline(time.Now().Add(s.idleTimeout))
n, err := conn.Read(readBuffer) n, err := conn.Read(readBuffer)
@@ -167,7 +172,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) {
} }
pending = remainder pending = remainder
for _, frame := range frames { for _, frame := range frames {
s.handleFrame(ctx, conn, frame, source) s.handleFrame(ctx, conn, frame, source, state)
} }
} }
if err != nil { if err != nil {
@@ -188,7 +193,7 @@ func (s *TCPServer) handleConnection(ctx context.Context, conn net.Conn) {
} }
} }
func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string) { func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte, source string, state *connectionState) {
frameCtx, cancelFrame := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout) frameCtx, cancelFrame := context.WithTimeout(context.WithoutCancel(ctx), frameOperationTimeout)
defer cancelFrame() defer cancelFrame()
@@ -217,6 +222,7 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte,
} else { } else {
env = resolved env = resolved
} }
enrichConnectionPlatform(&env, state)
} }
s.recordFrameMetric(env.ParseStatus) s.recordFrameMetric(env.ParseStatus)
@@ -254,6 +260,30 @@ func (s *TCPServer) handleFrame(ctx context.Context, conn net.Conn, raw []byte,
} }
} }
func enrichConnectionPlatform(env *envelope.FrameEnvelope, state *connectionState) {
if env == nil || state == nil || env.ParseStatus == envelope.ParseBadFrame {
return
}
if env.Fields == nil {
env.Fields = map[string]any{}
}
if current := strings.TrimSpace(fmt.Sprint(env.Fields["platform_account"])); current != "" && current != "<nil>" {
state.platformName = current
}
if state.platformName == "" {
return
}
if strings.TrimSpace(fmt.Sprint(env.Fields["platform_account"])) == "" || fmt.Sprint(env.Fields["platform_account"]) == "<nil>" {
env.Fields["platform_account"] = state.platformName
}
if env.Parsed == nil {
env.Parsed = map[string]any{}
}
if _, ok := env.Parsed["platform_name"]; !ok {
env.Parsed["platform_name"] = state.platformName
}
}
func (s *TCPServer) recordFrameMetric(status envelope.ParseStatus) { func (s *TCPServer) recordFrameMetric(status envelope.ParseStatus) {
if s.metrics == nil { if s.metrics == nil {
return return

6
go/vehicle-gateway/internal/gateway/tcp_server_test.go
View File

@@ -74,7 +74,7 @@ func TestTCPServerRecordsFrameMetrics(t *testing.T) {
t.Fatalf("NewTCPServer() error = %v", err) t.Fatalf("NewTCPServer() error = %v", err)
} }
server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808") server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})
text := registry.Render() text := registry.Render()
for _, want := range []string{ for _, want := range []string{
@@ -226,7 +226,7 @@ func TestTCPServerUsesUncancelledFrameContextForAlreadyReadFrame(t *testing.T) {
t.Fatalf("NewTCPServer() error = %v", err) t.Fatalf("NewTCPServer() error = %v", err)
} }
server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808") server.handleFrame(parent, nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})
if resolver.ctxErr != nil { if resolver.ctxErr != nil {
t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr) t.Fatalf("resolver saw cancelled context: %v", resolver.ctxErr)
@@ -262,7 +262,7 @@ func TestTCPServerPublishesUnifiedWhenExplicitlyEnabled(t *testing.T) {
}, sink) }, sink)
server.publishUnified = true server.publishUnified = true
server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808") server.handleFrame(context.Background(), nil, []byte{0x01}, "127.0.0.1:808", &connectionState{})
if len(sink.raw) != 1 || len(sink.unified) != 1 { if len(sink.raw) != 1 || len(sink.unified) != 1 {
t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified)) t.Fatalf("raw=%d unified=%d", len(sink.raw), len(sink.unified))

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

@@ -25,13 +25,16 @@ type RealtimeTableQuery struct {
} }
type SnapshotRow struct { type SnapshotRow struct {
Protocol string `json:"protocol"` Protocol string `json:"protocol"`
VIN string `json:"vin"` VIN string `json:"vin"`
Plate string `json:"plate"` Plate string `json:"plate"`
EventTime string `json:"event_time,omitempty"` PlatformName string `json:"platform_name,omitempty"`
ReceivedAt string `json:"received_at,omitempty"` Peer string `json:"peer,omitempty"`
EventID string `json:"event_id"` ParsedJSON string `json:"parsed_json,omitempty"`
UpdatedAt string `json:"updated_at"` EventTime string `json:"event_time,omitempty"`
ReceivedAt string `json:"received_at,omitempty"`
EventID string `json:"event_id"`
UpdatedAt string `json:"updated_at"`
} }
type LocationRow struct { type LocationRow struct {
@@ -73,7 +76,7 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable
query = normalizeRealtimeTableQuery(query) query = normalizeRealtimeTableQuery(query)
sqlText, args := buildRealtimeSelectSQL( sqlText, args := buildRealtimeSelectSQL(
"vehicle_realtime_snapshot", "vehicle_realtime_snapshot",
"protocol, vin, plate, event_time, received_at, event_id, updated_at", "protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at",
query, query,
) )
rows, err := r.db.QueryContext(ctx, sqlText, args...) rows, err := r.db.QueryContext(ctx, sqlText, args...)
@@ -86,9 +89,13 @@ func (r *SnapshotQueryRepository) Query(ctx context.Context, query RealtimeTable
for rows.Next() { for rows.Next() {
var row SnapshotRow var row SnapshotRow
var eventTime, receivedAt, updatedAt scanSQLDateTime var eventTime, receivedAt, updatedAt scanSQLDateTime
if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { var parsedJSON sql.NullString
if err := rows.Scan(&row.Protocol, &row.VIN, &row.Plate, &row.PlatformName, &row.Peer, &parsedJSON, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil {
return nil, err return nil, err
} }
if parsedJSON.Valid {
row.ParsedJSON = parsedJSON.String
}
row.EventTime = eventTime.String row.EventTime = eventTime.String
row.ReceivedAt = receivedAt.String row.ReceivedAt = receivedAt.String
row.UpdatedAt = updatedAt.String row.UpdatedAt = updatedAt.String

14
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"). mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_snapshot").
WithArgs("GB32960", "LB9A32A21R0LS1707"). WithArgs("GB32960", "LB9A32A21R0LS1707").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1))
mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). mock.ExpectQuery("SELECT protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot").
WithArgs("GB32960", "LB9A32A21R0LS1707", 20, 0). WithArgs("GB32960", "LB9A32A21R0LS1707", 20, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at", "protocol", "vin", "plate", "platform_name", "peer", "parsed_json", "event_time", "received_at", "event_id", "updated_at",
}).AddRow( }).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:04", "GB32960", "LB9A32A21R0LS1707", "浙A12345", "YueJin", "117.160.0.65:50945", `{"header":{"command":"0x02"}}`, "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)) handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))
@@ -38,7 +38,7 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
} }
body := response.Body.String() body := response.Body.String()
for _, want := range []string{`"protocol":"GB32960"`, `"vin":"LB9A32A21R0LS1707"`, `"plate":"浙A12345"`, `"total":1`, `"limit":20`} { for _, want := range []string{`"protocol":"GB32960"`, `"vin":"LB9A32A21R0LS1707"`, `"plate":"浙A12345"`, `"platform_name":"YueJin"`, `"peer":"117.160.0.65:50945"`, `"parsed_json":"{\"header\":{\"command\":\"0x02\"}}"`, `"total":1`, `"limit":20`} {
if !strings.Contains(body, want) { if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body) t.Fatalf("response missing %s: %s", want, body)
} }
@@ -101,12 +101,12 @@ func TestSnapshotQueryHandlerSkipsTotalCountByDefault(t *testing.T) {
} }
defer db.Close() defer db.Close()
mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot"). mock.ExpectQuery("SELECT protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot").
WithArgs("GB32960", 1, 0). WithArgs("GB32960", 1, 0).
WillReturnRows(sqlmock.NewRows([]string{ WillReturnRows(sqlmock.NewRows([]string{
"protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at", "protocol", "vin", "plate", "platform_name", "peer", "parsed_json", "event_time", "received_at", "event_id", "updated_at",
}).AddRow( }).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:04", "GB32960", "LB9A32A21R0LS1707", "浙A12345", "", "", nil, "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)) handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))

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

@@ -3,6 +3,7 @@ package realtime
import ( import (
"context" "context"
"database/sql" "database/sql"
"encoding/json"
"errors" "errors"
"strings" "strings"
"sync" "sync"
@@ -102,6 +103,11 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error {
if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil { if _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL); err != nil {
return err return err
} }
for _, statement := range realtimeSnapshotCompatibilitySQL {
if _, err := w.exec.ExecContext(ctx, statement); err != nil && !isDuplicateColumnError(err) {
return err
}
}
if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil {
return err return err
} }
@@ -122,10 +128,15 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope)
} }
eventTime := nullableTime(env.EventTimeMS) eventTime := nullableTime(env.EventTimeMS)
receivedAt := nullableTime(env.ReceivedAtMS) receivedAt := nullableTime(env.ReceivedAtMS)
platformName := platformNameFromEnvelope(env)
parsedJSON := parsedJSONFromEnvelope(env)
if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, if _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL,
string(env.Protocol), string(env.Protocol),
vin, vin,
plate, plate,
platformName,
env.SourceEndpoint,
parsedJSON,
eventTime, eventTime,
receivedAt, receivedAt,
env.StableEventID(), env.StableEventID(),
@@ -156,6 +167,43 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope)
return err return err
} }
func platformNameFromEnvelope(env envelope.FrameEnvelope) string {
if env.Fields != nil {
if value := strings.TrimSpace(stringValue(env.Fields["platform_account"])); value != "" {
return value
}
}
if env.Parsed != nil {
if value := strings.TrimSpace(stringValue(env.Parsed["platform_name"])); value != "" {
return value
}
if login, ok := env.Parsed["platform_login"].(map[string]any); ok {
return strings.TrimSpace(stringValue(login["username"]))
}
}
return ""
}
func stringValue(value any) string {
switch typed := value.(type) {
case string:
return typed
default:
return ""
}
}
func parsedJSONFromEnvelope(env envelope.FrameEnvelope) any {
if len(env.Parsed) == 0 {
return nil
}
data, err := json.Marshal(env.Parsed)
if err != nil {
return nil
}
return string(data)
}
func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.FrameEnvelope) (string, error) { func (w *SnapshotWriter) plateForEnvelope(ctx context.Context, env envelope.FrameEnvelope) (string, error) {
if plate := strings.TrimSpace(env.Plate); plate != "" { if plate := strings.TrimSpace(env.Plate); plate != "" {
return plate, nil return plate, nil
@@ -331,6 +379,9 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn
protocol VARCHAR(32) NOT NULL, protocol VARCHAR(32) NOT NULL,
vin VARCHAR(32) NOT NULL DEFAULT '', vin VARCHAR(32) NOT NULL DEFAULT '',
plate VARCHAR(32) NOT NULL DEFAULT '', plate VARCHAR(32) NOT NULL DEFAULT '',
platform_name VARCHAR(64) NOT NULL DEFAULT '',
peer VARCHAR(128) NOT NULL DEFAULT '',
parsed_json LONGTEXT NULL,
event_time DATETIME(3) NULL, event_time DATETIME(3) NULL,
received_at DATETIME(3) NULL, received_at DATETIME(3) NULL,
event_id VARCHAR(64) NOT NULL DEFAULT '', event_id VARCHAR(64) NOT NULL DEFAULT '',
@@ -340,12 +391,26 @@ const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_sn
KEY idx_protocol_updated (protocol, updated_at) KEY idx_protocol_updated (protocol, updated_at)
)` )`
var realtimeSnapshotCompatibilitySQL = []string{
"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name VARCHAR(64) NOT NULL DEFAULT '' AFTER plate",
"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer VARCHAR(128) NOT NULL DEFAULT '' AFTER platform_name",
"ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json LONGTEXT NULL AFTER peer",
}
func isDuplicateColumnError(err error) bool {
text := strings.ToLower(err.Error())
return strings.Contains(text, "duplicate column") || strings.Contains(text, "1060")
}
const upsertRealtimeSnapshotSQL = ` const upsertRealtimeSnapshotSQL = `
INSERT INTO vehicle_realtime_snapshot INSERT INTO vehicle_realtime_snapshot
(protocol, vin, plate, event_time, received_at, event_id) (protocol, vin, plate, platform_name, peer, parsed_json, event_time, received_at, event_id)
VALUES (?, ?, ?, ?, ?, ?) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON DUPLICATE KEY UPDATE ON DUPLICATE KEY UPDATE
plate = IF(VALUES(plate) <> '', VALUES(plate), plate), plate = IF(VALUES(plate) <> '', VALUES(plate), plate),
platform_name = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(platform_name), platform_name),
peer = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(peer), peer),
parsed_json = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(parsed_json), parsed_json),
event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_time), event_time), event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_time), event_time),
received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(received_at), received_at), received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(received_at), received_at),
event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_id), event_id), event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_snapshot.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_snapshot.event_time), VALUES(event_id), event_id),

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

@@ -40,18 +40,18 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
t.Fatalf("Update() error = %v", err) t.Fatalf("Update() error = %v", err)
} }
if len(exec.calls) != 3 { if len(exec.calls) != 6 {
t.Fatalf("exec calls = %d, want 3", len(exec.calls)) t.Fatalf("exec calls = %d, want 6", len(exec.calls))
} }
if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") { if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") {
t.Fatalf("schema query = %s", exec.calls[0].query) t.Fatalf("schema query = %s", exec.calls[0].query)
} }
if !strings.Contains(exec.calls[1].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { if !strings.Contains(exec.calls[4].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") {
t.Fatalf("location schema query = %s", exec.calls[1].query) t.Fatalf("location schema query = %s", exec.calls[4].query)
} }
for _, call := range exec.calls[:2] { for _, call := range []snapshotExecCall{exec.calls[0], exec.calls[4]} {
if strings.Contains(call.query, "parsed_json") || strings.Contains(call.query, "fields_json") { if strings.Contains(call.query, "fields_json") {
t.Fatalf("realtime schema should not contain json payload columns: %s", call.query) t.Fatalf("realtime schema should not contain fields_json: %s", call.query)
} }
for _, column := range []string{"id BIGINT", "created_at", "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) { if strings.Contains(call.query, column) {
@@ -62,15 +62,20 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query) t.Fatalf("realtime current-state table should key by protocol/vin: %s", call.query)
} }
} }
if strings.Contains(exec.calls[1].query, "idx_location") { for _, want := range []string{"platform_name", "peer", "parsed_json"} {
t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[1].query) if !strings.Contains(exec.calls[0].query, want) {
t.Fatalf("snapshot schema should contain %s: %s", want, exec.calls[0].query)
}
} }
upsert := exec.calls[2] if strings.Contains(exec.calls[4].query, "idx_location") {
t.Fatalf("realtime location table should not keep unused geo index: %s", exec.calls[4].query)
}
upsert := exec.calls[5]
if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") {
t.Fatalf("upsert query = %s", upsert.query) t.Fatalf("upsert query = %s", upsert.query)
} }
if strings.Contains(upsert.query, "parsed_json") || strings.Contains(upsert.query, "fields_json") { if !strings.Contains(upsert.query, "parsed_json") || !strings.Contains(upsert.query, "platform_name") || !strings.Contains(upsert.query, "peer") {
t.Fatalf("snapshot upsert should not write json payload columns: %s", upsert.query) t.Fatalf("snapshot upsert should write realtime context columns: %s", upsert.query)
} }
for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} { for _, column := range []string{"message_id", "sequence_id", "source_endpoint"} {
if strings.Contains(upsert.query, column) { if strings.Contains(upsert.query, column) {
@@ -83,8 +88,14 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) {
if got, want := upsert.args[1], "VIN001"; got != want { if got, want := upsert.args[1], "VIN001"; got != want {
t.Fatalf("vin arg = %#v, want %q", got, want) t.Fatalf("vin arg = %#v, want %q", got, want)
} }
if len(upsert.args) != 6 { if got, want := upsert.args[4], "1.2.3.4:32960"; got != want {
t.Fatalf("snapshot upsert args = %d, want 6", len(upsert.args)) t.Fatalf("peer arg = %#v, want %q", got, want)
}
if got := upsert.args[5]; !strings.Contains(got.(string), `"data_units"`) {
t.Fatalf("parsed json arg = %#v", got)
}
if len(upsert.args) != 9 {
t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args))
} }
} }
@@ -98,6 +109,12 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) {
mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot"). mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot").
WillReturnResult(sqlmock.NewResult(0, 0)) WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN platform_name").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN peer").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("ALTER TABLE vehicle_realtime_snapshot ADD COLUMN parsed_json").
WillReturnResult(sqlmock.NewResult(0, 0))
mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location"). mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location").
WillReturnResult(sqlmock.NewResult(0, 0)) WillReturnResult(sqlmock.NewResult(0, 0))