From fb9ab8525af039d539591d3ebbf200465822d8c5 Mon Sep 17 00:00:00 2001 From: lingniu Date: Thu, 2 Jul 2026 00:27:20 +0800 Subject: [PATCH] feat: add vehicle location history query --- go/vehicle-gateway/cmd/realtime-api/main.go | 11 +- go/vehicle-gateway/internal/history/query.go | 237 ++++++++++++++++++ .../internal/history/query_test.go | 60 +++++ 3 files changed, 304 insertions(+), 4 deletions(-) diff --git a/go/vehicle-gateway/cmd/realtime-api/main.go b/go/vehicle-gateway/cmd/realtime-api/main.go index a98411bc..f804686e 100644 --- a/go/vehicle-gateway/cmd/realtime-api/main.go +++ b/go/vehicle-gateway/cmd/realtime-api/main.go @@ -92,14 +92,17 @@ func main() { closeHistory = func() { _ = db.Close() } database := env("TDENGINE_DATABASE", history.DefaultDatabase) mux.Handle("/api/history/raw-frames", history.NewRawFrameHandler(history.NewRawFrameRepository(db, database))) - logger.Info("history raw frame query enabled", "driver", driver, "database", database) + mux.Handle("/api/history/locations", history.NewLocationHandler(history.NewLocationRepository(db, database))) + logger.Info("history query enabled", "driver", driver, "database", database) } else { - mux.HandleFunc("/api/history/raw-frames", func(w http.ResponseWriter, _ *http.Request) { + historyUnavailable := func(w http.ResponseWriter, _ *http.Request) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(http.StatusServiceUnavailable) _ = json.NewEncoder(w).Encode(map[string]any{"error": "TDENGINE_DSN is not configured"}) - }) - logger.Warn("TDENGINE_DSN is empty; raw frame query api disabled") + } + mux.HandleFunc("/api/history/raw-frames", historyUnavailable) + mux.HandleFunc("/api/history/locations", historyUnavailable) + logger.Warn("TDENGINE_DSN is empty; history query api disabled") } defer closeHistory() diff --git a/go/vehicle-gateway/internal/history/query.go b/go/vehicle-gateway/internal/history/query.go index 16062cda..0b41eeca 100644 --- a/go/vehicle-gateway/internal/history/query.go +++ b/go/vehicle-gateway/internal/history/query.go @@ -52,6 +52,38 @@ type RawFrameRow struct { DeviceID string `json:"device_id,omitempty"` } +type LocationQuery struct { + Protocol string + VehicleKey string + VIN string + Phone string + DeviceID string + DateFrom string + DateTo string + Limit int + Offset int +} + +type LocationRow struct { + TS string `json:"ts"` + EventID string `json:"event_id"` + FrameID string `json:"frame_id"` + ReceivedAt string `json:"received_at"` + Longitude float64 `json:"longitude"` + Latitude float64 `json:"latitude"` + AltitudeM *float64 `json:"altitude_m,omitempty"` + SpeedKMH *float64 `json:"speed_kmh,omitempty"` + DirectionDeg *int64 `json:"direction_deg,omitempty"` + AlarmFlag *int64 `json:"alarm_flag,omitempty"` + StatusFlag *int64 `json:"status_flag,omitempty"` + TotalMileageKM *float64 `json:"total_mileage_km,omitempty"` + Protocol string `json:"protocol"` + VehicleKey string `json:"vehicle_key"` + VIN string `json:"vin"` + Phone string `json:"phone,omitempty"` + DeviceID string `json:"device_id,omitempty"` +} + type RawFrameRepository struct { db Queryer database string @@ -68,6 +100,22 @@ func NewRawFrameRepository(db Queryer, database string) *RawFrameRepository { return &RawFrameRepository{db: db, database: database} } +type LocationRepository struct { + db Queryer + database string +} + +func NewLocationRepository(db Queryer, database string) *LocationRepository { + if db == nil { + panic("location query db must not be nil") + } + database = strings.TrimSpace(database) + if database != "" && !safeIdentifier(database) { + database = "" + } + return &LocationRepository{db: db, database: database} +} + func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]RawFrameRow, error) { query = normalizeRawFrameQuery(query) sqlText, args := buildRawFrameSQL(r.tableName(), query) @@ -115,6 +163,60 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([] return out, rows.Err() } +func (r *LocationRepository) Query(ctx context.Context, query LocationQuery) ([]LocationRow, error) { + query = normalizeLocationQuery(query) + sqlText, args := buildLocationSQL(r.tableName(), query) + rows, err := r.db.QueryContext(ctx, sqlText, args...) + if err != nil { + return nil, err + } + defer rows.Close() + + var out []LocationRow + for rows.Next() { + var row LocationRow + var ts scanDateTime + var receivedAt scanDateTime + var altitude sql.NullFloat64 + var speed sql.NullFloat64 + var direction sql.NullInt64 + var alarm sql.NullInt64 + var status sql.NullInt64 + var mileage sql.NullFloat64 + if err := rows.Scan( + &ts, + &row.EventID, + &row.FrameID, + &receivedAt, + &row.Longitude, + &row.Latitude, + &altitude, + &speed, + &direction, + &alarm, + &status, + &mileage, + &row.Protocol, + &row.VehicleKey, + &row.VIN, + &row.Phone, + &row.DeviceID, + ); err != nil { + return nil, err + } + row.TS = ts.String + row.ReceivedAt = receivedAt.String + row.AltitudeM = nullableFloat(altitude) + row.SpeedKMH = nullableFloat(speed) + row.DirectionDeg = nullableInt(direction) + row.AlarmFlag = nullableInt(alarm) + row.StatusFlag = nullableInt(status) + row.TotalMileageKM = nullableFloat(mileage) + out = append(out, row) + } + return out, rows.Err() +} + func (r *RawFrameRepository) tableName() string { if r.database == "" { return "raw_frames" @@ -122,6 +224,13 @@ func (r *RawFrameRepository) tableName() string { return r.database + ".raw_frames" } +func (r *LocationRepository) tableName() string { + if r.database == "" { + return "vehicle_locations" + } + return r.database + ".vehicle_locations" +} + func normalizeRawFrameQuery(query RawFrameQuery) RawFrameQuery { query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) query.VehicleKey = strings.TrimSpace(query.VehicleKey) @@ -137,6 +246,20 @@ func normalizeRawFrameQuery(query RawFrameQuery) RawFrameQuery { return query } +func normalizeLocationQuery(query LocationQuery) LocationQuery { + query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol)) + query.VehicleKey = strings.TrimSpace(query.VehicleKey) + query.VIN = strings.TrimSpace(query.VIN) + query.Phone = strings.TrimSpace(query.Phone) + query.DeviceID = strings.TrimSpace(query.DeviceID) + query.DateFrom = strings.TrimSpace(query.DateFrom) + query.DateTo = strings.TrimSpace(query.DateTo) + if query.Limit <= 0 { + query.Limit = 20 + } + return query +} + func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) { var where []string add := func(clause string) { @@ -176,10 +299,48 @@ func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) { return sqlText, nil } +func buildLocationSQL(table string, query LocationQuery) (string, []any) { + var where []string + add := func(clause string) { + where = append(where, clause) + } + if query.Protocol != "" { + add("protocol = '" + quote(query.Protocol) + "'") + } + if query.VehicleKey != "" { + add("vehicle_key = '" + quote(query.VehicleKey) + "'") + } + if query.VIN != "" { + add("vin = '" + quote(query.VIN) + "'") + } + if query.Phone != "" { + add("phone = '" + quote(query.Phone) + "'") + } + if query.DeviceID != "" { + add("device_id = '" + quote(query.DeviceID) + "'") + } + if query.DateFrom != "" { + add("ts >= '" + quote(query.DateFrom) + "'") + } + if query.DateTo != "" { + add("ts <= '" + quote(query.DateTo) + "'") + } + sqlText := `SELECT ts, event_id, frame_id, received_at, longitude, latitude, altitude_m, speed_kmh, direction_deg, alarm_flag, status_flag, total_mileage_km, protocol, vehicle_key, vin, phone, device_id FROM ` + table + if len(where) > 0 { + sqlText += " WHERE " + strings.Join(where, " AND ") + } + sqlText += " ORDER BY ts DESC LIMIT " + strconv.Itoa(query.Limit) + " OFFSET " + strconv.Itoa(query.Offset) + return sqlText, nil +} + type RawFrameHandler struct { repository *RawFrameRepository } +type LocationHandler struct { + repository *LocationRepository +} + func NewRawFrameHandler(repository *RawFrameRepository) *RawFrameHandler { if repository == nil { panic("raw frame repository must not be nil") @@ -187,6 +348,13 @@ func NewRawFrameHandler(repository *RawFrameRepository) *RawFrameHandler { return &RawFrameHandler{repository: repository} } +func NewLocationHandler(repository *LocationRepository) *LocationHandler { + if repository == nil { + panic("location repository must not be nil") + } + return &LocationHandler{repository: repository} +} + func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { if r.Method != http.MethodGet { writeHistoryError(w, http.StatusMethodNotAllowed, "method not allowed") @@ -215,6 +383,34 @@ func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { }) } +func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) { + if r.Method != http.MethodGet { + writeHistoryError(w, http.StatusMethodNotAllowed, "method not allowed") + return + } + if strings.Trim(r.URL.Path, "/") != "api/history/locations" { + writeHistoryError(w, http.StatusNotFound, "route not found") + return + } + query, err := parseLocationQuery(r) + if err != nil { + writeHistoryError(w, http.StatusBadRequest, err.Error()) + return + } + rows, err := h.repository.Query(r.Context(), query) + if err != nil { + writeHistoryError(w, http.StatusInternalServerError, err.Error()) + return + } + w.Header().Set("Content-Type", "application/json") + _ = json.NewEncoder(w).Encode(map[string]any{ + "items": rows, + "total": len(rows), + "limit": query.Limit, + "offset": query.Offset, + }) +} + func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) { values := r.URL.Query() limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit") @@ -248,6 +444,33 @@ func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) { return normalizeRawFrameQuery(query), nil } +func parseLocationQuery(r *http.Request) (LocationQuery, error) { + values := r.URL.Query() + limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit") + if err != nil { + return LocationQuery{}, err + } + offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset") + if err != nil { + return LocationQuery{}, err + } + query := LocationQuery{ + Protocol: values.Get("protocol"), + VehicleKey: values.Get("vehicleKey"), + VIN: values.Get("vin"), + Phone: values.Get("phone"), + DeviceID: values.Get("deviceId"), + DateFrom: values.Get("dateFrom"), + DateTo: values.Get("dateTo"), + Limit: limit, + Offset: offset, + } + if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) { + return LocationQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss") + } + return normalizeLocationQuery(query), nil +} + func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) { raw = strings.TrimSpace(raw) if raw == "" { @@ -323,6 +546,20 @@ func formatSQLTime(value any, layout string) string { } } +func nullableFloat(value sql.NullFloat64) *float64 { + if !value.Valid { + return nil + } + return &value.Float64 +} + +func nullableInt(value sql.NullInt64) *int64 { + if !value.Valid { + return nil + } + return &value.Int64 +} + func writeHistoryError(w http.ResponseWriter, status int, message string) { w.Header().Set("Content-Type", "application/json") w.WriteHeader(status) diff --git a/go/vehicle-gateway/internal/history/query_test.go b/go/vehicle-gateway/internal/history/query_test.go index dd6a9c1f..09911b36 100644 --- a/go/vehicle-gateway/internal/history/query_test.go +++ b/go/vehicle-gateway/internal/history/query_test.go @@ -131,6 +131,42 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) { } } +func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) { + db, mock, err := sqlmock.New() + if err != nil { + t.Fatalf("sqlmock.New() error = %v", err) + } + defer db.Close() + mock.ExpectQuery("vehicle_key = 'JT808:013307811350'"). + WillReturnRows(sqlmock.NewRows([]string{ + "ts", "event_id", "frame_id", "received_at", "longitude", "latitude", "altitude_m", "speed_kmh", + "direction_deg", "alarm_flag", "status_flag", "total_mileage_km", "protocol", "vehicle_key", "vin", "phone", "device_id", + }).AddRow( + "2026-07-02 00:18:22", "event-3", "go_frame", "2026-07-02 00:22:43", + 121.07764, 30.585928, 11.0, 8.0, 171, 0, 4718595, 8792.8, + "JT808", "JT808:013307811350", "", "013307811350", "", + )) + + handler := NewLocationHandler(NewLocationRepository(db, "lingniu_vehicle_ts")) + request := httptest.NewRequest(http.MethodGet, "/api/history/locations?vehicleKey=JT808:013307811350&protocol=JT808&limit=1", nil) + response := httptest.NewRecorder() + + handler.ServeHTTP(response, request) + + if response.Code != http.StatusOK { + t.Fatalf("status = %d body=%s", response.Code, response.Body.String()) + } + body := response.Body.String() + for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"longitude":121.07764`, `"total_mileage_km":8792.8`, `"total":1`} { + if !strings.Contains(body, want) { + t.Fatalf("response missing %s: %s", want, body) + } + } + if err := mock.ExpectationsWereMet(); err != nil { + t.Fatalf("sql expectations: %v", err) + } +} + func TestRawFrameHandlerRejectsInvalidLimit(t *testing.T) { handler := NewRawFrameHandler(NewRawFrameRepository(&sql.DB{}, "")) request := httptest.NewRequest(http.MethodGet, "/api/history/raw-frames?limit=501", nil) @@ -156,6 +192,30 @@ func TestParseMessageIDSupportsDecimalAndHex(t *testing.T) { } } +func TestBuildLocationSQLUsesLiteralsForTDengine(t *testing.T) { + sqlText, args := buildLocationSQL("lingniu_vehicle_ts.vehicle_locations", LocationQuery{ + Protocol: "JT808", + VehicleKey: "JT808:013307811350", + DateFrom: "2026-07-02 00:00:00", + DateTo: "2026-07-02 23:59:59", + Limit: 20, + Offset: 5, + }) + if len(args) != 0 { + t.Fatalf("expected no query args for TDengine, got %#v", args) + } + for _, want := range []string{ + "protocol = 'JT808'", + "vehicle_key = 'JT808:013307811350'", + "ts >= '2026-07-02 00:00:00'", + "LIMIT 20 OFFSET 5", + } { + if !strings.Contains(sqlText, want) { + t.Fatalf("sql missing %s: %s", want, sqlText) + } + } +} + func TestBuildRawFrameSQLUsesLiteralsForTDengine(t *testing.T) { sqlText, args := buildRawFrameSQL("lingniu_vehicle_ts.raw_frames", RawFrameQuery{ Protocol: "JT808",