diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md index ca44d2f5..66554e0f 100644 --- a/docs/architecture/storage-minimal-contract.md +++ b/docs/architecture/storage-minimal-contract.md @@ -64,6 +64,7 @@ - 索引:不在 `vehicle_realtime_location` 维护经纬度组合索引;实时表只回答当前态,地理范围和轨迹类查询走 TDengine 历史位置。 - 查询:`/api/realtime/snapshots`、`/api/realtime/locations` 默认不执行 `COUNT(*)`,`total` 表示本页返回条数;只有需要精确总数时传 `includeTotal=true`。 - 写入:车牌从 `vehicle_identity_binding` 按 VIN 主键直查,并使用 VIN 级短 TTL 内存缓存,避免实时帧每条都打 MySQL;缓存不作为事实来源。 + - 不写 MySQL KV 投影表;实时全量字段保留在 Redis KV,历史全量字段走 TDengine raw fields。 - 生产:项目上线前可直接重建这两张 MySQL 表,实时数据会从 Kafka 新消息继续投影。 7. Redis realtime 只服务已定位 VIN 的正式车辆。 diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index 0671ecea..62d10c19 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -115,7 +115,6 @@ func main() { mux.Handle("/api/stats/daily-metrics", stats.NewMetricHandler(stats.NewMetricRepository(db))) mux.Handle("/api/realtime/snapshots", realtime.NewSnapshotQueryHandler(realtime.NewSnapshotQueryRepository(db))) mux.Handle("/api/realtime/locations", realtime.NewLocationQueryHandler(realtime.NewLocationQueryRepository(db))) - mux.Handle("/api/realtime/kv", realtime.NewKVQueryHandler(realtime.NewKVQueryRepository(db))) logger.Info("stats mysql query enabled") } else { mysqlUnavailable := func(w http.ResponseWriter, _ *http.Request) { @@ -126,7 +125,6 @@ func main() { mux.HandleFunc("/api/stats/daily-metrics", mysqlUnavailable) mux.HandleFunc("/api/realtime/snapshots", mysqlUnavailable) mux.HandleFunc("/api/realtime/locations", mysqlUnavailable) - mux.HandleFunc("/api/realtime/kv", mysqlUnavailable) logger.Warn("MYSQL_DSN is empty; stats query api disabled") } defer closeStats() diff --git a/go/vehicle-gateway/cmd/realtime-api/main_test.go b/go/vehicle-gateway/cmd/realtime-api/main_test.go index 7807505a..3466fe25 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main_test.go +++ b/go/vehicle-gateway/cmd/realtime-api/main_test.go @@ -252,6 +252,16 @@ func TestRealtimeAPIExposesOperationalRealtimeRoutes(t *testing.T) { } } +func TestRealtimeAPIDoesNotExposeMySQLKVRoute(t *testing.T) { + source, err := os.ReadFile("main.go") + if err != nil { + t.Fatalf("read main.go: %v", err) + } + if strings.Contains(string(source), `"/api/realtime/kv"`) || strings.Contains(string(source), "NewKVQuery") { + t.Fatal("realtime api should not expose MySQL realtime kv route") + } +} + type contextCheckingRealtimeUpdater struct { ctxErr error count int diff --git a/go/vehicle-gateway/internal/realtime/mysql_query.go b/go/vehicle-gateway/internal/realtime/mysql_query.go index 52db3acd..8a0c5845 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query.go @@ -19,8 +19,6 @@ type RealtimeTableQuery struct { Protocol string VIN string Plate string - Domain string - Field string IncludeTotal bool Limit int Offset int @@ -58,19 +56,6 @@ type LocationRow struct { UpdatedAt string `json:"updated_at"` } -type KVRow struct { - Protocol string `json:"protocol"` - VIN string `json:"vin"` - Domain string `json:"domain"` - Field string `json:"field"` - Value string `json:"value"` - ValueType string `json:"value_type"` - EventTime string `json:"event_time,omitempty"` - ReceivedAt string `json:"received_at,omitempty"` - EventID string `json:"event_id"` - UpdatedAt string `json:"updated_at"` -} - type SnapshotQueryRepository struct { db mysqlQueryer } @@ -123,10 +108,6 @@ type LocationQueryRepository struct { db mysqlQueryer } -type KVQueryRepository struct { - db mysqlQueryer -} - func NewLocationQueryRepository(db mysqlQueryer) *LocationQueryRepository { if db == nil { panic("location query db must not be nil") @@ -134,13 +115,6 @@ func NewLocationQueryRepository(db mysqlQueryer) *LocationQueryRepository { return &LocationQueryRepository{db: db} } -func NewKVQueryRepository(db mysqlQueryer) *KVQueryRepository { - if db == nil { - panic("kv query db must not be nil") - } - return &KVQueryRepository{db: db} -} - func (r *LocationQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) { sqlText, args := buildRealtimeCountSQL("vehicle_realtime_location", normalizeRealtimeTableQuery(query)) return queryCount(ctx, r.db, sqlText, args) @@ -200,49 +174,6 @@ func (r *LocationQueryRepository) Query(ctx context.Context, query RealtimeTable return out, rows.Err() } -func (r *KVQueryRepository) Count(ctx context.Context, query RealtimeTableQuery) (int64, error) { - where, args := buildRealtimeKVWhere(normalizeRealtimeTableQuery(query)) - sqlText := "SELECT COUNT(*) FROM vehicle_realtime_kv" - if len(where) > 0 { - sqlText += " WHERE " + strings.Join(where, " AND ") - } - return queryCount(ctx, r.db, sqlText, args) -} - -func (r *KVQueryRepository) Query(ctx context.Context, query RealtimeTableQuery) ([]KVRow, error) { - query = normalizeRealtimeTableQuery(query) - where, args := buildRealtimeKVWhere(query) - sqlText := "SELECT protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id, updated_at FROM vehicle_realtime_kv" - if len(where) > 0 { - sqlText += " WHERE " + strings.Join(where, " AND ") - } - sqlText += " ORDER BY updated_at DESC LIMIT ? OFFSET ?" - args = append(args, query.Limit, query.Offset) - rows, err := r.db.QueryContext(ctx, sqlText, args...) - if err != nil { - return nil, err - } - defer rows.Close() - - out := make([]KVRow, 0) - for rows.Next() { - var row KVRow - var eventTime, receivedAt, updatedAt scanSQLDateTime - var value sql.NullString - if err := rows.Scan(&row.Protocol, &row.VIN, &row.Domain, &row.Field, &value, &row.ValueType, &eventTime, &receivedAt, &row.EventID, &updatedAt); err != nil { - return nil, err - } - if value.Valid { - row.Value = value.String - } - row.EventTime = eventTime.String - row.ReceivedAt = receivedAt.String - row.UpdatedAt = updatedAt.String - out = append(out, row) - } - return out, rows.Err() -} - type SnapshotQueryHandler struct { repository *SnapshotQueryRepository } @@ -291,10 +222,6 @@ type LocationQueryHandler struct { repository *LocationQueryRepository } -type KVQueryHandler struct { - repository *KVQueryRepository -} - func NewLocationQueryHandler(repository *LocationQueryRepository) *LocationQueryHandler { if repository == nil { panic("location query repository must not be nil") @@ -302,13 +229,6 @@ func NewLocationQueryHandler(repository *LocationQueryRepository) *LocationQuery return &LocationQueryHandler{repository: repository} } -func NewKVQueryHandler(repository *KVQueryRepository) *KVQueryHandler { - if repository == nil { - panic("kv query repository must not be nil") - } - return &KVQueryHandler{repository: repository} -} - func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeError(w, http.StatusMethodNotAllowed, "method not allowed") @@ -342,39 +262,6 @@ func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) writePage(w, rows, total, query) } -func (h *KVQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - writeError(w, http.StatusMethodNotAllowed, "method not allowed") - return - } - if strings.Trim(r.URL.Path, "/") != "api/realtime/kv" { - writeError(w, http.StatusNotFound, "route not found") - return - } - query, err := parseRealtimeTableQuery(r) - if err != nil { - writeError(w, http.StatusBadRequest, err.Error()) - return - } - var total int64 - if query.IncludeTotal { - total, err = h.repository.Count(r.Context(), query) - if err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - } - rows, err := h.repository.Query(r.Context(), query) - if err != nil { - writeError(w, http.StatusInternalServerError, err.Error()) - return - } - if !query.IncludeTotal { - total = int64(len(rows)) - } - writePage(w, rows, total, query) -} - func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) { values := r.URL.Query() limit, err := parseRealtimeBoundedInt(values.Get("limit"), 50, 1, 1000, "limit") @@ -389,8 +276,6 @@ func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) { Protocol: values.Get("protocol"), VIN: values.Get("vin"), Plate: values.Get("plate"), - Domain: values.Get("domain"), - Field: values.Get("field"), IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"), Limit: limit, Offset: offset, @@ -401,36 +286,12 @@ func normalizeRealtimeTableQuery(query RealtimeTableQuery) RealtimeTableQuery { query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) query.VIN = strings.TrimSpace(query.VIN) query.Plate = strings.TrimSpace(query.Plate) - query.Domain = strings.TrimSpace(query.Domain) - query.Field = strings.TrimSpace(query.Field) if query.Limit <= 0 { query.Limit = 50 } return query } -func buildRealtimeKVWhere(query RealtimeTableQuery) ([]string, []any) { - var where []string - var args []any - if query.Protocol != "" { - where = append(where, "protocol = ?") - args = append(args, query.Protocol) - } - if query.VIN != "" { - where = append(where, "vin = ?") - args = append(args, query.VIN) - } - if query.Domain != "" { - where = append(where, "domain_name = ?") - args = append(args, query.Domain) - } - if query.Field != "" { - where = append(where, "field_name = ?") - args = append(args, query.Field) - } - return where, args -} - func buildRealtimeCountSQL(table string, query RealtimeTableQuery) (string, []any) { where, args := buildRealtimeWhere(query) sqlText := "SELECT COUNT(*) FROM " + table diff --git a/go/vehicle-gateway/internal/realtime/mysql_query_test.go b/go/vehicle-gateway/internal/realtime/mysql_query_test.go index 8f0ca4f3..17d5adbe 100644 --- a/go/vehicle-gateway/internal/realtime/mysql_query_test.go +++ b/go/vehicle-gateway/internal/realtime/mysql_query_test.go @@ -94,44 +94,6 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) { } } -func TestKVQueryHandlerReturnsRealtimeKVFields(t *testing.T) { - db, mock, err := sqlmock.New() - if err != nil { - t.Fatalf("sqlmock.New() error = %v", err) - } - defer db.Close() - - mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_realtime_kv"). - WithArgs("GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c"). - WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(1)) - mock.ExpectQuery("SELECT protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id, updated_at FROM vehicle_realtime_kv"). - WithArgs("GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c", 10, 0). - WillReturnRows(sqlmock.NewRows([]string{ - "protocol", "vin", "domain_name", "field_name", "field_value", "value_type", "event_time", "received_at", "event_id", "updated_at", - }).AddRow( - "GB32960", "VIN001", "gd_fc_stack", "stack_water_outlet_temp_c", "63", "number", "2026-07-03 11:18:45.000", "2026-07-03 11:18:45.123", "evt-kv", "2026-07-03 11:18:46", - )) - - handler := NewKVQueryHandler(NewKVQueryRepository(db)) - request := httptest.NewRequest(http.MethodGet, "/api/realtime/kv?protocol=gb32960&vin=VIN001&domain=gd_fc_stack&field=stack_water_outlet_temp_c&includeTotal=true&limit=10", nil) - response := httptest.NewRecorder() - - handler.ServeHTTP(response, request) - - if response.Code != http.StatusOK { - t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) - } - body := response.Body.String() - for _, want := range []string{`"protocol":"GB32960"`, `"domain":"gd_fc_stack"`, `"field":"stack_water_outlet_temp_c"`, `"value":"63"`, `"value_type":"number"`, `"total":1`} { - if !strings.Contains(body, want) { - t.Fatalf("response missing %s: %s", want, body) - } - } - if err := mock.ExpectationsWereMet(); err != nil { - t.Fatalf("sql expectations: %v", err) - } -} - func TestSnapshotQueryHandlerSkipsTotalCountByDefault(t *testing.T) { db, mock, err := sqlmock.New() if err != nil { diff --git a/go/vehicle-gateway/internal/realtime/openapi.go b/go/vehicle-gateway/internal/realtime/openapi.go index 9457606d..f10c6b54 100644 --- a/go/vehicle-gateway/internal/realtime/openapi.go +++ b/go/vehicle-gateway/internal/realtime/openapi.go @@ -38,7 +38,7 @@ func realtimeOpenAPISpec() map[string]any { "info": map[string]any{ "title": "Lingniu Vehicle Realtime API", "version": "1.0.0", - "description": "MySQL realtime query APIs for vehicle_realtime_snapshot, vehicle_realtime_location, and vehicle_realtime_kv.", + "description": "MySQL realtime query APIs for vehicle_realtime_snapshot and vehicle_realtime_location.", }, "paths": map[string]any{ "/api/realtime/snapshots": map[string]any{ @@ -79,31 +79,11 @@ func realtimeOpenAPISpec() map[string]any { }, }, }, - "/api/realtime/kv": map[string]any{ - "get": map[string]any{ - "summary": "查询车辆实时 KV 字段", - "description": "从 vehicle_realtime_kv 查询实时字段投影,支持协议、VIN、domain、field 过滤和分页。适合 UI 局部刷新、快速字段查询和统计输入。", - "tags": []string{"Realtime KV API"}, - "x-table": "vehicle_realtime_kv", - "parameters": kvRealtimeParameters(), - "responses": map[string]any{ - "200": map[string]any{ - "description": "KV page", - "content": map[string]any{ - "application/json": map[string]any{ - "schema": map[string]any{"$ref": "#/components/schemas/KVPage"}, - }, - }, - }, - }, - }, - }, }, "components": map[string]any{ "schemas": map[string]any{ "SnapshotPage": pageSchema("#/components/schemas/SnapshotRow"), "LocationPage": pageSchema("#/components/schemas/LocationRow"), - "KVPage": pageSchema("#/components/schemas/KVRow"), "SnapshotRow": map[string]any{ "type": "object", "properties": map[string]any{ @@ -137,35 +117,11 @@ func realtimeOpenAPISpec() map[string]any { "updated_at": stringSchema("2026-07-02 16:11:04"), }, }, - "KVRow": map[string]any{ - "type": "object", - "properties": map[string]any{ - "protocol": stringSchema("GB32960"), - "vin": stringSchema("LNXNEGRR0SR321372"), - "domain": stringSchema("gd_fc_stack"), - "field": stringSchema("stack_water_outlet_temp_c"), - "value": stringSchema("63"), - "value_type": stringSchema("number"), - "event_time": stringSchema("2026-07-03 11:18:45.000"), - "received_at": stringSchema("2026-07-03 11:18:45.123"), - "event_id": stringSchema("event id"), - "updated_at": stringSchema("2026-07-03 11:18:46"), - }, - }, }, }, } } -func kvRealtimeParameters() []map[string]any { - params := commonRealtimeParameters() - params = append(params, - map[string]any{"name": "domain", "in": "query", "schema": map[string]any{"type": "string", "example": "gd_fc_stack"}, "required": false}, - map[string]any{"name": "field", "in": "query", "schema": map[string]any{"type": "string", "example": "stack_water_outlet_temp_c"}, "required": false}, - ) - return params -} - func commonRealtimeParameters() []map[string]any { return []map[string]any{ {"name": "protocol", "in": "query", "schema": map[string]any{"type": "string", "enum": []string{"GB32960", "JT808", "YUTONG_MQTT"}}, "required": false}, diff --git a/go/vehicle-gateway/internal/realtime/openapi_test.go b/go/vehicle-gateway/internal/realtime/openapi_test.go index a9960e86..e86cebd3 100644 --- a/go/vehicle-gateway/internal/realtime/openapi_test.go +++ b/go/vehicle-gateway/internal/realtime/openapi_test.go @@ -22,6 +22,11 @@ func TestOpenAPIHandlerDocumentsRealtimeSnapshotAndLocation(t *testing.T) { t.Fatalf("openapi missing %s: %s", want, body) } } + for _, removed := range []string{`"/api/realtime/kv"`, `"vehicle_realtime_kv"`, `"Realtime KV API"`} { + if strings.Contains(body, removed) { + t.Fatalf("openapi should not expose removed MySQL realtime kv API %s: %s", removed, body) + } + } for _, want := range []string{`"includeTotal"`, `默认不执行 COUNT(*)`, `total 默认表示本页返回条数`} { if !strings.Contains(body, want) { t.Fatalf("openapi should document lightweight total semantics, missing %s: %s", want, body) diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer.go b/go/vehicle-gateway/internal/realtime/snapshot_writer.go index 57d8c411..a4a82943 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer.go @@ -111,9 +111,6 @@ func (w *SnapshotWriter) EnsureSchema(ctx context.Context) error { if _, err := w.exec.ExecContext(ctx, realtimeLocationTableSQL); err != nil { return err } - if _, err := w.exec.ExecContext(ctx, realtimeKVTableSQL); err != nil { - return err - } return nil } @@ -152,9 +149,6 @@ func (w *SnapshotWriter) Update(ctx context.Context, env envelope.FrameEnvelope) ); err != nil { return err } - if err := w.upsertKV(ctx, env, parsed); err != nil { - return err - } location, ok := realtimeLocationFromEnvelope(env, vin, plate) if !ok { return nil @@ -196,31 +190,6 @@ func (w *SnapshotWriter) parsedForEnvelope(ctx context.Context, env envelope.Fra return parsed, nil } -func (w *SnapshotWriter) upsertKV(ctx context.Context, env envelope.FrameEnvelope, parsed map[string]any) error { - rows := realtimeKVFields(env, parsed) - if len(rows) == 0 { - return nil - } - placeholders := make([]string, 0, len(rows)) - args := make([]any, 0, len(rows)*9) - for _, row := range rows { - placeholders = append(placeholders, "(?, ?, ?, ?, ?, ?, ?, ?, ?)") - args = append(args, - string(row.Protocol), - row.VIN, - row.Domain, - row.Field, - row.Value, - row.ValueType, - nullableTime(row.EventTimeMS), - nullableTime(row.ReceivedAtMS), - row.EventID, - ) - } - _, err := w.exec.ExecContext(ctx, upsertRealtimeKVSQLPrefix+strings.Join(placeholders, ",")+upsertRealtimeKVSQLSuffix, args...) - return err -} - func platformNameFromEnvelope(env envelope.FrameEnvelope) string { if env.Fields != nil { if value := strings.TrimSpace(stringValue(env.Fields["platform_account"])); value != "" { @@ -533,37 +502,6 @@ const realtimeLocationTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_lo KEY idx_protocol_updated (protocol, updated_at) )` -const realtimeKVTableSQL = `CREATE TABLE IF NOT EXISTS vehicle_realtime_kv ( - protocol VARCHAR(32) NOT NULL, - vin VARCHAR(32) NOT NULL, - domain_name VARCHAR(64) NOT NULL, - field_name VARCHAR(160) NOT NULL, - field_value TEXT NULL, - value_type VARCHAR(16) NOT NULL DEFAULT '', - event_time DATETIME(3) NULL, - received_at DATETIME(3) NULL, - event_id VARCHAR(64) NOT NULL DEFAULT '', - updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, - PRIMARY KEY (protocol, vin, domain_name, field_name), - KEY idx_vin_domain (vin, domain_name), - KEY idx_protocol_domain (protocol, domain_name) -)` - -const upsertRealtimeKVSQLPrefix = ` -INSERT INTO vehicle_realtime_kv - (protocol, vin, domain_name, field_name, field_value, value_type, event_time, received_at, event_id) -VALUES ` - -const upsertRealtimeKVSQLSuffix = ` -ON DUPLICATE KEY UPDATE - field_value = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(field_value), field_value), - value_type = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(value_type), value_type), - event_time = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(event_time), event_time), - received_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(received_at), received_at), - event_id = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), VALUES(event_id), event_id), - updated_at = IF(VALUES(event_time) IS NOT NULL AND (vehicle_realtime_kv.event_time IS NULL OR VALUES(event_time) >= vehicle_realtime_kv.event_time), CURRENT_TIMESTAMP, updated_at) -` - const upsertRealtimeLocationSQL = ` INSERT INTO vehicle_realtime_location (protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, diff --git a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go index 3a898c87..595438f9 100644 --- a/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go +++ b/go/vehicle-gateway/internal/realtime/snapshot_writer_test.go @@ -42,8 +42,8 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 8 { - t.Fatalf("exec calls = %d, want 8", len(exec.calls)) + if len(exec.calls) != 6 { + 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") { t.Fatalf("schema query = %s", exec.calls[0].query) @@ -51,8 +51,10 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if !strings.Contains(exec.calls[4].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_location") { t.Fatalf("location schema query = %s", exec.calls[4].query) } - if !strings.Contains(exec.calls[5].query, "CREATE TABLE IF NOT EXISTS vehicle_realtime_kv") { - t.Fatalf("kv schema query = %s", exec.calls[5].query) + for _, call := range exec.calls { + if strings.Contains(call.query, "vehicle_realtime_kv") { + t.Fatalf("snapshot writer should not create or write MySQL realtime kv: %s", call.query) + } } for _, call := range []snapshotExecCall{exec.calls[0], exec.calls[4]} { if strings.Contains(call.query, "fields_json") { @@ -75,7 +77,7 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { 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[6] + upsert := exec.calls[5] if !strings.Contains(upsert.query, "ON DUPLICATE KEY UPDATE") { t.Fatalf("upsert query = %s", upsert.query) } @@ -102,10 +104,6 @@ func TestSnapshotWriterEnsuresSchemaAndUpsertsCoreSnapshot(t *testing.T) { if len(upsert.args) != 9 { t.Fatalf("snapshot upsert args = %d, want 9", len(upsert.args)) } - kvUpsert := exec.calls[7] - if !strings.Contains(kvUpsert.query, "INSERT INTO vehicle_realtime_kv") || !strings.Contains(kvUpsert.query, "ON DUPLICATE KEY UPDATE") { - t.Fatalf("kv upsert query = %s", kvUpsert.query) - } } func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { @@ -126,8 +124,6 @@ func TestSnapshotWriterEnsureSchemaOnlyCreatesTargetTables(t *testing.T) { WillReturnResult(sqlmock.NewResult(0, 0)) mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_location"). WillReturnResult(sqlmock.NewResult(0, 0)) - mock.ExpectExec("CREATE TABLE IF NOT EXISTS vehicle_realtime_kv"). - WillReturnResult(sqlmock.NewResult(0, 0)) if err := writer.EnsureSchema(context.Background()); err != nil { t.Fatalf("EnsureSchema() error = %v", err) @@ -169,10 +165,10 @@ func TestSnapshotWriterUpsertsRealtimeLocationWhenCoordinatesExist(t *testing.T) t.Fatalf("Update() error = %v", err) } - if len(exec.calls) != 3 { - t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) } - locationUpsert := exec.calls[2] + locationUpsert := exec.calls[1] if !strings.Contains(locationUpsert.query, "INSERT INTO vehicle_realtime_location") { t.Fatalf("location upsert query = %s", locationUpsert.query) } @@ -235,8 +231,6 @@ func TestSnapshotWriterMergesGB32960ParsedJSONAcrossSplitRealtimeFrames(t *testi sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec("INSERT INTO vehicle_realtime_kv"). - WillReturnResult(sqlmock.NewResult(0, 2)) err = writer.Update(context.Background(), envelope.FrameEnvelope{ Protocol: envelope.ProtocolGB32960, @@ -283,8 +277,6 @@ func TestSnapshotWriterCollapsesGB32960StackFragmentsWhenMergingParsedJSON(t *te sqlmock.AnyArg(), ). WillReturnResult(sqlmock.NewResult(0, 1)) - mock.ExpectExec("INSERT INTO vehicle_realtime_kv"). - WillReturnResult(sqlmock.NewResult(0, 3)) err = writer.Update(context.Background(), envelope.FrameEnvelope{ Protocol: envelope.ProtocolGB32960, @@ -379,13 +371,13 @@ func TestSnapshotWriterBackfillsPlateFromBindingByVIN(t *testing.T) { if resolver.vin != "VIN001" { t.Fatalf("resolver vin = %q", resolver.vin) } - if len(exec.calls) != 3 { - t.Fatalf("exec calls = %d, want 3", len(exec.calls)) + if len(exec.calls) != 2 { + t.Fatalf("exec calls = %d, want 2", len(exec.calls)) } if got, want := exec.calls[0].args[2], "沪A12345"; got != want { t.Fatalf("snapshot plate arg = %#v, want %q", got, want) } - if got, want := exec.calls[2].args[2], "沪A12345"; got != want { + if got, want := exec.calls[1].args[2], "沪A12345"; got != want { t.Fatalf("location plate arg = %#v, want %q", got, want) } } @@ -441,10 +433,10 @@ func TestSnapshotWriterCachesBindingPlateByVIN(t *testing.T) { if resolver.calls != 1 { t.Fatalf("plate resolver calls = %d, want 1", resolver.calls) } - if len(exec.calls) != 6 { - t.Fatalf("exec calls = %d, want 6", len(exec.calls)) + if len(exec.calls) != 4 { + t.Fatalf("exec calls = %d, want 4", len(exec.calls)) } - for _, index := range []int{0, 3} { + for _, index := range []int{0, 2} { if got, want := exec.calls[index].args[2], "沪A12345"; got != want { t.Fatalf("snapshot %d plate arg = %#v, want %q", index, got, want) }