From 5d844701ff12abb8c082695e62094a95d6da79ac Mon Sep 17 00:00:00 2001 From: lingniu Date: Wed, 1 Jul 2026 21:47:38 +0800 Subject: [PATCH] feat: add go tdengine history writer --- go/vehicle-gateway/cmd/history-writer/main.go | 123 +++++++ go/vehicle-gateway/go.mod | 15 +- go/vehicle-gateway/go.sum | 41 +++ go/vehicle-gateway/internal/history/schema.go | 71 +++++ go/vehicle-gateway/internal/history/writer.go | 300 ++++++++++++++++++ .../internal/history/writer_test.go | 118 +++++++ 6 files changed, 665 insertions(+), 3 deletions(-) create mode 100644 go/vehicle-gateway/cmd/history-writer/main.go create mode 100644 go/vehicle-gateway/internal/history/schema.go create mode 100644 go/vehicle-gateway/internal/history/writer.go create mode 100644 go/vehicle-gateway/internal/history/writer_test.go diff --git a/go/vehicle-gateway/cmd/history-writer/main.go b/go/vehicle-gateway/cmd/history-writer/main.go new file mode 100644 index 00000000..7a99a4a2 --- /dev/null +++ b/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 +} diff --git a/go/vehicle-gateway/go.mod b/go/vehicle-gateway/go.mod index 9e362c5e..f5c6af4f 100644 --- a/go/vehicle-gateway/go.mod +++ b/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/pierrec/lz4/v4 v4.1.15 // indirect - github.com/segmentio/kafka-go v0.4.49 // indirect + github.com/segmentio/kafka-go v0.4.49 + github.com/taosdata/driver-go/v3 v3.8.1 +) + +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 ) diff --git a/go/vehicle-gateway/go.sum b/go/vehicle-gateway/go.sum index 28ba83df..d5ca35d3 100644 --- a/go/vehicle-gateway/go.sum +++ b/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= diff --git a/go/vehicle-gateway/internal/history/schema.go b/go/vehicle-gateway/internal/history/schema.go new file mode 100644 index 00000000..3b48970e --- /dev/null +++ b/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) +)`, + } +} diff --git a/go/vehicle-gateway/internal/history/writer.go b/go/vehicle-gateway/internal/history/writer.go new file mode 100644 index 00000000..64333c97 --- /dev/null +++ b/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, "'", "''") +} diff --git a/go/vehicle-gateway/internal/history/writer_test.go b/go/vehicle-gateway/internal/history/writer_test.go new file mode 100644 index 00000000..753b0532 --- /dev/null +++ b/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 +}