diff --git a/docs/architecture/iot-data-platform-principles.md b/docs/architecture/iot-data-platform-principles.md index be9da809..9ea23b7d 100644 --- a/docs/architecture/iot-data-platform-principles.md +++ b/docs/architecture/iot-data-platform-principles.md @@ -41,6 +41,8 @@ flowchart LR ## Minimal Storage Contract +详细的落库边界见 [车辆数据最小落库合约](storage-minimal-contract.md)。 + | Store | Table or key | Purpose | Keep | Avoid | | --- | --- | --- | --- | --- | | TDengine | `raw_frames` | Replay and audit evidence | raw hex/text, full parsed JSON, parse status, protocol tags | business-only duplicated columns | diff --git a/docs/architecture/storage-minimal-contract.md b/docs/architecture/storage-minimal-contract.md new file mode 100644 index 00000000..2cb6fc30 --- /dev/null +++ b/docs/architecture/storage-minimal-contract.md @@ -0,0 +1,70 @@ +# 车辆数据最小落库合约 + +## 目标 + +车辆接入系统只持久化能回答业务问题、能回放纠错、能支撑高频查询的数据。协议细节默认保存在 raw,不向上层业务表扩散。 + +## 第一性原则 + +1. raw 是证据层,必须能证明收到过什么、解析成什么。 +2. Kafka 是回放层,消费者失败后优先靠 Kafka lag 追平。 +3. Redis 是当前态缓存,不是历史库。 +4. TDengine 只承担高写入时间序列:raw、位置历史。 +5. MySQL 只承担低基数业务状态:身份映射、实时快照、日指标。 +6. 同一个事实只落一个主表,其他表只存查询必要的投影。 +7. 新字段先进入 `parsed_json`,只有稳定查询需求出现后才提升为列。 + +## 目标表边界 + +| 存储 | 表/Key | 保留内容 | 不保留内容 | +| --- | --- | --- | --- | +| TDengine | `raw_frames` | raw hex/text、完整 `parsed_json`、解析状态、协议标签、车辆标识标签 | `fields_json` 这类可从 `parsed_json`/业务表再得到的重复字段 | +| TDengine | `raw_frame_payload_chunks` | 超长 raw/parsed payload 分片 | 业务查询字段 | +| TDengine | `vehicle_locations` | 时间、VIN/协议标签、经纬度、速度、方向、SOC、总里程等位置核心字段 | 完整协议 JSON、注册鉴权信息 | +| MySQL | `vehicle_realtime_snapshot` | 每个协议+VIN 的最新事件时间、接收时间、车牌、事件 ID | phone、device、source endpoint、完整 JSON | +| MySQL | `vehicle_realtime_location` | 每个协议+VIN 的最新位置核心字段 | raw、parsed JSON、消息头内部字段 | +| MySQL | `vehicle_daily_metric` | 日期、VIN/临时 vehicle key、协议、指标名、指标值、首末总里程、样本数 | 每帧细节、位置点列表 | +| MySQL | `vehicle_identity_binding` | 人工维护或导入的 VIN、车牌、phone、device_id 映射 | 注册历史 | +| MySQL | `jt808_registration` | JT808 phone 主键下的注册、鉴权、VIN 匹配状态 | GB32960/MQTT 注册信息 | +| Redis | `vehicle:realtime-raw:{protocol}:{vin}` | 每协议最新完整 parsed 状态 | 历史数据、统计结果 | + +## 当前应收敛的重复点 + +1. `vehicle_mileage_points` 与 `vehicle_locations.total_mileage_km` 重复。 + - 目标:停止写入 `vehicle_mileage_points`。 + - 兼容:`/api/history/mileage-points` 改为从 `vehicle_locations` 读取 `total_mileage_km IS NOT NULL`。 + - 删除:确认线上无直接读表后,再删除 TDengine stable 和子表。 + +2. `raw_frames.fields_json` 与 `raw_frames.parsed_json`、`vehicle_locations` 重复。 + - 目标:raw 只保留完整 `parsed_json`,核心查询字段进入 `vehicle_locations`。 + - 兼容:先让 raw 查询返回不依赖 `fields_json`。 + - 删除:等历史查询和前端不再展示 `fields_json` 后删除列。 + +3. `vehicle_daily_metric` 的 `vehicle_key` 与 `vin` 同时存在。 + - 当前保留原因:JT808 可能先用 phone 作为临时 key,后续才绑定 VIN。 + - 收敛方向:业务查询以 VIN 为主;没有 VIN 的数据只用于排查和待绑定,不作为正式车辆指标。 + +## 字段提升规则 + +协议字段进入系统后分三层: + +1. `parsed_json`:默认入口,保存完整结构化解析。 +2. 核心列:只有跨协议稳定查询需要时才提升,例如经纬度、速度、SOC、总里程。 +3. 指标表:只有聚合口径稳定、产品需要分页/排序/报表时才持久化。 + +反例: + +- 不因为某个协议字段存在就增加 MySQL 列。 +- 不为了方便调试在 snapshot/location 放完整 JSON。 +- 不为 telemetry field 配置服务提前复制一份字段表;字段配置服务应从 raw/Kafka 回放生成自己的结果。 + +## 删除或迁移顺序 + +1. 先写测试证明旧 API 能从新主表读到同等结果。 +2. 改代码停止写重复表或重复列。 +3. 部署后观察 Kafka lag、writer 成功计数、API 结果。 +4. 查询生产库确认旧表不再新增。 +5. 做一次备份或导出。 +6. 删除旧表或旧列。 + +任何一步无法证明安全,就停在兼容状态,不直接删生产数据。 diff --git a/docs/superpowers/plans/2026-07-02-storage-simplification.md b/docs/superpowers/plans/2026-07-02-storage-simplification.md new file mode 100644 index 00000000..21e01206 --- /dev/null +++ b/docs/superpowers/plans/2026-07-02-storage-simplification.md @@ -0,0 +1,376 @@ +# Storage Simplification Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 精简车辆接入落库模型,去掉重复持久化,同时保持 raw 可回放、实时可查、历史位置和日指标可用。 + +**Architecture:** TDengine 只保留 raw 证据和位置历史两类高写入时间序列;MySQL 只保留身份映射、实时当前态和日指标。先兼容 API,再停止重复写入,最后在生产验证后删除旧表/旧列。 + +**Tech Stack:** Go, TDengine, MySQL, Redis, Kafka, NATS JetStream, `go test`. + +--- + +## File Structure + +- Modify: `go/vehicle-gateway/internal/history/schema.go` + - Stop creating `vehicle_mileage_points` after API compatibility is in place. +- Modify: `go/vehicle-gateway/internal/history/writer.go` + - Stop calling `AppendMileagePoint`; keep total mileage in `vehicle_locations`. +- Modify: `go/vehicle-gateway/internal/history/query.go` + - Make `MileagePointRepository` read from `vehicle_locations` with `total_mileage_km IS NOT NULL`. +- Modify: `go/vehicle-gateway/internal/history/query_test.go` + - Update mileage point query expectations to `vehicle_locations`. +- Modify: `go/vehicle-gateway/internal/history/writer_test.go` + - Prove no `vehicle_mileage_points` stable/table is created or written. +- Modify: `go/vehicle-gateway/internal/history/schema.go` + - In a later task, remove `fields_json` from new `raw_frames` schema. +- Modify: `go/vehicle-gateway/internal/history/writer.go` + - In a later task, stop writing `fields_json`. +- Modify: `go/vehicle-gateway/internal/history/query.go` + - In a later task, make raw query tolerate schemas without `fields_json`. +- Modify: `docs/architecture/storage-minimal-contract.md` + - Keep the target storage contract updated as implementation lands. + +### Task 1: Move mileage point queries onto `vehicle_locations` + +**Files:** +- Modify: `go/vehicle-gateway/internal/history/query.go` +- Modify: `go/vehicle-gateway/internal/history/query_test.go` + +- [ ] **Step 1: Write the failing test** + +Update `TestMileagePointHandlerReturnsMileageByVehicleKey` in `go/vehicle-gateway/internal/history/query_test.go` so it expects `vehicle_locations`, not `vehicle_mileage_points`: + +```go +mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_locations"). + WillReturnRows(sqlmock.NewRows([]string{"count"}).AddRow(17)) +mock.ExpectQuery("SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.vehicle_locations"). + WillReturnRows(sqlmock.NewRows([]string{ + "ts", "event_id", "frame_id", "received_at", "total_mileage_km", "speed_kmh", "longitude", "latitude", + "protocol", "vehicle_key", "vin", "phone", "device_id", + }).AddRow("2026-07-02 10:00:00.000", "event-1", "frame-1", "2026-07-02 10:00:01.000", 8792.8, 8.0, 121.07764, 31.22436, "JT808", "JT808:13307811350", "VIN001", "13307811350", "dev001")) +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history -run 'TestMileagePointHandlerReturnsMileageByVehicleKey|TestBuildMileagePointSQLUsesLiteralsForTDengine' -count=1 +``` + +Expected: FAIL because `MileagePointRepository.tableName()` still returns `vehicle_mileage_points`. + +- [ ] **Step 3: Point mileage repository at location table** + +Change `MileagePointRepository.tableName()` in `go/vehicle-gateway/internal/history/query.go`: + +```go +func (r *MileagePointRepository) tableName() string { + if r.database == "" { + return "vehicle_locations" + } + return r.database + ".vehicle_locations" +} +``` + +Change `mileagePointWhere` so it filters only rows with total mileage: + +```go +func mileagePointWhere(query MileagePointQuery) []string { + where := []string{"total_mileage_km IS NOT NULL"} + add := func(condition string) { + where = append(where, condition) + } + if query.Protocol != "" { + add("protocol = '" + quote(query.Protocol) + "'") + } + if query.VehicleKey != "" { + add("vehicle_key = '" + quote(query.VehicleKey) + "'") + } + if query.VIN != "" { + add("vin = '" + quote(query.VIN) + "'") + } + if query.Phone != "" { + add("phone = '" + quote(query.Phone) + "'") + } + if query.DeviceID != "" { + add("device_id = '" + quote(query.DeviceID) + "'") + } + if query.DateFrom != "" { + add("ts >= '" + quote(normalizeDateTimeLiteral(query.DateFrom)) + "'") + } + if query.DateTo != "" { + add("ts <= '" + quote(normalizeDateTimeLiteral(query.DateTo)) + "'") + } + return where +} +``` + +- [ ] **Step 4: Run tests** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history -count=1 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add go/vehicle-gateway/internal/history/query.go go/vehicle-gateway/internal/history/query_test.go +git commit -m "refactor(go): read mileage points from locations" +``` + +### Task 2: Stop writing duplicate mileage point table + +**Files:** +- Modify: `go/vehicle-gateway/internal/history/schema.go` +- Modify: `go/vehicle-gateway/internal/history/writer.go` +- Modify: `go/vehicle-gateway/internal/history/writer_test.go` + +- [ ] **Step 1: Write the failing tests** + +Update `go/vehicle-gateway/internal/history/writer_test.go`: + +```go +for _, want := range []string{"CREATE DATABASE IF NOT EXISTS test_ts", "raw_frames", "vehicle_locations"} { + if got := countSQL(exec.calls, want); got == 0 { + t.Fatalf("expected schema statement containing %q", want) + } +} +if got := countSQL(exec.calls, "vehicle_mileage_points"); got != 0 { + t.Fatalf("schema should not create vehicle_mileage_points, got %d", got) +} +if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 0 { + t.Fatalf("history writer should not insert duplicate mileage point rows, got %d", got) +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history -run 'TestWriter' -count=1 +``` + +Expected: FAIL because schema and writer still create/write `vehicle_mileage_points`. + +- [ ] **Step 3: Remove duplicate write path** + +Change `AppendAll` in `go/vehicle-gateway/internal/history/writer.go`: + +```go +func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) error { + if err := w.AppendRawFrame(ctx, env); err != nil { + return err + } + return w.AppendLocation(ctx, env) +} +``` + +Remove `vehicle_mileage_points` from `SchemaStatements` in `go/vehicle-gateway/internal/history/schema.go`. Keep `AppendMileagePoint` only if tests still use it directly; otherwise remove the function and its helper after compile errors guide cleanup. + +- [ ] **Step 4: Run tests** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history ./cmd/history-writer -count=1 +``` + +Expected: PASS. + +- [ ] **Step 5: Commit** + +```bash +git add go/vehicle-gateway/internal/history/schema.go go/vehicle-gateway/internal/history/writer.go go/vehicle-gateway/internal/history/writer_test.go +git commit -m "refactor(go): stop writing duplicate mileage history" +``` + +### Task 3: Remove `fields_json` from new raw writes + +**Files:** +- Modify: `go/vehicle-gateway/internal/history/schema.go` +- Modify: `go/vehicle-gateway/internal/history/writer.go` +- Modify: `go/vehicle-gateway/internal/history/writer_test.go` +- Modify: `go/vehicle-gateway/internal/history/query.go` +- Modify: `go/vehicle-gateway/internal/history/query_test.go` + +- [ ] **Step 1: Write failing writer test** + +Add this assertion in the raw insert test: + +```go +rawInsert := findSQL(exec.calls, "INSERT INTO raw_") +if strings.Contains(rawInsert, "fields_json") { + t.Fatalf("raw insert should not write duplicate fields_json: %s", rawInsert) +} +``` + +- [ ] **Step 2: Run test to verify it fails** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history -run 'TestWriter' -count=1 +``` + +Expected: FAIL because raw insert still includes `fields_json`. + +- [ ] **Step 3: Remove `fields_json` from schema and writer** + +In `go/vehicle-gateway/internal/history/schema.go`, remove: + +```sql +fields_json BINARY(4096), +``` + +In `go/vehicle-gateway/internal/history/writer.go`, remove the `fieldsJSON` variable and `fields_json` insert column. Keep `parsed_json` complete. + +- [ ] **Step 4: Make raw query backward compatible** + +Keep API response field `fields_json,omitempty`, but query only columns that exist. Add a repository option so tests and startup can select the schema shape explicitly: + +```go +type RawFrameRepository struct { + db Queryer + database string + includeFieldsJSON bool +} + +func NewRawFrameRepository(db Queryer, database string) *RawFrameRepository { + return NewRawFrameRepositoryWithOptions(db, database, RawFrameRepositoryOptions{ + IncludeFieldsJSON: true, + }) +} + +type RawFrameRepositoryOptions struct { + IncludeFieldsJSON bool +} + +func NewRawFrameRepositoryWithOptions(db Queryer, database string, opts RawFrameRepositoryOptions) *RawFrameRepository { + if db == nil { + panic("raw frame query db must not be nil") + } + database = strings.TrimSpace(database) + if database != "" && !safeIdentifier(database) { + database = "" + } + return &RawFrameRepository{db: db, database: database, includeFieldsJSON: opts.IncludeFieldsJSON} +} +``` + +Then split raw SQL construction: + +```go +func buildRawFrameSQL(table string, query RawFrameQuery, includeFieldsJSON bool) (string, []any) { + fieldsColumn := "" + if includeFieldsJSON { + fieldsColumn = "fields_json, " + } + sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, ` + + fieldsColumn + + `parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table + where := rawFrameWhere(query) + if len(where) > 0 { + sqlText += " WHERE " + strings.Join(where, " AND ") + } + sqlText += " ORDER BY " + rawFrameOrderColumn(query.OrderBy) + " DESC LIMIT " + strconv.Itoa(query.Limit) + " OFFSET " + strconv.Itoa(query.Offset) + return sqlText, nil +} +``` + +The new-schema test must scan a row without `fields_json` and assert the JSON response omits `fields_json`. The old-schema test must use `IncludeFieldsJSON: true` and assert `fields_json` is hydrated. + +- [ ] **Step 5: Run tests** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history ./cmd/history-writer ./cmd/realtime-api -count=1 +``` + +Expected: PASS. + +- [ ] **Step 6: Commit** + +```bash +git add go/vehicle-gateway/internal/history/schema.go go/vehicle-gateway/internal/history/writer.go go/vehicle-gateway/internal/history/writer_test.go go/vehicle-gateway/internal/history/query.go go/vehicle-gateway/internal/history/query_test.go +git commit -m "refactor(go): remove duplicate raw fields json" +``` + +### Task 4: Production migration and cleanup gate + +**Files:** +- Modify: `docs/architecture/storage-minimal-contract.md` +- Modify: `docs/ops/vehicle-ingest-runbook.md` + +- [ ] **Step 1: Deploy the code changes** + +Run the existing build/deploy flow for Go services, then restart: + +```bash +systemctl restart lingniu-go-history-writer.service +systemctl restart lingniu-go-realtime-api.service +``` + +- [ ] **Step 2: Verify service health** + +Run on ECS: + +```bash +for port in 20212 20200; do + curl -fsS "http://127.0.0.1:${port}/readyz" + echo +done +curl -fsS http://127.0.0.1:20212/metrics | grep vehicle_history_kafka_lag +curl -fsS http://127.0.0.1:20212/metrics | grep vehicle_history_writes_total +``` + +Expected: + +- both readiness endpoints return `status=ok` +- history lag drains to `0` or remains bounded +- history writes continue increasing + +- [ ] **Step 3: Verify no new duplicate mileage table writes** + +Run TDengine checks on ECS: + +```sql +SELECT COUNT(*) FROM lingniu_vehicle_ts.vehicle_locations WHERE total_mileage_km IS NOT NULL; +SELECT COUNT(*) FROM lingniu_vehicle_ts.vehicle_mileage_points; +``` + +Expected: `vehicle_locations` grows; `vehicle_mileage_points` stops increasing after deployment time. + +- [ ] **Step 4: Drop old table only after observation** + +After at least one production observation window with no direct reads from `vehicle_mileage_points`, run: + +```sql +DROP STABLE IF EXISTS lingniu_vehicle_ts.vehicle_mileage_points; +``` + +- [ ] **Step 5: Commit docs update** + +```bash +git add docs/architecture/storage-minimal-contract.md docs/ops/vehicle-ingest-runbook.md +git commit -m "docs: record storage simplification migration" +``` + +## Self-Review + +- Spec coverage: The plan addresses minimal storage, duplicate mileage storage, raw field duplication, production verification, and delayed destructive cleanup. +- Red-flag scan: The plan has no open-ended task. Task 3 specifies the compatibility option and the new/old schema test expectations. +- Type consistency: The plan keeps existing public API types and changes only their backing tables.