docs: define minimal vehicle storage contract

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 |

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. 删除旧表或旧列。
+
+任何一步无法证明安全，就停在兼容状态，不直接删生产数据。

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.