refactor(go): make realtime totals opt in
This commit is contained in:
@@ -62,6 +62,7 @@
|
|||||||
- 原因:`vehicle_realtime_snapshot`、`vehicle_realtime_location` 都是每协议每 VIN 一行的当前态投影,业务主键就是 `(protocol, vin)`。
|
- 原因:`vehicle_realtime_snapshot`、`vehicle_realtime_location` 都是每协议每 VIN 一行的当前态投影,业务主键就是 `(protocol, vin)`。
|
||||||
- 状态:schema 使用 `PRIMARY KEY (protocol, vin)`,不保留自增 `id` 和 `created_at`,对外只暴露最新 `updated_at`。
|
- 状态:schema 使用 `PRIMARY KEY (protocol, vin)`,不保留自增 `id` 和 `created_at`,对外只暴露最新 `updated_at`。
|
||||||
- 索引:不在 `vehicle_realtime_location` 维护经纬度组合索引;实时表只回答当前态,地理范围和轨迹类查询走 TDengine 历史位置。
|
- 索引:不在 `vehicle_realtime_location` 维护经纬度组合索引;实时表只回答当前态,地理范围和轨迹类查询走 TDengine 历史位置。
|
||||||
|
- 查询:`/api/realtime/snapshots`、`/api/realtime/locations` 默认不执行 `COUNT(*)`,`total` 表示本页返回条数;只有需要精确总数时传 `includeTotal=true`。
|
||||||
- 写入:车牌从 `vehicle_identity_binding` 反查时使用 VIN 级短 TTL 内存缓存,避免实时帧每条都打 MySQL;缓存不作为事实来源。
|
- 写入:车牌从 `vehicle_identity_binding` 反查时使用 VIN 级短 TTL 内存缓存,避免实时帧每条都打 MySQL;缓存不作为事实来源。
|
||||||
- 生产:项目上线前可直接重建这两张 MySQL 表,实时数据会从 Kafka 新消息继续投影。
|
- 生产:项目上线前可直接重建这两张 MySQL 表,实时数据会从 Kafka 新消息继续投影。
|
||||||
|
|
||||||
|
|||||||
@@ -16,11 +16,12 @@ type mysqlQueryer interface {
|
|||||||
}
|
}
|
||||||
|
|
||||||
type RealtimeTableQuery struct {
|
type RealtimeTableQuery struct {
|
||||||
Protocol string
|
Protocol string
|
||||||
VIN string
|
VIN string
|
||||||
Plate string
|
Plate string
|
||||||
Limit int
|
IncludeTotal bool
|
||||||
Offset int
|
Limit int
|
||||||
|
Offset int
|
||||||
}
|
}
|
||||||
|
|
||||||
type SnapshotRow struct {
|
type SnapshotRow struct {
|
||||||
@@ -191,16 +192,22 @@ func (h *SnapshotQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
|
|||||||
writeError(w, http.StatusBadRequest, err.Error())
|
writeError(w, http.StatusBadRequest, err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
total, err := h.repository.Count(r.Context(), query)
|
var total int64
|
||||||
if err != nil {
|
if query.IncludeTotal {
|
||||||
writeError(w, http.StatusInternalServerError, err.Error())
|
total, err = h.repository.Count(r.Context(), query)
|
||||||
return
|
if err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
rows, err := h.repository.Query(r.Context(), query)
|
rows, err := h.repository.Query(r.Context(), query)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeError(w, http.StatusInternalServerError, err.Error())
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if !query.IncludeTotal {
|
||||||
|
total = int64(len(rows))
|
||||||
|
}
|
||||||
writePage(w, rows, total, query)
|
writePage(w, rows, total, query)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -229,16 +236,22 @@ func (h *LocationQueryHandler) ServeHTTP(w http.ResponseWriter, r *http.Request)
|
|||||||
writeError(w, http.StatusBadRequest, err.Error())
|
writeError(w, http.StatusBadRequest, err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
total, err := h.repository.Count(r.Context(), query)
|
var total int64
|
||||||
if err != nil {
|
if query.IncludeTotal {
|
||||||
writeError(w, http.StatusInternalServerError, err.Error())
|
total, err = h.repository.Count(r.Context(), query)
|
||||||
return
|
if err != nil {
|
||||||
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
|
return
|
||||||
|
}
|
||||||
}
|
}
|
||||||
rows, err := h.repository.Query(r.Context(), query)
|
rows, err := h.repository.Query(r.Context(), query)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
writeError(w, http.StatusInternalServerError, err.Error())
|
writeError(w, http.StatusInternalServerError, err.Error())
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
if !query.IncludeTotal {
|
||||||
|
total = int64(len(rows))
|
||||||
|
}
|
||||||
writePage(w, rows, total, query)
|
writePage(w, rows, total, query)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -253,11 +266,12 @@ func parseRealtimeTableQuery(r *http.Request) (RealtimeTableQuery, error) {
|
|||||||
return RealtimeTableQuery{}, err
|
return RealtimeTableQuery{}, err
|
||||||
}
|
}
|
||||||
return normalizeRealtimeTableQuery(RealtimeTableQuery{
|
return normalizeRealtimeTableQuery(RealtimeTableQuery{
|
||||||
Protocol: values.Get("protocol"),
|
Protocol: values.Get("protocol"),
|
||||||
VIN: values.Get("vin"),
|
VIN: values.Get("vin"),
|
||||||
Plate: values.Get("plate"),
|
Plate: values.Get("plate"),
|
||||||
Limit: limit,
|
IncludeTotal: strings.EqualFold(strings.TrimSpace(values.Get("includeTotal")), "true"),
|
||||||
Offset: offset,
|
Limit: limit,
|
||||||
|
Offset: offset,
|
||||||
}), nil
|
}), nil
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -29,7 +29,7 @@ func TestSnapshotQueryHandlerReturnsRealtimeSnapshots(t *testing.T) {
|
|||||||
))
|
))
|
||||||
|
|
||||||
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))
|
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))
|
||||||
request := httptest.NewRequest(http.MethodGet, "/api/realtime/snapshots?protocol=gb32960&vin=LB9A32A21R0LS1707&limit=20", nil)
|
request := httptest.NewRequest(http.MethodGet, "/api/realtime/snapshots?protocol=gb32960&vin=LB9A32A21R0LS1707&includeTotal=true&limit=20", nil)
|
||||||
response := httptest.NewRecorder()
|
response := httptest.NewRecorder()
|
||||||
|
|
||||||
handler.ServeHTTP(response, request)
|
handler.ServeHTTP(response, request)
|
||||||
@@ -72,7 +72,7 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) {
|
|||||||
))
|
))
|
||||||
|
|
||||||
handler := NewLocationQueryHandler(NewLocationQueryRepository(db))
|
handler := NewLocationQueryHandler(NewLocationQueryRepository(db))
|
||||||
request := httptest.NewRequest(http.MethodGet, "/api/realtime/locations?protocol=jt808&plate=粤B98765&limit=10&offset=10", nil)
|
request := httptest.NewRequest(http.MethodGet, "/api/realtime/locations?protocol=jt808&plate=粤B98765&includeTotal=true&limit=10&offset=10", nil)
|
||||||
response := httptest.NewRecorder()
|
response := httptest.NewRecorder()
|
||||||
|
|
||||||
handler.ServeHTTP(response, request)
|
handler.ServeHTTP(response, request)
|
||||||
@@ -94,6 +94,72 @@ func TestLocationQueryHandlerReturnsRealtimeLocations(t *testing.T) {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestSnapshotQueryHandlerSkipsTotalCountByDefault(t *testing.T) {
|
||||||
|
db, mock, err := sqlmock.New()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sqlmock.New() error = %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
mock.ExpectQuery("SELECT protocol, vin, plate, event_time, received_at, event_id, updated_at FROM vehicle_realtime_snapshot").
|
||||||
|
WithArgs("GB32960", 1, 0).
|
||||||
|
WillReturnRows(sqlmock.NewRows([]string{
|
||||||
|
"protocol", "vin", "plate", "event_time", "received_at", "event_id", "updated_at",
|
||||||
|
}).AddRow(
|
||||||
|
"GB32960", "LB9A32A21R0LS1707", "浙A12345", "2026-07-02 16:01:02.123", "2026-07-02 16:01:03.456", "evt-1", "2026-07-02 16:01:04",
|
||||||
|
))
|
||||||
|
|
||||||
|
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(db))
|
||||||
|
request := httptest.NewRequest(http.MethodGet, "/api/realtime/snapshots?protocol=gb32960&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())
|
||||||
|
}
|
||||||
|
if body := response.Body.String(); !strings.Contains(body, `"total":1`) {
|
||||||
|
t.Fatalf("response should use page size as total when total count is not requested: %s", body)
|
||||||
|
}
|
||||||
|
if err := mock.ExpectationsWereMet(); err != nil {
|
||||||
|
t.Fatalf("sql expectations: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestLocationQueryHandlerSkipsTotalCountByDefault(t *testing.T) {
|
||||||
|
db, mock, err := sqlmock.New()
|
||||||
|
if err != nil {
|
||||||
|
t.Fatalf("sqlmock.New() error = %v", err)
|
||||||
|
}
|
||||||
|
defer db.Close()
|
||||||
|
|
||||||
|
mock.ExpectQuery("SELECT protocol, vin, plate, event_time, latitude, longitude, speed_kmh, total_mileage_km, soc_percent, altitude_m, direction_deg, alarm_flag, status_flag, received_at, event_id, updated_at FROM vehicle_realtime_location").
|
||||||
|
WithArgs("JT808", 1, 0).
|
||||||
|
WillReturnRows(sqlmock.NewRows([]string{
|
||||||
|
"protocol", "vin", "plate", "event_time", "latitude", "longitude", "speed_kmh", "total_mileage_km",
|
||||||
|
"soc_percent", "altitude_m", "direction_deg", "alarm_flag", "status_flag", "received_at", "event_id", "updated_at",
|
||||||
|
}).AddRow(
|
||||||
|
"JT808", "LKLG7C4E3NA774736", "粤B98765", "2026-07-02 16:11:02.000", 30.123456, 120.654321, 54.3, 48798.9,
|
||||||
|
nil, 19.0, 88.0, int64(0), int64(3), "2026-07-02 16:11:03.000", "evt-2", "2026-07-02 16:11:04",
|
||||||
|
))
|
||||||
|
|
||||||
|
handler := NewLocationQueryHandler(NewLocationQueryRepository(db))
|
||||||
|
request := httptest.NewRequest(http.MethodGet, "/api/realtime/locations?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())
|
||||||
|
}
|
||||||
|
if body := response.Body.String(); !strings.Contains(body, `"total":1`) {
|
||||||
|
t.Fatalf("response should use page size as total when total count is not requested: %s", body)
|
||||||
|
}
|
||||||
|
if err := mock.ExpectationsWereMet(); err != nil {
|
||||||
|
t.Fatalf("sql expectations: %v", err)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
func TestRealtimeQueryHandlerRejectsInvalidPagination(t *testing.T) {
|
func TestRealtimeQueryHandlerRejectsInvalidPagination(t *testing.T) {
|
||||||
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(&sql.DB{}))
|
handler := NewSnapshotQueryHandler(NewSnapshotQueryRepository(&sql.DB{}))
|
||||||
request := httptest.NewRequest(http.MethodGet, "/api/realtime/snapshots?limit=1001", nil)
|
request := httptest.NewRequest(http.MethodGet, "/api/realtime/snapshots?limit=1001", nil)
|
||||||
|
|||||||
Reference in New Issue
Block a user