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

fix: return real pagination totals

Browse Source
This commit is contained in:
lingniu
2026-07-02 01:14:06 +08:00
parent 4f2cded760
commit 2c224396c9
4 changed files with 174 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -207,6 +207,12 @@ func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]
return out, rows.Err()
}
func (r *RawFrameRepository) Count(ctx context.Context, query RawFrameQuery) (int64, error) {
query = normalizeRawFrameQuery(query)
sqlText, args := buildRawFrameCountSQL(r.tableName(), query)
return countRows(ctx, r.db, sqlText, args...)
}
func (r *LocationRepository) Query(ctx context.Context, query LocationQuery) ([]LocationRow, error) {
query = normalizeLocationQuery(query)
sqlText, args := buildLocationSQL(r.tableName(), query)
@@ -261,6 +267,12 @@ func (r *LocationRepository) Query(ctx context.Context, query LocationQuery) ([]
return out, rows.Err()
}
func (r *LocationRepository) Count(ctx context.Context, query LocationQuery) (int64, error) {
query = normalizeLocationQuery(query)
sqlText, args := buildLocationCountSQL(r.tableName(), query)
return countRows(ctx, r.db, sqlText, args...)
}
func (r *MileagePointRepository) Query(ctx context.Context, query MileagePointQuery) ([]MileagePointRow, error) {
query = normalizeMileagePointQuery(query)
sqlText, args := buildMileagePointSQL(r.tableName(), query)
@@ -305,6 +317,27 @@ func (r *MileagePointRepository) Query(ctx context.Context, query MileagePointQu
return out, rows.Err()
}
func (r *MileagePointRepository) Count(ctx context.Context, query MileagePointQuery) (int64, error) {
query = normalizeMileagePointQuery(query)
sqlText, args := buildMileagePointCountSQL(r.tableName(), query)
return countRows(ctx, r.db, sqlText, args...)
}
func countRows(ctx context.Context, db Queryer, sqlText string, args ...any) (int64, error) {
rows, err := db.QueryContext(ctx, sqlText, args...)
if err != nil {
return 0, err
}
defer rows.Close()
var total int64
if rows.Next() {
if err := rows.Scan(&total); err != nil {
return 0, err
}
}
return total, rows.Err()
}
func (r *RawFrameRepository) tableName() string {
if r.database == "" {
return "raw_frames"
@@ -370,6 +403,60 @@ func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery {
}
func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) {
where := rawFrameWhere(query)
sqlText := `SELECT 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, 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
}
func buildRawFrameCountSQL(table string, query RawFrameQuery) (string, []any) {
sqlText := `SELECT COUNT(*) FROM ` + table
if where := rawFrameWhere(query); len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
return sqlText, nil
}
func buildLocationSQL(table string, query LocationQuery) (string, []any) {
where := locationWhere(query)
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
}
func buildLocationCountSQL(table string, query LocationQuery) (string, []any) {
sqlText := `SELECT COUNT(*) FROM ` + table
if where := locationWhere(query); len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
return sqlText, nil
}
func buildMileagePointSQL(table string, query MileagePointQuery) (string, []any) {
where := mileagePointWhere(query)
sqlText := `SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, 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
}
func buildMileagePointCountSQL(table string, query MileagePointQuery) (string, []any) {
sqlText := `SELECT COUNT(*) FROM ` + table
if where := mileagePointWhere(query); len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
return sqlText, nil
}
func rawFrameWhere(query RawFrameQuery) []string {
var where []string
add := func(clause string) {
where = append(where, clause)
@@ -400,15 +487,10 @@ func buildRawFrameSQL(table string, query RawFrameQuery) (string, []any) {
if query.DateTo != "" {
add("ts <= '" + quote(query.DateTo) + "'")
}
sqlText := `SELECT 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, 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
return where
}
func buildLocationSQL(table string, query LocationQuery) (string, []any) {
func locationWhere(query LocationQuery) []string {
var where []string
add := func(clause string) {
where = append(where, clause)
@@ -434,15 +516,10 @@ func buildLocationSQL(table string, query LocationQuery) (string, []any) {
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
return where
}
func buildMileagePointSQL(table string, query MileagePointQuery) (string, []any) {
func mileagePointWhere(query MileagePointQuery) []string {
var where []string
add := func(clause string) {
where = append(where, clause)
@@ -468,12 +545,7 @@ func buildMileagePointSQL(table string, query MileagePointQuery) (string, []any)
if query.DateTo != "" {
add("ts <= '" + quote(query.DateTo) + "'")
}
sqlText := `SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, 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
return where
}
type RawFrameHandler struct {
@@ -523,6 +595,11 @@ func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
writeHistoryError(w, http.StatusBadRequest, err.Error())
return
}
total, err := h.repository.Count(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
return
}
rows, err := h.repository.Query(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
@@ -531,7 +608,7 @@ func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": len(rows),
"total": total,
"limit": query.Limit,
"offset": query.Offset,
})
@@ -551,6 +628,11 @@ func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
writeHistoryError(w, http.StatusBadRequest, err.Error())
return
}
total, err := h.repository.Count(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
return
}
rows, err := h.repository.Query(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
@@ -559,7 +641,7 @@ func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": len(rows),
"total": total,
"limit": query.Limit,
"offset": query.Offset,
})
@@ -579,6 +661,11 @@ func (h *MileagePointHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
writeHistoryError(w, http.StatusBadRequest, err.Error())
return
}
total, err := h.repository.Count(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
return
}
rows, err := h.repository.Query(r.Context(), query)
if err != nil {
writeHistoryError(w, http.StatusInternalServerError, err.Error())
@@ -587,7 +674,7 @@ func (h *MileagePointHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": len(rows),
"total": total,
"limit": query.Limit,
"offset": query.Offset,
})

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

@@ -63,6 +63,8 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(38))
mock.ExpectQuery("SELECT 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, protocol, vehicle_key, vin, phone, device_id FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
@@ -84,7 +86,7 @@ func TestRawFrameHandlerReturnsRawFrames(t *testing.T) {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
body := response.Body.String()
for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"message_id_hex":"0x0002"`, `"total":1`} {
for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"message_id_hex":"0x0002"`, `"total":38`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
@@ -100,6 +102,8 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.raw_frames").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(11))
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "frame_id", "event_id", "message_id", "event_time", "received_at", "raw_size_bytes",
@@ -121,7 +125,7 @@ func TestRawFrameHandlerFiltersByVehicleKey(t *testing.T) {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
body := response.Body.String()
for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"phone":"013307811350"`, `"total":1`} {
for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"phone":"013307811350"`, `"total":11`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
@@ -137,6 +141,8 @@ func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM lingniu_vehicle_ts.vehicle_locations").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(17))
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "event_id", "frame_id", "received_at", "longitude", "latitude", "altitude_m", "speed_kmh",
@@ -157,7 +163,7 @@ func TestLocationHandlerReturnsLocationsByVehicleKey(t *testing.T) {
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`} {
for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"longitude":121.07764`, `"total_mileage_km":8792.8`, `"total":17`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
@@ -173,6 +179,8 @@ 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").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(19))
mock.ExpectQuery("vehicle_key = 'JT808:013307811350'").
WillReturnRows(sqlmock.NewRows([]string{
"ts", "event_id", "frame_id", "received_at", "total_mileage_km", "speed_kmh", "longitude", "latitude",
@@ -193,7 +201,7 @@ func TestMileagePointHandlerReturnsMileageByVehicleKey(t *testing.T) {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
body := response.Body.String()
for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"total_mileage_km":8792.8`, `"speed_kmh":8`, `"total":1`} {
for _, want := range []string{`"vehicle_key":"JT808:013307811350"`, `"total_mileage_km":8792.8`, `"speed_kmh":8`, `"total":19`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}

53
go/vehicle-gateway/internal/stats/query.go
View File

@@ -102,6 +102,24 @@ func (r *MetricRepository) Query(ctx context.Context, query MetricQuery) ([]Metr
return out, rows.Err()
}
func (r *MetricRepository) Count(ctx context.Context, query MetricQuery) (int64, error) {
query = normalizeMetricQuery(query)
sqlText, args := buildMetricCountSQL(query)
rows, err := r.db.QueryContext(ctx, sqlText, args...)
if err != nil {
return 0, err
}
defer rows.Close()
var total int64
if rows.Next() {
if err := rows.Scan(&total); err != nil {
return 0, err
}
}
return total, rows.Err()
}
func normalizeMetricQuery(query MetricQuery) MetricQuery {
query.VehicleKey = strings.TrimSpace(query.VehicleKey)
query.VIN = strings.TrimSpace(query.VIN)
@@ -116,6 +134,26 @@ func normalizeMetricQuery(query MetricQuery) MetricQuery {
}
func buildMetricSQL(query MetricQuery) (string, []any) {
where, args := buildMetricWhere(query)
sqlText := `SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric`
if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
sqlText += " ORDER BY stat_date DESC, vehicle_key ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?"
args = append(args, query.Limit, query.Offset)
return sqlText, args
}
func buildMetricCountSQL(query MetricQuery) (string, []any) {
where, args := buildMetricWhere(query)
sqlText := `SELECT COUNT(*) FROM vehicle_daily_metric`
if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
return sqlText, args
}
func buildMetricWhere(query MetricQuery) ([]string, []any) {
var where []string
var args []any
add := func(clause string, value any) {
@@ -140,13 +178,7 @@ func buildMetricSQL(query MetricQuery) (string, []any) {
if query.DateTo != "" {
add("stat_date <= ?", query.DateTo)
}
sqlText := `SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric`
if len(where) > 0 {
sqlText += " WHERE " + strings.Join(where, " AND ")
}
sqlText += " ORDER BY stat_date DESC, vehicle_key ASC, protocol ASC, metric_key ASC LIMIT ? OFFSET ?"
args = append(args, query.Limit, query.Offset)
return sqlText, args
return where, args
}
type MetricHandler struct {
@@ -174,6 +206,11 @@ func (h *MetricHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
writeMetricError(w, http.StatusBadRequest, err.Error())
return
}
total, err := h.repository.Count(r.Context(), query)
if err != nil {
writeMetricError(w, http.StatusInternalServerError, err.Error())
return
}
rows, err := h.repository.Query(r.Context(), query)
if err != nil {
writeMetricError(w, http.StatusInternalServerError, err.Error())
@@ -182,7 +219,7 @@ func (h *MetricHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": len(rows),
"total": total,
"limit": query.Limit,
"offset": query.Offset,
})

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

@@ -63,6 +63,9 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_daily_metric").
WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(42))
mock.ExpectQuery("SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric").
WithArgs("LB9A32A21R0LS1707", "GB32960", "2020-07-01", "2020-07-01", 50, 0).
WillReturnRows(sqlmock.NewRows([]string{
@@ -84,7 +87,7 @@ func TestMetricHandlerReturnsDailyMetrics(t *testing.T) {
t.Fatalf("status = %d body=%s", response.Code, response.Body.String())
}
body := response.Body.String()
for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"metric_key":"daily_mileage_km"`, `"total":1`} {
for _, want := range []string{`"vin":"LB9A32A21R0LS1707"`, `"metric_key":"daily_mileage_km"`, `"total":42`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
@@ -100,6 +103,9 @@ func TestMetricHandlerFiltersByVehicleKey(t *testing.T) {
t.Fatalf("sqlmock.New() error = %v", err)
}
defer db.Close()
mock.ExpectQuery("SELECT COUNT\\(\\*\\) FROM vehicle_daily_metric").
WithArgs("JT808:013307811254", "JT808").
WillReturnRows(sqlmock.NewRows([]string{"total"}).AddRow(12))
mock.ExpectQuery("SELECT vehicle_key, vin, stat_date, protocol, metric_key, metric_value, metric_unit, first_total_mileage_km, latest_total_mileage_km, sample_count, calculation_method, created_at, updated_at FROM vehicle_daily_metric").
WithArgs("JT808:013307811254", "JT808", 50, 0).
WillReturnRows(sqlmock.NewRows([]string{