From dac6718ee4a21e853841b678b1c5d6259d4cd0a7 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 16:06:29 +0800 Subject: [PATCH] feat(go): persist realtime snapshots to mysql --- go/vehicle-gateway/cmd/realtime-api/main.go | 40 +++++- .../cmd/realtime-api/main_test.go | 14 ++ .../internal/realtime/snapshot_writer.go | 128 ++++++++++++++++++ .../internal/realtime/snapshot_writer_test.go | 90 ++++++++++++ 4 files changed, 265 insertions(+), 7 deletions(-) create mode 100644 go/vehicle-gateway/internal/realtime/snapshot_writer.go create mode 100644 go/vehicle-gateway/internal/realtime/snapshot_writer_test.go diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index e55f56d3..c0e978bf 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -44,15 +44,11 @@ func main() { repository := realtime.NewRepository(client, realtime.Config{ OnlineTTL: time.Duration(envInt("ONLINE_TTL_SECONDS", 600)) * time.Second, }) - if brokers := splitCSV(os.Getenv("KAFKA_BROKERS")); len(brokers) > 0 { - go consumeKafka(ctx, logger, repository, brokers) - } else { - logger.Warn("KAFKA_BROKERS is empty; realtime api will serve existing redis data only") - } mux := http.NewServeMux() mux.Handle("/api/realtime/vehicles/", realtime.NewHandler(repository)) closeStats := func() {} + var updater realtimeUpdater = repository if dsn := strings.TrimSpace(os.Getenv("MYSQL_DSN")); dsn != "" { db, err := sql.Open("mysql", dsn) if err != nil { @@ -65,6 +61,16 @@ func main() { os.Exit(1) } closeStats = func() { _ = db.Close() } + snapshotWriter := realtime.NewSnapshotWriter(db) + if env("MYSQL_REALTIME_SNAPSHOT_ENABLED", "true") != "false" { + if err := snapshotWriter.EnsureSchema(ctx); err != nil { + _ = db.Close() + logger.Error("realtime snapshot mysql schema bootstrap failed", "error", err) + os.Exit(1) + } + updater = compositeRealtimeUpdater{primary: repository, secondary: snapshotWriter} + logger.Info("realtime mysql snapshot enabled", "table", "vehicle_realtime_snapshot") + } mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) logger.Info("stats mysql query enabled") } else { @@ -76,6 +82,11 @@ func main() { logger.Warn("MYSQL_DSN is empty; stats query api disabled") } defer closeStats() + if brokers := splitCSV(os.Getenv("KAFKA_BROKERS")); len(brokers) > 0 { + go consumeKafka(ctx, logger, updater, brokers) + } else { + logger.Warn("KAFKA_BROKERS is empty; realtime api will serve existing redis data only") + } closeHistory := func() {} if dsn := strings.TrimSpace(os.Getenv("TDENGINE_DSN")); dsn != "" { driver := env("TDENGINE_DRIVER", "taosWS") @@ -131,7 +142,7 @@ func consumeKafka(ctx context.Context, logger interface { Info(string, ...any) Error(string, ...any) Warn(string, ...any) -}, repository *realtime.Repository, brokers []string) { +}, updater realtimeUpdater, brokers []string) { reader := kafka.NewReader(kafka.ReaderConfig{ Brokers: brokers, GroupID: env("KAFKA_GROUP", "go-realtime-api"), @@ -150,7 +161,7 @@ func consumeKafka(ctx context.Context, logger interface { logger.Error("kafka fetch failed", "error", err) continue } - processRealtimeMessage(ctx, logger, repository, reader, message) + processRealtimeMessage(ctx, logger, updater, reader, message) } } @@ -160,6 +171,21 @@ type realtimeUpdater interface { Update(context.Context, envelope.FrameEnvelope) error } +type compositeRealtimeUpdater struct { + primary realtimeUpdater + secondary realtimeUpdater +} + +func (u compositeRealtimeUpdater) Update(ctx context.Context, env envelope.FrameEnvelope) error { + if err := u.primary.Update(ctx, env); err != nil { + return err + } + if u.secondary == nil { + return nil + } + return u.secondary.Update(ctx, env) +} + type kafkaMessageCommitter interface { CommitMessages(context.Context, ...kafka.Message) error } diff --git a/go/vehicle-gateway/cmd/realtime-api/main_test.go b/go/vehicle-gateway/cmd/realtime-api/main_test.go index d7102f7a..7a1c2fda 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go @@ -42,6 +42,20 @@ func TestProcessRealtimeMessageUsesUncancelledContextForUpdateAndCommit(t *testi } } +func TestCompositeRealtimeUpdaterUpdatesBothStores(t *testing.T) { + env := envelope.FrameEnvelope{Protocol: envelope.ProtocolGB32960, VIN: "VIN001"} + redisUpdater := &contextCheckingRealtimeUpdater{} + snapshotUpdater := &contextCheckingRealtimeUpdater{} + updater := compositeRealtimeUpdater{primary: redisUpdater, secondary: snapshotUpdater} + + if err := updater.Update(context.Background(), env); err != nil { + t.Fatalf("Update() error = %v", err) + } + if redisUpdater.count != 1 || snapshotUpdater.count != 1 { + t.Fatalf("updates primary=%d secondary=%d", redisUpdater.count, snapshotUpdater.count) + } +} + type contextCheckingRealtimeUpdater struct { ctxErr error count int diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go new file mode 100644 index 00000000..b7bf8c40 --- /dev/null +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -0,0 +1,128 @@ +package realtime + +import ( + "context" + "database/sql" + "encoding/json" + "strings" + "time" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type SnapshotExecer interface { + ExecContext(context.Context, string, ...any) (sql.Result, error) +} + +type SnapshotWriter struct { + exec SnapshotExecer +} + +func NewSnapshotWriter(exec SnapshotExecer) *SnapshotWriter { + if exec == nil { + panic("snapshot execer must not be nil") + } + return &SnapshotWriter{exec: exec} +} + +func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { + _, err := w.exec.ExecContext(ctx, realtimeSnapshotTableSQL) + return err +} + +func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) error { + vehicleKey := strings.TrimSpace(env.VehicleKey()) + if vehicleKey == "" || strings.HasSuffix(vehicleKey, ":unknown") { + return nil + } + fieldsJSON, err := marshalObject(env.Fields) + if err != nil { + return err + } + parsedJSON, err := marshalObject(env.Parsed) + if err != nil { + return err + } + eventTime := nullableTime(env.EventTimeMS) + receivedAt := nullableTime(env.ReceivedAtMS) + _, err = w.exec.ExecContext(ctx, upsertRealtimeSnapshotSQL, + string(env.Protocol), + vehicleKey, + strings.TrimSpace(env.VIN), + strings.TrimSpace(env.Phone), + strings.TrimSpace(env.DeviceID), + strings.TrimSpace(env.Plate), + strings.TrimSpace(env.MessageID), + env.Sequence, + strings.TrimSpace(env.SourceEndpoint), + eventTime, + parsedJSON, + fieldsJSON, + receivedAt, + env.StableEventID(), + ) + return err +} + +func marshalObject(value map[string]any) (string, error) { + if value == nil { + value = map[string]any{} + } + payload, err := json.Marshal(value) + if err != nil { + return "", err + } + return string(payload), nil +} + +func nullableTime(ms int64) any { + if ms <= 0 { + return nil + } + return time.UnixMilli(ms) +} + +const realtimeSnapshotTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + protocol VARCHAR(32) NOT NULL, + vehicle_key VARCHAR(96) NOT NULL, + vin VARCHAR(32) NOT NULL DEFAULT '', + phone VARCHAR(32) NOT NULL DEFAULT '', + device_id VARCHAR(96) NOT NULL DEFAULT '', + plate VARCHAR(32) NOT NULL DEFAULT '', + message_id VARCHAR(32) NOT NULL DEFAULT '', + sequence_id INT NOT NULL DEFAULT 0, + source_endpoint VARCHAR(128) NOT NULL DEFAULT '', + event_time DATETIME(3) NULL, + parsed_json JSON NOT NULL, + fields_json JSON NOT 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_vehicle (protocol, vehicle_key), + KEY idx_vehicle_key (vehicle_key), + KEY idx_vin (vin), + KEY idx_protocol_updated (protocol, updated_at) +)` + +const upsertRealtimeSnapshotSQL = ` +INSERT INTO vehicle_realtime_snapshot + (protocol, vehicle_key, vin, phone, device_id, plate, message_id, sequence_id, + source_endpoint, event_time, parsed_json, fields_json, received_at, event_id) +VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, CAST(? AS JSON), CAST(? AS JSON), ?, ?) +ON DUPLICATE KEY UPDATE + vin = IF(VALUES(vin) <> '', VALUES(vin), vin), + phone = IF(VALUES(phone) <> '', VALUES(phone), phone), + device_id = IF(VALUES(device_id) <> '', VALUES(device_id), device_id), + plate = IF(VALUES(plate) <> '', VALUES(plate), plate), + message_id = VALUES(message_id), + sequence_id = VALUES(sequence_id), + source_endpoint = VALUES(source_endpoint), + event_time = VALUES(event_time), + parsed_json = VALUES(parsed_json), + fields_json = VALUES(fields_json), + received_at = VALUES(received_at), + event_id = VALUES(event_id), + updated_at = CURRENT_TIMESTAMP +` diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go new file mode 100644 index 00000000..243e6792 --- /dev/null +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -0,0 +1,90 @@ +package realtime + +import ( + "context" + "database/sql" + "strings" + "testing" + + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +func TestSnapshotWriterEnsuresSchemaAndUpsertsParsedJSON(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.EnsureSchema(context.Background()); err != nil { + t.Fatalf("EnsureSchema() error = %v", err) + } + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolGB32960, + MessageID: "0x02", + VIN: "VIN001", + Phone: "13307795425", + DeviceID: "iccid-1", + SourceEndpoint: "1.2.3.4:32960", + EventTimeMS: 1782918600000, + ReceivedAtMS: 1782918601000, + Parsed: map[string]any{ + "data_units": []any{ + map[string]any{"name": "vehicle", "type": "0x01", "value": map[string]any{"soc_percent": 90}}, + map[string]any{"name": "gd_fc_vendor_tlv", "type": "vendor", "value": map[string]any{"foo": "bar"}}, + }, + }, + Fields: map[string]any{envelope.FieldSOCPercent: 90}, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) + } + if !strings.Contains(exec.calls[0].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_snapshot") { + t.Fatalf("schema query = %s", exec.calls[0].query) + } + upsert := exec.calls[1] + if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { + t.Fatalf("upsert query = %s", upsert.query) + } + if got, want := upsert.args[0], "GB32960"; got != want { + t.Fatalf("protocol arg = %#v, want %q", got, want) + } + if got, want := upsert.args[1], "VIN001"; got != want { + t.Fatalf("vehicle key arg = %#v, want %q", got, want) + } + parsedJSON, ok := upsert.args[10].(string) + if !ok { + t.Fatalf("parsed_json arg type = %T", upsert.args[10]) + } + if !strings.Contains(parsedJSON, "gd_fc_vendor_tlv") || !strings.Contains(parsedJSON, "vehicle") { + t.Fatalf("parsed_json should contain original and vendor parsed data: %s", parsedJSON) + } +} + +func TestSnapshotWriterSkipsUnknownVehicleKey(t *testing.T) { + exec := &recordingSnapshotExec{} + writer := NewSnapshotWriter(exec) + + if err := writer.Update(context.Background(), envelope.FrameEnvelope{ + Protocol: envelope.ProtocolJT808, + }); err != nil { + t.Fatalf("Update() error = %v", err) + } + if len(exec.calls) != 0 { + t.Fatalf("exec calls = %d, want 0", len(exec.calls)) + } +} + +type recordingSnapshotExec struct { + calls []snapshotExecCall +} + +type snapshotExecCall struct { + query string + args []any +} + +func (e *recordingSnapshotExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) { + e.calls = append(e.calls, snapshotExecCall{query: query, args: args}) + return nil, nil +}