Files

lingniu d148a4e3af docs: design go ingest redesign phase one

2026-07-01 21:36:07 +08:00

18 KiB

Raw Blame History

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

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:

import _ "github.com/taosdata/driver-go/v3/taosWS"

Default driver name:

TDENGINE_DRIVER=taosWS

Short-term compatibility:

TDENGINE_DRIVER=taosSql

Only use taosSql when ECS TDengine WebSocket is not available.

Step 4: Verify

Run:

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:

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:

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:

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:

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:

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.

18 KiB Raw Blame History

Go Vehicle Ingest Redesign Phase 1 Implementation Plan

File Structure

Task 1: Create Single Go Module

Task 2: Define Unified Envelope

Task 3: Implement JT808 Frame and 0200 Parser

Task 4: Implement GB32960 Frame and Data Unit Parser

Task 5: Implement Kafka Sink

Task 6: Implement TDengine History Writer

Task 7: Implement MySQL Daily Metric Writer

Task 8: Implement Redis Realtime API

Task 9: Wire Gateway Runtime

Task 10: Docker and ECS Deployment

Task 11: Production Verification

Self-Review Checklist

18 KiB

Raw Blame History