feat(go): persist realtime snapshots to mysql

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
 }

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

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
+`

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
+}