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

refactor(go): remove duplicate mileage history api

Browse Source
This commit is contained in:
lingniu
2026-07-02 21:34:12 +08:00
parent e27af63025
commit 7c41b81654
6 changed files with 24 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/architecture/storage-minimal-contract.md
View File

@@ -32,7 +32,7 @@
1. `vehicle_mileage_points` 与 `vehicle_locations.total_mileage_km` 重复。
- 状态：Go 写入链路已停止创建和写入 `vehicle_mileage_points`。
- 兼容：`/api/history/mileage-points` 已改为从 `vehicle_locations` 读取 `total_mileage_km IS NOT NULL`。
- 查询：不再暴露单独 `/api/history/mileage-points`，里程点直接从 `/api/history/locations` 的 `total_mileage_km` 获取。
- 生产：ECS TDengine 历史库已在上线前重建，`vehicle_mileage_points` 不再存在。
2. `raw_frames.fields_json` 与 `raw_frames.parsed_json`、`vehicle_locations` 重复。

3
docs/go-version-data-flow.html
View File

@@ -649,8 +649,7 @@
<thead><tr><th>接口类型</th><th>数据源</th><th>说明</th></tr></thead>
<tbody>
<tr><td>RAW 帧查询</td><td>TDengine raw_frames + chunks</td><td>按协议、VIN/phone、时间、消息类型分页。</td></tr>
<tr><td>位置历史</td><td>TDengine vehicle_locations</td><td>高频位置分页查询，避免每次扫完整 JSON。</td></tr>
<tr><td>里程历史</td><td>TDengine vehicle_locations</td><td>从位置核心表读取总里程点，不再单独维护里程点表。</td></tr>
<tr><td>位置/总里程历史</td><td>TDengine vehicle_locations</td><td>高频位置分页查询，包含 total_mileage_km，避免重复里程点接口。</td></tr>
<tr><td>每日里程</td><td>MySQL vehicle_daily_mileage</td><td>按日期、协议查询首末总里程差值结果。</td></tr>
</tbody>
</table>

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

@@ -122,7 +122,6 @@ func main() {
database := env("TDENGINE_DATABASE", history.DefaultDatabase)
mux.Handle("/api/history/raw-frames", history.NewRawFrameHandler(history.NewRawFrameRepository(db, database)))
mux.Handle("/api/history/locations", history.NewLocationHandler(history.NewLocationRepository(db, database)))
mux.Handle("/api/history/mileage-points", history.NewMileagePointHandler(history.NewMileagePointRepository(db, database)))
logger.Info("history query enabled", "driver", driver, "database", database)
} else {
historyUnavailable := func(w http.ResponseWriter, _ *http.Request) {
@@ -132,7 +131,6 @@ func main() {
}
mux.HandleFunc("/api/history/raw-frames", historyUnavailable)
mux.HandleFunc("/api/history/locations", historyUnavailable)
mux.HandleFunc("/api/history/mileage-points", historyUnavailable)
logger.Warn("TDENGINE_DSN is empty; history query api disabled")
}
defer closeHistory()

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

@@ -3,6 +3,7 @@ package main
import (
"context"
"encoding/json"
"os"
"strings"
"testing"
@@ -103,6 +104,16 @@ func TestKafkaTopicsFromEnvDefaultsToGoUnifiedTopic(t *testing.T) {
}
}
func TestRealtimeAPIDoesNotExposeDuplicateMileagePointRoute(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/mileage-points") {
t.Fatalf("realtime api should expose mileage through /api/history/locations only")
}
}
type contextCheckingRealtimeUpdater struct {
ctxErr error
count int

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

@@ -79,28 +79,6 @@ type LocationRow struct {
VIN string `json:"vin"`
}
type MileagePointQuery struct {
Protocol string
VIN string
DateFrom string
DateTo string
Limit int
Offset int
}
type MileagePointRow struct {
TS string `json:"ts"`
EventID string `json:"event_id"`
FrameID string `json:"frame_id"`
ReceivedAt string `json:"received_at"`
TotalMileageKM float64 `json:"total_mileage_km"`
SpeedKMH *float64 `json:"speed_kmh,omitempty"`
Longitude *float64 `json:"longitude,omitempty"`
Latitude *float64 `json:"latitude,omitempty"`
Protocol string `json:"protocol"`
VIN string `json:"vin"`
}
type RawFrameRepository struct {
db Queryer
database string
@@ -122,11 +100,6 @@ type LocationRepository struct {
database string
}
type MileagePointRepository struct {
db Queryer
database string
}
func NewLocationRepository(db Queryer, database string) *LocationRepository {
if db == nil {
panic("location query db must not be nil")
@@ -138,17 +111,6 @@ func NewLocationRepository(db Queryer, database string) *LocationRepository {
return &LocationRepository{db: db, database: database}
}
func NewMileagePointRepository(db Queryer, database string) *MileagePointRepository {
if db == nil {
panic("mileage point query db must not be nil")
}
database = strings.TrimSpace(database)
if database != "" && !safeIdentifier(database) {
database = ""
}
return &MileagePointRepository{db: db, database: database}
}
func (r *RawFrameRepository) Query(ctx context.Context, query RawFrameQuery) ([]RawFrameRow, error) {
query = normalizeRawFrameQuery(query)
sqlText, args := buildRawFrameSQL(r.tableName(), query)
@@ -264,53 +226,6 @@ func (r *LocationRepository) Count(ctx context.Context, query LocationQuery) (in
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)
rows, err := r.db.QueryContext(ctx, sqlText, args...)
if err != nil {
return nil, err
}
defer rows.Close()
out := make([]MileagePointRow, 0)
for rows.Next() {
var row MileagePointRow
var ts scanDateTime
var receivedAt scanDateTime
var speed sql.NullFloat64
var longitude sql.NullFloat64
var latitude sql.NullFloat64
if err := rows.Scan(
&ts,
&row.EventID,
&row.FrameID,
&receivedAt,
&row.TotalMileageKM,
&speed,
&longitude,
&latitude,
&row.Protocol,
&row.VIN,
); err != nil {
return nil, err
}
row.TS = ts.String
row.ReceivedAt = receivedAt.String
row.SpeedKMH = nullableFloat(speed)
row.Longitude = nullableFloat(longitude)
row.Latitude = nullableFloat(latitude)
out = append(out, row)
}
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 {
@@ -347,13 +262,6 @@ func (r *LocationRepository) tableName() string {
return r.database + ".vehicle_locations"
}
func (r *MileagePointRepository) 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)
@@ -381,17 +289,6 @@ func normalizeLocationQuery(query LocationQuery) LocationQuery {
return query
}
func normalizeMileagePointQuery(query MileagePointQuery) MileagePointQuery {
query.Protocol = strings.ToUpper(strings.TrimSpace(query.Protocol))
query.VIN = strings.TrimSpace(query.VIN)
query.DateFrom = normalizeDateTimeLiteral(query.DateFrom)
query.DateTo = normalizeDateTimeLiteral(query.DateTo)
if query.Limit <= 0 {
query.Limit = 20
}
return query
}
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, parse_status, parse_error, source_endpoint, protocol, vehicle_key, vin, phone, device_id FROM ` + table
@@ -548,24 +445,6 @@ func buildLocationCountSQL(table string, query LocationQuery) (string, []any) {
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, vin 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) {
@@ -637,26 +516,6 @@ func locationWhere(query LocationQuery) []string {
return where
}
func mileagePointWhere(query MileagePointQuery) []string {
where := []string{"total_mileage_km IS NOT NULL"}
add := func(clause string) {
where = append(where, clause)
}
if query.Protocol != "" {
add("protocol = '" + quote(query.Protocol) + "'")
}
if query.VIN != "" {
add("vin = '" + quote(query.VIN) + "'")
}
if query.DateFrom != "" {
add("ts >= '" + quote(normalizeDateTimeLiteral(query.DateFrom)) + "'")
}
if query.DateTo != "" {
add("ts <= '" + quote(normalizeDateTimeLiteral(query.DateTo)) + "'")
}
return where
}
type RawFrameHandler struct {
repository *RawFrameRepository
}
@@ -665,10 +524,6 @@ type LocationHandler struct {
repository *LocationRepository
}
type MileagePointHandler struct {
repository *MileagePointRepository
}
func NewRawFrameHandler(repository *RawFrameRepository) *RawFrameHandler {
if repository == nil {
panic("raw frame repository must not be nil")
@@ -683,13 +538,6 @@ func NewLocationHandler(repository *LocationRepository) *LocationHandler {
return &LocationHandler{repository: repository}
}
func NewMileagePointHandler(repository *MileagePointRepository) *MileagePointHandler {
if repository == nil {
panic("mileage point repository must not be nil")
}
return &MileagePointHandler{repository: repository}
}
func (h *RawFrameHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
if r.Method != http.MethodGet {
writeHistoryError(w, http.StatusMethodNotAllowed, "method not allowed")
@@ -762,39 +610,6 @@ func (h *LocationHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
})
}
func (h *MileagePointHandler) 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/mileage-points" {
writeHistoryError(w, http.StatusNotFound, "route not found")
return
}
query, err := parseMileagePointQuery(r)
if err != nil {
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())
return
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(map[string]any{
"items": rows,
"total": total,
"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")
@@ -830,30 +645,6 @@ func parseRawFrameQuery(r *http.Request) (RawFrameQuery, error) {
return normalizeRawFrameQuery(query), nil
}
func parseMileagePointQuery(r *http.Request) (MileagePointQuery, error) {
values := r.URL.Query()
limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit")
if err != nil {
return MileagePointQuery{}, err
}
offset, err := parseBoundedInt(values.Get("offset"), 0, 0, 1_000_000, "offset")
if err != nil {
return MileagePointQuery{}, err
}
query := MileagePointQuery{
Protocol: values.Get("protocol"),
VIN: values.Get("vin"),
DateFrom: values.Get("dateFrom"),
DateTo: values.Get("dateTo"),
Limit: limit,
Offset: offset,
}
if !validDateTime(query.DateFrom) || !validDateTime(query.DateTo) {
return MileagePointQuery{}, errors.New("dateFrom/dateTo must use YYYY-MM-DD or YYYY-MM-DD HH:mm:ss")
}
return normalizeMileagePointQuery(query), nil
}
func parseLocationQuery(r *http.Request) (LocationQuery, error) {
values := r.URL.Query()
limit, err := parseBoundedInt(values.Get("limit"), 20, 1, 500, "limit")

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

@@ -5,6 +5,7 @@ import (
"database/sql"
"net/http"
"net/http/httptest"
"os"
"strings"
"testing"
"time"
@@ -252,6 +253,16 @@ func TestRawFrameHandlerReturnsEmptyItemsArrayWhenNoRows(t *testing.T) {
}
}
func TestHistoryQueryPackageDoesNotKeepDuplicateMileagePointConcept(t *testing.T) {
source, err := os.ReadFile("query.go")
if err != nil {
t.Fatalf("read query.go: %v", err)
}
if strings.Contains(string(source), "MileagePoint") || strings.Contains(string(source), "mileage-points") {
t.Fatalf("history query should expose mileage through locations only")
}
}
func TestLocationHandlerReturnsLocationsByVIN(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
@@ -295,49 +306,6 @@ func TestLocationHandlerReturnsLocationsByVIN(t *testing.T) {
}
}
func TestMileagePointHandlerReturnsMileageByVIN(t *testing.T) {
db, mock, err := sqlmock.New()
if err != nil {
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(19))
mock.ExpectQuery("SELECT ts, event_id, frame_id, received_at, total_mileage_km, speed_kmh, longitude, latitude, protocol, vin 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", "vin",
}).AddRow(
"2026-07-02 00:18:22", "event-3", "go_frame", "2026-07-02 00:22:43",
8792.8, 8.0, 121.07764, 30.585928,
"JT808", "LKLG7C4E3NA774736",
))
handler := NewMileagePointHandler(NewMileagePointRepository(db, "lingniu_vehicle_ts"))
request := httptest.NewRequest(http.MethodGet, "/api/history/mileage-points?vin=LKLG7C4E3NA774736&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{`"vin":"LKLG7C4E3NA774736"`, `"total_mileage_km":8792.8`, `"speed_kmh":8`, `"total":19`} {
if !strings.Contains(body, want) {
t.Fatalf("response missing %s: %s", want, body)
}
}
for _, legacy := range []string{"vehicle_key", "phone", "device_id"} {
if strings.Contains(body, legacy) {
t.Fatalf("mileage response should not expose %s: %s", legacy, 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)
@@ -388,35 +356,6 @@ func TestParseMessageIDSupportsDecimalAndHex(t *testing.T) {
}
}
func TestBuildMileagePointSQLUsesLiteralsForTDengine(t *testing.T) {
sqlText, args := buildMileagePointSQL("lingniu_vehicle_ts.vehicle_locations", MileagePointQuery{
Protocol: "JT808",
VIN: "LKLG7C4E3NA774736",
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{
"FROM lingniu_vehicle_ts.vehicle_locations",
"total_mileage_km IS NOT NULL",
"protocol = 'JT808'",
"vin = 'LKLG7C4E3NA774736'",
"ts >= '2026-07-01 16:00:00'",
"LIMIT 20 OFFSET 5",
} {
if !strings.Contains(sqlText, want) {
t.Fatalf("sql missing %s: %s", want, sqlText)
}
}
if strings.Contains(sqlText, "vehicle_key") || strings.Contains(sqlText, "phone") || strings.Contains(sqlText, "device_id") {
t.Fatalf("mileage point sql should use vin-only identity filters: %s", sqlText)
}
}
func TestBuildLocationSQLUsesLiteralsForTDengine(t *testing.T) {
sqlText, args := buildLocationSQL("lingniu_vehicle_ts.vehicle_locations", LocationQuery{
Protocol: "JT808",