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

feat(go): add post raw frame query

Browse Source
This commit is contained in:
lingniu
2026-07-03 14:54:52 +08:00
parent c0d3b41550
commit 28c81611a1
4 changed files with 169 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
go/vehicle-gateway/cmd/realtime-api/main.go
View File

@@ -135,7 +135,9 @@ func main() {
healthChecks = append(healthChecks, health.Check{Name: "tdengine", Check: db.PingContext})
closeHistory = func() { _ = db.Close() }
database := env("TDENGINE_DATABASE", history.DefaultDatabase)
mux.Handle("/api/history/raw-frames", history.NewRawFrameHandler(history.NewRawFrameRepository(db, database)))
rawFrameHandler := history.NewRawFrameHandler(history.NewRawFrameRepository(db, database))
mux.Handle("/api/history/raw-frames", rawFrameHandler)
mux.Handle("/api/history/raw-frames/query", rawFrameHandler)
mux.Handle("/api/history/locations", history.NewLocationHandler(history.NewLocationRepository(db, database)))
logger.Info("history query enabled", "driver", driver, "database", database)
} else {
@@ -145,6 +147,7 @@ func main() {
_ = json.NewEncoder(w).Encode(map[string]any{"error": "TDENGINE_DSN is not configured"})
}
mux.HandleFunc("/api/history/raw-frames", historyUnavailable)
mux.HandleFunc("/api/history/raw-frames/query", historyUnavailable)
mux.HandleFunc("/api/history/locations", historyUnavailable)
logger.Warn("TDENGINE_DSN is empty; history query api disabled")
}

10
go/vehicle-gateway/cmd/realtime-api/main_test.go
View File

@@ -164,6 +164,16 @@ func TestRealtimeAPICachesBindingPlateResolver(t *testing.T) {
}
}
func TestRealtimeAPIExposesPostRawFrameQueryRoute(t *testing.T) {
source, err := os.ReadFile("main.go")
if err != nil {
t.Fatalf("read main.go: %v", err)
}
if !strings.Contains(string(source), `"/api/history/raw-frames/query"`) {
t.Fatalf("realtime api should expose POST raw frame query route")
}
}
type contextCheckingRealtimeUpdater struct {
ctxErr error
count int

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

@@ -7,6 +7,7 @@ import (
"encoding/json"
"errors"
"fmt"
"io"
"net/http"
"strconv"
"strings"
@@ -83,6 +84,31 @@ type LocationRow struct {
VIN string `json:"vin"`
}
type rawFrameQueryRequest struct {
Protocol string `json:"protocol"`
VehicleKey string `json:"vehicleKey"`
VehicleKeyAlt string `json:"vehicle_key"`
VIN string `json:"vin"`
Phone string `json:"phone"`
DeviceID string `json:"deviceId"`
DeviceIDAlt string `json:"device_id"`
MessageID string `json:"messageId"`
MessageIDAlt string `json:"message_id"`
OrderBy string `json:"orderBy"`
OrderByAlt string `json:"order_by"`
IncludeFields bool `json:"includeFields"`
IncludePayload bool `json:"includePayload"`
IncludeTotal bool `json:"includeTotal"`
Fields []string `json:"fields"`
ParsedFields []string `json:"parsedFields"`
DateFrom string `json:"dateFrom"`
DateFromAlt string `json:"date_from"`
DateTo string `json:"dateTo"`
DateToAlt string `json:"date_to"`
Limit int `json:"limit"`
Offset int `json:"offset"`
}
type RawFrameRepository struct {
db Queryer
database string
@@ -647,15 +673,16 @@ func NewLocationHandler(repository *LocationRepository) *LocationHandler {
}
func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
if r.Method != http.MethodGet && r.Method != http.MethodPost {
writeHistoryError(w, http.StatusMethodNotAllowed, "method not allowed")
return
}
if strings.Trim(r.URL.Path, "/") != "api/history/raw-frames" {
path := strings.Trim(r.URL.Path, "/")
if path != "api/history/raw-frames" && path != "api/history/raw-frames/query" {
writeHistoryError(w, http.StatusNotFound, "route not found")
return
}
query, err := parseRawFrameQuery(r)
query, err := rawFrameQueryFromRequest(r)
if err != nil {
writeHistoryError(w, http.StatusBadRequest, err.Error())
return
@@ -735,6 +762,13 @@ func writeHistoryJSON(w http.ResponseWriter, r *http.Request, value any) {
_ = json.NewEncoder(w).Encode(value)
}
func rawFrameQueryFromRequest(r *http.Request) (RawFrameQuery, error) {
if r.Method == http.MethodPost {
return parseRawFrameQueryBody(r)
}
return parseRawFrameQuery(r)
}
func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) {
values := r.URL.Query()
limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit")
@@ -773,6 +807,72 @@ func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) {
return normalizeRawFrameQuery(query), nil
}
func parseRawFrameQueryBody(r *http.Request) (RawFrameQuery, error) {
defer r.Body.Close()
decoder := json.NewDecoder(http.MaxBytesReader(nil, r.Body, 1<<20))
decoder.DisallowUnknownFields()
var body rawFrameQueryRequest
if err := decoder.Decode(&body); err != nil {
return RawFrameQuery{}, fmt.Errorf("invalid JSON body: %w", err)
}
var extra any
if err := decoder.Decode(&extra); err != io.EOF {
return RawFrameQuery{}, errors.New("invalid JSON body")
}
if body.VehicleKey == "" {
body.VehicleKey = body.VehicleKeyAlt
}
if body.DeviceID == "" {
body.DeviceID = body.DeviceIDAlt
}
if body.MessageID == "" {
body.MessageID = body.MessageIDAlt
}
if body.OrderBy == "" {
body.OrderBy = body.OrderByAlt
}
if body.DateFrom == "" {
body.DateFrom = body.DateFromAlt
}
if body.DateTo == "" {
body.DateTo = body.DateToAlt
}
limit, err := boundedInt(body.Limit, 20, 1, 500, "limit")
if err != nil {
return RawFrameQuery{}, err
}
offset, err := boundedInt(body.Offset, 0, 0, 1_000_000, "offset")
if err != nil {
return RawFrameQuery{}, err
}
query := RawFrameQuery{
Protocol: body.Protocol,
VehicleKey: body.VehicleKey,
VIN: body.VIN,
Phone: body.Phone,
DeviceID: body.DeviceID,
MessageID: body.MessageID,
OrderBy: body.OrderBy,
IncludeFields: body.IncludeFields,
IncludePayload: body.IncludePayload,
IncludeTotal: body.IncludeTotal,
ParsedFields: append(body.Fields, body.ParsedFields...),
DateFrom: body.DateFrom,
DateTo: body.DateTo,
Limit: limit,
Offset: offset,
}
if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) {
return RawFrameQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss")
}
if strings.TrimSpace(query.MessageID) != "" {
if _, ok := parseMessageID(query.MessageID); !ok {
return RawFrameQuery{}, errors.New("messageId must be decimal or hex like 0x0200")
}
}
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")
@@ -798,6 +898,16 @@ func parseLocationQuery(r *http.Request) (LocationQuery, error) {
return normalizeLocationQuery(query), nil
}
func boundedInt(value int, fallback int, min int, max int, name string) (int, error) {
if value == 0 {
return fallback, nil
}
if value < min || value > max {
return 0, fmt.Errorf("%s must be between %d and %d", name, min, max)
}
return value, nil
}
func parseBoundedInt(raw string, fallback int, min int, max int, name string) (int, error) {
raw = strings.TrimSpace(raw)
if raw == "" {

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

@@ -1,6 +1,7 @@
package history
import (
"bytes"
"context"
"database/sql"
"net/http"
@@ -186,6 +187,47 @@ func TestRawFrameHandlerFiltersParsedFields(t *testing.T) {
}
}
func TestRawFrameHandlerAcceptsPostQueryBody(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT ts, frame_id, event_id, message_id, event_time, received_at, raw_size_bytes, .* AS parsed_fields, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_gb32960_").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
"raw_hex", "raw_text", "parsed_fields", "parse_status", "parse_error", "source_endpoint",
"protocol", "vehicle_key", "vin", "phone", "device_id",
}).AddRow(
"2026-07-01 22:28:25", "go_frame", "event-2", 2, "2026-07-01 22:28:25", "2026-07-01 22:28:25",
128, "2323", "", `{"gb32960.vehicle.soc_percent":"85","gb32960.vehicle.speed_kmh":"30","gb32960.vehicle.total_mileage_km":"10000"}`,
"OK", "", "8.134.95.166:53702", "GB32960", "LB9A32A21R0LS1707", "LB9A32A21R0LS1707", "", "",
))
handler := NewRawFrameHandler(NewRawFrameRepository(db, "lingniu_vehicle_ts"))
body := []byte(`{"protocol":"GB32960","vin":"LB9A32A21R0LS1707","limit":1,"fields":["gb32960.vehicle.soc_percent","gb32960.vehicle.speed_kmh"]}`)
request := httptest.NewRequest(http.MethodPost, "/api/history/raw-frames/query", bytes.NewReader(body))
response := httptest.NewRecorder()
handler.ServeHTTP(response, request)
if response.Code != http.StatusOK {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
text := response.Body.String()
for _, want := range []string{`"gb32960.vehicle.soc_percent":"85"`, `"gb32960.vehicle.speed_kmh":"30"`} {
if !strings.Contains(text, want) {
t.Fatalf("response missing %s: %s", want, text)
}
}
if strings.Contains(text, "gb32960.vehicle.total_mileage_km") {
t.Fatalf("response should filter unrequested parsed field: %s", text)
}
if err := mock.ExpectationsWereMet(); err != nil {
t.Fatalf("sql expectations: %v", err)
}
}
func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {