diff --git a/docs/superpowers/plans/2026-07-01-go-ingest-redesign-phase1.md b/docs/superpowers/plans/2026-07-01-go-ingest-redesign-phase1.md new file mode 100644 index 00000000..20314115 --- /dev/null +++ b/docs/superpowers/plans/2026-07-01-go-ingest-redesign-phase1.md @@ -0,0 +1,720 @@ +# Go Vehicle Ingest Redesign Phase 1 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:** Build the first production-capable Go runtime for GB32960, JT808, and Yutong MQTT ingestion with unified Kafka, TDengine, MySQL statistics, and Redis realtime state boundaries. + +**Architecture:** Create a new single Go module at `go/vehicle-gateway` and migrate only the useful pieces from the existing `go/ingest-edge` and `go/vehicle-state` prototypes. The gateway produces unified envelopes to Kafka; independent consumers write TDengine history, MySQL daily metrics, and Redis realtime state. + +**Tech Stack:** Go 1.26+, Kafka (`segmentio/kafka-go`), Redis (`redis/go-redis`), MySQL (`go-sql-driver/mysql`), TDengine official Go connector, MQTT (`eclipse/paho.mqtt.golang`), standard library TCP. + +--- + +## File Structure + +- Create: `go/vehicle-gateway/go.mod` +- Create: `go/vehicle-gateway/cmd/gateway/main.go` +- Create: `go/vehicle-gateway/cmd/history-writer/main.go` +- Create: `go/vehicle-gateway/cmd/stat-writer/main.go` +- Create: `go/vehicle-gateway/cmd/realtime-api/main.go` +- Create: `go/vehicle-gateway/internal/envelope/envelope.go` +- Create: `go/vehicle-gateway/internal/envelope/envelope_test.go` +- Create: `go/vehicle-gateway/internal/protocol/jt808/*` +- Create: `go/vehicle-gateway/internal/protocol/gb32960/*` +- Create: `go/vehicle-gateway/internal/protocol/yutongmqtt/*` +- Create: `go/vehicle-gateway/internal/gateway/*` +- Create: `go/vehicle-gateway/internal/identity/*` +- Create: `go/vehicle-gateway/internal/eventbus/*` +- Create: `go/vehicle-gateway/internal/history/*` +- Create: `go/vehicle-gateway/internal/stats/*` +- Create: `go/vehicle-gateway/internal/realtime/*` +- Create: `go/vehicle-gateway/internal/observability/*` +- Modify: `README.md` +- Modify: `docs/target-architecture.md` +- Create: `deploy/portainer/docker-compose-go.yml` + +The existing `go/ingest-edge` and `go/vehicle-state` directories are migration sources only. After phase 1 is verified, remove or mark them superseded. + +--- + +### Task 1: Create Single Go Module + +**Files:** +- Create: `go/vehicle-gateway/go.mod` +- Create: `go/vehicle-gateway/internal/observability/logger.go` +- Create: `go/vehicle-gateway/cmd/gateway/main.go` + +- [ ] **Step 1: Create module manifest** + +Add `go/vehicle-gateway/go.mod`: + +```go +module lingniu-vehicle-ingest/go/vehicle-gateway + +go 1.26 + +require ( + github.com/eclipse/paho.mqtt.golang v1.5.1 + github.com/go-sql-driver/mysql v1.9.3 + github.com/redis/go-redis/v9 v9.17.2 + github.com/segmentio/kafka-go v0.4.49 + github.com/taosdata/driver-go/v3 v3.8.1 +) +``` + +- [ ] **Step 2: Add logger helper** + +Add `go/vehicle-gateway/internal/observability/logger.go`: + +```go +package observability + +import ( + "log/slog" + "os" +) + +func NewLogger(service string) *slog.Logger { + handler := slog.NewJSONHandler(os.Stdout, &slog.HandlerOptions{AddSource: true}) + return slog.New(handler).With("service", service) +} +``` + +- [ ] **Step 3: Add temporary gateway entrypoint** + +Add `go/vehicle-gateway/cmd/gateway/main.go`: + +```go +package main + +import "lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability" + +func main() { + logger := observability.NewLogger("vehicle-gateway") + logger.Info("vehicle gateway scaffold started") +} +``` + +- [ ] **Step 4: Verify scaffold builds** + +Run: + +```bash +cd go/vehicle-gateway +go mod tidy +go test ./... +go build ./cmd/gateway +``` + +Expected: all commands exit `0`. + +--- + +### Task 2: Define Unified Envelope + +**Files:** +- Create: `go/vehicle-gateway/internal/envelope/envelope.go` +- Create: `go/vehicle-gateway/internal/envelope/envelope_test.go` + +- [ ] **Step 1: Write envelope tests** + +Add `go/vehicle-gateway/internal/envelope/envelope_test.go`: + +```go +package envelope + +import "testing" + +func TestFrameEnvelopeVehicleKeyPrefersVIN(t *testing.T) { + e := FrameEnvelope{Protocol: ProtocolJT808, VIN: "LNBVIN00000000001", Phone: "013307795425"} + if got := e.VehicleKey(); got != "LNBVIN00000000001" { + t.Fatalf("VehicleKey() = %q", got) + } +} + +func TestFrameEnvelopeVehicleKeyFallsBackToPhone(t *testing.T) { + e := FrameEnvelope{Protocol: ProtocolJT808, Phone: "013307795425"} + if got := e.VehicleKey(); got != "JT808:013307795425" { + t.Fatalf("VehicleKey() = %q", got) + } +} + +func TestFrameEnvelopeEventIDStable(t *testing.T) { + e := FrameEnvelope{ + Protocol: ProtocolJT808, + MessageID: "0x0200", + Phone: "013307795425", + Sequence: 1, + EventTimeMS: 1782745114000, + ReceivedAtMS: 1782745114999, + RawHex: "7e02000000ff7e", + } + a := e.StableEventID() + b := e.StableEventID() + if a == "" || a != b { + t.Fatalf("event id must be non-empty and stable: %q %q", a, b) + } +} +``` + +- [ ] **Step 2: Run tests and confirm failure** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/envelope +``` + +Expected: fail because `FrameEnvelope` is not defined. + +- [ ] **Step 3: Implement envelope** + +Add `go/vehicle-gateway/internal/envelope/envelope.go`: + +```go +package envelope + +import ( + "crypto/sha256" + "encoding/hex" + "encoding/json" + "fmt" + "strings" +) + +type Protocol string + +const ( + ProtocolGB32960 Protocol = "GB32960" + ProtocolJT808 Protocol = "JT808" + ProtocolYutongMQTT Protocol = "YUTONG_MQTT" +) + +type ParseStatus string + +const ( + ParseOK ParseStatus = "OK" + ParsePartial ParseStatus = "PARTIAL" + ParseBadFrame ParseStatus = "BAD_FRAME" +) + +type FrameEnvelope struct { + EventID string `json:"event_id"` + TraceID string `json:"trace_id"` + Protocol Protocol `json:"protocol"` + MessageID string `json:"message_id"` + Sequence uint16 `json:"sequence"` + VIN string `json:"vin,omitempty"` + VehicleKeyHint string `json:"vehicle_key,omitempty"` + Phone string `json:"phone,omitempty"` + DeviceID string `json:"device_id,omitempty"` + Plate string `json:"plate,omitempty"` + SourceEndpoint string `json:"source_endpoint,omitempty"` + EventTimeMS int64 `json:"event_time_ms"` + ReceivedAtMS int64 `json:"received_at_ms"` + RawHex string `json:"raw_hex,omitempty"` + RawText string `json:"raw_text,omitempty"` + Parsed map[string]any `json:"parsed,omitempty"` + Fields map[string]any `json:"fields,omitempty"` + ParseStatus ParseStatus `json:"parse_status"` + ParseError string `json:"parse_error,omitempty"` +} + +func (e FrameEnvelope) VehicleKey() string { + if key := strings.TrimSpace(e.VIN); key != "" { + return key + } + if key := strings.TrimSpace(e.VehicleKeyHint); key != "" { + return key + } + if key := strings.TrimSpace(e.Phone); key != "" { + return string(e.Protocol) + ":" + key + } + if key := strings.TrimSpace(e.DeviceID); key != "" { + return string(e.Protocol) + ":" + key + } + return string(e.Protocol) + ":unknown" +} + +func (e FrameEnvelope) StableEventID() string { + if strings.TrimSpace(e.EventID) != "" { + return e.EventID + } + input := fmt.Sprintf("%s|%s|%s|%d|%d|%s", + e.Protocol, e.MessageID, e.VehicleKey(), e.Sequence, e.EventTimeMS, e.RawHex) + sum := sha256.Sum256([]byte(input)) + return hex.EncodeToString(sum[:16]) +} + +func (e FrameEnvelope) MarshalJSONBytes() ([]byte, error) { + if e.EventID == "" { + e.EventID = e.StableEventID() + } + if e.ParseStatus == "" { + e.ParseStatus = ParseOK + } + return json.Marshal(e) +} +``` + +- [ ] **Step 4: Verify tests pass** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/envelope +``` + +Expected: pass. + +--- + +### Task 3: Implement JT808 Frame and 0200 Parser + +**Files:** +- Create: `go/vehicle-gateway/internal/protocol/jt808/parser.go` +- Create: `go/vehicle-gateway/internal/protocol/jt808/parser_test.go` + +- [ ] **Step 1: Add sample frame tests** + +Use the production sample provided in the thread: + +```text +7E020000320133077954250001000000000048000301D2C4C707376139000A00E6004F26063016235701040001900C2504000000000202000030011F31010F867E +``` + +Expected assertions: + +- message id is `0x0200` +- phone keeps normalized BCD string `013307795425` +- sequence is `1` +- `fields.total_mileage_km` is parsed from additional item `0x01` +- latitude, longitude, speed, direction, alarm, status, and device time are present + +- [ ] **Step 2: Implement parser** + +Implementation rules: + +- strip `0x7e` start/end delimiters +- unescape `0x7d 0x02 -> 0x7e` +- unescape `0x7d 0x01 -> 0x7d` +- verify XOR checksum +- parse 2011/2013 common header +- parse BCD phone from 6 bytes and keep both raw and normalized values in `parsed.header` +- parse 0200 fixed body +- parse additional item list as `id,length,value_hex` +- parse additional `0x01` as `total_mileage_km = uint32 / 10` + +- [ ] **Step 3: Verify** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/protocol/jt808 +``` + +Expected: pass. + +--- + +### Task 4: Implement GB32960 Frame and Data Unit Parser + +**Files:** +- Create: `go/vehicle-gateway/internal/protocol/gb32960/parser.go` +- Create: `go/vehicle-gateway/internal/protocol/gb32960/parser_test.go` + +- [ ] **Step 1: Add parser tests** + +Tests must cover: + +- `##` frame boundary +- command id +- response flag +- 17-byte VIN +- encryption flag +- payload length +- BCC verification +- realtime data command `0x02` +- reissue data command `0x03` +- data unit `0x01` vehicle status fields +- data unit `0x05` position fields + +- [ ] **Step 2: Implement parser** + +Implementation rules: + +- parse header without allocating large temporary buffers +- keep RAW as hex in envelope +- write all recognized data units to `Parsed` +- write core fields to `Fields` +- unsupported data unit stays in `Parsed["unknown_units"]` + +- [ ] **Step 3: Verify** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/protocol/gb32960 +``` + +Expected: pass. + +--- + +### Task 5: Implement Kafka Sink + +**Files:** +- Create: `go/vehicle-gateway/internal/eventbus/kafka_sink.go` +- Create: `go/vehicle-gateway/internal/eventbus/kafka_sink_test.go` + +- [ ] **Step 1: Add sink interface** + +Create a sink interface: + +```go +package eventbus + +import ( + "context" + "lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope" +) + +type Sink interface { + PublishRaw(context.Context, envelope.FrameEnvelope) error + PublishUnified(context.Context, envelope.FrameEnvelope) error + Close() error +} +``` + +- [ ] **Step 2: Implement Kafka topic routing** + +Rules: + +- `GB32960 -> vehicle.raw.gb32960.v1` +- `JT808 -> vehicle.raw.jt808.v1` +- `YUTONG_MQTT -> vehicle.raw.yutong-mqtt.v1` +- unified topic is `vehicle.event.unified.v1` +- message key is `env.VehicleKey()` + +- [ ] **Step 3: Verify routing with unit tests** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/eventbus +``` + +Expected: pass. + +--- + +### Task 6: Implement TDengine History Writer + +**Files:** +- Create: `go/vehicle-gateway/internal/history/schema.go` +- Create: `go/vehicle-gateway/internal/history/writer.go` +- Create: `go/vehicle-gateway/internal/history/writer_test.go` +- Create: `go/vehicle-gateway/cmd/history-writer/main.go` + +- [ ] **Step 1: Add schema bootstrap SQL** + +Implement schema strings for: + +- database `lingniu_vehicle_ts` +- stable `raw_frames` +- stable `vehicle_locations` +- stable `vehicle_mileage_points` + +- [ ] **Step 2: Implement writer** + +Rules: + +- `AppendRawFrame` always writes one raw row. +- `AppendLocation` writes only when longitude and latitude exist. +- `AppendMileagePoint` writes only when `total_mileage_km` exists. +- child table name is deterministic hash of `protocol + vehicle_key`. +- escape tag values. + +- [ ] **Step 3: Use TDengine official driver** + +Import WebSocket driver in command: + +```go +import _ "github.com/taosdata/driver-go/v3/taosWS" +``` + +Default driver name: + +```text +TDENGINE_DRIVER=taosWS +``` + +Short-term compatibility: + +```text +TDENGINE_DRIVER=taosSql +``` + +Only use `taosSql` when ECS TDengine WebSocket is not available. + +- [ ] **Step 4: Verify** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/history +go build ./cmd/history-writer +``` + +Expected: pass. + +--- + +### Task 7: Implement MySQL Daily Metric Writer + +**Files:** +- Create: `go/vehicle-gateway/internal/stats/schema.go` +- Create: `go/vehicle-gateway/internal/stats/daily_metric.go` +- Create: `go/vehicle-gateway/internal/stats/daily_metric_test.go` +- Create: `go/vehicle-gateway/cmd/stat-writer/main.go` + +- [ ] **Step 1: Add schema bootstrap** + +Implement `vehicle_daily_metric` schema from the design spec. + +- [ ] **Step 2: Add metric derivation tests** + +Test cases: + +- no `total_mileage_km` produces no metric +- one sample produces: + - `daily_mileage_km = 0` + - `daily_total_mileage_km = sample` +- later larger sample updates: + - `latest_total_mileage_km` + - `daily_mileage_km` + - `daily_total_mileage_km` +- out-of-order smaller sample updates: + - `first_total_mileage_km` + - `daily_mileage_km` + +- [ ] **Step 3: Implement MySQL upsert** + +Use one table and one idempotent upsert: + +```sql +INSERT INTO vehicle_daily_metric + (vin, stat_date, protocol, metric_key, metric_value, metric_unit, + first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method) +VALUES (?, ?, ?, ?, ?, 'km', ?, ?, 1, 'TOTAL_MILEAGE_DIFF') +ON DUPLICATE KEY UPDATE + first_total_mileage_km = LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)), + latest_total_mileage_km = GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)), + metric_value = CASE + WHEN metric_key = 'daily_mileage_km' + THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)) + - LEAST(first_total_mileage_km, VALUES(first_total_mileage_km)) + WHEN metric_key = 'daily_total_mileage_km' + THEN GREATEST(latest_total_mileage_km, VALUES(latest_total_mileage_km)) + ELSE VALUES(metric_value) + END, + sample_count = sample_count + 1, + updated_at = CURRENT_TIMESTAMP +``` + +- [ ] **Step 4: Verify** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/stats +go build ./cmd/stat-writer +``` + +Expected: pass. + +--- + +### Task 8: Implement Redis Realtime API + +**Files:** +- Create: `go/vehicle-gateway/internal/realtime/repository.go` +- Create: `go/vehicle-gateway/internal/realtime/repository_test.go` +- Create: `go/vehicle-gateway/internal/realtime/http.go` +- Create: `go/vehicle-gateway/cmd/realtime-api/main.go` + +- [ ] **Step 1: Add repository contract** + +Methods: + +- `Update(ctx, envelope.FrameEnvelope) error` +- `GetMerged(ctx, vin string) (Snapshot, error)` +- `GetProtocol(ctx, vin string, protocol envelope.Protocol) (Snapshot, error)` +- `IsOnline(ctx, vin string) (OnlineStatus, error)` + +- [ ] **Step 2: Implement merge logic** + +Rules: + +- update `vehicle:latest:{vin}:{protocol}` +- update `vehicle:latest:{vin}` with newest fields +- update `vehicle:online:{vin}` +- update sorted set `vehicle:last_seen` + +- [ ] **Step 3: Implement HTTP API** + +Routes: + +- `GET /api/realtime/vehicles/{vin}` +- `GET /api/realtime/vehicles/{vin}/online` +- `GET /api/realtime/vehicles/{vin}/protocols/{protocol}` + +- [ ] **Step 4: Verify** + +Run: + +```bash +cd go/vehicle-gateway +go test ./internal/realtime +go build ./cmd/realtime-api +``` + +Expected: pass. + +--- + +### Task 9: Wire Gateway Runtime + +**Files:** +- Create: `go/vehicle-gateway/internal/gateway/tcp_server.go` +- Create: `go/vehicle-gateway/internal/gateway/mqtt_client.go` +- Modify: `go/vehicle-gateway/cmd/gateway/main.go` + +- [ ] **Step 1: Implement TCP server** + +Rules: + +- one goroutine per accepted connection +- bounded max connections +- read timeout and idle timeout +- protocol-specific frame extractor +- structured peer endpoint +- graceful shutdown on SIGTERM + +- [ ] **Step 2: Implement MQTT client** + +Rules: + +- connect with official production config from environment +- subscribe configured topic list +- convert each message into envelope +- publish raw and unified events +- reconnect with backoff + +- [ ] **Step 3: Verify local JSON mode** + +Run without Kafka: + +```bash +cd go/vehicle-gateway +GB32960_TCP_ADDR=:132960 JT808_TCP_ADDR=:18080 go run ./cmd/gateway +``` + +Expected: service starts and logs configured listeners. + +--- + +### Task 10: Docker and ECS Deployment + +**Files:** +- Create: `go/vehicle-gateway/Dockerfile` +- Create: `deploy/portainer/docker-compose-go.yml` +- Modify: `docs/operations/current-ecs-deployment.md` + +- [ ] **Step 1: Add multi-stage Dockerfile** + +Build all commands: + +- `gateway` +- `history-writer` +- `stat-writer` +- `realtime-api` + +- [ ] **Step 2: Add Portainer compose** + +Services: + +- `go-vehicle-gateway` +- `go-history-writer` +- `go-stat-writer` +- `go-realtime-api` + +Each service must include: + +- restart policy +- memory limit +- Kafka env +- MySQL/TDengine/Redis env as needed +- logging options + +- [ ] **Step 3: Verify Linux build** + +Run: + +```bash +cd go/vehicle-gateway +GOOS=linux GOARCH=amd64 go build ./cmd/gateway +GOOS=linux GOARCH=amd64 go build ./cmd/history-writer +GOOS=linux GOARCH=amd64 go build ./cmd/stat-writer +GOOS=linux GOARCH=amd64 go build ./cmd/realtime-api +``` + +Expected: all commands exit `0`. + +--- + +### Task 11: Production Verification + +**Files:** +- Create: `docs/operations/go-vehicle-gateway-verification.md` + +- [ ] **Step 1: Record test commands** + +Document commands to verify: + +- gateway process health +- Kafka topic consumption +- TDengine row counts +- MySQL daily metric rows +- Redis realtime lookup + +- [ ] **Step 2: Validate real traffic** + +Evidence required: + +- one real 32960 VIN with RAW, location, mileage point, daily metric, Redis snapshot +- one real JT808 phone/VIN with RAW, location, mileage point, daily metric, Redis snapshot +- one real Yutong MQTT VIN with RAW and Redis snapshot + +- [ ] **Step 3: Keep old Java services until evidence is captured** + +Only disable Java equivalents after the evidence file contains successful command outputs and timestamps. + +--- + +## Self-Review Checklist + +- The plan creates one new Go module instead of extending scattered prototypes. +- The plan covers GB32960, JT808, and Yutong MQTT ingress. +- The plan covers Kafka, TDengine, MySQL, and Redis. +- The plan includes 32960 and 808 daily mileage and daily total mileage. +- The plan includes local and ECS verification. +- The plan does not restore Xinda Push. +- The plan keeps Java services untouched until Go evidence exists. diff --git a/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md b/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md new file mode 100644 index 00000000..49bf5cd7 --- /dev/null +++ b/docs/superpowers/specs/2026-07-01-go-ingest-redesign-design.md @@ -0,0 +1,486 @@ +# Go 车辆接入重构设计规格 + +## 目标 + +使用 Go 重新设计车辆数据接入运行面，让系统回到第一性原则： + +- 接入层只负责可靠收包、协议解析、应答、身份解析和投递。 +- Kafka 是唯一在线消息通道。 +- TDengine 保存高吞吐时序明细和完整 RAW 解析 JSON。 +- MySQL 保存低频配置、身份映射和每日统计窄表。 +- Redis 保存可重建的准实时状态。 +- 32960、JT808、宇通 MQTT 使用同一套数据层接口，避免每个协议各自落库。 + +第一阶段只做生产链路的骨架和核心数据闭环：接收、解析、Kafka、TDengine、MySQL 统计、Redis 实时查询、ECS 验证。历史查询 API 和业务侧复杂分析放在后续阶段。 + +## 参考原则 + +本设计只吸收开源项目的边界思想，不复制外部实现代码。 + +- GB32960 参考 `DarkInno/gb32960-go-sdk`：连接管理、Handler、Forwarder、鉴权接口分离。 +- JT808 参考 `cuteLittleDevil/go-jt808`：高并发 Go 原生 TCP、协议核心小而可扩展、事件/适配器机制。 +- JT808 参考 `fakeyanss/jt808-server-go`：`FramePayload -> PacketData -> JT808Msg -> Reply` 的分层处理、Gateway 模式、保活和版本兼容。 +- TDengine 参考官方数据模型：按数据采集点建立 supertable，静态维度做 tags，明细数据按子表写入；Go 连接优先使用 WebSocket 驱动。 + +## 非目标 + +- 第一阶段不继续扩展 Java 服务。 +- 第一阶段不做统一大查询 API。 +- 第一阶段不恢复信达 Push。 +- 第一阶段不把所有协议字段展开成一张超宽 TDengine 表。 +- Redis 不作为历史事实源。 +- MySQL 不保存高频原始明细。 + +## 目标目录 + +新的 Go 运行面放在 `go/vehicle-gateway`，不继续沿用实验性的多 module 分散结构。现有 `go/ingest-edge` 和 `go/vehicle-state` 可以作为迁移素材，最终归并或删除。 + +```text +go/vehicle-gateway/ + cmd/ + gateway/ # 协议接入进程：32960 TCP、JT808 TCP、宇通 MQTT + history-writer/ # Kafka -> TDengine RAW/location/mileage points + stat-writer/ # Kafka -> MySQL 每日统计窄表 + realtime-api/ # Redis 准实时查询 API + internal/ + config/ # 环境变量、配置文件、校验 + gateway/ # TCP/MQTT 生命周期、连接限制、优雅停机 + protocol/ + gb32960/ # 32960 编解码 + jt808/ # 808 编解码 + yutongmqtt/ # 宇通 MQTT payload 解析 + identity/ # VIN、phone、device_id、plate 绑定 + envelope/ # 统一事件模型 + eventbus/ # Kafka producer/consumer + history/ # TDengine adapter + stats/ # 每日里程聚合 + realtime/ # Redis adapter 与快照合并 + observability/ # 日志、metrics、健康检查 +``` + +## 协议接入设计 + +### GB32960 + +职责： + +- 监听 TCP `32960`。 +- 支持车辆登入、实时信息上报、补发信息上报、车辆登出、平台登入/登出、心跳、校时。 +- 正确返回协议应答。 +- 完整解析 2016 数据单元： + - 整车数据 + - 驱动电机数据 + - 燃料电池数据 + - 发动机数据 + - 车辆位置数据 + - 极值数据 + - 报警数据 + - 可充电储能装置电压数据 + - 可充电储能装置温度数据 +- RAW 中保存完整 parsed JSON。 +- 关键字段标准化为统一 field： + - `speed_kmh` + - `total_mileage_km` + - `soc_percent` + - `longitude` + - `latitude` + - `vehicle_status` + - `charge_status` + +### JT808 + +职责： + +- 监听 TCP `808`。 +- 支持 2011/2013 基础兼容，预留 2019 版本标记。 +- 分层处理： + - Frame：`0x7e` 包边界、转义、校验。 + - Packet：消息头、手机号 BCD、流水号、分包。 + - Message：注册、鉴权、心跳、注销、位置、批量位置、未知上行。 + - Reply：注册应答、平台通用应答。 +- 终端手机号按协议头 BCD 解析，内部同时保留原始 BCD、规范化 phone、去前导零 phone。 +- 注册帧和鉴权帧写入 808 registration 表。 +- 如果设备直接上报位置帧，也要走 identity 解析，尝试用 phone、device_id、plate 找到 VIN。 +- 0200 附加信息必须解析表 27： + - `0x01` GPS 总里程，DWORD，单位 0.1 km。 + - 其他已知附加项结构化放入 `parsed_json.additional`。 +- 标准化字段： + - `speed_kmh` + - `total_mileage_km` + - `longitude` + - `latitude` + - `altitude_m` + - `direction_deg` + - `alarm_flag` + - `status_flag` + +### 宇通 MQTT + +职责： + +- 使用正式 MQTT 配置订阅生产 topic。 +- 接收 payload 后解析为统一 envelope。 +- 按 payload 中的车辆标识解析 VIN、车牌、设备号。 +- 保存完整 payload 和 parsed JSON。 +- 标准化字段向 32960/808 对齐： + - `speed_kmh` + - `total_mileage_km` + - `longitude` + - `latitude` + - `soc_percent` + - `vehicle_status` + +## 统一 Envelope + +所有协议接入后输出同一结构。 + +```json +{ + "event_id": "string", + "trace_id": "string", + "protocol": "GB32960|JT808|YUTONG_MQTT", + "message_id": "string", + "vin": "string", + "vehicle_key": "string", + "phone": "string", + "device_id": "string", + "plate": "string", + "source_endpoint": "ip:port or mqtt://broker/topic", + "event_time_ms": 0, + "received_at_ms": 0, + "raw_hex": "string", + "raw_text": "string", + "parsed": {}, + "fields": { + "speed_kmh": 0, + "total_mileage_km": 0, + "longitude": 0, + "latitude": 0 + }, + "parse_status": "OK|PARTIAL|BAD_FRAME", + "parse_error": "" +} +``` + +规则： + +- `event_id` 使用协议、设备标识、消息流水、事件时间、RAW hash 生成，保证幂等。 +- Kafka partition key 优先使用 VIN；没有 VIN 时使用 `protocol:phone/device_id`。 +- `parsed` 保存协议完整结构化字段。 +- `fields` 只保存跨协议核心字段，用于统计、位置和实时快照。 + +## Kafka Topic + +生产只支持 Kafka。 + +| Topic | 内容 | Key | +|---|---|---| +| `vehicle.raw.gb32960.v1` | 32960 完整 RAW envelope | VIN 或 vehicle_key | +| `vehicle.raw.jt808.v1` | 808 完整 RAW envelope | VIN 或 phone | +| `vehicle.raw.yutong-mqtt.v1` | 宇通 MQTT 完整 RAW envelope | VIN 或 vehicle_key | +| `vehicle.event.unified.v1` | 统一轻量事件，用于业务消费 | VIN 或 vehicle_key | + +接入层必须先写 RAW topic，再写 unified event。后续消费者只依赖 Kafka，不直接依赖接入进程内存。 + +## TDengine 数据库设计 + +如果现有 TDengine 库不符合目标，可以新建库并迁移。建议库名： + +```sql +CREATE DATABASE IF NOT EXISTS lingniu_vehicle_ts KEEP 7300 DURATION 10 BUFFER 256; +``` + +### raw_frames + +保存完整 RAW 和完整 parsed JSON。 + +```sql +CREATE STABLE IF NOT EXISTS raw_frames ( + ts TIMESTAMP, + frame_id NCHAR(64), + event_id NCHAR(64), + message_id INT, + event_time TIMESTAMP, + received_at TIMESTAMP, + raw_size_bytes INT, + raw_hex BINARY(16374), + raw_text BINARY(16374), + parsed_json BINARY(16374), + fields_json BINARY(4096), + parse_status NCHAR(16), + parse_error BINARY(1024), + source_endpoint NCHAR(128) +) TAGS ( + protocol NCHAR(32), + vehicle_key NCHAR(64), + vin NCHAR(32), + phone NCHAR(32), + device_id NCHAR(64) +); +``` + +说明： + +- 子表按 `protocol + vehicle_key hash` 创建。 +- RAW 查询优先按 tags 和时间范围过滤。 +- `parsed_json` 是完整协议字段；查询 API 可选择是否返回，避免默认大字段拖慢。 + +### vehicle_locations + +只保存位置查询核心字段。 + +```sql +CREATE STABLE IF NOT EXISTS vehicle_locations ( + ts TIMESTAMP, + event_id NCHAR(64), + frame_id NCHAR(64), + received_at TIMESTAMP, + longitude DOUBLE, + latitude DOUBLE, + altitude_m DOUBLE, + speed_kmh DOUBLE, + direction_deg INT, + alarm_flag BIGINT, + status_flag BIGINT, + total_mileage_km DOUBLE +) TAGS ( + protocol NCHAR(32), + vehicle_key NCHAR(64), + vin NCHAR(32), + phone NCHAR(32), + device_id NCHAR(64) +); +``` + +### vehicle_mileage_points + +保存总里程采样点，用于回放重算和查询。 + +```sql +CREATE STABLE IF NOT EXISTS vehicle_mileage_points ( + ts TIMESTAMP, + event_id NCHAR(64), + frame_id NCHAR(64), + received_at TIMESTAMP, + total_mileage_km DOUBLE, + speed_kmh DOUBLE, + longitude DOUBLE, + latitude DOUBLE +) TAGS ( + protocol NCHAR(32), + vehicle_key NCHAR(64), + vin NCHAR(32), + phone NCHAR(32), + device_id NCHAR(64) +); +``` + +## MySQL 数据库设计 + +MySQL 只保存配置、身份映射、808 注册鉴权和每日统计窄表。 + +### vehicle_identity_binding + +用于把外部标识定位到 VIN。 + +```sql +CREATE TABLE vehicle_identity_binding ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + vin VARCHAR(32) NOT NULL, + plate VARCHAR(32) NULL, + phone VARCHAR(32) NULL, + device_id VARCHAR(64) NULL, + remark VARCHAR(255) NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_vin (vin), + KEY idx_plate (plate), + KEY idx_phone (phone), + KEY idx_device_id (device_id) +); +``` + +### jt808_registration + +只记录 808 独有的注册、鉴权和在线身份信息。 + +```sql +CREATE TABLE jt808_registration ( + phone VARCHAR(32) PRIMARY KEY, + vin VARCHAR(32) NULL, + plate VARCHAR(32) NULL, + device_id VARCHAR(64) NULL, + manufacturer_id VARCHAR(32) NULL, + terminal_model VARCHAR(64) NULL, + terminal_id VARCHAR(64) NULL, + auth_code VARCHAR(128) NULL, + source_endpoint VARCHAR(128) NULL, + first_registered_at DATETIME NULL, + latest_registered_at DATETIME NULL, + latest_authed_at DATETIME NULL, + latest_seen_at DATETIME NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + KEY idx_vin (vin), + KEY idx_plate (plate), + KEY idx_device_id (device_id) +); +``` + +### vehicle_daily_metric + +32960 和 808 的每日里程、每日总里程都进入同一张窄表。 + +```sql +CREATE TABLE vehicle_daily_metric ( + id BIGINT PRIMARY KEY AUTO_INCREMENT, + vin VARCHAR(32) NOT NULL, + stat_date DATE NOT NULL, + protocol VARCHAR(32) NOT NULL, + metric_key VARCHAR(64) NOT NULL, + metric_value DECIMAL(18,3) NOT NULL, + metric_unit VARCHAR(16) NOT NULL, + first_total_mileage_km DECIMAL(18,3) NULL, + latest_total_mileage_km DECIMAL(18,3) NULL, + sample_count BIGINT NOT NULL DEFAULT 0, + calculation_method VARCHAR(64) NOT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + UNIQUE KEY uk_daily_metric (vin, stat_date, protocol, metric_key), + KEY idx_stat_date (stat_date), + KEY idx_protocol_metric (protocol, metric_key) +); +``` + +指标： + +- `daily_mileage_km = max(total_mileage_km) - min(total_mileage_km)` +- `daily_total_mileage_km = max(total_mileage_km)` + +统计消费者可幂等重放。每个样本 upsert 时只维护当天 `min/max total_mileage_km`，不依赖进程内状态。 + +## Redis 实时数据设计 + +Redis 保存可从 Kafka 重建的数据。 + +| Key | 内容 | +|---|---| +| `vehicle:online:{vin}` | 在线状态、最后活跃时间、协议列表 | +| `vehicle:latest:{vin}` | 合并后的 VIN 最新快照 | +| `vehicle:latest:{vin}:{protocol}` | 单协议最新快照 | +| `vehicle:last_seen` | VIN 到最后接收时间的 sorted set | + +合并规则： + +- 位置以最新 event_time 为准。 +- 总里程取最新上报值，不做递增假设。 +- 32960 多帧字段按 field timestamp 合并。 +- 808 和 MQTT 不覆盖其他协议独有字段。 +- Redis TTL 用于在线判断，不作为数据删除依据。 + +## 数据流 + +```mermaid +flowchart LR + source["车辆 / 平台"] --> gateway["Go gateway"] + gateway --> parser["protocol parser"] + parser --> identity["identity resolver"] + identity --> kafkaRaw["Kafka RAW topics"] + identity --> kafkaEvent["Kafka unified event"] + kafkaRaw --> history["history-writer"] + kafkaRaw --> stats["stat-writer"] + kafkaEvent --> realtime["realtime consumer"] + history --> td["TDengine raw_frames / locations / mileage_points"] + stats --> mysql["MySQL vehicle_daily_metric"] + realtime --> redis["Redis latest state"] +``` + +## 错误处理 + +- 协议坏帧仍写 RAW topic，`parse_status=BAD_FRAME`。 +- 可部分解析的帧写 `parse_status=PARTIAL`，保留 parse error。 +- Kafka 写失败时接入层不确认业务成功；协议是否应答按协议要求处理，但必须打错误日志和 metrics。 +- TDengine 写失败不提交 Kafka offset。 +- MySQL 统计写失败不提交 Kafka offset。 +- Redis 写失败不影响历史和统计，但要通过 metrics 暴露。 +- 808 相同 phone 新连接时，旧连接应被替换并记录 latest_seen。 + +## 部署设计 + +生产 ECS 运行 Go 二进制或容器： + +- `vehicle-gateway` +- `vehicle-history-writer` +- `vehicle-stat-writer` +- `vehicle-realtime-api` + +配置来源： + +- 环境变量优先。 +- Nacos 可作为后续配置中心，但 Go 第一阶段必须能只靠环境变量启动，便于 Portainer 部署和故障恢复。 + +TDengine 连接： + +- 优先 WebSocket `taosWS`。 +- 若当前 ECS 只开放原生端口，可短期兼容 `taosSql`，但代码配置必须显式标记为兼容模式。 + +## 验证计划 + +### 本地验证 + +- 用样例 808 报文验证： + - phone BCD 解析。 + - 0200 经纬度、速度、方向、时间。 + - 表 27 `0x01` GPS 总里程。 +- 用真实 32960 报文验证： + - 登录应答正确。 + - 实时帧和补发帧都能写 RAW。 + - 位置和总里程字段进入统一 fields。 +- 用宇通 MQTT payload 验证： + - payload 完整保存。 + - 核心字段归一化。 +- 不连 Kafka 时可输出 JSON log。 +- 连接 Kafka 时 RAW topic 可消费到 envelope。 + +### ECS 验证 + +- 部署前先旁路接收或短窗口切流。 +- 验证端口： + - `32960` + - `808` + - MQTT 订阅连接。 +- 验证 Kafka topic 有持续写入。 +- 验证 TDengine： + - `raw_frames` 有三种协议数据。 + - `vehicle_locations` 能按 VIN、协议、时间分页查询。 + - `vehicle_mileage_points` 有 32960 和 808 总里程采样。 +- 验证 MySQL： + - `vehicle_daily_metric` 有 32960 和 808 的 `daily_mileage_km`、`daily_total_mileage_km`。 +- 验证 Redis： + - 查询 VIN 在线。 + - 查询 VIN 合并实时快照。 + - 查询 VIN 分协议实时快照。 + +## 验收标准 + +- Go gateway 能在 ECS 上接收真实 32960、808、宇通 MQTT 数据。 +- 三种协议 RAW 都进入 Kafka。 +- 三种协议 RAW 和 parsed JSON 都进入 TDengine。 +- 32960 和 808 的位置进入 `vehicle_locations`。 +- 32960 和 808 的总里程采样进入 `vehicle_mileage_points`。 +- 32960 和 808 的每日里程、每日总里程进入 MySQL 窄表。 +- Redis 可以查询 VIN 在线和实时数据。 +- 任一消费者宕机后可通过 Kafka offset 继续恢复。 +- 查询和统计验证使用 ECS 上的真实数据完成。 + +## 实施顺序 + +1. 创建 `go/vehicle-gateway` 单 module，迁移现有 Go 原型中可用的解析、Kafka、TDengine、MySQL、Redis 代码。 +2. 先写协议解析单元测试，覆盖 808 样例报文和 32960 核心字段。 +3. 完成统一 envelope 和 Kafka sink。 +4. 完成 TDengine schema bootstrap 与 history writer。 +5. 完成 MySQL schema bootstrap 与 stat writer。 +6. 完成 Redis realtime writer/API。 +7. 本地启动 gateway，用 ECS 转发流量验证。 +8. 构建 Linux 镜像，部署 ECS。 +9. 按验收标准逐项验证并记录证据。