feat: add go tdengine history writer

go/vehicle-gateway/cmd/history-writer/main.go

@@ -0,0 +1,123 @@
+package main
+
+import (
+	"context"
+	"database/sql"
+	"encoding/json"
+	"os"
+	"os/signal"
+	"strings"
+	"syscall"
+
+	"github.com/segmentio/kafka-go"
+	_ "github.com/taosdata/driver-go/v3/taosWS"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/history"
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/observability"
+)
+
+func main() {
+	logger := observability.NewLogger("vehicle-history-writer")
+	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer stop()
+
+	cfg := loadConfig()
+	db, err := sql.Open(cfg.TDengineDriver, cfg.TDengineDSN)
+	if err != nil {
+		logger.Error("tdengine open failed", "error", err)
+		os.Exit(1)
+	}
+	defer db.Close()
+	if err := db.PingContext(ctx); err != nil {
+		logger.Error("tdengine ping failed", "error", err)
+		os.Exit(1)
+	}
+
+	writer := history.NewWriter(db)
+	if cfg.EnsureSchema {
+		if err := writer.EnsureSchema(ctx, cfg.TDengineDatabase); err != nil {
+			logger.Error("tdengine schema bootstrap failed", "error", err)
+			os.Exit(1)
+		}
+	}
+
+	reader := kafka.NewReader(kafka.ReaderConfig{
+		Brokers:     cfg.KafkaBrokers,
+		GroupID:     cfg.KafkaGroup,
+		GroupTopics: cfg.KafkaTopics,
+		MinBytes:    1,
+		MaxBytes:    10e6,
+	})
+	defer reader.Close()
+
+	logger.Info("history writer started",
+		"driver", cfg.TDengineDriver,
+		"group", cfg.KafkaGroup,
+		"topics", strings.Join(cfg.KafkaTopics, ","))
+
+	for {
+		message, err := reader.FetchMessage(ctx)
+		if err != nil {
+			if ctx.Err() != nil {
+				return
+			}
+			logger.Error("kafka fetch failed", "error", err)
+			continue
+		}
+		var env envelope.FrameEnvelope
+		if err := json.Unmarshal(message.Value, &env); err != nil {
+			logger.Warn("skip invalid envelope json", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
+			_ = reader.CommitMessages(ctx, message)
+			continue
+		}
+		if err := writer.AppendAll(ctx, env); err != nil {
+			logger.Error("tdengine append failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "event_id", env.StableEventID(), "error", err)
+			continue
+		}
+		if err := reader.CommitMessages(ctx, message); err != nil {
+			logger.Error("kafka commit failed", "topic", message.Topic, "partition", message.Partition, "offset", message.Offset, "error", err)
+		}
+	}
+}
+
+type config struct {
+	KafkaBrokers     []string
+	KafkaTopics      []string
+	KafkaGroup       string
+	TDengineDriver   string
+	TDengineDSN      string
+	TDengineDatabase string
+	EnsureSchema     bool
+}
+
+func loadConfig() config {
+	return config{
+		KafkaBrokers:     splitCSV(env("KAFKA_BROKERS", "127.0.0.1:9092")),
+		KafkaTopics:      splitCSV(env("KAFKA_TOPICS", "vehicle.raw.gb32960.v1,vehicle.raw.jt808.v1,vehicle.raw.yutong-mqtt.v1")),
+		KafkaGroup:       env("KAFKA_GROUP", "go-history-writer"),
+		TDengineDriver:   env("TDENGINE_DRIVER", "taosWS"),
+		TDengineDSN:      env("TDENGINE_DSN", ""),
+		TDengineDatabase: env("TDENGINE_DATABASE", history.DefaultDatabase),
+		EnsureSchema:     env("TDENGINE_ENSURE_SCHEMA", "true") != "false",
+	}
+}
+
+func env(key string, fallback string) string {
+	value := strings.TrimSpace(os.Getenv(key))
+	if value == "" {
+		return fallback
+	}
+	return value
+}
+
+func splitCSV(value string) []string {
+	var out []string
+	for _, item := range strings.Split(value, ",") {
+		item = strings.TrimSpace(item)
+		if item != "" {
+			out = append(out, item)
+		}
+	}
+	return out
+}

go/vehicle-gateway/go.mod

@@ -3,7 +3,16 @@ module lingniu-vehicle-ingest/go/vehicle-gateway
 go 1.26
 
 require (
-	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/segmentio/kafka-go v0.4.49
-	github.com/pierrec/lz4/v4 v4.1.15 // indirect
+	github.com/taosdata/driver-go/v3 v3.8.1
-	github.com/segmentio/kafka-go v0.4.49 // indirect
+)
+
+require (
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/gorilla/websocket v1.5.0 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/compress v1.15.9 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/pierrec/lz4/v4 v4.1.15 // indirect
 )

go/vehicle-gateway/go.sum

@@ -1,6 +1,47 @@
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gorilla/websocket v1.5.0 h1:PPwGk2jz7EePpoHN/+ClbZu8SPxiqlu12wZP/3sWmnc=
+github.com/gorilla/websocket v1.5.0/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
 github.com/klauspost/compress v1.15.9 h1:wKRjX6JRtDdrE9qwa4b/Cip7ACOshUI4smpCQanqjSY=
 github.com/klauspost/compress v1.15.9/go.mod h1:PhcZ0MbTNciWF3rruxRgKxI5NkcHHrHUDtV4Yw2GlzU=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OHLH3mGKHDcjJRFFRrJa6eAM5H+CtDdOsPc=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
 github.com/pierrec/lz4/v4 v4.1.15 h1:MO0/ucJhngq7299dKLwIMtgTfbkoSPF6AoMYDd8Q4q0=
 github.com/pierrec/lz4/v4 v4.1.15/go.mod h1:gZWDp/Ze/IJXGXf23ltt2EXimqmTUXEy0GFuRQyBid4=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/segmentio/kafka-go v0.4.49 h1:GJiNX1d/g+kG6ljyJEoi9++PUMdXGAxb7JGPiDCuNmk=
 github.com/segmentio/kafka-go v0.4.49/go.mod h1:Y1gn60kzLEEaW28YshXyk2+VCUKbJ3Qr6DrnT3i4+9E=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0 h1:1zr/of2m5FGMsad5YfcqgdqdWrIhu+EBEJRhR1U7z/c=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
+github.com/stretchr/testify v1.8.2/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/taosdata/driver-go/v3 v3.8.1 h1:kkd4ABsGiU+oXDbsw/sic985LKAvnpF9Gb/TEunTnLE=
+github.com/taosdata/driver-go/v3 v3.8.1/go.mod h1:S6OGOinfR0xxxaMGsvBi9cLkYxEIW1p6qqr8QJATTlg=
+github.com/xdg-go/pbkdf2 v1.0.0 h1:Su7DPu48wXMwC3bs7MCNG+z4FhcyEuz5dlvchbq0B0c=
+github.com/xdg-go/pbkdf2 v1.0.0/go.mod h1:jrpuAogTd400dnrH08LKmI/xc1MbPOebTwRqcT5RDeI=
+github.com/xdg-go/scram v1.1.2 h1:FHX5I5B4i4hKRVRBCFRxq1iQRej7WO3hhBuJf+UUySY=
+github.com/xdg-go/scram v1.1.2/go.mod h1:RT/sEzTbU5y00aCK8UOx6R7YryM0iF1N2MOmC3kKLN4=
+github.com/xdg-go/stringprep v1.0.4 h1:XLI/Ng3O1Atzq0oBs3TWm+5ZVgkq2aqdlvP9JtoZ6c8=
+github.com/xdg-go/stringprep v1.0.4/go.mod h1:mPGuuIYwz7CmR2bT9j4GbQqutWS1zV24gijq1dTyGkM=
+golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8=
+golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8=
+golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY=
+golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

go/vehicle-gateway/internal/history/schema.go

@@ -0,0 +1,71 @@
+package history
+
+const DefaultDatabase = "lingniu_vehicle_ts"
+
+func SchemaStatements(database string) []string {
+	if database == "" {
+		database = DefaultDatabase
+	}
+	return []string{
+		"CREATE DATABASE IF NOT EXISTS " + database + " KEEP 7300 DURATION 10 BUFFER 256",
+		"USE " + database,
+		`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)
+)`,
+		`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)
+)`,
+		`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)
+)`,
+	}
+}

go/vehicle-gateway/internal/history/writer.go

@@ -0,0 +1,300 @@
+package history
+
+import (
+	"context"
+	"crypto/sha1"
+	"database/sql"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+	"time"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+type Execer interface {
+	ExecContext(context.Context, string, ...any) (sql.Result, error)
+}
+
+type Writer struct {
+	exec  Execer
+	cache tableCache
+}
+
+type tableCache struct {
+	mu   sync.Mutex
+	seen map[string]struct{}
+}
+
+func NewWriter(exec Execer) *Writer {
+	if exec == nil {
+		panic("history execer must not be nil")
+	}
+	return &Writer{exec: exec, cache: tableCache{seen: map[string]struct{}{}}}
+}
+
+func (w *Writer) EnsureSchema(ctx context.Context, database string) error {
+	for _, statement := range SchemaStatements(database) {
+		if _, err := w.exec.ExecContext(ctx, statement); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) error {
+	if err := w.AppendRawFrame(ctx, env); err != nil {
+		return err
+	}
+	if err := w.AppendLocation(ctx, env); err != nil {
+		return err
+	}
+	return w.AppendMileagePoint(ctx, env)
+}
+
+func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) error {
+	table := tableName("raw", env)
+	if err := w.ensureChild(ctx, table, "raw_frames", env); err != nil {
+		return err
+	}
+	_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
+  (ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes,
+   raw_hex, raw_text, parsed_json, fields_json, parse_status, parse_error, source_endpoint)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, table), rawValues(env)...)
+	return err
+}
+
+func (w *Writer) AppendLocation(ctx context.Context, env envelope.FrameEnvelope) error {
+	longitude, okLon := floatField(env, envelope.FieldLongitude)
+	latitude, okLat := floatField(env, envelope.FieldLatitude)
+	if !okLon || !okLat {
+		return nil
+	}
+	table := tableName("loc", env)
+	if err := w.ensureChild(ctx, table, "vehicle_locations", env); err != nil {
+		return err
+	}
+	_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
+  (ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh,
+   direction_deg, alarm_flag, status_flag, total_mileage_km)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, table), locationValues(env, longitude, latitude)...)
+	return err
+}
+
+func (w *Writer) AppendMileagePoint(ctx context.Context, env envelope.FrameEnvelope) error {
+	totalMileage, ok := floatField(env, envelope.FieldTotalMileageKM)
+	if !ok {
+		return nil
+	}
+	table := tableName("mil", env)
+	if err := w.ensureChild(ctx, table, "vehicle_mileage_points", env); err != nil {
+		return err
+	}
+	_, err := w.exec.ExecContext(ctx, fmt.Sprintf(`INSERT INTO %s
+  (ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude)
+VALUES (?, ?, ?, ?, ?, ?, ?, ?)`, table), mileageValues(env, totalMileage)...)
+	return err
+}
+
+func (w *Writer) ensureChild(ctx context.Context, table string, stable string, env envelope.FrameEnvelope) error {
+	key := stable + "." + table
+	w.cache.mu.Lock()
+	_, ok := w.cache.seen[key]
+	w.cache.mu.Unlock()
+	if ok {
+		return nil
+	}
+	statement := fmt.Sprintf("CREATE TABLE IF NOT EXISTS %s USING %s TAGS ('%s', '%s', '%s', '%s', '%s')",
+		table, stable,
+		quote(string(env.Protocol)),
+		quote(env.VehicleKey()),
+		quote(env.VIN),
+		quote(env.Phone),
+		quote(env.DeviceID))
+	if _, err := w.exec.ExecContext(ctx, statement); err != nil {
+		return err
+	}
+	w.cache.mu.Lock()
+	w.cache.seen[key] = struct{}{}
+	w.cache.mu.Unlock()
+	return nil
+}
+
+func rawValues(env envelope.FrameEnvelope) []any {
+	received := millis(env.ReceivedAtMS)
+	eventTime := millis(env.EventTimeMS)
+	return []any{
+		received,
+		frameID(env),
+		env.StableEventID(),
+		messageIDInt(env.MessageID),
+		eventTime,
+		received,
+		rawSize(env.RawHex),
+		env.RawHex,
+		env.RawText,
+		jsonString(env.Parsed),
+		jsonString(env.Fields),
+		string(env.ParseStatus),
+		env.ParseError,
+		env.SourceEndpoint,
+	}
+}
+
+func locationValues(env envelope.FrameEnvelope, longitude float64, latitude float64) []any {
+	received := millis(env.ReceivedAtMS)
+	return []any{
+		eventTimeOrReceived(env),
+		env.StableEventID(),
+		frameID(env),
+		received,
+		longitude,
+		latitude,
+		floatFieldOrNil(env, "altitude_m"),
+		floatFieldOrNil(env, envelope.FieldSpeedKMH),
+		intFieldOrNil(env, "direction_deg"),
+		intFieldOrNil(env, "alarm_flag"),
+		intFieldOrNil(env, "status_flag"),
+		floatFieldOrNil(env, envelope.FieldTotalMileageKM),
+	}
+}
+
+func mileageValues(env envelope.FrameEnvelope, totalMileage float64) []any {
+	received := millis(env.ReceivedAtMS)
+	return []any{
+		eventTimeOrReceived(env),
+		env.StableEventID(),
+		frameID(env),
+		received,
+		totalMileage,
+		floatFieldOrNil(env, envelope.FieldSpeedKMH),
+		floatFieldOrNil(env, envelope.FieldLongitude),
+		floatFieldOrNil(env, envelope.FieldLatitude),
+	}
+}
+
+func frameID(env envelope.FrameEnvelope) string {
+	return "go_" + env.StableEventID()
+}
+
+func tableName(prefix string, env envelope.FrameEnvelope) string {
+	return prefix + "_" + strings.ToLower(string(env.Protocol)) + "_" + hash16(env.VehicleKey())
+}
+
+func hash16(value string) string {
+	sum := sha1.Sum([]byte(value))
+	return hex.EncodeToString(sum[:8])
+}
+
+func messageIDInt(value string) int64 {
+	value = strings.TrimSpace(strings.ToLower(value))
+	value = strings.TrimPrefix(value, "0x")
+	parsed, err := strconv.ParseInt(value, 16, 32)
+	if err == nil {
+		return parsed
+	}
+	parsed, _ = strconv.ParseInt(value, 10, 32)
+	return parsed
+}
+
+func rawSize(rawHex string) int {
+	rawHex = strings.TrimSpace(rawHex)
+	if len(rawHex)%2 != 0 {
+		return len(rawHex) / 2
+	}
+	return len(rawHex) / 2
+}
+
+func jsonString(value any) string {
+	data, err := json.Marshal(value)
+	if err != nil {
+		return "{}"
+	}
+	return string(data)
+}
+
+func floatField(env envelope.FrameEnvelope, key string) (float64, bool) {
+	if env.Fields == nil {
+		return 0, false
+	}
+	value, ok := env.Fields[key]
+	if !ok || value == nil {
+		return 0, false
+	}
+	switch typed := value.(type) {
+	case float64:
+		return typed, true
+	case float32:
+		return float64(typed), true
+	case int:
+		return float64(typed), true
+	case int64:
+		return float64(typed), true
+	case uint16:
+		return float64(typed), true
+	case uint32:
+		return float64(typed), true
+	case string:
+		parsed, err := strconv.ParseFloat(strings.TrimSpace(typed), 64)
+		return parsed, err == nil
+	default:
+		return 0, false
+	}
+}
+
+func floatFieldOrNil(env envelope.FrameEnvelope, key string) any {
+	value, ok := floatField(env, key)
+	if !ok {
+		return nil
+	}
+	return value
+}
+
+func intFieldOrNil(env envelope.FrameEnvelope, key string) any {
+	if env.Fields == nil {
+		return nil
+	}
+	value, ok := env.Fields[key]
+	if !ok || value == nil {
+		return nil
+	}
+	switch typed := value.(type) {
+	case int:
+		return typed
+	case int64:
+		return typed
+	case uint16:
+		return int64(typed)
+	case uint32:
+		return int64(typed)
+	case float64:
+		return int64(typed)
+	case string:
+		parsed, err := strconv.ParseInt(strings.TrimSpace(typed), 10, 64)
+		if err == nil {
+			return parsed
+		}
+	}
+	return nil
+}
+
+func eventTimeOrReceived(env envelope.FrameEnvelope) time.Time {
+	if env.EventTimeMS > 0 {
+		return millis(env.EventTimeMS)
+	}
+	return millis(env.ReceivedAtMS)
+}
+
+func millis(value int64) time.Time {
+	if value <= 0 {
+		return time.UnixMilli(time.Now().UnixMilli()).UTC()
+	}
+	return time.UnixMilli(value).UTC()
+}
+
+func quote(value string) string {
+	return strings.ReplaceAll(value, "'", "''")
+}

go/vehicle-gateway/internal/history/writer_test.go

@@ -0,0 +1,118 @@
+package history
+
+import (
+	"context"
+	"database/sql"
+	"strings"
+	"testing"
+
+	"lingniu-vehicle-ingest/go/vehicle-gateway/internal/envelope"
+)
+
+func TestSchemaStatementsCreateCoreStables(t *testing.T) {
+	statements := strings.Join(SchemaStatements("test_ts"), "\n")
+	for _, want := range []string{"CREATE DATABASE IF NOT EXISTS test_ts", "raw_frames", "vehicle_locations", "vehicle_mileage_points"} {
+		if !strings.Contains(statements, want) {
+			t.Fatalf("schema missing %q:\n%s", want, statements)
+		}
+	}
+}
+
+func TestWriterAppendsRawLocationAndMileage(t *testing.T) {
+	exec := &recordingExec{}
+	writer := NewWriter(exec)
+	env := sampleEnvelope()
+
+	if err := writer.AppendAll(context.Background(), env); err != nil {
+		t.Fatalf("AppendAll() error = %v", err)
+	}
+
+	if got := countSQL(exec.calls, "USING raw_frames"); got != 1 {
+		t.Fatalf("raw child create count = %d", got)
+	}
+	if got := countSQL(exec.calls, "USING vehicle_locations"); got != 1 {
+		t.Fatalf("location child create count = %d", got)
+	}
+	if got := countSQL(exec.calls, "USING vehicle_mileage_points"); got != 1 {
+		t.Fatalf("mileage child create count = %d", got)
+	}
+	if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
+		t.Fatalf("raw insert count = %d", got)
+	}
+	if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 1 {
+		t.Fatalf("location insert count = %d", got)
+	}
+	if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 1 {
+		t.Fatalf("mileage insert count = %d", got)
+	}
+}
+
+func TestWriterSkipsSparseDerivedRows(t *testing.T) {
+	exec := &recordingExec{}
+	writer := NewWriter(exec)
+	env := sampleEnvelope()
+	env.Fields = map[string]any{}
+
+	if err := writer.AppendAll(context.Background(), env); err != nil {
+		t.Fatalf("AppendAll() error = %v", err)
+	}
+
+	if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
+		t.Fatalf("raw insert count = %d", got)
+	}
+	if got := countSQL(exec.calls, "INSERT INTO loc_"); got != 0 {
+		t.Fatalf("location insert count = %d", got)
+	}
+	if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 0 {
+		t.Fatalf("mileage insert count = %d", got)
+	}
+}
+
+func sampleEnvelope() envelope.FrameEnvelope {
+	return envelope.FrameEnvelope{
+		Protocol:       envelope.ProtocolJT808,
+		MessageID:      "0x0200",
+		Sequence:       1,
+		VIN:            "LNBVIN00000000001",
+		Phone:          "013307795425",
+		SourceEndpoint: "127.0.0.1:18080",
+		EventTimeMS:    1782745114000,
+		ReceivedAtMS:   1782745114999,
+		RawHex:         "7E0200",
+		Parsed:         map[string]any{"message": "location"},
+		Fields: map[string]any{
+			envelope.FieldLongitude:      121.069881,
+			envelope.FieldLatitude:       30.590151,
+			envelope.FieldSpeedKMH:       23.0,
+			envelope.FieldTotalMileageKM: 10241.2,
+			"direction_deg":              uint16(79),
+			"alarm_flag":                 uint32(0),
+			"status_flag":                uint32(72),
+		},
+		ParseStatus: envelope.ParseOK,
+	}
+}
+
+func countSQL(calls []execCall, pattern string) int {
+	var count int
+	for _, call := range calls {
+		if strings.Contains(call.query, pattern) {
+			count++
+		}
+	}
+	return count
+}
+
+type execCall struct {
+	query string
+	args  []any
+}
+
+type recordingExec struct {
+	calls []execCall
+}
+
+func (e *recordingExec) ExecContext(_ context.Context, query string, args ...any) (sql.Result, error) {
+	e.calls = append(e.calls, execCall{query: query, args: args})
+	return nil, nil
+}