337 lines
12 KiB
Markdown
337 lines
12 KiB
Markdown
# 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`
|
||
|
||
- [x] **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"))
|
||
```
|
||
|
||
- [x] **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`.
|
||
|
||
- [x] **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
|
||
}
|
||
```
|
||
|
||
- [x] **Step 4: Run tests**
|
||
|
||
Run:
|
||
|
||
```bash
|
||
cd go/vehicle-gateway
|
||
go test ./internal/history -count=1
|
||
```
|
||
|
||
Expected: PASS.
|
||
|
||
- [x] **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`
|
||
|
||
- [x] **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)
|
||
}
|
||
```
|
||
|
||
- [x] **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`.
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **Step 4: Run tests**
|
||
|
||
Run:
|
||
|
||
```bash
|
||
cd go/vehicle-gateway
|
||
go test ./internal/history ./cmd/history-writer -count=1
|
||
```
|
||
|
||
Expected: PASS.
|
||
|
||
- [x] **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`
|
||
|
||
- [x] **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)
|
||
}
|
||
```
|
||
|
||
- [x] **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`.
|
||
|
||
- [x] **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.
|
||
|
||
- [x] **Step 4: Make raw query backward compatible**
|
||
|
||
Query only columns that exist in both old and new schemas. Old production tables may still contain `fields_json`, but the API no longer selects or returns it:
|
||
|
||
```go
|
||
func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) {
|
||
sqlText := `SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, raw_hex, raw_text, parsed_json, 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 scans a row without `fields_json` and asserts the JSON response omits `fields_json`. This also works on old schemas because selecting a subset of columns is valid when the table has extra columns.
|
||
|
||
- [x] **Step 5: Run tests**
|
||
|
||
Run:
|
||
|
||
```bash
|
||
cd go/vehicle-gateway
|
||
go test ./internal/history ./cmd/history-writer ./cmd/realtime-api -count=1
|
||
```
|
||
|
||
Expected: PASS.
|
||
|
||
- [x] **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`
|
||
|
||
- [x] **Step 1: Deploy the code changes**
|
||
|
||
Built Linux amd64 binaries for `history-writer` and `realtime-api`, uploaded them to the current ECS release, and restarted:
|
||
|
||
```bash
|
||
systemctl restart lingniu-go-history-writer.service
|
||
systemctl restart lingniu-go-realtime-api.service
|
||
```
|
||
|
||
- [x] **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
|
||
|
||
- [x] **Step 3: Verify no new duplicate mileage table writes**
|
||
|
||
Because the system is not live yet and historical data can be discarded, the TDengine history database was dropped and recreated from the new schema:
|
||
|
||
```sql
|
||
DROP DATABASE IF EXISTS lingniu_vehicle_ts;
|
||
DESCRIBE lingniu_vehicle_ts.raw_frames;
|
||
SHOW lingniu_vehicle_ts.STABLES;
|
||
```
|
||
|
||
Observed stables after recreation: `raw_frames`, `raw_frame_payload_chunks`, `vehicle_locations`. `vehicle_mileage_points` is gone.
|
||
|
||
- [x] **Step 4: Drop old table only after observation**
|
||
|
||
Old history data was intentionally discarded before launch. No separate observation window is required for `vehicle_mileage_points`.
|
||
|
||
- [x] **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.
|