shishengliang/lingniu-vehicle-ingest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(go): stop duplicate mileage history writes

Browse Source
This commit is contained in:
lingniu
2026-07-02 19:42:53 +08:00
parent be559c0f99
commit 0bbe87e6fc
7 changed files with 30 additions and 73 deletions
Download Patch File Download Diff File Expand all files Collapse all files

6
go/vehicle-gateway/internal/history/query.go
View File

@@ -369,9 +369,9 @@ func (r *LocationRepository) tableName() string {
func (r *MileagePointRepository) tableName() string {
if r.database == "" {
return "vehicle_mileage_points"
return "vehicle_locations"
}
return r.database + ".vehicle_mileage_points"
return r.database + ".vehicle_locations"
}
func normalizeRawFrameQuery(query RawFrameQuery) RawFrameQuery {
@@ -676,7 +676,7 @@ func locationWhere(query LocationQuery) []string {
}
func mileagePointWhere(query MileagePointQuery) []string {
var where []string
where := []string{"total_mileage_km IS NOT NULL"}
add := func(clause string) {
where = append(where, clause)
}

8
go/vehicle-gateway/internal/history/query_test.go
View File

@@ -293,9 +293,9 @@ func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_mileage_points").
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_locations").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(19))
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
mock.ExpectQuery("SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.vehicle_locations").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "event_id", "frame_id", "received_at", "total_mileage_km", "speed_kmh", "longitude", "latitude",
"protocol", "vehicle_key", "vin", "phone", "device_id",
@@ -376,7 +376,7 @@ func TestParseMessageIDSupportsDecimalAndHex(t *testing.T) {
}
func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
sqlText, args := buildMileagePointSQL("lingniu_vehicle_ts.vehicle_mileage_points", MileagePointQuery{
sqlText, args := buildMileagePointSQL("lingniu_vehicle_ts.vehicle_locations", MileagePointQuery{
Protocol: "JT808",
VehicleKey: "JT808:013307811350",
DateFrom: "2026-07-02 00:00:00",
@@ -388,6 +388,8 @@ func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
t.Fatalf("expected no query args for TDengine, got %#v", args)
}
for _, want := range []string{
"FROM lingniu_vehicle_ts.vehicle_locations",
"total_mileage_km IS NOT NULL",
"protocol = 'JT808'",
"vehicle_key = 'JT808:013307811350'",
"ts >= '2026-07-01 16:00:00'",

16
go/vehicle-gateway/internal/history/schema.go
View File

@@ -66,22 +66,6 @@ func SchemaStatements(database string) []string {
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)
)`,
}
}

34
go/vehicle-gateway/internal/history/writer.go
View File

@@ -62,10 +62,7 @@ func (w *Writer) AppendAll(ctx context.Context, env envelope.FrameEnvelope) erro
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)
return w.AppendLocation(ctx, env)
}
func (w *Writer) AppendRawFrame(ctx context.Context, env envelope.FrameEnvelope) error {
@@ -121,21 +118,6 @@ VALUES (%s)`, table, joinLiterals(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 (%s)`, table, joinLiterals(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()
@@ -259,20 +241,6 @@ func locationValues(env envelope.FrameEnvelope, longitude float64, latitude floa
}
}
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()
}

15
go/vehicle-gateway/internal/history/writer_test.go
View File

@@ -11,14 +11,17 @@ import (
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"} {
for _, want := range []string{"CREATE DATABASE IF NOT EXISTS test_ts", "raw_frames", "vehicle_locations"} {
if !strings.Contains(statements, want) {
t.Fatalf("schema missing %q:\n%s", want, statements)
}
}
if strings.Contains(statements, "vehicle_mileage_points") {
t.Fatalf("schema should not create duplicate mileage stable:\n%s", statements)
}
}
func TestWriterAppendsRawLocationAndMileage(t *testing.T) {
func TestWriterAppendsRawAndLocationOnly(t *testing.T) {
exec := &recordingExec{}
writer := NewWriter(exec)
env := sampleEnvelope()
@@ -33,8 +36,8 @@ func TestWriterAppendsRawLocationAndMileage(t *testing.T) {
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, "USING vehicle_mileage_points"); got != 0 {
t.Fatalf("duplicate mileage child create count = %d", got)
}
if got := countSQL(exec.calls, "INSERT INTO raw_"); got != 1 {
t.Fatalf("raw insert count = %d", got)
@@ -42,8 +45,8 @@ func TestWriterAppendsRawLocationAndMileage(t *testing.T) {
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)
if got := countSQL(exec.calls, "INSERT INTO mil_"); got != 0 {
t.Fatalf("duplicate mileage insert count = %d", got)
}
rawInsert := findSQL(exec.calls, "INSERT INTO raw_")
if !strings.Contains(rawInsert, "'go_") || !strings.Contains(rawInsert, "'2e") && !strings.Contains(rawInsert, "'") {